From a3963611f07d4d77b4ec5f6368e633f652fbafbc Mon Sep 17 00:00:00 2001 From: retoor Date: Sun, 26 Jul 2026 16:45:38 +0200 Subject: [PATCH] ipdate --- devplacepy/database/schema.py | 25 ++ devplacepy/routers/admin/devii_tasks.py | 27 ++- devplacepy/schemas/admin.py | 1 + devplacepy/services/devii/CLAUDE.md | 33 ++- devplacepy/services/devii/service.py | 57 ++++- devplacepy/services/devii/session/core.py | 11 + devplacepy/services/devii/session/prompts.py | 18 ++ devplacepy/services/devii/tasks/actions.py | 18 +- devplacepy/services/devii/tasks/context.py | 22 ++ devplacepy/services/devii/tasks/controller.py | 31 ++- devplacepy/services/devii/tasks/guards.py | 75 ++++-- devplacepy/services/devii/tasks/limits.py | 215 +++++++++++++++++ devplacepy/services/devii/tasks/scheduler.py | 80 ++++++- devplacepy/services/devii/tasks/store.py | 105 +++++++-- devplacepy/templates/admin_devii_tasks.html | 16 +- tests/api/admin/deviitasks.py | 25 ++ tests/unit/services/devii/actions/catalog.py | 23 +- tests/unit/services/devii/session.py | 55 +++++ tests/unit/services/devii/tasks/context.py | 104 ++++++++ tests/unit/services/devii/tasks/guards.py | 160 ++++++++++--- tests/unit/services/devii/tasks/limits.py | 196 +++++++++++++++ tests/unit/services/devii/tasks/scheduler.py | 223 ++++++++++++++---- tests/unit/services/devii/tasks/store.py | 170 ++++++++++--- 23 files changed, 1500 insertions(+), 190 deletions(-) create mode 100644 devplacepy/services/devii/tasks/context.py create mode 100644 devplacepy/services/devii/tasks/limits.py create mode 100644 tests/unit/services/devii/tasks/context.py create mode 100644 tests/unit/services/devii/tasks/limits.py diff --git a/devplacepy/database/schema.py b/devplacepy/database/schema.py index 0bef69b0..48da4f59 100644 --- a/devplacepy/database/schema.py +++ b/devplacepy/database/schema.py @@ -421,6 +421,29 @@ def init_db(): ) except Exception as e: # noqa: BLE001 logger.warning(f"Could not backfill devii_tasks.failure_count: {e}") + task_runs = get_table("devii_task_runs") + for column, example in ( + ("uid", ""), + ("owner_kind", ""), + ("owner_id", ""), + ("task_uid", ""), + ("created_at", ""), + ): + if not task_runs.has_column(column): + task_runs.create_column_by_example(column, example) + _index( + db, + "devii_task_runs", + "idx_devii_task_runs_owner_time", + ["owner_kind", "owner_id", "created_at"], + ) + _index(db, "devii_task_runs", "idx_devii_task_runs_time", ["created_at"]) + _index( + db, + "devii_tasks", + "idx_devii_tasks_owner_created", + ["owner_kind", "owner_id", "created_at"], + ) _index(db, "devii_tasks", "idx_devii_tasks_owner", ["owner_kind", "owner_id"]) _index( db, "devii_tasks", "idx_devii_tasks_due", ["enabled", "status", "next_run_at"] @@ -1677,6 +1700,8 @@ def backfill_api_keys() -> int: users = db["users"] if not users.has_column("api_key"): users.create_column_by_example("api_key", "") + if not users.has_column("created_at"): + users.create_column_by_example("created_at", "") if not users.has_column("cust_disable_global"): users.create_column_by_example("cust_disable_global", 0) if not users.has_column("cust_disable_pagetype"): diff --git a/devplacepy/routers/admin/devii_tasks.py b/devplacepy/routers/admin/devii_tasks.py index 3f0a92df..7a3f1c8e 100644 --- a/devplacepy/routers/admin/devii_tasks.py +++ b/devplacepy/routers/admin/devii_tasks.py @@ -11,7 +11,9 @@ from devplacepy.schemas import AdminDeviiTasksOut from devplacepy.seo import base_seo_context, site_url, website_schema from devplacepy.services.audit import record as audit from devplacepy.services.devii import config as devii_config +from devplacepy.services.devii.tasks import limits from devplacepy.services.devii.tasks.guards import DEFAULT_MAX_PER_OWNER +from devplacepy.services.devii.tasks.schedule import now_utc from devplacepy.services.devii.tasks.store import TABLE, TaskStore from devplacepy.utils import not_found, require_admin @@ -43,9 +45,25 @@ def _rows(state: str) -> list[dict]: return rows +def _quotas(owner_uids: set[str]) -> dict[str, dict]: + reference = now_utc() + quotas = {} + for owner_uid in owner_uids: + runs = limits.run_quota(db, "user", owner_uid, reference) + creations = limits.create_quota(db, "user", owner_uid, reference) + quotas[owner_uid] = { + "runs_used": runs.used, + "runs_limit": runs.limit, + "creates_used": creations.used, + "creates_limit": creations.limit, + } + return quotas + + def _items(rows: list[dict]) -> list[dict]: owners = get_users_by_uids([row.get("owner_id") for row in rows if row.get("owner_id")]) admins = get_admin_uids() + quotas = _quotas({str(row.get("owner_id") or "") for row in rows if row.get("owner_id")}) items = [] for row in rows: owner_uid = row.get("owner_id") or "" @@ -57,6 +75,7 @@ def _items(rows: list[dict]) -> list[dict]: "owner_uid": owner_uid, "owner": owner["username"] if owner else owner_uid, "owner_is_admin": owner_uid in admins, + "quota": quotas.get(owner_uid, {}), "schedule": _schedule_text(row), "status": row.get("status") or "", "enabled": bool(row.get("enabled")), @@ -85,7 +104,7 @@ async def admin_devii_tasks(request: Request, state: str = "active"): state = "active" rows = _rows(state) items = _items(rows) - limits = { + bounds = { "max_concurrent": get_int_setting( devii_config.FIELD_TASK_MAX_CONCURRENT, devii_config.DEFAULT_TASK_MAX_CONCURRENT, @@ -100,6 +119,10 @@ async def admin_devii_tasks(request: Request, state: str = "active"): "idle_days": get_int_setting( devii_config.FIELD_TASK_IDLE_DAYS, devii_config.DEFAULT_TASK_IDLE_DAYS ), + "member_create_24h": limits.create_limit(False), + "member_runs_24h": limits.run_limit(False), + "admin_create_24h": limits.create_limit(True), + "admin_runs_24h": limits.run_limit(True), } tabs = [ {"key": key, "label": key.capitalize(), "active": key == state} @@ -127,7 +150,7 @@ async def admin_devii_tasks(request: Request, state: str = "active"): "items": items, "state": state, "tabs": tabs, - "limits": limits, + "limits": bounds, "admin_section": "devii-tasks", }, model=AdminDeviiTasksOut, diff --git a/devplacepy/schemas/admin.py b/devplacepy/schemas/admin.py index b8fcc610..79406434 100644 --- a/devplacepy/schemas/admin.py +++ b/devplacepy/schemas/admin.py @@ -80,6 +80,7 @@ class AdminDeviiTaskItemOut(_Out): owner_uid: str = "" owner: str = "" owner_is_admin: bool = False + quota: dict = {} schedule: str = "" status: str = "" enabled: bool = False diff --git a/devplacepy/services/devii/CLAUDE.md b/devplacepy/services/devii/CLAUDE.md index a1584be9..3e4dd656 100644 --- a/devplacepy/services/devii/CLAUDE.md +++ b/devplacepy/services/devii/CLAUDE.md @@ -45,13 +45,36 @@ Conversation history persists to `devii_conversations` (rehydrated on reconnect, `create_task` queues a self-contained prompt that a fresh agent runs later (`services/devii/tasks/`): `kind=once` (`delay_seconds` for relative, `run_at` UTC for absolute), `interval` (`every_seconds`), or `cron`. The result is broadcast to any connected tab and buffered (`type:"task"` frame) when the terminal is closed. -**Scheduling is an administrator privilege, enforced at three independent chokepoints** (one alone is not enough - the tool gate is a UI control, the store gate covers every writer, and the claim gate is the only one that also retires rows already in the database): +**Any signed-in account may schedule; two rolling 24-hour quotas bound it, and role decides the size.** Guests never can (`automation_allowed` requires `owner_kind == "user"`). Defaults: a member may CREATE `devii_task_member_create_24h` (5) tasks and EXECUTE `devii_task_member_runs_24h` (10) runs per 24h; an administrator gets `devii_task_admin_create_24h` (5) and `devii_task_admin_runs_24h` (100). All four are admin-editable on the Devii service; `0` means unlimited. -1. **Tool visibility and dispatch.** `create_task`/`update_task`/`run_task_now` are `requires_admin=True`, so `Catalog.tool_schemas_for` never offers them to a member and `Dispatcher.dispatch` refuses them server-side (audited `security.authz.denied`). `list_tasks`/`get_task`/`delete_task` stay open so a member keeps inspection and cleanup rights over tasks they already own. -2. **`TaskStore.create` / `TaskStore.update`.** The store raises `AutomationDenied` when a non-admin owner tries to persist a task or to ENABLE one (disabling is always allowed, which is what lets the scheduler and admins stop things). This sits below the dispatcher, so it also covers the standalone `devii` CLI - whose store passes `operator=True`, the one deliberate exemption for the local operator - and anything added later. -3. **`Scheduler`/`GlobalScheduler` claim.** `tasks.guards.refusal(row, now, budget_exceeded, max_failures)` runs before every execution and re-checks owner privilege, expiry, run limit, failure streak, and automation budget. **Privilege is therefore evaluated at execution time, not creation time**, so demoting an admin retires their tasks on the next tick with no migration and no manual step. +**The two quotas are enforced by a single atomic conditional INSERT each, never a check-then-act** (`tasks/limits.py`): -**Privilege is live, never frozen into the session.** `DeviiSession.is_admin`/`is_primary_admin` are properties resolving `get_admin_uids()`/`get_primary_admin_uid()` (both TTL-cached and cross-worker invalidated). `_refresh_privileges()` runs at the top of every turn AND inside the scheduled executor: it rewrites the base system prompt when the role changed and pushes the flags into the dispatcher via `Dispatcher.set_admin`. Regression to avoid: the old code captured `is_admin` as a bool at session construction, and a session with pending tasks was exempt from `gc_idle`, so a **demoted admin kept admin tools indefinitely**. +- `reserve_run(...)` -> `INSERT INTO devii_task_runs ... SELECT ... WHERE (SELECT COUNT(*) ... created_at >= :cutoff) < :limit`, decided on the driver's real `rowcount`. The scheduler calls it AFTER `claim` and BEFORE dispatch; a refusal releases the claim via `defer`. +- `insert_task_within_quota(...)` -> the same shape for the task row itself, so two concurrent `create_task` calls (main + telegram channel, or two workers) can never both slip past a check. `TaskStore.create` pre-checks for a good error message, then relies on this insert for the actual guarantee. + +Both are proven with real multi-process races (16 processes -> exactly 10 runs for a member, 12 processes -> exactly 5 creations). **Never replace either with a count-then-insert.** + +**Creation counts the task row itself, so deleting a task does NOT give the slot back** - the window counts `devii_tasks.created_at` regardless of `deleted_at`. Runs are counted in the dedicated `devii_task_runs` ledger (a derived counter table, NOT in `SOFT_DELETE_TABLES`; hard-pruned to 48h by `run_once`). + +**A quota refusal postpones, it never disables.** `retire_reason` (terminal: guest owner, expiry, `max_runs`, failure streak) disables the task; `run_deferral`/`budget_deferral` push `next_run_at` to when the slot frees (`Quota.free_at` = the oldest run in the window + 24h) and leave the task enabled, audited `devii.task.deferred`. Keep that split: a rate limit that disabled tasks would silently destroy a user's automation. + +**The task-run context is the unhackable part** (`tasks/context.py`). `task_run_scope()` sets a `contextvars.ContextVar` around the executor inside `run_row`, so: + +- Everything the scheduled agent does - including subagents, `delegate`, and virtual tools, which all share the dispatcher - runs inside that context. +- A concurrent interactive turn runs in its own asyncio Task context and can never see or clear it (asyncio copies the context at task creation). +- **There is no tool, argument, or prompt that can write it.** The model only ever influences tool arguments; the flag lives in process state set by the scheduler. This is why the nesting rule is enforced on the ContextVar rather than on anything the model can say. + +`nesting_allowed(owner_id)` = `not in_task_run() or owner_is_admin(owner_id)`. `TaskStore.create` and `TaskStore.update`-when-enabling both consult it, so a member's task run cannot create a task, re-enable one, or `run_task_now` - while an administrator's task may schedule follow-up work. The session also TELLS the agent where it is: `_compose_system_prompt()` injects the `# TASK RUN ENVIRONMENT` block (`TASK_RUN_MEMBER`/`TASK_RUN_ADMIN`) only when `in_task_run()`. + +Beyond the quotas, three chokepoints still gate scheduling (one alone is not enough - the tool gate is a UI control, the store gate covers every writer, and the claim gate is the only one that also retires rows already in the database): + +1. **Tool visibility and dispatch.** `create_task`/`update_task`/`run_task_now` are `requires_auth=True`, so guests are never offered them. `list_tasks`/`get_task`/`delete_task` stay open. +2. **`TaskStore.create` / `TaskStore.update`.** Raises `AutomationDenied` (with `retry_at` when a quota is the cause) for a guest owner, a nested member creation, or a spent creation quota. Enabling a task takes the same nesting check; disabling is always allowed, which is what lets the scheduler and admins stop things. This sits below the dispatcher, so it also covers the standalone `devii` CLI - whose store passes `operator=True`, the one deliberate exemption for the local operator - and anything added later. +3. **`Scheduler`/`GlobalScheduler` claim.** `retire_reason` runs before every execution and re-checks owner kind, expiry, run limit, and failure streak, so a task that must stop is retired at execution time with no migration and no manual step. + +**`TaskStore._ensure_schema` declares the FULL column set** (`TASK_COLUMNS`) rather than leaning on dataset's lazy schema - the atomic INSERT names columns explicitly, and dataset skips a `None`-valued key when creating columns, so a lazily-built table would be missing `run_at`/`cron`/`start_at`. Both `_ensure_schema` and `_ensure_indexes` tolerate a concurrent `duplicate column` / `already exists` error: several processes construct a `TaskStore` at once and SQLite DDL is not idempotent (this was a real crash, found by racing 12 processes). + +**Role is live, never frozen into the session.** `DeviiSession.is_admin`/`is_primary_admin` are properties resolving `get_admin_uids()`/`get_primary_admin_uid()` (both TTL-cached and cross-worker invalidated). `_refresh_privileges()` runs at the top of every turn AND inside the scheduled executor: it rewrites the base system prompt when the role changed and pushes the flags into the dispatcher via `Dispatcher.set_admin`. Regression to avoid: the old code captured `is_admin` as a bool at session construction, and a session with pending tasks was exempt from `gc_idle`, so a **demoted admin kept admin tools indefinitely**. **One global scheduler, not one per owner.** `DeviiService.scheduler()` owns a single `GlobalScheduler` started in `on_enable` on the lock-owner worker. Each 1s tick issues ONE indexed query (`tasks.store.due_rows`, served by `idx_devii_tasks_due` on `(enabled, status, next_run_at)`) instead of one query per owner per second, and hands out at most `devii_task_max_concurrent` slots **round-robin, one at a time per owner**, so no owner can monopolise the scheduler. A row is taken with an atomic conditional claim (`tasks.store.claim` -> `UPDATE ... WHERE uid=? AND status='pending' AND enabled=1 AND deleted_at IS NULL`, decided on the driver's real `rowcount`), which closes the TOCTOU between a tick and `run_task_now`. Sessions are materialized lazily by `_resolve_task_owner` only when a task actually fires, so `gc_idle` now reaps on `session.is_busy()` (a turn in flight) rather than exempting every task-owning session forever. The per-store `Scheduler` class remains for the standalone `devii` CLI; both share `run_row`/`retire`/`refusal`, so there is one implementation of the guard logic. diff --git a/devplacepy/services/devii/service.py b/devplacepy/services/devii/service.py index 7e783c9b..dd11c93e 100644 --- a/devplacepy/services/devii/service.py +++ b/devplacepy/services/devii/service.py @@ -12,6 +12,7 @@ from .config import ( build_settings, ) from .cost.tracker import Pricing +from .tasks import limits from .tasks.guards import DEFAULT_MAX_PER_OWNER, REASON_OWNER_IDLE from .tasks.scheduler import GlobalScheduler, retire @@ -267,6 +268,44 @@ class DeviiService(BaseService): help="How many enabled tasks one administrator may hold at once. 0 disables the cap.", group="Automation", ), + ConfigField( + limits.FIELD_MEMBER_CREATE, + "Member tasks created / 24h", + type="int", + default=limits.DEFAULT_MEMBER_CREATE, + minimum=0, + help="How many tasks a member may create in a rolling 24 hours. Deleting a task does " + "not give the slot back. 0 = unlimited.", + group="Automation", + ), + ConfigField( + limits.FIELD_MEMBER_RUNS, + "Member task runs / 24h", + type="int", + default=limits.DEFAULT_MEMBER_RUNS, + minimum=0, + help="How many scheduled runs a member's tasks may execute in a rolling 24 hours. " + "A run over the limit is postponed until a slot frees, never dropped. 0 = unlimited.", + group="Automation", + ), + ConfigField( + limits.FIELD_ADMIN_CREATE, + "Administrator tasks created / 24h", + type="int", + default=limits.DEFAULT_ADMIN_CREATE, + minimum=0, + help="Same rolling 24-hour creation quota for administrators. 0 = unlimited.", + group="Automation", + ), + ConfigField( + limits.FIELD_ADMIN_RUNS, + "Administrator task runs / 24h", + type="int", + default=limits.DEFAULT_ADMIN_RUNS, + minimum=0, + help="Rolling 24-hour run quota for administrators. 0 = unlimited.", + group="Automation", + ), ConfigField( config.FIELD_TASK_DAILY_USD, "Max USD per owner / 24h for automation", @@ -396,15 +435,27 @@ class DeviiService(BaseService): hub = self.hub() self._configure_scheduler() idle = self._disable_idle_owner_tasks() + runs_pruned = self._prune_task_runs() pruned = hub.ledger.prune(48) removed = await hub.gc_idle() lessons_pruned = self._prune_old_lessons() - if pruned or removed or idle or lessons_pruned: + if pruned or removed or idle or lessons_pruned or runs_pruned: self.log( - f"Housekeeping: disabled {idle} idle-owner task(s), pruned {pruned} " - f"ledger rows, pruned {lessons_pruned} lessons, closed {removed} idle sessions" + f"Housekeeping: disabled {idle} idle-owner task(s), pruned {runs_pruned} task-run " + f"rows, pruned {pruned} ledger rows, pruned {lessons_pruned} lessons, closed " + f"{removed} idle sessions" ) + def _prune_task_runs(self) -> int: + from devplacepy.database import db + from .tasks.schedule import now_utc + + try: + return limits.prune_runs(db, now_utc()) + except Exception: # noqa: BLE001 - a bad read must not stop housekeeping + logger.exception("Failed to prune Devii task-run rows") + return 0 + def _prune_old_lessons(self) -> int: try: from devplacepy.database import db diff --git a/devplacepy/services/devii/session/core.py b/devplacepy/services/devii/session/core.py index fffdfcbd..3531867a 100644 --- a/devplacepy/services/devii/session/core.py +++ b/devplacepy/services/devii/session/core.py @@ -29,6 +29,7 @@ from ..interaction.capabilities import ( from ..llm import LLMClient from ..registry import CATALOG from ..tasks import TaskController, TaskStore +from ..tasks.context import in_task_run from ..virtual_tools import VirtualToolController, VirtualToolStore from ..text import normalize_newlines from ._helpers import _format_offset, _now_iso, _repair_history @@ -38,6 +39,9 @@ from .prompts import ( DOCS_SYSTEM_PROMPT, DOCS_TOOLS, LOGIN_REQUEST, + TASK_RUN_ADMIN, + TASK_RUN_HEADER, + TASK_RUN_MEMBER, TOPIC_CLASSIFIER_PROMPT, _parse_topic, _system_prompt_for, @@ -537,10 +541,17 @@ class DeviiSession: f"{self._system_prompt}\n\n" f"{CA_IWP_SYSTEM_FRAGMENT}\n\n" f"{channel_block}\n\n" + f"{self._task_run_block()}" f"{section}\n\n" f"{self._clock_line()}" ) + def _task_run_block(self) -> str: + if not in_task_run(): + return "" + body = TASK_RUN_ADMIN if self.is_admin else TASK_RUN_MEMBER + return f"{TASK_RUN_HEADER}\n{body}\n\n" + def _clock_line(self) -> str: from datetime import datetime, timedelta, timezone diff --git a/devplacepy/services/devii/session/prompts.py b/devplacepy/services/devii/session/prompts.py index 8ded5276..88b4dd3d 100644 --- a/devplacepy/services/devii/session/prompts.py +++ b/devplacepy/services/devii/session/prompts.py @@ -18,6 +18,24 @@ NON_ADMIN_COST_RULE = ( "quota used and the turn count." ) +TASK_RUN_HEADER = "# TASK RUN ENVIRONMENT" + +TASK_RUN_MEMBER = ( + "You are running as a SCHEDULED TASK, not in a live conversation. Nobody is watching this " + "turn; your final message becomes the recorded task result. You may NOT schedule further " + "work from here: create_task, enabling a task, and run_task_now are refused inside a task " + "run for this account. Do the work the prompt asks for and finish. Every run also consumes " + "one slot of this account's rolling 24-hour run quota." +) + +TASK_RUN_ADMIN = ( + "You are running as a SCHEDULED TASK, not in a live conversation. Nobody is watching this " + "turn; your final message becomes the recorded task result. As an administrator you MAY " + "schedule follow-up work from here, but do so only when the prompt asks for it - each new " + "task and each run counts against this account's rolling 24-hour quotas, and a task that " + "schedules itself repeatedly will exhaust them." +) + DOCS_TOOLS = frozenset({"search_docs"}) TOPIC_CLASSIFIER_PROMPT = ( diff --git a/devplacepy/services/devii/tasks/actions.py b/devplacepy/services/devii/tasks/actions.py index 65da94db..19cdcd30 100644 --- a/devplacepy/services/devii/tasks/actions.py +++ b/devplacepy/services/devii/tasks/actions.py @@ -88,12 +88,13 @@ TASK_ACTIONS: tuple[Action, ...] = ( summary="Queue a prompt to run autonomously on a schedule with full tool access", description=( "The prompt is executed later by a fresh agent that has every tool available, " - "running inside the current authenticated session. Scheduling is restricted to " - "administrators, is capped per owner, and every recurring task stops by itself." + "running inside the current authenticated session. Two rolling 24-hour quotas apply " + "per account: how many tasks may be created, and how many task runs may execute. " + "Every recurring task also stops by itself. While a task run is executing, members " + "may not schedule further work from inside it; administrators may." ), handler="task", requires_auth=True, - requires_admin=True, params=( field( "prompt", @@ -143,12 +144,11 @@ TASK_ACTIONS: tuple[Action, ...] = ( path="", summary="Update a task's prompt, label, enabled state, or schedule", description=( - "Provide schedule fields together with kind to reschedule the task. Restricted to " - "administrators." + "Provide schedule fields together with kind to reschedule the task. Enabling a task " + "from inside a running task is refused for members." ), handler="task", requires_auth=True, - requires_admin=True, params=( field("uid", "Uid of the task.", required=True), field("prompt", "New prompt."), @@ -183,10 +183,12 @@ TASK_ACTIONS: tuple[Action, ...] = ( method="LOCAL", path="", summary="Trigger a task to execute immediately on the next scheduler tick", - description="Restricted to administrators.", + description=( + "The run still consumes a slot of the account's rolling 24-hour run quota, and is " + "postponed when that quota is spent." + ), handler="task", requires_auth=True, - requires_admin=True, params=(field("uid", "Uid of the task.", required=True),), ), ) diff --git a/devplacepy/services/devii/tasks/context.py b/devplacepy/services/devii/tasks/context.py new file mode 100644 index 00000000..80929c25 --- /dev/null +++ b/devplacepy/services/devii/tasks/context.py @@ -0,0 +1,22 @@ +# retoor + +from __future__ import annotations + +import contextvars +from contextlib import contextmanager +from typing import Iterator + +_TASK_RUN = contextvars.ContextVar("devii_task_run", default=False) + + +def in_task_run() -> bool: + return _TASK_RUN.get() + + +@contextmanager +def task_run_scope() -> Iterator[None]: + token = _TASK_RUN.set(True) + try: + yield + finally: + _TASK_RUN.reset(token) diff --git a/devplacepy/services/devii/tasks/controller.py b/devplacepy/services/devii/tasks/controller.py index 2cbbe661..d5868af6 100644 --- a/devplacepy/services/devii/tasks/controller.py +++ b/devplacepy/services/devii/tasks/controller.py @@ -93,7 +93,7 @@ class TaskController: prompt = str(arguments.get("prompt", "")).strip() if not prompt: raise ToolInputError("create_task requires a non-empty prompt.") - self._require_automation() + self._require_creation() self._require_capacity() schedule = self._build_schedule(arguments) reference = now_utc() @@ -118,7 +118,7 @@ class TaskController: try: self._store.create(record) except AutomationDenied as exc: - raise ToolInputError(f"Task refused: {exc.reason}.") from exc + raise self._denied(exc) from exc return json.dumps( {"status": "created", "task": _serialize(record, preview=True)}, ensure_ascii=False, @@ -144,7 +144,7 @@ class TaskController: def update_task(self, arguments: dict[str, Any]) -> str: row = self._require_task(arguments) - self._require_automation() + self._require_scheduling() reference = now_utc() changes: dict[str, Any] = {} @@ -210,7 +210,7 @@ class TaskController: def run_task_now(self, arguments: dict[str, Any]) -> str: row = self._require_task(arguments) - self._require_automation() + self._require_scheduling() self._store.update( row["uid"], {"enabled": True, "status": "pending", "next_run_at": to_iso(now_utc())}, @@ -224,13 +224,26 @@ class TaskController: ensure_ascii=False, ) - def _require_automation(self) -> None: + @staticmethod + def _denied(exc: AutomationDenied) -> ToolInputError: + if exc.retry_at is None: + return ToolInputError(f"Task refused: {exc.reason}.") + return ToolInputError( + f"Task refused: {exc.reason}. The next slot frees up at " + f"{to_iso(exc.retry_at)} UTC." + ) + + def _require_creation(self) -> None: try: - self._store.require_automation() + self._store.require_creation_allowed() except AutomationDenied as exc: - raise ToolInputError( - f"Scheduled tasks are restricted to administrators: {exc.reason}." - ) from exc + raise self._denied(exc) from exc + + def _require_scheduling(self) -> None: + try: + self._store.require_scheduling_allowed() + except AutomationDenied as exc: + raise self._denied(exc) from exc def _require_capacity(self) -> None: limit = max_active_per_owner() diff --git a/devplacepy/services/devii/tasks/guards.py b/devplacepy/services/devii/tasks/guards.py index 5a1cb935..67df5cd9 100644 --- a/devplacepy/services/devii/tasks/guards.py +++ b/devplacepy/services/devii/tasks/guards.py @@ -3,8 +3,10 @@ from __future__ import annotations from datetime import datetime, timedelta -from typing import Any, Callable, Optional +from typing import Any, Callable, NamedTuple, Optional +from .context import in_task_run +from .limits import create_quota, owner_is_admin, run_quota from .schedule import ( DEFAULT_MAX_RUNS, MAX_LIFETIME_DAYS, @@ -14,31 +16,41 @@ from .schedule import ( to_iso, ) -REASON_NOT_ADMIN = "owner is not an administrator" +REASON_NOT_A_USER = "tasks belong to a signed-in account" REASON_EXPIRED = "task lifetime expired" REASON_MAX_RUNS = "run limit reached" -REASON_BUDGET = "automation budget reached" REASON_FAILURES = "too many consecutive failures" REASON_OWNER_IDLE = "owner has been inactive" +REASON_NESTED = "a task run may not schedule more work" +REASON_RUN_QUOTA = "daily run limit reached" +REASON_BUDGET = "automation budget reached" +REASON_CREATE_QUOTA = "daily task limit reached" BudgetProbe = Callable[[str, str], bool] FIELD_MAX_PER_OWNER = "devii_task_max_per_owner" DEFAULT_MAX_PER_OWNER = 10 +BUDGET_RETRY_HOURS = 1 + + +class Deferral(NamedTuple): + reason: str + retry_at: datetime class AutomationDenied(Exception): - def __init__(self, reason: str) -> None: + def __init__(self, reason: str, retry_at: Optional[datetime] = None) -> None: super().__init__(reason) self.reason = reason + self.retry_at = retry_at def automation_allowed(owner_kind: str, owner_id: str) -> bool: - if owner_kind != "user" or not owner_id: - return False - from devplacepy.database import get_admin_uids + return owner_kind == "user" and bool(owner_id) - return owner_id in get_admin_uids() + +def nesting_allowed(owner_id: str) -> bool: + return not in_task_run() or owner_is_admin(owner_id) def max_active_per_owner() -> int: @@ -68,16 +80,13 @@ def expiry_of(row: dict[str, Any]) -> Optional[datetime]: return None -def refusal( - row: dict[str, Any], - reference: datetime, - budget_exceeded: Optional[BudgetProbe] = None, - max_failures: int = 0, +def retire_reason( + row: dict[str, Any], reference: datetime, max_failures: int = 0 ) -> Optional[str]: owner_kind = str(row.get("owner_kind") or "") owner_id = str(row.get("owner_id") or "") if not automation_allowed(owner_kind, owner_id): - return REASON_NOT_ADMIN + return REASON_NOT_A_USER expiry = expiry_of(row) if expiry is not None and reference >= expiry: @@ -90,7 +99,39 @@ def refusal( if max_failures > 0 and int(row.get("failure_count") or 0) >= max_failures: return REASON_FAILURES - if budget_exceeded is not None and budget_exceeded(owner_kind, owner_id): - return REASON_BUDGET - + return None + + +def budget_deferral( + row: dict[str, Any], + reference: datetime, + budget_exceeded: Optional[BudgetProbe] = None, +) -> Optional[Deferral]: + if budget_exceeded is None: + return None + owner_kind = str(row.get("owner_kind") or "") + owner_id = str(row.get("owner_id") or "") + if not budget_exceeded(owner_kind, owner_id): + return None + return Deferral(REASON_BUDGET, reference + timedelta(hours=BUDGET_RETRY_HOURS)) + + +def run_deferral(db: Any, row: dict[str, Any], reference: datetime) -> Deferral: + owner_kind = str(row.get("owner_kind") or "") + owner_id = str(row.get("owner_id") or "") + quota = run_quota(db, owner_kind, owner_id, reference) + retry_at = quota.free_at or reference + timedelta(hours=BUDGET_RETRY_HOURS) + return Deferral(REASON_RUN_QUOTA, retry_at) + + +def creation_denial( + db: Any, owner_kind: str, owner_id: str, reference: datetime +) -> Optional[AutomationDenied]: + if not automation_allowed(owner_kind, owner_id): + return AutomationDenied(REASON_NOT_A_USER) + if not nesting_allowed(owner_id): + return AutomationDenied(REASON_NESTED) + quota = create_quota(db, owner_kind, owner_id, reference) + if quota.exceeded: + return AutomationDenied(REASON_CREATE_QUOTA, quota.free_at) return None diff --git a/devplacepy/services/devii/tasks/limits.py b/devplacepy/services/devii/tasks/limits.py new file mode 100644 index 00000000..62200a25 --- /dev/null +++ b/devplacepy/services/devii/tasks/limits.py @@ -0,0 +1,215 @@ +# retoor + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta +from typing import Any, NamedTuple, Optional + +from .schedule import from_iso, to_iso + +logger = logging.getLogger("devii.tasks.limits") + +RUNS_TABLE = "devii_task_runs" +TASKS_TABLE = "devii_tasks" +WINDOW_HOURS = 24 + +FIELD_MEMBER_CREATE = "devii_task_member_create_24h" +FIELD_MEMBER_RUNS = "devii_task_member_runs_24h" +FIELD_ADMIN_CREATE = "devii_task_admin_create_24h" +FIELD_ADMIN_RUNS = "devii_task_admin_runs_24h" + +DEFAULT_MEMBER_CREATE = 5 +DEFAULT_MEMBER_RUNS = 10 +DEFAULT_ADMIN_CREATE = 5 +DEFAULT_ADMIN_RUNS = 100 + +RUNS_SQL = ( + f"SELECT created_at FROM {RUNS_TABLE} WHERE owner_kind = :kind AND owner_id = :owner " + "AND created_at >= :cutoff ORDER BY created_at LIMIT :cap" +) +CREATIONS_SQL = ( + f"SELECT created_at FROM {TASKS_TABLE} WHERE owner_kind = :kind AND owner_id = :owner " + "AND created_at >= :cutoff ORDER BY created_at LIMIT :cap" +) +RESERVE_SQL = ( + f"INSERT INTO {RUNS_TABLE} (uid, owner_kind, owner_id, task_uid, created_at) " + "SELECT :uid, :kind, :owner, :task, :now WHERE (" + f"SELECT COUNT(*) FROM {RUNS_TABLE} WHERE owner_kind = :kind AND owner_id = :owner " + "AND created_at >= :cutoff) < :limit" +) +SAMPLE_CAP = 1000 + + +class Quota(NamedTuple): + used: int + limit: int + free_at: Optional[datetime] + + @property + def exceeded(self) -> bool: + return self.limit > 0 and self.used >= self.limit + + @property + def remaining(self) -> int: + if self.limit <= 0: + return -1 + return max(0, self.limit - self.used) + + +def owner_is_admin(owner_id: str) -> bool: + from devplacepy.database import get_admin_uids + + return bool(owner_id) and owner_id in get_admin_uids() + + +def _setting(name: str, default: int) -> int: + from devplacepy.database import get_int_setting + + return get_int_setting(name, default) + + +def create_limit(is_admin: bool) -> int: + if is_admin: + return _setting(FIELD_ADMIN_CREATE, DEFAULT_ADMIN_CREATE) + return _setting(FIELD_MEMBER_CREATE, DEFAULT_MEMBER_CREATE) + + +def run_limit(is_admin: bool) -> int: + if is_admin: + return _setting(FIELD_ADMIN_RUNS, DEFAULT_ADMIN_RUNS) + return _setting(FIELD_MEMBER_RUNS, DEFAULT_MEMBER_RUNS) + + +def window_start(reference: datetime) -> datetime: + return reference - timedelta(hours=WINDOW_HOURS) + + +def _stamps(db: Any, sql: str, table: str, owner_kind: str, owner_id: str, cutoff: str) -> list[str]: + if table not in db.tables: + return [] + rows = db.query( + sql, kind=owner_kind, owner=owner_id, cutoff=cutoff, cap=SAMPLE_CAP + ) + return [str(row["created_at"]) for row in rows if row.get("created_at")] + + +def _quota(stamps: list[str], limit: int) -> Quota: + used = len(stamps) + if limit <= 0 or used < limit: + return Quota(used, limit, None) + oldest_kept = stamps[used - limit] + try: + free_at = from_iso(oldest_kept[:19]) + timedelta(hours=WINDOW_HOURS) + except ValueError: + free_at = None + return Quota(used, limit, free_at) + + +def run_quota(db: Any, owner_kind: str, owner_id: str, reference: datetime) -> Quota: + limit = run_limit(owner_is_admin(owner_id)) + stamps = _stamps( + db, RUNS_SQL, RUNS_TABLE, owner_kind, owner_id, to_iso(window_start(reference)) + ) + return _quota(stamps, limit) + + +def create_quota(db: Any, owner_kind: str, owner_id: str, reference: datetime) -> Quota: + limit = create_limit(owner_is_admin(owner_id)) + stamps = _stamps( + db, + CREATIONS_SQL, + TASKS_TABLE, + owner_kind, + owner_id, + to_iso(window_start(reference)), + ) + return _quota(stamps, limit) + + +def record_run(db: Any, owner_kind: str, owner_id: str, task_uid: str, reference: datetime) -> None: + from devplacepy.utils import generate_uid + + db[RUNS_TABLE].insert( + { + "uid": generate_uid(), + "owner_kind": owner_kind, + "owner_id": owner_id, + "task_uid": task_uid, + "created_at": to_iso(reference), + } + ) + + +def reserve_run( + db: Any, owner_kind: str, owner_id: str, task_uid: str, reference: datetime +) -> bool: + import sqlalchemy + + from devplacepy.utils import generate_uid + + limit = run_limit(owner_is_admin(owner_id)) + if RUNS_TABLE not in db.tables: + record_run(db, owner_kind, owner_id, task_uid, reference) + return True + if limit <= 0: + record_run(db, owner_kind, owner_id, task_uid, reference) + return True + params = { + "uid": generate_uid(), + "kind": owner_kind, + "owner": owner_id, + "task": task_uid, + "now": to_iso(reference), + "cutoff": to_iso(window_start(reference)), + "limit": limit, + } + with db: + result = db.executable.execute(sqlalchemy.text(RESERVE_SQL), params) + return result.rowcount == 1 + + +def insert_task_within_quota( + db: Any, + record: dict[str, Any], + owner_kind: str, + owner_id: str, + reference: datetime, +) -> bool: + import sqlalchemy + + limit = create_limit(owner_is_admin(owner_id)) + if limit <= 0 or TASKS_TABLE not in db.tables: + db[TASKS_TABLE].insert(record) + return True + + columns = list(record) + statement = sqlalchemy.text( + f"INSERT INTO {TASKS_TABLE} ({', '.join(columns)}) " + f"SELECT {', '.join(':' + name for name in columns)} WHERE (" + f"SELECT COUNT(*) FROM {TASKS_TABLE} WHERE owner_kind = :q_kind " + "AND owner_id = :q_owner AND created_at >= :q_cutoff) < :q_limit" + ) + params = dict(record) + params.update( + { + "q_kind": owner_kind, + "q_owner": owner_id, + "q_cutoff": to_iso(window_start(reference)), + "q_limit": limit, + } + ) + with db: + result = db.executable.execute(statement, params) + return result.rowcount == 1 + + +def prune_runs(db: Any, reference: datetime, keep_hours: int = WINDOW_HOURS * 2) -> int: + if RUNS_TABLE not in db.tables: + return 0 + cutoff = to_iso(reference - timedelta(hours=keep_hours)) + table = db[RUNS_TABLE] + stale = table.count(created_at={"<": cutoff}) + if stale: + table.delete(created_at={"<": cutoff}) + return stale diff --git a/devplacepy/services/devii/tasks/scheduler.py b/devplacepy/services/devii/tasks/scheduler.py index 8d11fbf0..ffafcf2d 100644 --- a/devplacepy/services/devii/tasks/scheduler.py +++ b/devplacepy/services/devii/tasks/scheduler.py @@ -6,8 +6,10 @@ import asyncio import logging from typing import Any, Awaitable, Callable, Optional +from .context import task_run_scope from .controller import compute_followup -from .guards import BudgetProbe, refusal +from .guards import BudgetProbe, Deferral, budget_deferral, run_deferral, retire_reason +from .limits import reserve_run from .schedule import next_run, now_utc, to_iso from .store import TaskStore, claim, due_rows @@ -68,6 +70,46 @@ DEFAULT_MAX_FAILURES = 3 DUE_BATCH_LIMIT = 200 +def _audit_task_deferred(row: dict[str, Any], reason: str, retry_at: str) -> None: + from devplacepy.services.audit import record as audit + + owner_kind = row.get("owner_kind") or "user" + owner_id = row.get("owner_id") or "" + audit.record_system( + "devii.task.deferred", + actor_kind="user" if owner_kind == "user" else owner_kind, + actor_uid=owner_id if owner_kind == "user" else None, + actor_role="user" if owner_kind == "user" else owner_kind, + origin="scheduler", + via_agent=1, + result="denied", + target_type="task", + target_uid=row.get("uid"), + summary=f"Devii task {row.get('uid')} postponed - {reason}", + metadata={"reason": reason, "retry_at": retry_at}, + links=[audit.task(row.get("uid"))], + ) + + +def defer(store: TaskStore, row: dict[str, Any], postponement: Deferral) -> None: + retry_at = to_iso(postponement.retry_at) + store.update( + row["uid"], + { + "status": "pending", + "next_run_at": retry_at, + "last_error": postponement.reason, + }, + ) + logger.info( + "Task uid=%s postponed to %s: %s", + row.get("uid"), + retry_at, + postponement.reason, + ) + _audit_task_deferred(row, postponement.reason, retry_at) + + def retire(store: TaskStore, row: dict[str, Any], reason: str) -> None: store.update( row["uid"], @@ -94,7 +136,8 @@ async def run_row( logger.info("Executing task uid=%s", uid) try: - result = await executor(row["prompt"]) + with task_run_scope(): + result = await executor(row["prompt"]) except Exception as exc: # noqa: BLE001 - surfaced into the task record logger.exception("Task uid=%s crashed", uid) store.update(uid, _failure_changes(row, str(exc), max_failures)) @@ -175,14 +218,26 @@ class Scheduler: async def _tick(self) -> None: now = now_utc() for row in self._store.due(to_iso(now)): - reason = refusal(row, now) + reason = retire_reason(row, now) if reason is not None: retire(self._store, row, reason) continue if not claim(self._store.db, row["uid"]): continue + if not self._reserve(row, now): + defer(self._store, row, run_deferral(self._store.db, row, now)) + continue await run_row(row, self._store, self._executor, self._on_event) + def _reserve(self, row: dict[str, Any], reference: Any) -> bool: + return reserve_run( + self._store.db, + str(row.get("owner_kind") or "user"), + str(row.get("owner_id") or ""), + row["uid"], + reference, + ) + class GlobalScheduler: def __init__( @@ -242,7 +297,7 @@ class GlobalScheduler: now = now_utc() free = self._max_concurrent - len(self._running) for row in due_rows(self._db, to_iso(now), DUE_BATCH_LIMIT): - reason = refusal(row, now, self._budget_exceeded, self._max_failures) + reason = retire_reason(row, now, self._max_failures) if reason is not None: retire(self._store_for(row), row, reason) continue @@ -250,10 +305,14 @@ class GlobalScheduler: continue if str(row.get("owner_id") or "") in self._running: continue - if self._dispatch(row): + postponement = budget_deferral(row, now, self._budget_exceeded) + if postponement is not None: + defer(self._store_for(row), row, postponement) + continue + if self._dispatch(row, now): free -= 1 - def _dispatch(self, row: dict[str, Any]) -> bool: + def _dispatch(self, row: dict[str, Any], reference: Any) -> bool: resolved = self._resolve(row) if resolved is None: retire(self._store_for(row), row, "owner could not be resolved") @@ -262,6 +321,15 @@ class GlobalScheduler: if not claim(self._db, row["uid"]): return False + if not reserve_run( + self._db, + str(row.get("owner_kind") or "user"), + str(row.get("owner_id") or ""), + row["uid"], + reference, + ): + defer(store, row, run_deferral(self._db, row, reference)) + return False owner_id = str(row.get("owner_id") or "") task = asyncio.create_task( diff --git a/devplacepy/services/devii/tasks/store.py b/devplacepy/services/devii/tasks/store.py index 679d26dd..518566f1 100644 --- a/devplacepy/services/devii/tasks/store.py +++ b/devplacepy/services/devii/tasks/store.py @@ -9,7 +9,17 @@ from typing import Any import dataset import sqlalchemy -from .guards import AutomationDenied, REASON_NOT_ADMIN, automation_allowed +from .guards import ( + AutomationDenied, + REASON_CREATE_QUOTA, + REASON_NESTED, + REASON_NOT_A_USER, + automation_allowed, + creation_denial, + nesting_allowed, +) +from .limits import create_quota, insert_task_within_quota +from .schedule import now_utc logger = logging.getLogger("devii.tasks.store") @@ -21,7 +31,27 @@ INDEXED_COLUMNS = ( ["next_run_at"], ["status"], ) -ADDED_COLUMNS = ( +TASK_COLUMNS = ( + ("uid", ""), + ("owner_kind", ""), + ("owner_id", ""), + ("label", ""), + ("prompt", ""), + ("enabled", True), + ("status", ""), + ("created_at", ""), + ("next_run_at", ""), + ("last_run_at", ""), + ("run_count", 0), + ("last_result", ""), + ("last_error", ""), + ("kind", ""), + ("run_at", ""), + ("delay_seconds", 0), + ("every_seconds", 0), + ("start_at", ""), + ("cron", ""), + ("max_runs", 0), ("deleted_at", ""), ("deleted_by", ""), ("notify", 0), @@ -73,12 +103,27 @@ class TaskStore: def _ensure_indexes(self) -> None: if TABLE not in self._db.tables: return - table = self._db[TABLE] - for column, example in ADDED_COLUMNS: - if not table.has_column(column): - table.create_column_by_example(column, example) + self._ensure_schema() for columns in INDEXED_COLUMNS: - table.create_index(columns) + try: + self._table.create_index(columns) + except sqlalchemy.exc.OperationalError as exc: + if "already exists" not in str(exc).lower(): + raise + logger.debug("Index on %s was created by another process", columns) + + def _ensure_schema(self) -> None: + table = self._db[TABLE] + for column, example in TASK_COLUMNS: + if table.has_column(column): + continue + try: + table.create_column_by_example(column, example) + except sqlalchemy.exc.OperationalError as exc: + if "duplicate column" not in str(exc).lower(): + raise + logger.debug("Column %s was added by another process", column) + table._reflect_table() @property def db(self) -> Any: @@ -91,7 +136,23 @@ class TaskStore: def require_automation(self) -> None: if not self.automation_allowed(): - raise AutomationDenied(REASON_NOT_ADMIN) + raise AutomationDenied(REASON_NOT_A_USER) + + def require_scheduling_allowed(self) -> None: + self.require_automation() + if self._operator: + return + if not nesting_allowed(self._owner_id): + raise AutomationDenied(REASON_NESTED) + + def require_creation_allowed(self) -> None: + if self._operator: + return + denial = creation_denial( + self._db, self._owner_kind, self._owner_id, now_utc() + ) + if denial is not None: + raise denial def count_active(self) -> int: rows = self._table.find(enabled=True, deleted_at=None, **self._scope) @@ -106,16 +167,22 @@ class TaskStore: return {"owner_kind": self._owner_kind, "owner_id": self._owner_id} def create(self, record: dict[str, Any]) -> None: - self.require_automation() - self._table.insert( - { - "deleted_at": None, - "deleted_by": None, - "failure_count": 0, - **record, - **self._scope, - } - ) + self.require_creation_allowed() + self._ensure_schema() + row = { + "deleted_at": None, + "deleted_by": None, + "failure_count": 0, + **record, + **self._scope, + } + if self._operator: + self._table.insert(row) + elif not insert_task_within_quota( + self._db, row, self._owner_kind, self._owner_id, now_utc() + ): + quota = create_quota(self._db, self._owner_kind, self._owner_id, now_utc()) + raise AutomationDenied(REASON_CREATE_QUOTA, quota.free_at) logger.info( "Task created uid=%s owner=%s/%s", record.get("uid"), @@ -139,7 +206,7 @@ class TaskStore: def update(self, uid: str, changes: dict[str, Any]) -> None: if changes.get("enabled"): - self.require_automation() + self.require_scheduling_allowed() changes = {**changes, "uid": uid, **self._scope} self._table.update(changes, ["uid", "owner_kind", "owner_id"]) logger.debug("Task updated uid=%s changes=%s", uid, list(changes)) diff --git a/devplacepy/templates/admin_devii_tasks.html b/devplacepy/templates/admin_devii_tasks.html index 173e7c3b..ccfeb033 100644 --- a/devplacepy/templates/admin_devii_tasks.html +++ b/devplacepy/templates/admin_devii_tasks.html @@ -6,10 +6,14 @@

- Scheduling is restricted to administrators. At most {{ limits['max_concurrent'] }} tasks run at - the same time across all owners, one at a time per owner. An owner may hold + Members may create {{ limits['member_create_24h'] }} tasks and execute + {{ limits['member_runs_24h'] }} runs per rolling 24 hours; administrators + {{ limits['admin_create_24h'] }} and {{ limits['admin_runs_24h'] }}. A run over the quota is + postponed until a slot frees, never dropped. At most {{ limits['max_concurrent'] }} tasks run at + the same time across all owners, one at a time per owner, and an owner may hold {{ limits['max_per_owner'] }} active tasks. A task stops itself after {{ limits['max_failures'] }} consecutive failures, or when its owner has been inactive for {{ limits['idle_days'] }} days. + Only administrators may schedule new work from inside a running task.