diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ed187a..e387e8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,19 @@ named rather than smoothed. ## [Unreleased] ### Added +- Run admission control on `delegate_task` (#101). The model may declare + `execution_mode` (`exclusive`, the default, or `parallel_read_only`) and up + to eight normalized `resource_keys` naming what a task touches — a repo + checkout, a deployment target. Two live runs that share a key never overlap + unless both are read-only; the check runs before a run id is minted or an + acceptance record is written, so a refused job burns nothing and can never + surface as `lost`. The refusal is a spoken tool result naming the run in + the way and the shared key, never a hang or a silent queue, and + `check_work` reads out what each running job holds. New knob + `TALK_TRUST_DECLARED_READ_ONLY`, default off: until the operator sets it, + `parallel_read_only` is downgraded to `exclusive` and recorded that way, + because the declaration is the delegating model's own claim, not a + sandbox. A task that names no keys is exactly the task Talk always ran. - The contributor experience, written down. `CONTRIBUTING.md` now ranks what we take first (bug fixes on live lanes, then provider and host compatibility, security hardening, cross-platform, new providers behind diff --git a/README.md b/README.md index 7fb1080..1cbaf85 100644 --- a/README.md +++ b/README.md @@ -516,6 +516,42 @@ silently does less than you asked: lane that always exists as long as `hermes` is on the PATH. 4. None available — a refusal naming all three missing lanes. +### Two jobs, one checkout — admission control + +Delegate two tasks that both touch the same repository and, left alone, they +race: two agents editing one checkout, two deploys to one target. Since #101 +the model can say what a task touches, and the run registry refuses the second +job instead of letting it collide: + +- `delegate_task` takes two optional arguments. `resource_keys` names up to + eight stable things the task touches — an absolute repo path, a deployment + target, a service name (whitespace-collapsed and case-folded, so two + spellings of one path are one key). `execution_mode` is `exclusive` (the + default) or `parallel_read_only`. +- Two live runs that share any key never overlap unless **both** are + `parallel_read_only`. The check happens before a run id is minted, before + the acceptance record is written, before the worker starts — a refused job + burns nothing and leaves no `lost` record behind. +- A refusal is a spoken tool result naming the run in the way: "run 4 (audit + the repo) is still running and touches the same resource ('/srv/app'); wait + for it, stop it, or re-delegate without that key." Never a hang, never a + silent queue. `check_work` reads out what each running job is holding. +- **`parallel_read_only` is believed only when you say so.** The declaration + is the delegating model's own claim about work it has not done yet — policy + input, not a sandbox — so by default it is downgraded to `exclusive` and + recorded that way. `TALK_TRUST_DECLARED_READ_ONLY=true` lets read-only jobs + on a shared key run together; the knob is read at admission time, so turning + it back off closes every overlap it had allowed. It is the only thing that + can widen behavior. +- No keys means no fence, in either direction: a task that names nothing is + exactly the task Talk always ran, record and all. + +The fence is per process and covers the api-server and detached lanes, whose +runs this registry owns. Inside `/talk`, the host's own delegation registry +runs the child: the job is still checked against the keys this registry +holds — never started on top of one — but it holds none itself afterwards, +and its `WORK_STARTED` receipt says so. + ### Redirecting work that's already running Say "tell that audit to focus on the token refresh instead" and `steer_agent` @@ -666,6 +702,7 @@ with defaults and failure modes: [docs/OPERATING.md](docs/OPERATING.md#configura | `TALK_API_SERVER_URL` | `http://127.0.0.1:8642` | Where the api-server lane looks | | `TALK_API_SERVER_KEY` | `API_SERVER_KEY` | Key for the api-server lane (blank = send none) | | `TALK_AGENT_TIMEOUT_S` | `1800` | Budget for one background run, and its watcher | +| `TALK_TRUST_DECLARED_READ_ONLY` | `false` | Believe a delegated task's `parallel_read_only` declaration, letting read-only runs share a `resource_key`; off downgrades every run to `exclusive` | | `TALK_IDENTITY_INCLUDE` | all | Which identity sections ride the prompt | | `TALK_MEMORY_SEARCH_TIMEOUT_S` | `10.0` | Wait bound for the in-process remembered-context (Honcho) lookup | | `TALK_SESSION_KEY` | unset | Stable operator scope sent as `X-Hermes-Session-Key` on api-server runs, so host-side memory survives `/clear` (blank = send none). **Not a session boundary: every voice-channel participant shares this scope** — memory reads are not gated by the operator ledger, so do not set it in multi-user channels until per-speaker scoping lands | diff --git a/docs/OPERATING.md b/docs/OPERATING.md index 40e14d8..a6a3aa7 100644 --- a/docs/OPERATING.md +++ b/docs/OPERATING.md @@ -356,6 +356,7 @@ allowlist returns a non-sensitive spoken denial without running the handler. |---|---|---| | `TALK_AGENT_PROFILE` | auto-detect | Hermes profile for the detached spawn. **Set-but-blank = explicit opt-out** (never pass `--profile`). Full story: [README](../README.md#talk_agent_profile--which-profile-the-background-agent-runs-under). | | `TALK_AGENT_TIMEOUT_S` | `1800` | Wall-clock budget for one background run and its watcher. Junk or ≤0 silently takes the default. | +| `TALK_TRUST_DECLARED_READ_ONLY` | `false` | Whether a delegated task's `parallel_read_only` declaration is believed (hermes-talk#101). Off (the default, and anything other than `1`/`true`/`yes`/`on`): every run is admitted as `exclusive`, so two runs sharing a `resource_key` never overlap. On: read-only runs may share a key; a read-only run still never overlaps an exclusive holder. Read at admission time — turning it off closes overlaps admitted earlier. The only knob that can widen behavior; the [README](../README.md#two-jobs-one-checkout--admission-control) has the model-facing contract. | | `TALK_APPROVAL_PROMPT_TIMEOUT_S` | `60.0` | How long a spoken approval question (the capability bridge) stays open before it resolves as **deny** — fail closed, silence is not consent. Sized under the host's own approval wait (300s) so the voice lane's deny lands first and the run unwinds on the operator's answer-or-silence. Junk or ≤0 silently takes the default. | | `TALK_CATALOG_STARTUP_WAIT_S` | `2.5` | Bounded head start a session start gives the first capability-catalog read, so a cold process still mints the live-catalog prompt section deterministically. `0` is honored and disables the wait (fire-and-forget); on expiry the session starts with the section omitted — logged, never a stall. Junk or negative silently takes the default. | | `TALK_MEMORY_SEARCH_TIMEOUT_S` | `10.0` | Wait bound for the in-process remembered-context (Honcho) tier of `search_memory`. On timeout the model speaks a retryable failure instead of the tool pipeline blocking; the transcript tier (`session_search`, a local FTS5 read) is not bounded. Junk or ≤0 silently takes the default. | diff --git a/talk_config.py b/talk_config.py index afcffb9..e7b4be1 100644 --- a/talk_config.py +++ b/talk_config.py @@ -870,6 +870,21 @@ def approval_prompt_timeout_s() -> float: ) +def trust_declared_read_only() -> bool: + """Whether a model's ``parallel_read_only`` declaration is believed. + + Default **off** (hermes-talk#101): the declaration is the delegating + model's own claim about work it has not done yet — policy input, not a + sandbox — so by default every delegated run is treated as ``exclusive`` + and two runs sharing a resource key never overlap. Only an explicit + ``TALK_TRUST_DECLARED_READ_ONLY=true`` lets read-only runs on a shared + key run together; anything else, junk included, keeps the fence. + """ + + raw = (os.environ.get("TALK_TRUST_DECLARED_READ_ONLY") or "").strip().lower() + return raw in {"1", "true", "yes", "on"} + + __all__ = [ "DEFAULT_AGENT_TIMEOUT_S", "DEFAULT_API_SERVER_POLL_S", @@ -936,5 +951,6 @@ def approval_prompt_timeout_s() -> float: "talk_model", "talk_provider", "talk_voice", + "trust_declared_read_only", "voice_mode", ] diff --git a/talk_host.py b/talk_host.py index 1c72ade..47cffb7 100644 --- a/talk_host.py +++ b/talk_host.py @@ -222,6 +222,15 @@ def _catalog_from_host_modules() -> dict | None: #: aloud on every such answer. REMEMBERED_PREFIX = "from remembered context: " +#: Appended to a tier-1 receipt when the model declared resource keys +#: (hermes-talk#101). The host's own delegation registry ran the child and +#: this plugin's run registry does not hold it, so a later run naming the +#: same key would not be fenced against it — said rather than assumed. +HOST_LOOP_ADMISSION_NOTE = ( + " (this lane runs inside Hermes's own delegation registry, which does " + "not hold resource keys — the fence applies to api-server and detached runs)" +) + MAX_TOOL_OUTPUT_CHARS = 2_000 @@ -1022,7 +1031,14 @@ def _search_memory_via_api_server(self, query: str, limit: int) -> str: "Hermes agent through the api server; I'll tell you what it finds." ) - def run_agent(self, prompt: str, background: bool = True) -> str: + def run_agent( + self, + prompt: str, + background: bool = True, + *, + execution_mode: str | None = None, + resource_keys: Any = None, + ) -> str: """Hand a self-contained task to a background Hermes agent. Four backends, tried in order, and every fall-through is ANNOUNCED in @@ -1044,10 +1060,23 @@ def run_agent(self, prompt: str, background: bool = True) -> str: ``background`` is accepted for the caller's mental model but never forwarded: Hermes documents the tool's own flag as deprecated and ignored, and every real backend is asynchronous regardless. + + ``execution_mode`` / ``resource_keys`` are the admission declaration + (hermes-talk#101), enforced by the run registry on tiers 2 and 3. Tier + 1 hands the child to Hermes's own delegation registry, which this + plugin does not own: it is still checked against the keys the run + registry holds — never started on top of one — but it holds none + itself afterwards, and the receipt says so. """ ctx = get_ctx() if ctx is not None: + try: + talk_runs.check_admission(execution_mode, resource_keys) + except talk_runs.AdmissionRefused as exc: + return f"I can't start that yet — {exc}." + except ValueError as exc: + return f"I couldn't start that work: {exc}" try: raw = ctx.dispatch_tool(DELEGATE_TOOL_NAME, {"goal": prompt}) except Exception as exc: # noqa: BLE001 — the model speaks the failure @@ -1058,14 +1087,27 @@ def run_agent(self, prompt: str, background: bool = True) -> str: # A host refusal (paused delegation, depth limit) must # never ride behind a WORK_STARTED prefix. return f"I couldn't start that work — {spoken}" - return f"WORK_STARTED — {spoken}" + started = f"WORK_STARTED — {spoken}" + if talk_runs.normalize_resource_keys(resource_keys): + started += HOST_LOOP_ADMISSION_NOTE + return started - via_api_server = self._run_api_server_agent(prompt) + via_api_server = self._run_api_server_agent( + prompt, execution_mode=execution_mode, resource_keys=resource_keys + ) if via_api_server is not None: return via_api_server - return self._run_detached_agent(prompt) + return self._run_detached_agent( + prompt, execution_mode=execution_mode, resource_keys=resource_keys + ) - def _run_api_server_agent(self, prompt: str) -> str | None: + def _run_api_server_agent( + self, + prompt: str, + *, + execution_mode: str | None = None, + resource_keys: Any = None, + ) -> str | None: """Tier 2: run the task on a real agent over the api_server. ``None`` means the lane is unavailable and the caller should fall @@ -1077,9 +1119,13 @@ def _run_api_server_agent(self, prompt: str) -> str | None: label = prompt.strip()[:60] try: run_id = talk_runs.start_run( - "agent", label, _api_server_worker(prompt, session_id=None) + "agent", + label, + _api_server_worker(prompt, session_id=None), + execution_mode=execution_mode, + resource_keys=resource_keys, ) - except talk_runs.RoutingUnavailable as exc: + except (talk_runs.RoutingUnavailable, talk_runs.AdmissionRefused) as exc: return f"I can't start that yet — {exc}." except Exception as exc: # noqa: BLE001 — the model speaks the failure return f"I couldn't start that work: {type(exc).__name__}: {exc}" @@ -1088,7 +1134,13 @@ def _run_api_server_agent(self, prompt: str) -> str | None: "Hermes agent through the api server; I'll tell you when it lands." ) - def _run_detached_agent(self, prompt: str) -> str: + def _run_detached_agent( + self, + prompt: str, + *, + execution_mode: str | None = None, + resource_keys: Any = None, + ) -> str: """Tier 3/4: run the task as a detached ``hermes -z`` one-shot.""" binary = hermes_binary() @@ -1101,9 +1153,13 @@ def _run_detached_agent(self, prompt: str) -> str: label = prompt.strip()[:60] try: run_id = talk_runs.start_run( - "agent", label, _detached_agent_worker(prompt, binary) + "agent", + label, + _detached_agent_worker(prompt, binary), + execution_mode=execution_mode, + resource_keys=resource_keys, ) - except talk_runs.RoutingUnavailable as exc: + except (talk_runs.RoutingUnavailable, talk_runs.AdmissionRefused) as exc: return f"I can't start that yet — {exc}." except Exception as exc: # noqa: BLE001 — the model speaks the failure return f"I couldn't start that work: {type(exc).__name__}: {exc}" @@ -1827,6 +1883,7 @@ def host() -> HostAdapter: "AGENT_LOOP_ABSENT_MARKERS", "DELEGATE_TOOL_NAME", "HERMES_BINARY", + "HOST_LOOP_ADMISSION_NOTE", "LANE_API_SERVER", "LANE_ATTACHED", "LANE_NONE", diff --git a/talk_runs.py b/talk_runs.py index fd855f9..968a9dc 100644 --- a/talk_runs.py +++ b/talk_runs.py @@ -121,6 +121,199 @@ class RoutingUnavailable(RuntimeError): """ +# -- admission control (hermes-talk#101) -------------------------------------- +# +# A delegating model may declare each run ``exclusive`` or +# ``parallel_read_only`` and name up to MAX_RESOURCE_KEYS stable resources it +# touches (an absolute repo path, a deployment target). Two LIVE runs in this +# process that share a key never overlap unless both are read-only AND the +# operator has opted into believing that declaration +# (``TALK_TRUST_DECLARED_READ_ONLY``). No keys means no fence, in either +# direction — a run that names nothing is exactly today's run. +# +# The fence is per PROCESS. The registry is; a run in the dashboard lane's +# process is invisible here in every other respect too (see the module +# docstring), and a history-only "running" record is a dead process's, not +# a live holder. + +EXECUTION_EXCLUSIVE = "exclusive" +EXECUTION_PARALLEL_READ_ONLY = "parallel_read_only" +EXECUTION_MODES = (EXECUTION_EXCLUSIVE, EXECUTION_PARALLEL_READ_ONLY) +MAX_RESOURCE_KEYS = 8 +MAX_RESOURCE_KEY_CHARS = 200 + +#: Keys reserved between the admission check and the registry insert. The +#: acceptance write in between takes the file lock, which must never nest +#: inside ``_RUN_LOCK``, so the check cannot simply hold the lock across it — +#: and two tool-pool workers admitted in that gap would both start. Guarded +#: by ``_RUN_LOCK``. +_RESERVATIONS: dict[int, dict] = {} +_RESERVATION_SEQ = 0 + + +class AdmissionRefused(RuntimeError): + """A live run already holds a resource this one names (hermes-talk#101). + + Raised BEFORE acceptance: no run id is burned, no history row is written, + no thread starts. The message names the run in the way so the model can + offer to wait for it, stop it, or re-delegate without that key — a + refusal the operator can act on, never a silent queue. + """ + + def __init__(self, *, run_id: int | None, label: str, keys: tuple[str, ...]) -> None: + self.run_id = run_id + self.label = label + self.keys = keys + quoted = ", ".join(f"'{key}'" for key in keys) + noun = "resource" if len(keys) == 1 else "resources" + # A None run id is a reservation: a sibling accepted a breath ago and + # has no number yet. It is as live as any run — say so, without one. + holder = ( + f"run {run_id} ({label})" if run_id is not None else f"a run just accepted ({label})" + ) + super().__init__( + f"{holder} is still running and touches the same " + f"{noun} ({quoted}); wait for it, stop it, or re-delegate without that key" + ) + + +def normalize_resource_keys(keys: Any) -> tuple[str, ...]: + """The canonical form of a declared key list; raises ``ValueError`` when unusable. + + Whitespace is stripped and collapsed, case is folded, blanks dropped, + duplicates removed with order kept. Folding case is the SAFE direction: + two spellings of one path become one key, so the fence errs toward a + refusal, never toward an overlap. More than :data:`MAX_RESOURCE_KEYS` + after normalization is refused rather than truncated — a key dropped on + the floor would be a silent hole in the fence. A bare string is one key. + """ + + if keys is None: + return () + if isinstance(keys, str): + keys = [keys] + if not isinstance(keys, (list, tuple)): + raise ValueError("resource_keys must be a list of short strings") + seen: list[str] = [] + for raw in keys: + if not isinstance(raw, str): + raise ValueError("resource_keys must be a list of short strings") + key = " ".join(raw.split()).casefold() + if not key: + continue + if len(key) > MAX_RESOURCE_KEY_CHARS: + raise ValueError( + f"a resource key must be at most {MAX_RESOURCE_KEY_CHARS} characters" + ) + if key not in seen: + seen.append(key) + if len(seen) > MAX_RESOURCE_KEYS: + raise ValueError(f"at most {MAX_RESOURCE_KEYS} resource_keys are allowed") + return tuple(seen) + + +def resolve_execution_mode(mode: str | None) -> str: + """The mode a run is ADMITTED under, after the trust knob has its say. + + ``None`` is ``exclusive``. ``parallel_read_only`` survives only when the + operator set ``TALK_TRUST_DECLARED_READ_ONLY`` (read at call time, Rule + 1); otherwise it is downgraded to ``exclusive`` — the model's claim is + kept as policy input, never as a sandbox. Anything else is a caller bug. + """ + + if mode is None: + return EXECUTION_EXCLUSIVE + if mode not in EXECUTION_MODES: + raise ValueError(f"unknown execution_mode: {mode!r}") + if mode == EXECUTION_PARALLEL_READ_ONLY and not talk_config.trust_declared_read_only(): + return EXECUTION_EXCLUSIVE + return mode + + +def _effective_mode(admission: dict | None) -> str: + """A holder's mode under the knob as it is NOW, not as it was at admission.""" + + if not admission or admission.get("mode") != EXECUTION_PARALLEL_READ_ONLY: + return EXECUTION_EXCLUSIVE + return ( + EXECUTION_PARALLEL_READ_ONLY + if talk_config.trust_declared_read_only() + else EXECUTION_EXCLUSIVE + ) + + +def _admission_conflict_locked(mode: str, keys: tuple[str, ...]) -> dict | None: + """The first live holder this run may not overlap, or ``None``. Caller holds the lock. + + Live means a non-terminal registry entry or a reservation mid-acceptance. + Two runs may share a key only when BOTH are read-only under the knob as + it stands now — a holder admitted as read-only under an earlier + configuration is judged again, so turning the knob off closes every + overlap it had allowed. + """ + + if not keys: + return None + holders = [ + (run_id, run) for run_id, run in _RUNS.items() if run["status"] not in TERMINAL_STATUSES + ] + holders.extend((None, reservation) for reservation in _RESERVATIONS.values()) + for run_id, holder in holders: + admission = holder.get("admission") or {} + held = admission.get("keys") or () + shared = tuple(key for key in keys if key in held) + if not shared: + continue + if ( + mode == EXECUTION_PARALLEL_READ_ONLY + and _effective_mode(admission) == EXECUTION_PARALLEL_READ_ONLY + ): + continue + return {"runId": run_id, "label": holder.get("label") or "", "keys": shared} + return None + + +def check_admission(execution_mode: str | None, resource_keys: Any) -> None: + """Raise :class:`AdmissionRefused` if this declaration could not start now. + + A dry check — nothing is reserved. For the lane that runs a child inside + the host's own delegation registry (which this fence cannot hold), it is + still the half that CAN be enforced: never start on top of a registry + run that holds the key. ``ValueError`` for an unusable declaration. + """ + + mode = resolve_execution_mode(execution_mode) + keys = normalize_resource_keys(resource_keys) + with _RUN_LOCK: + conflict = _admission_conflict_locked(mode, keys) + if conflict is not None: + raise AdmissionRefused( + run_id=conflict["runId"], label=conflict["label"], keys=conflict["keys"] + ) + + +def _reserve_admission_locked(label: str, mode: str, keys: tuple[str, ...]) -> int: + """Hold ``keys`` until the entry lands in the registry. Caller holds the lock.""" + + global _RESERVATION_SEQ + _RESERVATION_SEQ += 1 + _RESERVATIONS[_RESERVATION_SEQ] = { + "label": label, + "admission": {"mode": mode, "keys": list(keys)}, + } + return _RESERVATION_SEQ + + +def _copy_admission(admission: dict) -> dict: + """A snapshot's own copy of an admission — the key list included. + + Snapshots are handed to callers who may mutate them; a shared list would + let a reader widen or narrow a LIVE holder's fence. + """ + + return {**admission, "keys": list(admission.get("keys") or ())} + + # The ambient ticket for the currently attached Talk connection. Same shape and # contract as talk_lifecycle's attach/detach: one connection at a time, last # attach wins, fail closed while unbound. Module-level state is per PROCESS, so @@ -456,6 +649,8 @@ def start_run( worker: Callable[[int], str], *, meta: dict | None = None, + execution_mode: str | None = None, + resource_keys: Any = None, ) -> int: """Register a run and spawn its daemon worker thread. @@ -468,10 +663,18 @@ def start_run( ``ALLOW_EPHEMERAL_ENV`` opt-in. All three mean the same thing: there is no exact place to send the result, so accepting the job would be a promise this process cannot keep. + + ``execution_mode`` and ``resource_keys`` are the admission declaration + (hermes-talk#101). Raises :class:`AdmissionRefused` — also before any + work, before even a run id — when a live run holds a named key this one + may not share. Both absent is exactly the run this function always + accepted: no fence, no extra record field. """ if kind not in RUN_KINDS: raise ValueError(f"unknown run kind: {kind!r}") + mode = resolve_execution_mode(execution_mode) + keys = normalize_resource_keys(resource_keys) owner = current_owner() if owner is None: @@ -494,12 +697,38 @@ def start_run( "ts": now, "updated": now, } + declared = execution_mode is not None or bool(keys) + if declared: + entry["admission"] = {"mode": mode, "keys": list(keys)} + # Admission BEFORE acceptance: a refused run must burn no id and leave no + # history row — a "running" record for work that never started would + # surface as `lost` on the next reconnect. The keys are reserved under + # the same lock the check ran under, because the acceptance write below + # cannot happen inside it (lock ordering) and two workers admitted in + # that gap would both start. + with _RUN_LOCK: + conflict = _admission_conflict_locked(mode, keys) + if conflict is not None: + raise AdmissionRefused( + run_id=conflict["runId"], label=conflict["label"], keys=conflict["keys"] + ) + reservation = _reserve_admission_locked(label, mode, keys) if keys else None # Durability FIRST, then the registry, then the worker. The old order wrote # history last and fail-open, so a failed write still returned a run id and # the caller still spoke WORK_STARTED — a receipt for a run nothing could # ever route. - run_id = _accept_run(entry) + try: + run_id = _accept_run(entry) + except BaseException: + if reservation is not None: + with _RUN_LOCK: + _RESERVATIONS.pop(reservation, None) + raise with _RUN_LOCK: + if reservation is not None: + # Same critical section as the insert: the keys pass from the + # reservation to the entry with no instant in which nobody holds them. + _RESERVATIONS.pop(reservation, None) _RUNS[run_id] = entry # Evict AFTER inserting so the cap holds for the registry as it now # stands; the entry just added is running, so it is never a candidate. @@ -873,6 +1102,8 @@ def list_undelivered_for_session( for run_id, run in _RUNS.items(): snapshot = dict(run) snapshot["meta"] = dict(run["meta"]) + if "admission" in run: + snapshot["admission"] = _copy_admission(run["admission"]) snapshot["runId"] = run_id live[run_id] = snapshot # Deliberately NOT list_runs(limit=100, ...): that limit is a UI display @@ -978,6 +1209,8 @@ def get_run(run_id: int) -> dict | None: return None snapshot = dict(run) snapshot["meta"] = dict(run["meta"]) + if "admission" in run: + snapshot["admission"] = _copy_admission(run["admission"]) snapshot["runId"] = run_id return snapshot @@ -998,6 +1231,8 @@ def list_runs(limit: int = 10, include_history: bool = False) -> list[dict]: for run_id, run in _RUNS.items(): snapshot = dict(run) snapshot["meta"] = dict(run["meta"]) + if "admission" in run: + snapshot["admission"] = _copy_admission(run["admission"]) snapshot["runId"] = run_id live[run_id] = snapshot @@ -1121,10 +1356,12 @@ def reset_for_tests() -> None: fail-closed behaviour hermes-talk#35 added. """ - global _RUN_SEQ + global _RUN_SEQ, _RESERVATION_SEQ with _RUN_LOCK: _RUNS.clear() + _RESERVATIONS.clear() _RUN_SEQ = 0 + _RESERVATION_SEQ = 0 with _PROCESS_LOCK: _PROCESSES.clear() detach_owner() @@ -1135,12 +1372,19 @@ def reset_for_tests() -> None: "DELIVERED", "DELIVERY_CLAIMED", "DELIVERY_PENDING", + "EXECUTION_EXCLUSIVE", + "EXECUTION_MODES", + "EXECUTION_PARALLEL_READ_ONLY", "HISTORY_OUTPUT_CAP", + "MAX_RESOURCE_KEYS", + "MAX_RESOURCE_KEY_CHARS", "RUN_KINDS", "TERMINAL_STATUSES", + "AdmissionRefused", "RoutingUnavailable", "annotate_run", "attach_owner", + "check_admission", "claim_delivery", "current_owner", "detach_owner", @@ -1150,9 +1394,11 @@ def reset_for_tests() -> None: "list_runs", "list_undelivered_for_session", "mark_delivered", + "normalize_resource_keys", "register_process", "release_process", "reset_for_tests", + "resolve_execution_mode", "start_run", "started_sentinel", "terminate_process", diff --git a/talk_tools.py b/talk_tools.py index f94bbcf..aa64b9e 100644 --- a/talk_tools.py +++ b/talk_tools.py @@ -140,7 +140,11 @@ "Hand a real task to a background Hermes agent and keep talking. The " "agent starts fresh and never sees this call, so write the whole task " "out: what to do, where it lives, and what done looks like. Returns a " - "WORK_STARTED receipt — say it is running and move on." + "WORK_STARTED receipt — say it is running and move on. If the task " + "touches something other work might also touch — a repository " + "checkout, a deployment target — name it in resource_keys so two jobs " + "never collide; a refusal names the run in the way, so offer to wait " + "for it, stop it, or retry without that key." ), "parameters": { "type": "object", @@ -155,6 +159,30 @@ "type": "boolean", "description": "Run without blocking the call (default true).", }, + "execution_mode": { + "type": "string", + "enum": ["exclusive", "parallel_read_only"], + "description": ( + "How this task may share its resource_keys with other running " + "work. 'exclusive' (the default): nothing else touching the " + "same key runs at the same time. 'parallel_read_only': the " + "task only reads, so it may overlap other read-only work on " + "the same key — honored only when the operator has chosen to " + "trust that declaration. Use exclusive unless the task is " + "certainly read-only." + ), + }, + "resource_keys": { + "type": "array", + "items": {"type": "string"}, + "maxItems": 8, + "description": ( + "Stable names for what the task touches: an absolute " + "repository path, a deployment target, a service name. Two " + "tasks that share a key never run together unless both are " + "parallel_read_only. Omit when the task touches nothing shared." + ), + }, }, "required": ["task"], "additionalProperties": False, @@ -462,7 +490,21 @@ def _handle_delegate_task(arguments: dict) -> str: if not task: return "delegate_task needs a task to hand off." background = arguments.get("background") - return talk_host.host().run_agent(task, background is not False) + # The admission declaration (hermes-talk#101) is validated HERE, before + # any backend is consulted: a malformed declaration must not fall through + # to a lane that would then run the task unfenced. + mode = arguments.get("execution_mode") + if mode is not None: + mode = str(mode).strip().lower() or None + if mode is not None and mode not in talk_runs.EXECUTION_MODES: + return "delegate_task's execution_mode must be 'exclusive' or 'parallel_read_only'." + try: + keys = talk_runs.normalize_resource_keys(arguments.get("resource_keys")) + except ValueError as exc: + return f"delegate_task could not use those resource_keys: {exc}." + return talk_host.host().run_agent( + task, background is not False, execution_mode=mode, resource_keys=keys + ) def _describe_age(run: dict) -> str: @@ -483,6 +525,12 @@ def _describe_run(run: dict) -> str: line = f"run {run.get('runId')} ({run.get('kind')}) {run.get('status')}" if run.get("status") == "running": line += _describe_age(run) + # What a live run holds (hermes-talk#101), so "why was that refused?" + # has an answer the model can read out. + admission = run.get("admission") if isinstance(run.get("admission"), dict) else {} + held = [key for key in admission.get("keys") or () if isinstance(key, str)] + if held: + line += " holding " + ", ".join(f"'{key}'" for key in held) if run.get("status") == "lost": line += " (started before this session — I can't see how it ended)" # A stop verb's detached confirmation lands in meta (hermes-talk#2) — diff --git a/tests/test_admission.py b/tests/test_admission.py new file mode 100644 index 0000000..a2a8692 --- /dev/null +++ b/tests/test_admission.py @@ -0,0 +1,513 @@ +"""Run admission control — execution_mode + resource_keys (hermes-talk#101). + +What is being proved: two live runs that share a resource key never +overlap; disjoint keys and undeclared runs are untouched; a read-only pair +may share a key only when the operator has opted into believing the +declaration; a refusal happens BEFORE acceptance (no run id burned, no +history row) and names the run in the way; the reservation closes the +check-then-accept gap between two tool-pool workers; and the declaration +rides delegate_task through the host adapter into the registry. +""" + +from __future__ import annotations + +import json +import threading +import time +from pathlib import Path + +import pytest + +import talk_config +import talk_host +import talk_runs +import talk_tools + +pytestmark = pytest.mark.usefixtures("_registry") + + +@pytest.fixture +def _registry(monkeypatch): + talk_host.bind_ctx(None) + talk_runs.reset_for_tests() + talk_runs.attach_owner( + talk_session_id="ts-test", + generation_id="gen-test", + hermes_session_id="sess-test", + operator="test", + profile=None, + ) + monkeypatch.delenv("TALK_TRUST_DECLARED_READ_ONLY", raising=False) + monkeypatch.setattr(talk_host, "hermes_binary", lambda: None) + yield + talk_host.bind_ctx(None) + talk_runs.reset_for_tests() + + +class _Gate: + """Workers that stay running until released, so admission has something to hit.""" + + def __init__(self): + self.release = threading.Event() + + def worker(self, _run_id: int) -> str: + self.release.wait(5.0) + return "done" + + +def _wait_terminal(run_id: int, timeout: float = 3.0) -> dict: + deadline = time.time() + timeout + while time.time() < deadline: + run = talk_runs.get_run(run_id) + if run and run["status"] in talk_runs.TERMINAL_STATUSES: + return run + time.sleep(0.01) + raise AssertionError(f"run {run_id} never finished") + + +# -- the fence ---------------------------------------------------------------- + + +def test_disjoint_keys_run_together(): + gate = _Gate() + first = talk_runs.start_run("agent", "audit", gate.worker, resource_keys=["/repo/a"]) + second = talk_runs.start_run("agent", "deploy", gate.worker, resource_keys=["/repo/b"]) + + assert talk_runs.get_run(first)["status"] == "running" + assert talk_runs.get_run(second)["status"] == "running" + gate.release.set() + _wait_terminal(first) + _wait_terminal(second) + + +def test_shared_keys_serialize_until_the_holder_finishes(): + gate = _Gate() + first = talk_runs.start_run("agent", "audit the repo", gate.worker, resource_keys=["/repo/a"]) + + with pytest.raises(talk_runs.AdmissionRefused) as refused: + talk_runs.start_run("agent", "deploy", gate.worker, resource_keys=["/repo/b", "/repo/a"]) + assert refused.value.run_id == first + assert refused.value.keys == ("/repo/a",) + + gate.release.set() + _wait_terminal(first) + # A terminal holder holds nothing: the same declaration is admitted now. + second_gate = _Gate() + second = talk_runs.start_run( + "agent", "deploy", second_gate.worker, resource_keys=["/repo/a"] + ) + assert talk_runs.get_run(second)["status"] == "running" + second_gate.release.set() + + +def test_the_refusal_names_the_run_and_the_key_and_what_to_do(): + gate = _Gate() + first = talk_runs.start_run("agent", "audit the repo", gate.worker, resource_keys=["/repo/a"]) + + with pytest.raises(talk_runs.AdmissionRefused) as refused: + talk_runs.start_run("agent", "deploy", gate.worker, resource_keys=["/repo/a"]) + + message = str(refused.value) + assert message.startswith(f"run {first} (audit the repo) is still running") + assert "the same resource ('/repo/a')" in message + assert "wait for it, stop it, or re-delegate without that key" in message + gate.release.set() + + +def test_two_shared_keys_are_plural_in_the_refusal(): + gate = _Gate() + talk_runs.start_run("agent", "a", gate.worker, resource_keys=["x", "y", "z"]) + # Named in the order the REFUSED run declared them — the model reads its + # own declaration back, not the holder's. + with pytest.raises(talk_runs.AdmissionRefused, match=r"resources \('z', 'x'\)"): + talk_runs.start_run("agent", "b", gate.worker, resource_keys=["z", "x"]) + gate.release.set() + + +def test_read_only_pairs_overlap_only_with_the_knob_on(monkeypatch): + gate = _Gate() + talk_runs.start_run( + "agent", "read a", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + # Default: the declaration is not trusted, so the holder was admitted as + # exclusive and the sibling is refused. + with pytest.raises(talk_runs.AdmissionRefused): + talk_runs.start_run( + "agent", "read b", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + gate.release.set() + talk_runs.reset_for_tests() + talk_runs.attach_owner( + talk_session_id="ts-test", generation_id="g", hermes_session_id="s", + operator="test", profile=None, + ) + + monkeypatch.setenv("TALK_TRUST_DECLARED_READ_ONLY", "true") + gate = _Gate() + first = talk_runs.start_run( + "agent", "read a", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + second = talk_runs.start_run( + "agent", "read b", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + assert talk_runs.get_run(first)["admission"]["mode"] == "parallel_read_only" + assert talk_runs.get_run(second)["status"] == "running" + # Read-only never overlaps a MUTATING holder of the same key, knob or not. + with pytest.raises(talk_runs.AdmissionRefused): + talk_runs.start_run("agent", "write a", gate.worker, resource_keys=["/repo/a"]) + gate.release.set() + + +def test_the_declaration_is_downgraded_not_dropped_when_untrusted(): + gate = _Gate() + run_id = talk_runs.start_run( + "agent", "read", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + admission = talk_runs.get_run(run_id)["admission"] + assert admission == {"mode": "exclusive", "keys": ["/repo/a"]} + gate.release.set() + + +def test_turning_the_knob_off_closes_overlaps_it_had_allowed(monkeypatch): + """A holder admitted as read-only under an earlier configuration is judged + under the knob as it stands NOW — the safer default cannot be escaped by + state minted before it was set.""" + + monkeypatch.setenv("TALK_TRUST_DECLARED_READ_ONLY", "1") + gate = _Gate() + talk_runs.start_run( + "agent", "read a", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + monkeypatch.delenv("TALK_TRUST_DECLARED_READ_ONLY") + with pytest.raises(talk_runs.AdmissionRefused): + talk_runs.start_run( + "agent", "read b", gate.worker, + execution_mode="parallel_read_only", resource_keys=["/repo/a"], + ) + gate.release.set() + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (None, False), + ("", False), + ("0", False), + ("junk", False), + ("1", True), + ("true", True), + ("YES", True), + (" on ", True), + ], +) +def test_the_trust_knob_is_off_unless_explicitly_on(monkeypatch, raw, expected): + if raw is None: + monkeypatch.delenv("TALK_TRUST_DECLARED_READ_ONLY", raising=False) + else: + monkeypatch.setenv("TALK_TRUST_DECLARED_READ_ONLY", raw) + assert talk_config.trust_declared_read_only() is expected + + +# -- the declaration ---------------------------------------------------------- + + +def test_keys_are_normalized_and_capped(): + assert talk_runs.normalize_resource_keys(None) == () + assert talk_runs.normalize_resource_keys("one key") == ("one key",) + assert talk_runs.normalize_resource_keys( + [" C:/Repo/A ", "c:/repo/a", "Deploy Target", "", " "] + ) == ("c:/repo/a", "deploy target") + + assert talk_runs.normalize_resource_keys([f"k{i}" for i in range(8)]) == tuple( + f"k{i}" for i in range(8) + ) + with pytest.raises(ValueError, match="at most 8"): + talk_runs.normalize_resource_keys([f"k{i}" for i in range(9)]) + with pytest.raises(ValueError, match="at most 200 characters"): + talk_runs.normalize_resource_keys(["x" * 201]) + with pytest.raises(ValueError, match="list of short strings"): + talk_runs.normalize_resource_keys([1, 2]) + with pytest.raises(ValueError, match="list of short strings"): + talk_runs.normalize_resource_keys({"a": 1}) + + +def test_case_folded_keys_collide_on_purpose(): + """Two spellings of one path are one key — the fence errs toward refusal.""" + + gate = _Gate() + talk_runs.start_run("agent", "a", gate.worker, resource_keys=["C:\\Repo"]) + with pytest.raises(talk_runs.AdmissionRefused, match="'c:\\\\repo'"): + talk_runs.start_run("agent", "b", gate.worker, resource_keys=["c:\\REPO"]) + gate.release.set() + + +def test_an_unknown_mode_is_a_caller_bug(): + with pytest.raises(ValueError, match="execution_mode"): + talk_runs.start_run("agent", "a", lambda _r: "x", execution_mode="yolo") + assert talk_runs.list_runs() == [] + + +# -- byte-for-byte compatibility ---------------------------------------------- + + +def test_undeclared_runs_are_exactly_yesterdays_runs(monkeypatch, tmp_path: Path): + state = tmp_path / "state" + state.mkdir() + monkeypatch.setattr(talk_runs, "_history_path", lambda: state / "talk-runs.jsonl") + monkeypatch.setattr(talk_runs, "_history_enabled", lambda: True) + gate = _Gate() + + first = talk_runs.start_run("agent", "a", gate.worker) + second = talk_runs.start_run("agent", "b", gate.worker) + third = talk_runs.start_run("agent", "c", gate.worker, resource_keys=["/repo"]) + # Undeclared runs never fence and are never fenced: all three are live. + assert all(talk_runs.get_run(rid)["status"] == "running" for rid in (first, second, third)) + assert "admission" not in talk_runs.get_run(first) + assert "admission" not in talk_runs.get_run(second) + + records = [ + json.loads(line) + for line in (state / "talk-runs.jsonl").read_text(encoding="utf-8").splitlines() + ] + undeclared = next(rec for rec in records if rec["runId"] == first) + assert set(undeclared) == { + "runId", "kind", "label", "status", "output", "meta", "ticket", "delivery", "ts", "updated", + } + declared = next(rec for rec in records if rec["runId"] == third) + assert declared["admission"] == {"mode": "exclusive", "keys": ["/repo"]} + gate.release.set() + + +def test_a_refusal_burns_no_run_id_and_writes_no_history_row(monkeypatch, tmp_path: Path): + state = tmp_path / "state" + state.mkdir() + monkeypatch.setattr(talk_runs, "_history_path", lambda: state / "talk-runs.jsonl") + monkeypatch.setattr(talk_runs, "_history_enabled", lambda: True) + gate = _Gate() + first = talk_runs.start_run("agent", "a", gate.worker, resource_keys=["/repo"]) + lines_before = (state / "talk-runs.jsonl").read_text(encoding="utf-8").splitlines() + + with pytest.raises(talk_runs.AdmissionRefused): + talk_runs.start_run("agent", "b", gate.worker, resource_keys=["/repo"]) + + assert (state / "talk-runs.jsonl").read_text(encoding="utf-8").splitlines() == lines_before + gate.release.set() + _wait_terminal(first) + # The next accepted run takes the very next id — nothing was burned. + assert talk_runs.start_run("agent", "c", gate.worker, resource_keys=["/repo"]) == first + 1 + gate.release.set() + + +# -- the reservation ---------------------------------------------------------- + + +def test_the_reservation_closes_the_check_then_accept_gap(monkeypatch): + """Two tool-pool workers admitted in the gap between the check and the + registry insert would both start. The first holds its keys from the + moment it passes the check, even while its acceptance write is in + flight — and releases them if that write fails.""" + + entered = threading.Event() + proceed = threading.Event() + real_accept = talk_runs._accept_run + outcome: dict = {} + + def slow_accept(entry): + if entry["label"] == "slow": + entered.set() + proceed.wait(5.0) + return real_accept(entry) + + monkeypatch.setattr(talk_runs, "_accept_run", slow_accept) + gate = _Gate() + + def start_slow(): + try: + outcome["slow"] = talk_runs.start_run( + "agent", "slow", gate.worker, resource_keys=["/repo"] + ) + except Exception as exc: # noqa: BLE001 — recorded for the assertion + outcome["slow"] = exc + + slow = threading.Thread(target=start_slow) + slow.start() + assert entered.wait(2.0) + # Mid-acceptance: not yet in the registry, but already holding the key. + with pytest.raises(talk_runs.AdmissionRefused) as refused: + talk_runs.start_run("agent", "fast", gate.worker, resource_keys=["/repo"]) + assert refused.value.run_id is None + assert "a run just accepted (slow)" in str(refused.value) + + proceed.set() + slow.join(2.0) + assert isinstance(outcome["slow"], int) + assert talk_runs._RESERVATIONS == {} + gate.release.set() + + +def test_a_failed_acceptance_releases_its_reservation(): + def failing_accept(_entry): + raise talk_runs.RoutingUnavailable("disk on fire") + + gate = _Gate() + # Scoped so the real acceptance path is back for the second call. + with pytest.MonkeyPatch.context() as failing: + failing.setattr(talk_runs, "_accept_run", failing_accept) + with pytest.raises(talk_runs.RoutingUnavailable): + talk_runs.start_run("agent", "a", gate.worker, resource_keys=["/repo"]) + assert talk_runs._RESERVATIONS == {} + + # Nothing lingers: the same key is free for the next caller. + run_id = talk_runs.start_run("agent", "b", gate.worker, resource_keys=["/repo"]) + assert talk_runs.get_run(run_id)["status"] == "running" + gate.release.set() + + +def test_snapshots_carry_the_admission_and_check_work_reads_it_out(): + gate = _Gate() + run_id = talk_runs.start_run("agent", "audit", gate.worker, resource_keys=["/repo", "prod"]) + + listed = next(run for run in talk_runs.list_runs() if run["runId"] == run_id) + assert listed["admission"] == {"mode": "exclusive", "keys": ["/repo", "prod"]} + listed["admission"]["keys"].append("tampered") + assert talk_runs.get_run(run_id)["admission"]["keys"] == ["/repo", "prod"] + + spoken = talk_tools.execute_talk_tool("check_work", {}) + assert f"run {run_id} (agent) running" in spoken + assert "holding '/repo', 'prod'" in spoken + gate.release.set() + + +# -- the tool and the host ---------------------------------------------------- + + +def test_delegate_task_advertises_the_declaration(): + schema = next(t for t in talk_tools.default_talk_tools() if t["name"] == "delegate_task") + properties = schema["parameters"]["properties"] + assert properties["execution_mode"]["enum"] == ["exclusive", "parallel_read_only"] + assert properties["resource_keys"]["maxItems"] == talk_runs.MAX_RESOURCE_KEYS + assert properties["resource_keys"]["items"] == {"type": "string"} + assert schema["parameters"]["required"] == ["task"] + + +def test_delegate_task_threads_the_declaration_into_the_host(monkeypatch): + seen: dict = {} + + class _Host: + def run_agent(self, task, background=True, *, execution_mode=None, resource_keys=None): + seen.update( + task=task, background=background, mode=execution_mode, keys=resource_keys + ) + return "WORK_STARTED #1 kind=agent (x)" + + monkeypatch.setattr(talk_host, "host", lambda: _Host()) + result = talk_tools.execute_talk_tool( + "delegate_task", + { + "task": "audit it", + "execution_mode": "Parallel_Read_Only", + "resource_keys": [" /Repo ", "/repo", "prod"], + }, + ) + assert result.startswith("WORK_STARTED") + assert seen == { + "task": "audit it", + "background": True, + "mode": "parallel_read_only", + "keys": ("/repo", "prod"), + } + + # Absent: exactly the call the host always received. + talk_tools.execute_talk_tool("delegate_task", {"task": "plain"}) + assert seen["mode"] is None and seen["keys"] == () + + +def test_delegate_task_refuses_a_bad_declaration_before_any_lane_runs(monkeypatch): + def never(*_a, **_k): # pragma: no cover - must not be reached + raise AssertionError("a malformed declaration reached a backend") + + monkeypatch.setattr(talk_host, "host", lambda: type("H", (), {"run_agent": never})()) + + assert "'exclusive' or 'parallel_read_only'" in talk_tools.execute_talk_tool( + "delegate_task", {"task": "x", "execution_mode": "yolo"} + ) + too_many = talk_tools.execute_talk_tool( + "delegate_task", {"task": "x", "resource_keys": [f"k{i}" for i in range(9)]} + ) + assert "at most 8" in too_many + + +class _StubCtx: + def __init__(self, result): + self.calls: list = [] + self.result = result + + def dispatch_tool(self, tool_name, args, **kwargs): + self.calls.append((tool_name, args)) + return self.result + + +def test_the_registry_lanes_speak_the_refusal(monkeypatch): + """Tier 2/3: the refusal is spoken in the same shape as a routing refusal — + nothing was accepted, so nothing is in flight to check on.""" + + monkeypatch.setattr(talk_host.talk_apiserver, "is_available", lambda: True) + monkeypatch.setattr( + talk_host, "_api_server_worker", lambda prompt, session_id=None: _Gate().worker + ) + gate = _Gate() + first = talk_runs.start_run("agent", "audit the repo", gate.worker, resource_keys=["/repo"]) + + spoken = talk_host.host().run_agent("deploy the repo", resource_keys=["/repo"]) + + assert spoken.startswith("I can't start that yet — ") + assert f"run {first} (audit the repo)" in spoken + assert "WORK_STARTED" not in spoken + assert len(talk_runs.list_runs()) == 1 + gate.release.set() + + +def test_the_registry_lanes_admit_and_record_the_declaration(monkeypatch): + monkeypatch.setattr(talk_host.talk_apiserver, "is_available", lambda: True) + monkeypatch.setattr( + talk_host, "_api_server_worker", lambda prompt, session_id=None: (lambda _r: "ok") + ) + + spoken = talk_host.host().run_agent( + "read the repo", execution_mode="parallel_read_only", resource_keys=["/repo"] + ) + + assert spoken.startswith("WORK_STARTED #") + run_id = int(spoken.split("#", 1)[1].split(" ", 1)[0]) + assert talk_runs.get_run(run_id)["admission"] == {"mode": "exclusive", "keys": ["/repo"]} + _wait_terminal(run_id) + + +def test_the_host_loop_lane_is_checked_but_says_it_holds_nothing(): + """Tier 1 hands the child to Hermes's own delegation registry: it is never + started on top of a registry holder, and the receipt says the fence does + not hold it afterwards — never silently unfenced.""" + + ctx = _StubCtx(json.dumps({"success": True, "result": "subagent 4 started"})) + talk_host.bind_ctx(ctx) + gate = _Gate() + first = talk_runs.start_run("agent", "audit", gate.worker, resource_keys=["/repo"]) + + refused = talk_host.host().run_agent("deploy", resource_keys=["/repo"]) + assert f"run {first} (audit)" in refused + assert ctx.calls == [] + + started = talk_host.host().run_agent("deploy elsewhere", resource_keys=["/other"]) + assert started.startswith("WORK_STARTED — subagent 4 started") + assert started.endswith(talk_host.HOST_LOOP_ADMISSION_NOTE) + assert ctx.calls == [("delegate_task", {"goal": "deploy elsewhere"})] + + plain = talk_host.host().run_agent("no keys at all") + assert plain == "WORK_STARTED — subagent 4 started" + gate.release.set()