diff --git a/README.md b/README.md index 90d147c..8369f17 100644 --- a/README.md +++ b/README.md @@ -67,12 +67,16 @@ Tmux owns PTYs and interactive terminal lifecycle. Python is restricted to evaluation and provenance; it does not implement a second production workflow or acceptance gate. -On production Linux, separate Unix identities isolate the orchestrator, the -single active writer, read-only agents, and the authority supervisor. The +On production Linux, separate Unix identities isolate the orchestrator, +path-scoped writers, read-only agents, and the authority supervisor. The orchestrator can read worker and reviewer state but cannot write the target repository or protected lifecycle state. Completion is a request to the supervisor, which checks every gate under the lifecycle lock before changing -the phase to `complete`. +the phase to `complete`. A Rust lifecycle reconciler also performs mechanical +process polling, terminal cleanup, assignment settlement, stale validation +lease handling, and only pre-budgeted infrastructure retries. It owns no +semantic policy and sends protected mutations to the authority supervisor; +prompts retain semantic routing but no longer implement those fixed operations. ## Common Commands diff --git a/docs/architecture.md b/docs/architecture.md index 8778c67..2f9f5ff 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -86,7 +86,7 @@ Production Linux runs four process identities: | Identity | May read target | May write target | Protected-state authority | | --- | --- | --- | --- | | Orchestrator | yes | no | typed requests only | -| Writer | yes | assigned paths while leased | no | +| Writers | yes | assigned paths while leased | no | | Scout/reviewer | yes | no | sealed evidence only | | Supervisor | metadata needed for gates | grants/revokes writer paths | yes | @@ -106,6 +106,15 @@ workspace. The supervisor isolates connection failures. A disconnected client or broken pipe terminates that request, not the authority service. +A dedicated Rust lifecycle reconciler runs in the tmux session under the +non-writing orchestrator identity, because that identity owns the tmux socket. +It observes panes, persists terminal output, and closes completed processes. +Assignment and validation-lease settlement crosses the typed socket to the +authority supervisor; the reconciler cannot write protected state directly. It +may restore a recoverable infrastructure exit only when the agent was launched +with remaining `infra_retry_budget`. It never chooses the worker count, task +split, owned paths, or semantic repair. + ## Lifecycle The normal state sequence is: @@ -127,10 +136,22 @@ pre-implementation -> implementation -> post-implementation -> complete ### Implementation -The supervisor authorizes one writer for predeclared paths. The writer receives -the complete approved context and registered contract. Assignment checks reject -changed files outside the declared scope. Read-only exploration can still run -in parallel. +The orchestrator builds an adaptive task graph and chooses worker +responsibilities. The supervisor does not choose a worker count or task split; +it authorizes each writer for predeclared paths only after the implementation +gate is ready. The DAG reports dependency-ready nodes for orchestrator routing. +Assignment checks reject changed files outside the declared scope. + +Disjoint writers can run concurrently when Linux Landlock provides a distinct +per-process write boundary. Overlapping ownership is rejected. On systems +without that boundary, the same workflow remains valid but writer execution is +serialized as a security fallback. Read-only agents are not subject to the +writer lease. + +This separates decisions from mechanics: the orchestrator reacts to `blocked` +or exhausted recovery states, while the supervisor continuously reconciles +observable process state without needing prompt instructions for polling, +finalization, ownership release, or lease cleanup. ### Post-implementation diff --git a/docs/decisions.md b/docs/decisions.md index 0f07ec7..5f595b1 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -77,19 +77,31 @@ state, create work requests, and ask for completion, but it cannot directly edit the target or protected state. A bypassed high-level CLI still runs under the same non-writing Unix identity. -## Permit One Path-Scoped Writer - -**Decision:** Only one writer may be active. Its launch authorization binds the -workflow, assignment, backend, prompt, and predeclared owned paths. Ownership is -granted for the role lifetime and revoked after exit or cancellation. - -**Why:** Multiple writers complicate overlap detection and permit detached or -late writes after cancellation. Sequential publication gives one consistent -snapshot and acceptance boundary while agents may still explore in parallel as -read-only roles. - -**Consequence:** Multiagent prioritizes deterministic authority over maximum -write concurrency. +The tmux-owning identity also hosts a fixed Rust lifecycle reconciler. That +process may observe and close panes, but protected assignment and lease changes +still cross the supervisor's typed authority socket. This avoids granting the +shared worker/reviewer group access to the tmux control socket. + +## Keep Topology Adaptive and Writing Path-Scoped + +**Decision:** The orchestrator chooses the task graph, worker count, and each +worker's responsibility. The supervisor does not encode a preferred topology. +It admits write-capable workers only when their durable assignments own +non-overlapping paths and the lifecycle gate is open. + +On Linux kernels with Landlock, disjoint writers may run concurrently under +per-process write allowlists. If that isolation is unavailable, the supervisor +falls back to a single active writer because shared Unix ownership cannot +safely distinguish two processes using the writer UID. + +**Why:** Task decomposition is semantic and belongs to the orchestrator; +non-overlap, lifecycle readiness, and isolation capability are mechanical and +belong to the supervisor. A hard-coded worker count makes simple work expensive +and parallelizable work unnecessarily slow. + +**Consequence:** Multiagent imposes no fixed worker count or responsibility +catalog. Available isolation and actual path conflicts determine concurrency, +while completion still evaluates one canonical diff. ## Bind Semantics and Reviews to Immutable Evidence diff --git a/docs/getting-started.md b/docs/getting-started.md index 61fbbea..c8bcf3b 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -50,7 +50,8 @@ multiagent launch \ --root /absolute/path/to/target-repo ``` -The default launch creates one tmux orchestrator window. It is a clean launch: +The default launch creates an orchestrator window and a Rust lifecycle-supervisor +window. It is a clean launch: persisted subagents are not automatically restored. Resume after an interrupted run: @@ -190,8 +191,6 @@ multiagent subagent spawn contract-scout-01 \ --role scout \ --instruction "Extract structured must and must-not contract rules. Do not edit." -multiagent subagent wait contract-scout-01 --timeout 900 -multiagent subagent finalize contract-scout-01 multiagent workflow contract-register "$MULTIAGENT_WORKFLOW_ID" \ --scout contract-scout-01 ``` @@ -239,15 +238,18 @@ multiagent subagent spawn worker-01 \ --workflow-id "$MULTIAGENT_WORKFLOW_ID" \ --decision-id DEC-001 \ --plan-id PLAN-A \ + --infra-retries 1 \ --instruction-file /absolute/path/to/worker-instruction.md -multiagent subagent wait worker-01 --timeout 1800 multiagent subagent assignment-check worker-01 ``` Only the supervisor-authorized writer receives temporary access to its existing -owned paths. The global writer lease prevents a second writer from becoming -active at the same time. +owned paths. On Linux, disjoint path-scoped writers may run concurrently; +overlapping writers are rejected. The lifecycle supervisor observes completion, +settles ownership and validation state, and applies only the explicitly +budgeted infrastructure retry. Use `multiagent status` to decide semantic next +steps such as repairing blocked work. Update a durable checkpoint during long work: @@ -269,7 +271,8 @@ multiagent snapshot --root "$MULTIAGENT_ROOT" --base HEAD --format json Transition to post-implementation with the reported hash, then run read-only scope, technical, decision-drift, and reflection reviews. Review instructions must include the original task, registered contract, approved context, and -canonical diff. Finalize each reviewer so the supervisor can seal its output. +canonical diff. The lifecycle supervisor finalizes terminal reviewer processes +so their output can be sealed. Record review findings and todos through `multiagent workflow` and `multiagent subagent` commands. A changed diff invalidates previous acceptance. diff --git a/evaluation/evalscope_multiagent_native_runner.py b/evaluation/evalscope_multiagent_native_runner.py index 1d0d6bf..3b20f9e 100644 --- a/evaluation/evalscope_multiagent_native_runner.py +++ b/evaluation/evalscope_multiagent_native_runner.py @@ -82,7 +82,10 @@ def solver_internal_timeout(agent_timeout: float) -> int: - reserve = int(os.environ.get("EVAL_NATIVE_SOLVER_TIMEOUT_RESERVE", "600")) + # Trace export and official verification happen after the solver process + # returns, so this reserve only needs to cover orderly tmux shutdown and + # workspace ownership restoration before AgentEnvironment's hard timeout. + reserve = int(os.environ.get("EVAL_NATIVE_SOLVER_TIMEOUT_RESERVE", "180")) reserve = max(90, min(reserve, int(agent_timeout) - 300)) return max(300, int(agent_timeout) - reserve) diff --git a/evaluation/native_solver/swe_prod_lifecycle.py b/evaluation/native_solver/swe_prod_lifecycle.py index 6ce2595..a40342b 100644 --- a/evaluation/native_solver/swe_prod_lifecycle.py +++ b/evaluation/native_solver/swe_prod_lifecycle.py @@ -313,14 +313,20 @@ def run_prod_solver(prompt_path: str | None, workdir: Path, repo_root: Path, tim deadline = time.monotonic() + timeout resume_count = 0 + final_phase: str | None = None try: while time.monotonic() < deadline: while time.monotonic() < deadline and tmux_has_orchestrator(session): time.sleep(5) phase = active_workflow_phase() - if phase in {None, "complete"} or time.monotonic() >= deadline: + final_phase = phase + if phase == "complete" or time.monotonic() >= deadline: break + if phase is None: + raise RuntimeError( + "production multiagent orchestrator exited without a persisted workflow lifecycle" + ) resume_count += 1 log( @@ -343,8 +349,15 @@ def run_prod_solver(prompt_path: str | None, workdir: Path, repo_root: Path, tim finally: if tmux_has_session(session): run(["tmux", "-S", str(TMUX_SOCKET), "kill-session", "-t", session], timeout=30) + restore_workspace_owner(workdir) + + if final_phase != "complete": + rendered_phase = final_phase or "missing" + raise RuntimeError( + "production multiagent workflow did not reach supervisor-owned completion before " + f"the solver deadline (phase={rendered_phase}); refusing workspace handoff" + ) - restore_workspace_owner(workdir) materialize_committed_changes(workdir, start_head) mark_untracked_intent_to_add(workdir, baseline_untracked=baseline_untracked) log("workspace prepared for EvalScope submission") diff --git a/evaluation/native_solver/swe_prod_repository.py b/evaluation/native_solver/swe_prod_repository.py index 9264f5f..ef48f89 100644 --- a/evaluation/native_solver/swe_prod_repository.py +++ b/evaluation/native_solver/swe_prod_repository.py @@ -59,12 +59,23 @@ def list_untracked_files(cwd: Path) -> list[str]: return [line.strip() for line in others.stdout.splitlines() if line.strip()] +def is_framework_internal_path(path: str) -> bool: + """Return whether an untracked path belongs to multiagent's control plane.""" + + normalized = path[2:] if path.startswith("./") else path + return normalized == ".multiagent" or normalized.startswith(".multiagent/") + + def mark_untracked_intent_to_add(cwd: Path, *, baseline_untracked: set[str] | None = None) -> list[str]: """Expose newly created solver files without submitting image residue.""" baseline = baseline_untracked or set() untracked = list_untracked_files(cwd) - intent_to_add = [path for path in untracked if (cwd / path).is_file()] + intent_to_add = [ + path + for path in untracked + if (cwd / path).is_file() and not is_framework_internal_path(path) + ] intent_to_add = [path for path in intent_to_add if path not in baseline] if intent_to_add: result = run(["git", "add", "-N", "--", *intent_to_add], cwd=cwd, timeout=120) diff --git a/orchestrator_prompt.md b/orchestrator_prompt.md index ce3a41e..5710014 100644 --- a/orchestrator_prompt.md +++ b/orchestrator_prompt.md @@ -5,8 +5,9 @@ You are the orchestrator, a commander running on Codex CLI. You run inside a dedicated tmux window. Your job is to coordinate worker agents and long-running subagents running in other tmux windows. You do not implement code yourself. You plan, spawn agents, monitor them, coordinate handoffs, -finalize results, kill finished or stuck agents, spawn more agents when needed, -and report status. +make semantic decisions for blocked or exhausted work, spawn more agents when +needed, and report status. The Rust lifecycle reconciler performs polling, +terminal cleanup, assignment settlement, and budgeted infrastructure retry. ## Role @@ -142,20 +143,22 @@ Use clear names: - Verifiers: `verifier-01-short-task` - Long-running subagents: `subagent-build-watch` -Use one verifier window per worker assignment at a time. A verifier is a -read-only reviewer, not a second implementer. +Choose verifier topology from the evidence boundary. A verifier may review one +assignment or a consolidated diff; the framework does not require one verifier +per worker. A verifier is a read-only reviewer, not a second implementer. -Before spawning a replacement worker for the same owned files, poll the existing -worker and either finalize/kill it or explicitly wait. Prefer the bounded -`multiagent subagent wait NAME --timeout SECONDS` command when a result is -required before continuing; one immediate poll is not evidence that an agent is -stalled. If validation ownership +Before spawning a replacement worker for the same owned files, inspect the +supervisor-observed state of the existing worker. The lifecycle supervisor owns +polling, terminal cleanup, assignment settlement, abandoned validation leases, +and explicitly budgeted infrastructure retry. The orchestrator decides whether +a blocked or exhausted task needs replacement; one immediate observation is not +evidence that an agent is stalled. If validation ownership is unclear, use the validation coordinator role before adding more workers. ## Role Routing Load `$PROMPT_DIR/prompts/playbooks/orchestration-routing.md` before spawning, -verifying, replacing, or finalizing agents. It owns the detailed role-routing +verifying, replacing, or accepting agent results. It owns the detailed role-routing workflow, progress/status procedure, safety rules, and optional playbook selection. diff --git a/prompts/playbooks/agent-spawning.md b/prompts/playbooks/agent-spawning.md index 76ae64d..788bada 100644 --- a/prompts/playbooks/agent-spawning.md +++ b/prompts/playbooks/agent-spawning.md @@ -1,7 +1,7 @@ # Agent Spawning Playbook -Use this playbook whenever the orchestrator is about to create, monitor, -replace, verify, or finalize worker windows or named subagents. +Use this playbook whenever the orchestrator is about to create, inspect, +replace, or verify worker windows or named subagents. ## Worker First Instruction @@ -43,14 +43,18 @@ SUBAGENT_CLI="$WORKER_CLI" multiagent subagent spawn worker-01-task \ --decision-id DECISION_ID \ --plan-id PLAN_ID \ --branch BRANCH \ + --infra-retries 1 \ --instruction-file WORKER_INSTRUCTION -multiagent subagent wait worker-01-task --timeout 1800 ``` The supervisor creates the assignment under its lock, completes authority -registration, and only then launches the trusted workspace-write worker. The -orchestrator remains unable to edit the target workspace. Inspect a terminal -`blocked` or `failed` result instead of treating it as completion. Separate git +registration, and only then launches the trusted workspace-write worker. Its +lifecycle reconciler observes the process, closes terminal windows, updates the +assignment, and marks any abandoned validation lease stale. `--infra-retries` +is an optional per-task policy; the supervisor may restore only that many +recoverable infrastructure exits and never invents repair work. The orchestrator +remains unable to edit the target workspace. Inspect a terminal `blocked` or +`failed` result instead of treating it as completion. Separate git worktrees remain available for intentionally parallel, disjoint assignments, but require an explicit integration step before completion; do not use them for the normal SWE single-writer path. @@ -62,11 +66,13 @@ persists context: ```bash multiagent subagent spawn subagent-build-watch --instruction "FIRST_INSTRUCTION_TEXT" -multiagent subagent wait subagent-build-watch --timeout 900 multiagent subagent inspect subagent-build-watch --lines 160 -multiagent subagent finalize subagent-build-watch ``` +The lifecycle supervisor monitors and finalizes the process automatically. +Use `multiagent status` when a semantic decision depends on its result; do not +reimplement polling or terminal cleanup in the prompt. + For a bounded worker in the current worktree, `spawn` can create the durable assignment and worker in one command: @@ -85,17 +91,18 @@ progress, before stopping, and whenever a blocker appears. ## Scout To Worker Handoff Read-only scouts are temporary evidence gatherers. Before spawning the first -edit-capable worker, poll or inspect any active scout once, persist the useful -ledger/findings, then finalize or kill the scout if it is still running. Do not -let an active generic scout block `multiagent subagent spawn` for the implementation -worker. Use `MULTIAGENT_ALLOW_PARALLEL_WORKERS=1` only when you intentionally -want parallel disjoint workers and have recorded non-overlapping ownership. +edit-capable worker, inspect the supervisor-observed status and persist the +useful ledger/findings. Do not +let an active generic scout block `multiagent subagent spawn` for implementation. +The orchestrator may select any number of workers and task-specific +responsibilities. Give every concurrent writer a durable path-scoped +assignment; the supervisor admits disjoint leases and rejects overlap +mechanically. -For a contract scout, finalize it and register its sealed output before any -worker or reviewer starts: +For a contract scout, wait for supervisor-finalized output and register its +sealed output before any worker or reviewer starts: ```bash -multiagent subagent finalize CONTRACT_SCOUT_NAME multiagent workflow contract-register "$MULTIAGENT_WORKFLOW_ID" \ --scout CONTRACT_SCOUT_NAME ``` @@ -107,8 +114,8 @@ injects the supervisor-owned original task and registered contract into every later worker and reviewer instruction. Give a live contract scout one bounded wait of at least 300 seconds before -classifying it as stalled. Do not kill or finalize a running scout merely -because one short poll has no final message. If it exits with an empty sealed +classifying it as stalled. Do not cancel a running scout merely because one +short status observation has no final message. If it exits with an empty sealed artifact, allow at most one replacement with a narrower source list and an explicit "return the structured artifact before any ninth tool call" reminder. If that replacement also has no artifact, stop with a recorded infrastructure @@ -185,11 +192,10 @@ continue indefinitely: the next state must be a source diff, `required-path-outside-owned: RELATIVE_PATH`, `validation-repair-needed:`, or blocked status with a source-visible reason. -After `multiagent subagent kill NAME` or `multiagent subagent finalize NAME`, ensure the -assignment no longer owns paths before reusing them. If needed, run -`multiagent subagent assignment-status NAME failed` for killed workers or -`multiagent subagent assignment-status NAME done` for finalized workers before -creating the replacement assignment. +After a worker reaches a terminal state, wait for the lifecycle supervisor to +settle its assignment before reusing those paths. Do not manually reproduce the +assignment or validation-lease cleanup sequence. A blocked worker is not +terminal; replacement remains an explicit orchestrator decision. Before final acceptance, run: diff --git a/prompts/playbooks/dag.md b/prompts/playbooks/dag.md index f60732a..4655c78 100644 --- a/prompts/playbooks/dag.md +++ b/prompts/playbooks/dag.md @@ -39,12 +39,13 @@ pending -> ready -> running -> done Only the orchestrator updates node status. Agents report their state; the orchestrator translates reports into DAG state. -## Typical Role Dependencies +## Responsibilities -- Exploration nodes usually depend only on initial architecture or research. -- Exploitation nodes depend on the selected decision and required architecture. -- QA/verifier nodes depend on the implementation nodes they verify. -- Reflection nodes depend on implementation, QA, or metrics nodes. +`--responsibility` is orchestrator-defined text, not a framework enum. Describe +the concrete output of the node, such as `repair storage adapter` or `verify API +compatibility`. Security authority is selected separately when the agent is +spawned. The DAG provides structure and dependency tracking; it does not automatically -spawn agents. The orchestrator remains the active workflow controller. +spawn agents or prescribe a topology. The orchestrator remains the active +workflow controller. diff --git a/prompts/playbooks/implementation-lifecycle.md b/prompts/playbooks/implementation-lifecycle.md index 6e5cd49..b01baa1 100644 --- a/prompts/playbooks/implementation-lifecycle.md +++ b/prompts/playbooks/implementation-lifecycle.md @@ -70,12 +70,13 @@ Spawn that review read-only through the normal subagent path, for example: ```bash SUBAGENT_CLI="$VERIFIER_CLI" multiagent subagent spawn decision-authority-reviewer \ --role reviewer --instruction-file AUTHORITY_REVIEW_INPUT -multiagent subagent wait decision-authority-reviewer --timeout 900 +multiagent status ``` -Do not continue merely because an immediate poll still reports `running`. -Inspect the completed or blocked result after the bounded wait and persist its -actual authority finding before preparing implementation. +The lifecycle supervisor monitors and finalizes the reviewer. Do not continue +merely because one observation still reports `running`; wait for its completed +or blocked state and persist the actual authority finding before preparing +implementation. Create an approved implementation context document containing the selected plan, decision and plan IDs, authority and approval basis, intended outcome, @@ -100,7 +101,9 @@ workers. ## Implementation -Spawn bounded exploitation workers only after the implementation gate passes. +Spawn bounded write-capable workers only after the implementation gate passes. +Choose their count and responsibilities from the approved plan rather than a +framework default. Concurrent writers must have disjoint owned paths. Every assignment must reference the active workflow, decision, and plan. The worker's first instruction must contain the complete current approved implementation context; a decision ID alone is insufficient. @@ -128,7 +131,8 @@ Run independent reviews against the frozen candidate diff: Every reviewer final message must include an exact durable marker on its own line: `review-record: type=TYPE verdict=pass|findings diff=DIFF_HASH` (use `diff=-` for decision-authority). A Markdown list prefix or enclosing backticks -are accepted as cosmetic formatting, but surrounding prose is not. Wait for and finalize that reviewer before +are accepted as cosmetic formatting, but surrounding prose is not. Wait for the +supervisor-finalized reviewer result before recording its result. Record each review with `multiagent workflow record-review ... --reviewer REVIEWER_NAME`; the supervisor rejects an orchestrator-authored verdict that is not backed by the finalized read-only diff --git a/prompts/playbooks/orchestration-routing.md b/prompts/playbooks/orchestration-routing.md index eaa7cd5..a1dc9e8 100644 --- a/prompts/playbooks/orchestration-routing.md +++ b/prompts/playbooks/orchestration-routing.md @@ -47,10 +47,12 @@ but it must not narrow, replace, or contradict the scout's historical ledger. If the proposed worker scope cannot address every output in that ledger, widen the bounded ownership to the actual transition owner or create explicit todos for the remaining outputs before implementation. -Before spawning the edit-capable implementation worker, poll or inspect any -active scout once, persist useful findings, then finalize or kill the scout if it -is still running. Do not let an active generic scout block the implementation -worker spawn; enable parallel workers only for explicit disjoint ownership. +Before spawning the edit-capable implementation worker, inspect the +supervisor-observed state of active scouts and persist useful findings. If a +generic scout remains genuinely stalled after the bounded wait, explicitly +cancel it; terminal cleanup is automatic. Do not let an active generic scout block the implementation +worker spawn. Choose the implementation topology from the task and record +explicit, non-overlapping ownership for every concurrent writer. When a task may add, remove, rename, or move source symbols, the worker first instruction must include `source-owner-ledger:` with `selected-owner=...`, all plausible `candidate-owner=...`, rejected-owner reasons, and @@ -86,9 +88,8 @@ intended validation commands. SUBAGENT_CLI="$VERIFIER_CLI" multiagent subagent spawn validation-coordinator-01-task --instruction "FIRST_INSTRUCTION_TEXT" ``` -Use the coordinator's lease report to decide whether to wait, poll, -kill/finalize stale panes, release a validation lease, or route a bounded -follow-up worker. +Use the coordinator's lease report to decide whether to wait for the lifecycle +supervisor, cancel genuinely stuck work, or route a bounded follow-up worker. ## Required Worker First Instruction @@ -130,8 +131,8 @@ Before spawning the verifier, load `prompts/playbooks/validation-scheduling.md` if the worker ran or is running expensive validation. Do not spawn the verifier until the worker's validation lease has a captured passed, failed, timed-out, stale, or released state. If the worker final message appears before its -validation command exits, poll the worker/process list instead of starting a -verifier that may duplicate the command. +validation command exits, wait for the supervisor-observed process and lease +state instead of starting a verifier that may duplicate the command. The orchestrator decides which findings become accepted follow-up; never pass raw verifier findings directly to the worker as orders. Accepted blocking @@ -157,8 +158,8 @@ test, fixture, compile, package, component, or source-derived probe failed after the patch. This is a repair signal, not acceptance evidence. 1. Capture the exact failing command, return code, and output tail. -2. Record or release the validation lease for the package/path before starting - replacement work. +2. Require a terminal validation-lease record before starting replacement work; + the lifecycle supervisor marks an abandoned active lease stale. 3. Derive the implicated source paths from the failing command, stack trace, fixture name, changed files, and contract ledger. 4. Spawn a fresh bounded repair worker with those paths in `--owned`; do not @@ -170,7 +171,7 @@ the patch. This is a repair signal, not acceptance evidence. 6. Only after the repair worker returns should a verifier decide acceptance, residual risk, or a bounded second follow-up. -Do not finalize on source review, compile-only checks, or synthetic helper +Do not accept on source review, compile-only checks, or synthetic helper probes while a relevant visible validation command is still failing. A stale visible expectation can be accepted only when the repair/verifier transcript contains both the source-visible reason and a replacement probe for the exact @@ -205,14 +206,13 @@ and use its progress/status procedure. source mapping: the next state must be a source diff, `required-path-outside-owned: RELATIVE_PATH`, `validation-repair-needed:`, or blocked status with a source-visible reason. -- After killing or finalizing a worker, release its assignment ownership before - reusing paths: `multiagent subagent assignment-status NAME failed` for killed - workers or `multiagent subagent assignment-status NAME done` for finalized - workers, then create the replacement assignment. +- The lifecycle supervisor settles terminal assignments and abandoned + validation leases. Wait for that observed state before reusing paths; do not + duplicate the cleanup commands in orchestration instructions. - Never let a verifier receive writable ownership for a worker's owned paths. - Before accepting completed worker or subagent work, run `multiagent subagent assignment-check NAME`. - Always capture final output before killing a worker. -- Always poll or inspect a long-running subagent before finalizing it. +- Use supervisor-observed state rather than manual poll/finalize sequences. - Do not delete `$MULTIAGENT_STATE_DIR`; it is durable context. - Prefer killing and respawning a stuck worker over manually untangling a confused one. - Keep a state table of active agents, owned files, branch names, status, and state directory. @@ -221,9 +221,11 @@ and use its progress/status procedure. 1. Plan: understand intent, run a contract scout when risk justifies it, update the contract ledger, split work, assign owner/branch/scope. 2. Spawn: create assignment metadata, load the right prompt module, start the agent, send the assignment. -3. Monitor: use `multiagent status`, inspect busy/blocked/done states, update checkpoints. +3. Monitor: use `multiagent status`; the lifecycle supervisor performs process + polling, terminal cleanup, assignment settlement, and budgeted infrastructure + retry while the orchestrator handles only semantic busy/blocked/done routing. 4. Coordinate: resolve blockers, prevent ownership conflicts, maintain validation leases, run scope guard when diff shape is risky, route verification, spawn independent follow-ups. -5. Accept: run `assignment-check`, review verifier findings, close accepted todo resolutions with `multiagent subagent todo-close ...` after reverification or reopen them, run `multiagent subagent gate-check`, finalize agents. +5. Accept: run `assignment-check`, review verifier findings, close accepted todo resolutions with `multiagent subagent todo-close ...` after reverification or reopen them, and run `multiagent subagent gate-check`; terminal process cleanup is automatic. 6. Report: summarize status, branches, commits, blockers, state paths, validation, and residual risk. ## Optional Playbooks diff --git a/prompts/playbooks/parallel-execution.md b/prompts/playbooks/parallel-execution.md index c871a81..132dc99 100644 --- a/prompts/playbooks/parallel-execution.md +++ b/prompts/playbooks/parallel-execution.md @@ -5,10 +5,16 @@ uncertainty justifies parallel exploration before implementation. ## Fan-Out Rule -Default to broad safe fan-out. Build a dependency graph from true blocking -artifacts, not vague ordering preferences. When multiple useful workers are -ready and their owned paths do not overlap, spawn them in the same wave and -consolidate their outputs later. +Choose topology from the task. The framework does not prescribe a worker count, +fixed worker responsibilities, or a default fan-out width. Build a dependency +graph from true blocking artifacts, not vague ordering preferences. When +multiple useful workers are ready and their owned paths do not overlap, they may +run in the same wave and consolidate their outputs later. A small cohesive task +may still use one worker. + +Record each task-specific responsibility in assignment metadata. Security roles +such as writer, scout, and reviewer describe authority only; they are not a +catalog of what an agent is allowed to reason about. If one subtree is blocked, keep spawning every other ready subtree. If work runs sequentially, state the exact dependency that prevents safe parallelism. diff --git a/prompts/playbooks/recovery.md b/prompts/playbooks/recovery.md index 192d3cb..1935363 100644 --- a/prompts/playbooks/recovery.md +++ b/prompts/playbooks/recovery.md @@ -23,6 +23,11 @@ subagents persisted memory. ## Recovery Actions +The lifecycle supervisor already restores exits covered by each agent's +persisted `infra_retry_budget`. Rows that remain in this plan exhausted that +budget or require semantic judgment; the supervisor deliberately does not +choose a new worker, scope, or repair strategy. + - `restore`: closed subagent with recoverable context. Report the restore, then run `multiagent subagent restore NAME` when appropriate. - `skip-open`: active tmux window already exists. Poll or inspect it; do not restore it. - `skip-finalized`: appears done, finalized, killed, or intentionally stopped. Do not restore by default. diff --git a/prompts/playbooks/validation-scheduling.md b/prompts/playbooks/validation-scheduling.md index ea5aadb..e772b66 100644 --- a/prompts/playbooks/validation-scheduling.md +++ b/prompts/playbooks/validation-scheduling.md @@ -48,17 +48,18 @@ not silently take a second lease for the same package/path. ## Routing Rules -- If a package/path has a running lease, poll that owner before starting another - equivalent command. +- If a package/path has a running lease, inspect its supervisor-observed owner + state before starting another equivalent command. - Before starting an expensive command, acquire a validation lease. If the - helper reports a conflict, do not run the duplicate command; poll or inspect - the named owner and report `blocked-validations:`. + helper reports a conflict, do not run the duplicate command; inspect the + named owner's recorded state and report `blocked-validations:`. - Do not spawn a verifier for a worker while that worker still owns a running - validation lease. First capture/poll the worker until the leased command - reaches passed, failed, timed-out, stale, or released. Then pass the captured + validation lease. Wait until the supervisor-observed lease reaches passed, + failed, timed-out, stale, or released. Then pass the captured result to the verifier. -- If the owner is stale, capture the pane and process list, then explicitly - kill/finalize or release the lease before replacement work starts. +- If an owner terminates without recording a result, the lifecycle supervisor + marks its active lease `stale`. Inspect that recorded state before choosing a + replacement; do not manually race terminal cleanup. - If the lease result is failed and the command is relevant to the changed source or contract ledger, route a bounded repair worker before acceptance. Pass the failing command, output tail, changed files, and lease target to that diff --git a/prompts/roles/contract-scout.md b/prompts/roles/contract-scout.md index ffd3885..dc1845f 100644 --- a/prompts/roles/contract-scout.md +++ b/prompts/roles/contract-scout.md @@ -67,6 +67,18 @@ When task text contrasts embedding with an internal or named field, emit both the positive named-field obligation and a separate negative rule forbidding the old anonymous embedding. Do this independently for each named type (for example, configuration and router); a generic alias-cleanup rule is not enough. +Use these machine-readable fields on the paired rules: + +- positive: `structure=positive owner=OWNER member=FIELD member-type=TYPE` +- negative: `structure=negative owner=OWNER embedded-type=TYPE` + +`OWNER` and `TYPE` must match across the pair. Inspect the current source and +nearby codebase conventions to name the concrete replacement field. For +example, replacing `type Forwarder struct { ForwarderConfig }` should produce a +positive rule for a named `cfg ForwarderConfig` member and a negative rule that +`Forwarder` must not anonymously embed `ForwarderConfig`. A rule that merely +lists fields exposed by `ForwarderConfig`, or forbids unrelated router +embedding, does not satisfy this structural pair. If an issue, visible test, doc, source path, or user message includes literal expected values, command argv, serialized output, error text, ordered lists, or diff --git a/prompts/roles/validation-coordinator.md b/prompts/roles/validation-coordinator.md index 8632697..2f9de32 100644 --- a/prompts/roles/validation-coordinator.md +++ b/prompts/roles/validation-coordinator.md @@ -32,8 +32,8 @@ for the orchestrator. or memory, or hide the real failure behind timeout noise. - Assign or recommend a single validation lease owner for each package/path, command family, and resource boundary. -- Recommend whether the orchestrator should wait, poll, kill/finalize a stale - pane, or route a follow-up worker. +- Recommend whether the orchestrator should wait for lifecycle settlement, + cancel genuinely stuck work, or route a follow-up worker. ## Output @@ -43,11 +43,11 @@ Report compactly to the orchestrator: when known. 2. `overlaps:` duplicate or risky validators, including why they conflict. 3. `validation-leases:` package/path, command, owner, state, and resource risk. -4. `stale-agents:` panes that should be captured and finalized or killed before - replacement work is spawned. +4. `stale-agents:` work that needs cancellation or a semantic replacement + decision before replacement is spawned. 5. `released-leases:` completed or stale leases safe to replace. -6. `routing:` exact next orchestrator action: wait, poll, kill/finalize, release - a lease, spawn a verifier, or spawn a bounded follow-up worker. +6. `routing:` exact next semantic action: wait for supervisor settlement, + cancel stuck work, spawn a verifier, or spawn a bounded follow-up worker. Keep the report short enough for the orchestrator to paste into a worker or verifier instruction when needed. diff --git a/prompts/verifier.md b/prompts/verifier.md index c3e43c1..23cd18a 100644 --- a/prompts/verifier.md +++ b/prompts/verifier.md @@ -102,6 +102,14 @@ environment dependency, report that exact command and dependency as unresolved risk; do not silently replace it with `node --check`, `git diff --check`, or an equally weak proxy. +When that blocked direct test is visible in the checkout, read its exact setup, +inputs, boundary values, and assertion. Reproduce those same values and +assertion through the narrowest runnable stubbed or source-level probe; a weaker +hand-picked happy path is not equivalent. Do not accept unless the exact replay +passes or source-visible task evidence proves the assertion intentionally +changed. Report a passing replay as `visible-test-replay-passed:` with the test +name, exact boundary values, command, and return code. + Report a compact verifier contract ledger: - intended outcome @@ -470,9 +478,10 @@ own line for `ACCEPTED`, or Prefer the exact line marker above. If structured JSON is also emitted, use `final_diff_sha256`, `compile_clean: true`, and a non-empty `commands` array whose every entry records `rc: 0`; the gate treats that as equivalent evidence. -The orchestrator must poll or finalize every launched verifier before the final -gate; any verifier still marked `running`, `starting`, or `pending` blocks -submission even when an earlier verifier accepted. +The lifecycle supervisor monitors and finalizes every launched verifier. The +orchestrator must wait for that recorded terminal state before the final gate; +any verifier still marked `running`, `starting`, or `pending` blocks submission +even when an earlier verifier accepted. ## Miss Taxonomy @@ -493,3 +502,7 @@ start/build the real router and make a request-level probe. Loading the edited module, checking syntax, or invoking a handler through a hand-written stub does not prove that production registration order, mount point, middleware, or URL reachability works. Treat stub-only validation as a blocking validation gap. +For a route/router diff, acceptance must include +`route-integration-probe-passed: final-diff-sha256=HASH command=... returncode=0` +for the exact live diff. The supervisor rejects route changes without this +marker; never emit it for a stub-only handler or isolated registration probe. diff --git a/prompts/worker.md b/prompts/worker.md index 344f4bf..99efa4b 100644 --- a/prompts/worker.md +++ b/prompts/worker.md @@ -159,6 +159,12 @@ implicated source paths, and the next bounded repair assignment. Source review, compile-only checks, or a weaker synthetic probe cannot clear a still-failing nearby visible command. +If a direct visible test is blocked by a missing service or other environment +dependency, read the test and replay its exact setup, boundary values, and +assertion with the narrowest runnable stubbed probe. Do not substitute easier +values. Report `visible-test-replay-passed:` only when that exact assertion +passes; otherwise return `validation-repair-needed:`. + For any code diff, final validation must include hash-bound build evidence for the final patch: `build-verification-passed: final-diff-sha256=... changed-files=N @@ -339,6 +345,9 @@ start/build the real router and issue a request-level probe. Syntax checks, module loading, and hand-written handler stubs are useful diagnostics but are not completion evidence because they do not prove registration order, mount point, middleware, or URL reachability. +For a route/router diff, report successful production-entrypoint validation as +`route-integration-probe-passed: final-diff-sha256=HASH command=... returncode=0`. +Do not emit this marker for a stub-only handler or isolated registration probe. For option/argument propagation across wrappers, validation must observe the next layer receiving the value for both the declared default and one override. An implementation that omits the default keyword/field and relies on the next diff --git a/src/authority.rs b/src/authority.rs new file mode 100644 index 0000000..2e2bf3d --- /dev/null +++ b/src/authority.rs @@ -0,0 +1,273 @@ +use crate::config; +use serde::{Deserialize, Serialize}; + +/// The complete mutation surface accepted by the authority supervisor. +/// +/// CLI parsing happens before a request crosses the Unix socket. The server +/// authorizes this enum instead of independently interpreting command strings, +/// so routing and role policy cannot drift apart. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct AuthorityRequest { + operation: AuthorityOperation, + args: Vec, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "kebab-case")] +enum AuthorityOperation { + Workflow, + Decision, + Dag, + OrchestratorComplete, + SupervisorRegisterLaunch, + SupervisorRenewLaunch, + SupervisorSettleAgent, + SupervisorShutdown, + AssignmentCreate, + AssignmentShow, + AssignmentStatus, + AssignmentCheck, + CheckpointUpdate, + CheckpointShow, + FindingCreate, + FindingShow, + FindingList, + FindingDismiss, + TodoCreate, + TodoShow, + TodoList, + TodoAssign, + TodoStatus, + ResolutionCreate, + TodoClose, + ValidationLeaseAcquire, + ValidationLeaseStatus, + ValidationLeaseShow, + ValidationLeaseList, + GateCheck, +} + +impl AuthorityRequest { + pub fn from_cli(command: &str, args: &[String]) -> Option { + let (operation, forwarded) = match command { + "workflow" => (AuthorityOperation::Workflow, args), + "decision" => (AuthorityOperation::Decision, args), + "dag" => (AuthorityOperation::Dag, args), + "orchestrator" if args == ["complete"] => { + (AuthorityOperation::OrchestratorComplete, &args[1..]) + } + "supervisor" => match args.first().map(String::as_str) { + Some("register-launch") => { + (AuthorityOperation::SupervisorRegisterLaunch, &args[1..]) + } + Some("renew-launch") => (AuthorityOperation::SupervisorRenewLaunch, &args[1..]), + Some("settle-agent") => (AuthorityOperation::SupervisorSettleAgent, &args[1..]), + Some("stop") => (AuthorityOperation::SupervisorShutdown, &args[1..]), + _ => return None, + }, + "subagent" => { + let operation = match args.first().map(String::as_str) { + Some("assignment-create") => AuthorityOperation::AssignmentCreate, + Some("assignment-show") => AuthorityOperation::AssignmentShow, + Some("assignment-status") => AuthorityOperation::AssignmentStatus, + Some("assignment-check") => AuthorityOperation::AssignmentCheck, + Some("checkpoint-update") => AuthorityOperation::CheckpointUpdate, + Some("checkpoint-show") => AuthorityOperation::CheckpointShow, + Some("finding-create") => AuthorityOperation::FindingCreate, + Some("finding-show") => AuthorityOperation::FindingShow, + Some("finding-list") => AuthorityOperation::FindingList, + Some("finding-dismiss") => AuthorityOperation::FindingDismiss, + Some("todo-create") => AuthorityOperation::TodoCreate, + Some("todo-show") => AuthorityOperation::TodoShow, + Some("todo-list") => AuthorityOperation::TodoList, + Some("todo-assign") => AuthorityOperation::TodoAssign, + Some("todo-status") => AuthorityOperation::TodoStatus, + Some("resolution-create") => AuthorityOperation::ResolutionCreate, + Some("todo-close") => AuthorityOperation::TodoClose, + Some("validation-lease-acquire") => AuthorityOperation::ValidationLeaseAcquire, + Some("validation-lease-status") => AuthorityOperation::ValidationLeaseStatus, + Some("validation-lease-show") => AuthorityOperation::ValidationLeaseShow, + Some("validation-lease-list") => AuthorityOperation::ValidationLeaseList, + Some("gate-check") => AuthorityOperation::GateCheck, + _ => return None, + }; + (operation, &args[1..]) + } + _ => return None, + }; + Some(Self { + operation, + args: forwarded.to_vec(), + }) + } + + pub fn shutdown() -> Self { + Self { + operation: AuthorityOperation::SupervisorShutdown, + args: Vec::new(), + } + } + + #[cfg_attr(not(target_os = "linux"), allow(dead_code))] + pub fn is_shutdown(&self) -> bool { + self.operation == AuthorityOperation::SupervisorShutdown + } + + #[cfg_attr(not(target_os = "linux"), allow(dead_code))] + pub fn authorized_for(&self, uid: u32) -> bool { + if uid == 0 { + return true; + } + match self.operation { + AuthorityOperation::Workflow + | AuthorityOperation::Decision + | AuthorityOperation::Dag + | AuthorityOperation::OrchestratorComplete + | AuthorityOperation::SupervisorRegisterLaunch + | AuthorityOperation::SupervisorRenewLaunch + | AuthorityOperation::SupervisorSettleAgent + | AuthorityOperation::SupervisorShutdown + | AuthorityOperation::AssignmentCreate + | AuthorityOperation::AssignmentShow + | AuthorityOperation::AssignmentStatus + | AuthorityOperation::AssignmentCheck + | AuthorityOperation::TodoCreate + | AuthorityOperation::TodoAssign + | AuthorityOperation::TodoStatus + | AuthorityOperation::GateCheck => uid == config::ORCHESTRATOR_UID, + AuthorityOperation::FindingCreate => uid == config::READER_UID, + AuthorityOperation::FindingDismiss | AuthorityOperation::TodoClose => { + matches!(uid, config::ORCHESTRATOR_UID | config::READER_UID) + } + AuthorityOperation::ResolutionCreate => uid == config::WRITER_UID, + AuthorityOperation::CheckpointUpdate | AuthorityOperation::CheckpointShow => matches!( + uid, + config::ORCHESTRATOR_UID | config::WRITER_UID | config::READER_UID + ), + AuthorityOperation::FindingShow + | AuthorityOperation::FindingList + | AuthorityOperation::TodoShow + | AuthorityOperation::TodoList + | AuthorityOperation::ValidationLeaseShow + | AuthorityOperation::ValidationLeaseList => matches!( + uid, + config::ORCHESTRATOR_UID | config::WRITER_UID | config::READER_UID + ), + AuthorityOperation::ValidationLeaseAcquire + | AuthorityOperation::ValidationLeaseStatus => { + matches!(uid, config::WRITER_UID | config::READER_UID) + } + } + } + + pub fn into_cli(self) -> (String, Vec) { + let (command, subcommand) = match self.operation { + AuthorityOperation::Workflow => ("workflow", None), + AuthorityOperation::Decision => ("decision", None), + AuthorityOperation::Dag => ("dag", None), + AuthorityOperation::OrchestratorComplete => ("orchestrator", Some("complete")), + AuthorityOperation::SupervisorRegisterLaunch => ("supervisor", Some("register-launch")), + AuthorityOperation::SupervisorRenewLaunch => ("supervisor", Some("renew-launch")), + AuthorityOperation::SupervisorSettleAgent => ("supervisor", Some("settle-agent")), + AuthorityOperation::SupervisorShutdown => ("supervisor", Some("shutdown")), + AuthorityOperation::AssignmentCreate => ("subagent", Some("assignment-create")), + AuthorityOperation::AssignmentShow => ("subagent", Some("assignment-show")), + AuthorityOperation::AssignmentStatus => ("subagent", Some("assignment-status")), + AuthorityOperation::AssignmentCheck => ("subagent", Some("assignment-check")), + AuthorityOperation::CheckpointUpdate => ("subagent", Some("checkpoint-update")), + AuthorityOperation::CheckpointShow => ("subagent", Some("checkpoint-show")), + AuthorityOperation::FindingCreate => ("subagent", Some("finding-create")), + AuthorityOperation::FindingShow => ("subagent", Some("finding-show")), + AuthorityOperation::FindingList => ("subagent", Some("finding-list")), + AuthorityOperation::FindingDismiss => ("subagent", Some("finding-dismiss")), + AuthorityOperation::TodoCreate => ("subagent", Some("todo-create")), + AuthorityOperation::TodoShow => ("subagent", Some("todo-show")), + AuthorityOperation::TodoList => ("subagent", Some("todo-list")), + AuthorityOperation::TodoAssign => ("subagent", Some("todo-assign")), + AuthorityOperation::TodoStatus => ("subagent", Some("todo-status")), + AuthorityOperation::ResolutionCreate => ("subagent", Some("resolution-create")), + AuthorityOperation::TodoClose => ("subagent", Some("todo-close")), + AuthorityOperation::ValidationLeaseAcquire => { + ("subagent", Some("validation-lease-acquire")) + } + AuthorityOperation::ValidationLeaseStatus => { + ("subagent", Some("validation-lease-status")) + } + AuthorityOperation::ValidationLeaseShow => ("subagent", Some("validation-lease-show")), + AuthorityOperation::ValidationLeaseList => ("subagent", Some("validation-lease-list")), + AuthorityOperation::GateCheck => ("subagent", Some("gate-check")), + }; + let mut args = self.args; + if let Some(subcommand) = subcommand { + args.insert(0, subcommand.to_string()); + } + (command.to_string(), args) + } + + #[cfg_attr(not(target_os = "linux"), allow(dead_code))] + pub fn display(&self) -> String { + let (command, args) = self.clone().into_cli(); + match args.first() { + Some(subcommand) => format!("{command} {subcommand}"), + None => command, + } + } +} + +#[cfg(test)] +mod tests { + use super::AuthorityRequest; + use crate::config; + + fn strings(values: &[&str]) -> Vec { + values.iter().map(|value| (*value).to_string()).collect() + } + + #[test] + fn typed_api_excludes_runtime_and_arbitrary_execution() { + assert!(AuthorityRequest::from_cli("workflow", &strings(&["status"])).is_some()); + assert!(AuthorityRequest::from_cli("subagent", &strings(&["assignment-create"])).is_some()); + assert!(AuthorityRequest::from_cli("agent", &strings(&["run"])).is_none()); + assert!(AuthorityRequest::from_cli("role-exec", &[]).is_none()); + assert!(AuthorityRequest::from_cli("subagent", &strings(&["spawn"])).is_none()); + assert!(AuthorityRequest::from_cli("supervisor", &strings(&["reconcile"])).is_none()); + assert!(AuthorityRequest::from_cli("subagent", &strings(&["worktree-create"])).is_none()); + assert!(AuthorityRequest::from_cli("subagent", &strings(&["validation-run"])).is_none()); + } + + #[test] + fn authority_mutations_are_role_typed() { + let workflow = AuthorityRequest::from_cli("workflow", &strings(&["transition"])) + .expect("workflow request"); + let finding = AuthorityRequest::from_cli("subagent", &strings(&["finding-create"])) + .expect("finding request"); + let close = AuthorityRequest::from_cli("subagent", &strings(&["todo-close"])) + .expect("close request"); + let settle = AuthorityRequest::from_cli( + "supervisor", + &strings(&["settle-agent", "worker-01", "done"]), + ) + .expect("settle request"); + assert!(workflow.authorized_for(config::ORCHESTRATOR_UID)); + assert!(!finding.authorized_for(config::ORCHESTRATOR_UID)); + assert!(finding.authorized_for(config::READER_UID)); + assert!(close.authorized_for(config::ORCHESTRATOR_UID)); + assert!(settle.authorized_for(config::ORCHESTRATOR_UID)); + assert!(!settle.authorized_for(config::WRITER_UID)); + assert!(!workflow.authorized_for(config::WRITER_UID)); + } + + #[test] + fn request_round_trips_to_the_legacy_cli_contract() { + let original = strings(&[ + "todo-create", + "todo-1", + "--task", + "repair the implementation", + ]); + let request = AuthorityRequest::from_cli("subagent", &original).expect("typed request"); + let (command, args) = request.into_cli(); + assert_eq!(command, "subagent"); + assert_eq!(args, original); + } +} diff --git a/src/dag.rs b/src/dag.rs index 4a67d65..826835a 100644 --- a/src/dag.rs +++ b/src/dag.rs @@ -1,5 +1,4 @@ -use crate::config; -use chrono::{SecondsFormat, Utc}; +use crate::{config, state::atomic_write, state::read_env, state::timestamp}; use fs2::FileExt; use std::collections::{BTreeMap, BTreeSet}; use std::fs::{self, File, OpenOptions}; @@ -7,24 +6,14 @@ use std::io::Write; use std::path::{Path, PathBuf}; const NODE_HEADER: &str = - "node_id\tagent\tassignment_id\trole\tbranch\towned_paths\tstatus\tdecision_id\tplan_id\tadded_at"; + "node_id\tagent\tassignment_id\tresponsibility\tbranch\towned_paths\tstatus\tdecision_id\tplan_id\tadded_at"; const EDGE_HEADER: &str = "from_node\tto_node\tadded_at"; const STATUSES: &[&str] = &[ "pending", "ready", "running", "blocked", "done", "failed", "skipped", ]; -const ROLES: &[&str] = &[ - "exploitation", - "exploration", - "reflection", - "architecture", - "qa", - "verifier", - "scout", -]; - const USAGE: &str = r#"Usage: multiagent dag init WORKFLOW_ID --title TEXT [--owner NAME] - multiagent dag add-node WORKFLOW_ID NODE_ID --agent NAME --assignment-id ID --role ROLE --branch BRANCH --owned PATH[,PATH...] [--depends-on NODE[,NODE...]] [--status STATUS] [--decision-id ID] [--plan-id ID] + multiagent dag add-node WORKFLOW_ID NODE_ID --agent NAME --assignment-id ID --responsibility TEXT --branch BRANCH --owned PATH[,PATH...] [--depends-on NODE[,NODE...]] [--status STATUS] [--decision-id ID] [--plan-id ID] multiagent dag status WORKFLOW_ID NODE_ID STATUS [--reason TEXT] multiagent dag ready WORKFLOW_ID multiagent dag blocked WORKFLOW_ID @@ -109,7 +98,7 @@ struct Node { node_id: String, agent: String, assignment_id: String, - role: String, + responsibility: String, branch: String, owned_paths: String, status: String, @@ -126,7 +115,7 @@ impl Node { node_id: fields[0].to_string(), agent: fields[1].to_string(), assignment_id: fields[2].to_string(), - role: fields[3].to_string(), + responsibility: fields[3].to_string(), branch: fields[4].to_string(), owned_paths: fields[5].to_string(), status: fields[6].to_string(), @@ -141,7 +130,7 @@ impl Node { self.node_id.as_str(), self.agent.as_str(), self.assignment_id.as_str(), - self.role.as_str(), + self.responsibility.as_str(), self.branch.as_str(), self.owned_paths.as_str(), self.status.as_str(), @@ -223,6 +212,7 @@ fn add_node(args: &[String]) -> Result<(), String> { &[ "agent", "assignment-id", + "responsibility", "role", "branch", "owned", @@ -238,19 +228,27 @@ fn add_node(args: &[String]) -> Result<(), String> { "assignment-id", "add-node requires --assignment-id", )?; - let role = required(&options, "role", "add-node requires --role")?; + let responsibility = match ( + options.get("responsibility").map(String::as_str), + options.get("role").map(String::as_str), + ) { + (Some(_), Some(_)) => { + return Err("add-node accepts only one of --responsibility or --role".into()) + } + (Some(value), None) | (None, Some(value)) if !value.is_empty() => value, + _ => return Err("add-node requires --responsibility".into()), + }; let branch = required(&options, "branch", "add-node requires --branch")?; let owned = required(&options, "owned", "add-node requires --owned")?; let status = options .get("status") .map(String::as_str) .unwrap_or("pending"); - validate_role(role)?; validate_status(status)?; for (label, current) in [ ("--agent", agent), ("--assignment-id", assignment_id), - ("--role", role), + ("--responsibility", responsibility), ("--branch", branch), ("--owned", owned), ("--depends-on", value(&options, "depends-on")), @@ -294,7 +292,7 @@ fn add_node(args: &[String]) -> Result<(), String> { node_id: node_id.clone(), agent: agent.to_string(), assignment_id: assignment_id.to_string(), - role: role.to_string(), + responsibility: responsibility.to_string(), branch: branch.to_string(), owned_paths: owned.to_string(), status: status.to_string(), @@ -522,19 +520,9 @@ fn validate_status(status: &str) -> Result<(), String> { } } -fn validate_role(role: &str) -> Result<(), String> { - if ROLES.contains(&role) { - Ok(()) - } else { - Err(format!( - "invalid role: {role} (expected exploitation|exploration|reflection|architecture|qa|verifier|scout)" - )) - } -} - fn reject_newline(label: &str, current: &str) -> Result<(), String> { - if current.contains('\n') || current.contains('\r') { - Err(format!("{label} may not contain newlines")) + if current.contains(['\n', '\r', '\t']) { + Err(format!("{label} may not contain tabs or newlines")) } else { Ok(()) } @@ -568,15 +556,6 @@ fn write_edges(path: &Path, edges: &[Edge]) -> Result<(), String> { atomic_write(path, &text) } -fn read_env(path: &Path) -> Result, String> { - let text = fs::read_to_string(path).map_err(io_error("read workflow metadata"))?; - Ok(text - .lines() - .filter_map(|line| line.split_once('=')) - .map(|(key, current)| (key.to_string(), current.to_string())) - .collect()) -} - fn has_cycle(edges: &[Edge]) -> bool { let mut adjacency: BTreeMap<&str, Vec<&str>> = BTreeMap::new(); let mut nodes = BTreeSet::new(); @@ -625,30 +604,6 @@ fn print_section(label: &str, path: &Path, header_only_empty: bool) -> Result<() Ok(()) } -fn atomic_write(path: &Path, text: &str) -> Result<(), String> { - let parent = path - .parent() - .ok_or_else(|| format!("path has no parent: {}", path.display()))?; - fs::create_dir_all(parent).map_err(io_error("create workflow state directory"))?; - let temporary = parent.join(format!( - ".{}.{}.tmp", - path.file_name() - .and_then(|name| name.to_str()) - .unwrap_or("state"), - std::process::id() - )); - let mut file = File::create(&temporary).map_err(io_error("create workflow temporary state"))?; - file.write_all(text.as_bytes()) - .map_err(io_error("write workflow temporary state"))?; - file.sync_all() - .map_err(io_error("sync workflow temporary state"))?; - fs::rename(&temporary, path).map_err(io_error("replace workflow state")) -} - -fn timestamp() -> String { - Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) -} - fn io_error(context: &'static str) -> impl FnOnce(std::io::Error) -> String { move |error| format!("{context}: {error}") } diff --git a/src/decision.rs b/src/decision.rs index 5c9831a..80a071c 100644 --- a/src/decision.rs +++ b/src/decision.rs @@ -1,5 +1,4 @@ -use crate::config; -use chrono::{SecondsFormat, Utc}; +use crate::{config, state::atomic_write, state::read_env, state::timestamp}; use fs2::FileExt; use std::collections::BTreeMap; use std::fs::{self, File, OpenOptions}; @@ -555,15 +554,6 @@ fn require_status( Ok(()) } -fn read_env(path: &Path) -> Result, String> { - let text = fs::read_to_string(path).map_err(io_error("read decision state"))?; - Ok(text - .lines() - .filter_map(|line| line.split_once('=')) - .map(|(key, current)| (key.to_string(), current.to_string())) - .collect()) -} - fn tsv_first_column_contains(path: &Path, expected: &str) -> Result { let text = fs::read_to_string(path).map_err(io_error("read decision table"))?; Ok(text @@ -597,25 +587,6 @@ fn rewrite_status(path: &Path, appended: &[&str]) -> Result<(), String> { atomic_write(path, &output) } -fn atomic_write(path: &Path, text: &str) -> Result<(), String> { - let parent = path - .parent() - .ok_or_else(|| format!("path has no parent: {}", path.display()))?; - fs::create_dir_all(parent).map_err(io_error("create state directory"))?; - let temporary = parent.join(format!( - ".{}.{}.tmp", - path.file_name() - .and_then(|name| name.to_str()) - .unwrap_or("state"), - std::process::id() - )); - let mut file = File::create(&temporary).map_err(io_error("create temporary state"))?; - file.write_all(text.as_bytes()) - .map_err(io_error("write temporary state"))?; - file.sync_all().map_err(io_error("sync temporary state"))?; - fs::rename(&temporary, path).map_err(io_error("replace state")) -} - fn print_section(label: &str, path: &Path, header_only_empty: bool) -> Result<(), String> { println!("\n{label}:"); let text = fs::read_to_string(path).unwrap_or_default(); @@ -628,10 +599,6 @@ fn print_section(label: &str, path: &Path, header_only_empty: bool) -> Result<() Ok(()) } -fn timestamp() -> String { - Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) -} - fn io_error(context: &'static str) -> impl FnOnce(std::io::Error) -> String { move |error| format!("{context}: {error}") } diff --git a/src/main.rs b/src/main.rs index 4cd3ae6..b93374a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,5 @@ mod agent; +mod authority; mod config; mod dag; mod decision; @@ -7,6 +8,7 @@ mod prompt_bundle; mod role_sandbox; mod runtime; mod snapshot; +mod state; mod subagent; mod supervisor; mod workflow; @@ -75,6 +77,9 @@ fn main() -> ExitCode { .map(|_| ExitCode::SUCCESS) .map_err(|message| ("snapshot", message)), "subagent" => subagent::run(&args).map_err(|message| ("subagent", message)), + "supervisor" if args.first().map(String::as_str) == Some("reconcile") => { + runtime::supervisor_reconcile(&args[1..]).map_err(|message| ("supervisor", message)) + } "supervisor" => supervisor::run(&args).map_err(|message| ("supervisor", message)), "workflow" => workflow::run(&args) .map(|_| ExitCode::SUCCESS) diff --git a/src/policy.rs b/src/policy.rs index 048aee4..6adbb04 100644 --- a/src/policy.rs +++ b/src/policy.rs @@ -1,5 +1,4 @@ -use crate::config; -use chrono::{SecondsFormat, Utc}; +use crate::{config, state::atomic_write, state::timestamp}; use fs2::FileExt; use std::env; use std::fs::{self, File, OpenOptions}; @@ -362,29 +361,6 @@ fn reject_record_field(label: &str, current: &str) -> Result<(), String> { } } -fn atomic_write(path: &Path, text: &str) -> Result<(), String> { - let parent = path - .parent() - .ok_or_else(|| format!("path has no parent: {}", path.display()))?; - fs::create_dir_all(parent).map_err(io_error("create policy directory"))?; - let temporary = parent.join(format!( - ".{}.{}.tmp", - path.file_name() - .and_then(|name| name.to_str()) - .unwrap_or("policy"), - std::process::id() - )); - let mut file = File::create(&temporary).map_err(io_error("create temporary policy"))?; - file.write_all(text.as_bytes()) - .map_err(io_error("write temporary policy"))?; - file.sync_all().map_err(io_error("sync temporary policy"))?; - fs::rename(&temporary, path).map_err(io_error("replace policy")) -} - -fn timestamp() -> String { - Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) -} - fn io_error(context: &'static str) -> impl FnOnce(std::io::Error) -> String { move |error| format!("{context}: {error}") } diff --git a/src/prompt_bundle.rs b/src/prompt_bundle.rs index 20269a4..93d0079 100644 --- a/src/prompt_bundle.rs +++ b/src/prompt_bundle.rs @@ -1,6 +1,6 @@ +use crate::state::atomic_write; use std::collections::BTreeMap; -use std::fs::{self, File}; -use std::io::Write; +use std::fs; use std::path::Path; const USAGE:&str="Usage:\n multiagent prompt-bundle --orchestrator PATH --lifecycle PATH --output PATH\n\nBuilds the canonical initial orchestrator prompt from the role prompt and the\nmandatory implementation lifecycle playbook."; @@ -54,18 +54,6 @@ fn parse_options(args: &[String]) -> Result, String> { } Ok(values) } -fn atomic_write(path: &Path, text: &str) -> Result<(), String> { - let parent = path - .parent() - .ok_or_else(|| format!("output path has no parent: {}", path.display()))?; - fs::create_dir_all(parent).map_err(io_error("create output directory"))?; - let temporary = path.with_file_name(format!(".orchestrator-prompt.{}.tmp", std::process::id())); - let mut file = File::create(&temporary).map_err(io_error("create prompt bundle"))?; - file.write_all(text.as_bytes()) - .map_err(io_error("write prompt bundle"))?; - file.sync_all().map_err(io_error("sync prompt bundle"))?; - fs::rename(temporary, path).map_err(io_error("publish prompt bundle")) -} fn io_error(action: &'static str) -> impl Fn(std::io::Error) -> String { move |error| format!("{action}: {error}") } diff --git a/src/role_sandbox.rs b/src/role_sandbox.rs index 5658997..f9d10bb 100644 --- a/src/role_sandbox.rs +++ b/src/role_sandbox.rs @@ -192,6 +192,20 @@ fn landlock_unavailable(error: &str) -> bool { || error.contains("Protocol not supported")) } +/// Return whether this kernel can enforce a distinct write allowlist for each +/// writer process. Callers may use this capability to admit disjoint writers +/// concurrently; the filesystem ownership fallback is safe only for one +/// writer at a time. +#[cfg(target_os = "linux")] +pub fn supports_scoped_writers() -> bool { + linux::abi_version().is_some() +} + +#[cfg(not(target_os = "linux"))] +pub fn supports_scoped_writers() -> bool { + false +} + #[cfg(unix)] extern "C" fn terminate_supervised_child(_signal: libc::c_int) { let child = SUPERVISED_CHILD.load(Ordering::SeqCst); @@ -313,6 +327,18 @@ mod linux { const ACCESS_TRUNCATE: u64 = 1 << 14; const ACCESS_IOCTL_DEV: u64 = 1 << 15; + pub fn abi_version() -> Option { + let abi = unsafe { + libc::syscall( + libc::SYS_landlock_create_ruleset, + std::ptr::null::(), + 0, + LANDLOCK_CREATE_RULESET_VERSION, + ) + }; + (abi >= 1).then_some(abi) + } + #[repr(C)] struct RulesetAttr { handled_access_fs: u64, @@ -325,20 +351,12 @@ mod linux { } pub fn restrict_writes(write_roots: &[PathBuf]) -> Result<(), String> { - let abi = unsafe { - libc::syscall( - libc::SYS_landlock_create_ruleset, - std::ptr::null::(), - 0, - LANDLOCK_CREATE_RULESET_VERSION, - ) - }; - if abi < 1 { + let Some(abi) = abi_version() else { return Err(format!( "Landlock is unavailable; refusing to run without role write enforcement: {}", io::Error::last_os_error() )); - } + }; let handled_access = handled_access_for_abi(abi); let ruleset_attr = RulesetAttr { diff --git a/src/runtime.rs b/src/runtime.rs index 7d61c6a..9b9c153 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,13 +1,14 @@ use crate::{ agent::{self, AgentRequest, BackendId, BackendPaths, InvocationMode, RoleAccess}, - config, policy, role_sandbox, supervisor, + config, policy, role_sandbox, + state::{atomic_write as write_state, read_env, timestamp}, + supervisor, }; -use chrono::{Local, SecondsFormat, Utc}; -use fs2::FileExt; +use chrono::{Local, Utc}; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet}; use std::env; -use std::fs::{self, File, OpenOptions}; +use std::fs::{self, OpenOptions}; use std::io::Write; use std::path::{Path, PathBuf}; use std::process::{Command, ExitCode, Output}; @@ -108,22 +109,17 @@ pub fn role_agent_exec(args: &[String]) -> Result { let cfg = RuntimeConfig::load()?; supervisor::validate_runtime_state(&cfg.state)?; let dir = cfg.state.join("subagents").join(name); - let writer_lock = if supervisor::launch_requires_writer(&cfg.state, name)? { - let lock_path = cfg.state.join("launch-authorizations/.writer.lock"); - let lock = OpenOptions::new() - .create(true) - .truncate(false) - .read(true) - .write(true) - .open(&lock_path) - .map_err(io_error("open secure writer lock"))?; - lock.try_lock_exclusive() - .map_err(|_| "another workspace writer is already active".to_string())?; - Some(lock) + let requires_writer = supervisor::launch_requires_writer(&cfg.state, name)?; + let authorization = supervisor::claim_launch(&cfg.state, name)?; + let writer_lease = if requires_writer { + Some(supervisor::acquire_writer_lease( + &cfg.state, + name, + &authorization.owned_paths, + )?) } else { None }; - let authorization = supervisor::claim_launch(&cfg.state, name)?; let cli = &authorization.cli; validate_cli(cli)?; if !cfg.headless(cli) { @@ -174,9 +170,16 @@ pub fn role_agent_exec(args: &[String]) -> Result { } else { READER_UID }; - if access == CodexAccess::WorkspaceWrite { - prepare_workspace_write_boundary(&cfg.state, &cfg.root, &authorization.owned_paths)?; - } + let mut write_boundary = if access == CodexAccess::WorkspaceWrite { + Some(WorkspaceWriteBoundary::acquire( + &cfg.state, + &cfg.root, + &authorization.owned_paths, + writer_lease.as_ref().is_some_and(|lease| lease.scoped()), + )?) + } else { + None + }; let output = supervisor::prepare_private_output(&cfg.state, name, role_uid)?; let runner_args = build_agent_runner_args( cli, @@ -204,11 +207,10 @@ pub fn role_agent_exec(args: &[String]) -> Result { &runner_args, ); let _ = fs::remove_file(supervisor_pid); - let revoked = if access == CodexAccess::WorkspaceWrite { - revoke_workspace_writes(&cfg.state, &cfg.root, &authorization.owned_paths) - } else { - Ok(()) - }; + let revoked = write_boundary + .as_mut() + .map(WorkspaceWriteBoundary::close) + .unwrap_or(Ok(())); let sealed = supervisor::seal_role_output( &cfg.state, name, @@ -218,9 +220,9 @@ pub fn role_agent_exec(args: &[String]) -> Result { &public_output, ); supervisor::finish_launch(&cfg.state, name)?; - drop(writer_lock); revoked?; sealed?; + drop(writer_lease); result } @@ -514,6 +516,10 @@ pub fn launch(args: &[String]) -> Result { resume, )?; if env::var("MULTIAGENT_UID_SANDBOX").as_deref() == Ok("1") { + let ready = state_dir.join("runtime_state/tmux-access-ready"); + if ready.is_file() { + fs::remove_file(&ready).map_err(io_error("remove stale tmux access marker"))?; + } supervisor::register_runtime_state(&state_dir)?; supervisor::prepare_state_permissions(&state_dir)?; if !log_dir.starts_with(&state_dir) { @@ -562,6 +568,23 @@ pub fn launch(args: &[String]) -> Result { "tmux access marker", )?; } + if !window_exists(&session, "supervisor") { + let reconcile_command = format!( + "{} supervisor reconcile --watch --interval 2", + shell_escape(&executable.display().to_string()) + ); + tmux_checked(&[ + "new-window", + "-d", + "-t", + &session, + "-n", + "supervisor", + &reconcile_command, + ])?; + pipe_log(&session, "supervisor", &log_dir)?; + } + tmux_checked(&["select-window", "-t", &format!("{session}:orchestrator")])?; println!("Started tmux session: {session}"); println!("Attach with: tmux attach -t {session}"); @@ -571,6 +594,7 @@ pub fn launch(args: &[String]) -> Result { println!("Prompt bundle: {}", prompt_bundle.display()); println!("Subagent state: {}", state_dir.display()); println!("Logs: {}", log_dir.display()); + println!("Lifecycle reconciler: supervisor window (automatic)"); println!( "Dashboard: MULTIAGENT_SESSION={} MULTIAGENT_ROOT={} {} watch", shell_escape(&session), @@ -592,7 +616,7 @@ pub fn launch(args: &[String]) -> Result { fn print_launch_usage() { println!( - "Usage: multiagent launch [--session NAME] [--root DIR] [--resume] [--attach|--no-attach]\n\nStarts a tmux multi-agent session with one orchestrator window." + "Usage: multiagent launch [--session NAME] [--root DIR] [--resume] [--attach|--no-attach]\n\nStarts a tmux multi-agent session with orchestrator and lifecycle-supervisor windows." ); } @@ -903,6 +927,251 @@ pub fn status(args: &[String]) -> Result { Ok(ExitCode::SUCCESS) } +#[derive(Clone, Debug, Eq, PartialEq)] +struct ReconcileRow { + name: String, + observed: String, + action: String, + detail: String, +} + +impl ReconcileRow { + fn line(&self) -> String { + format!( + "{}\t{}\t{}\t{}", + self.name, self.observed, self.action, self.detail + ) + } +} + +/// Reconcile process observations with protected lifecycle state. +/// +/// This loop intentionally contains no worker-topology or semantic repair +/// policy. It only performs operations whose result follows mechanically from +/// observable process state and the retry budget recorded at spawn time. +pub fn supervisor_reconcile(args: &[String]) -> Result { + let cfg = RuntimeConfig::load()?; + let mut watch = false; + let mut interval = 2.0f64; + let mut index = 0; + while index < args.len() { + match args[index].as_str() { + "--watch" => { + watch = true; + index += 1; + } + "--interval" => { + interval = required_value(args, index, "reconcile --interval")? + .parse::() + .map_err(|_| "reconcile --interval must be a positive number".to_string())?; + if !interval.is_finite() || interval <= 0.0 { + return Err("reconcile --interval must be a positive number".into()); + } + index += 2; + } + "-h" | "--help" => { + println!( + "Usage: multiagent supervisor reconcile [--watch] [--interval SECONDS]\n\nObserves coding-agent processes, finalizes terminal sessions, synchronizes assignments, marks abandoned validation leases stale, and applies only explicitly budgeted infrastructure retries." + ); + return Ok(ExitCode::SUCCESS); + } + other => return Err(format!("unknown reconcile argument: {other}")), + } + } + + require_command("tmux")?; + let mut previous = BTreeMap::::new(); + loop { + let rows = reconcile_once(&cfg)?; + if !watch { + println!("NAME\tOBSERVED\tACTION\tDETAIL"); + } + let current = rows + .iter() + .map(|row| (row.name.clone(), row.clone())) + .collect::>(); + for row in rows { + if !watch || previous.get(&row.name) != Some(&row) { + println!("{}", row.line()); + } + } + if !watch { + return Ok(ExitCode::SUCCESS); + } + previous = current; + thread::sleep(Duration::from_secs_f64(interval)); + } +} + +fn reconcile_once(cfg: &RuntimeConfig) -> Result, String> { + let mut rows = Vec::new(); + for dir in sorted_directories(&cfg.state.join("subagents"))? { + let name = file_name(&dir)?; + let open = window_exists(&cfg.session, &name); + if open { + if let Err(error) = capture_subagent(cfg, &name) { + rows.push(ReconcileRow { + name, + observed: "unknown".into(), + action: "error".into(), + detail: error.replace(['\n', '\t'], " "), + }); + continue; + } + let observed = infer_reconcile_status(cfg, &name); + set_subagent_status(cfg, &name, &observed)?; + if matches!(observed.as_str(), "done" | "failed") { + settle_terminal_process(cfg, &name, &observed)?; + rows.push(ReconcileRow { + name, + observed, + action: "finalized".into(), + detail: "terminal-process-cleaned".into(), + }); + } else { + rows.push(ReconcileRow { + name, + action: if observed == "blocked" { + "needs-decision".into() + } else { + "observed".into() + }, + observed, + detail: "window-open".into(), + }); + } + continue; + } + + // A headless tmux window may close between observations. Reconstruct + // current.txt from the final message/transcript before classifying it. + let persisted = read_trimmed(&dir.join("status")).unwrap_or_else(|| "unknown".into()); + if dir.join("finalized_at").is_file() + && matches!( + persisted.as_str(), + "done" | "failed" | "finalized" | "killed" | "cancelled" | "canceled" + ) + { + rows.push(ReconcileRow { + name, + observed: persisted, + action: "unchanged".into(), + detail: "already-settled".into(), + }); + continue; + } + let _ = capture_subagent(cfg, &name); + let inferred = infer_reconcile_status(cfg, &name); + if matches!(inferred.as_str(), "done" | "failed") { + set_subagent_status(cfg, &name, &inferred)?; + settle_terminal_process(cfg, &name, &inferred)?; + rows.push(ReconcileRow { + name, + observed: inferred, + action: "settled".into(), + detail: "durable-terminal-output".into(), + }); + continue; + } + if matches!(persisted.as_str(), "killed" | "cancelled" | "canceled") { + run_self_quiet(&["supervisor", "settle-agent", &name, "killed"])?; + atomic_write( + &dir.join("finalized_at"), + &format!("{}\n", timestamp()), + "finalized timestamp", + )?; + rows.push(ReconcileRow { + name, + observed: persisted, + action: "settled".into(), + detail: "intentional-stop".into(), + }); + continue; + } + + let recovery = classify_recovery(cfg, &name)?; + if recovery.action == "restore" && consume_infra_retry(&dir)? { + match restore(cfg, std::slice::from_ref(&name)) { + Ok(()) => rows.push(ReconcileRow { + name, + observed: persisted, + action: "restored".into(), + detail: "budgeted-infrastructure-retry".into(), + }), + Err(error) => rows.push(ReconcileRow { + name, + observed: persisted, + action: "error".into(), + detail: error.replace(['\n', '\t'], " "), + }), + } + } else { + rows.push(ReconcileRow { + name, + observed: persisted, + action: match recovery.action.as_str() { + "restore" => "needs-recovery", + "skip-blocked" => "needs-decision", + _ => "unchanged", + } + .into(), + detail: recovery.reason, + }); + } + } + Ok(rows) +} + +fn settle_terminal_process(cfg: &RuntimeConfig, name: &str, status: &str) -> Result<(), String> { + let supervisor_pid = read_supervisor_pid(cfg, name); + if window_exists(&cfg.session, name) { + let _ = capture_subagent(cfg, name); + tmux_checked(&["kill-window", "-t", &format!("{}:{name}", cfg.session)])?; + } + if let Some(pid) = supervisor_pid { + wait_for_process_exit(pid, name)?; + } + record_supervisor_termination( + cfg, + name, + if status == "done" { + "completed" + } else { + "failed" + }, + )?; + set_subagent_status(cfg, name, status)?; + atomic_write( + &cfg.state.join("subagents").join(name).join("finalized_at"), + &format!("{}\n", timestamp()), + "finalized timestamp", + )?; + run_self_quiet(&["supervisor", "settle-agent", name, status]) +} + +fn consume_infra_retry(dir: &Path) -> Result { + let path = dir.join("meta.env"); + let mut metadata = read_env(&path).unwrap_or_default(); + let budget = metadata + .get("infra_retry_budget") + .and_then(|value| value.parse::().ok()) + .unwrap_or(0); + let count = metadata + .get("infra_retry_count") + .and_then(|value| value.parse::().ok()) + .unwrap_or(0); + if count >= budget { + return Ok(false); + } + metadata.insert("infra_retry_count".into(), (count + 1).to_string()); + let body = metadata + .iter() + .map(|(key, value)| format!("{key}={value}\n")) + .collect::(); + atomic_write(&path, &body, "subagent metadata")?; + Ok(true) +} + fn status_text() -> Result { require_command("tmux")?; let cfg = RuntimeConfig::load()?; @@ -917,7 +1186,9 @@ fn status_text() -> Result { .collect::>(); let mut result = STATUS_HEADER.to_string(); for name in &window_names { - if name == "orchestrator" || cfg.state.join("subagents").join(name).is_dir() { + if matches!(name.as_str(), "orchestrator" | "supervisor") + || cfg.state.join("subagents").join(name).is_dir() + { continue; } let capture = capture_window(&cfg.session, name, 300).unwrap_or_default(); @@ -1227,7 +1498,7 @@ pub fn subagent(args: &[String]) -> Result { fn print_subagent_usage() { println!( - "Usage:\n multiagent subagent spawn NAME [--own PATH[,PATH...] ...] [--assignment-id ID] [--workflow-id ID --decision-id ID --plan-id ID] [--branch BRANCH] [--start-commit COMMIT] [--role ROLE] [--instruction TEXT | --instruction-file PATH | -- TEXT]\n multiagent subagent list|recover-plan|restore-all|gate-check\n multiagent subagent poll|inspect|restore|finalize|kill NAME [OPTIONS]\n multiagent subagent wait NAME [--timeout SECONDS] [--poll-interval SECONDS]\n\nAll durable state and tmux subprocess orchestration are implemented by the Rust CLI." + "Usage:\n multiagent subagent spawn NAME [--own PATH[,PATH...] ...] [--assignment-id ID] [--workflow-id ID --decision-id ID --plan-id ID] [--branch BRANCH] [--start-commit COMMIT] [--role ROLE] [--responsibility TEXT] [--infra-retries N] [--instruction TEXT | --instruction-file PATH | -- TEXT]\n multiagent subagent list|recover-plan|restore-all|gate-check\n multiagent subagent poll|inspect|restore|finalize|kill NAME [OPTIONS]\n multiagent subagent wait NAME [--timeout SECONDS] [--poll-interval SECONDS]\n\nAll durable state and tmux subprocess orchestration are implemented by the Rust CLI." ); } @@ -1241,6 +1512,7 @@ fn spawn(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { let mut instruction_file = None::; let mut owned = Vec::new(); let mut role = String::new(); + let mut infra_retries = 0u32; let mut assignment_values = BTreeMap::::new(); let mut index = 1; while index < args.len() { @@ -1256,8 +1528,16 @@ fn spawn(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { } index += 2; } + "--infra-retries" => { + infra_retries = required_value(args, index, "spawn --infra-retries")? + .parse::() + .map_err(|_| { + "spawn --infra-retries must be a non-negative integer".to_string() + })?; + index += 2; + } "--assignment-id" | "--workflow-id" | "--decision-id" | "--plan-id" | "--branch" - | "--start-commit" => { + | "--start-commit" | "--responsibility" => { assignment_values.insert( args[index].clone(), required_value(args, index, "spawn assignment metadata")?.to_string(), @@ -1340,7 +1620,6 @@ fn spawn(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { if window_exists(&cfg.session, name) { return Err(format!("subagent window already exists: {name}")); } - reject_parallel_generic_worker_spawn(cfg, name)?; if owned.is_empty() && !assignment_values.is_empty() { return Err("spawn assignment metadata requires --own PATH".into()); } @@ -1355,6 +1634,7 @@ fn spawn(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { ("--plan-id", "plan_id"), ("--branch", "branch"), ("--start-commit", "start_commit"), + ("--responsibility", "responsibility"), ] { if let Some(requested) = assignment_values.get(flag) { if metadata.get(key) != Some(requested) { @@ -1410,6 +1690,7 @@ fn spawn(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { "--decision-id", "--plan-id", "--start-commit", + "--responsibility", ] { if let Some(value) = assignment_values.get(flag) { command.push(flag.to_string()); @@ -1428,7 +1709,7 @@ fn spawn(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { fs::create_dir_all(&cfg.logs).map_err(io_error("create subagent log directory"))?; let executable = env::current_exe().map_err(io_error("resolve multiagent executable"))?; let metadata = format!( - "name={name}\nsession={}\nroot={}\nrole={}\naccess={}\ncodex_access={}\nworkflow_id={}\nwrite_policy={}\nlog_file={}\ntrace_dir={}\ncli={cli}\ncli_bin={binary}\nhelper={}\ncreated_at={}\n", + "name={name}\nsession={}\nroot={}\nrole={}\naccess={}\ncodex_access={}\nworkflow_id={}\nwrite_policy={}\nlog_file={}\ntrace_dir={}\ncli={cli}\ncli_bin={binary}\nhelper={}\ninfra_retry_budget={infra_retries}\ninfra_retry_count=0\ncreated_at={}\n", cfg.session, cfg.root.display(), authority_role, @@ -2400,27 +2681,6 @@ fn implementation_context(cfg: &RuntimeConfig, name: &str) -> Result Result<(), String> { - if env::var("MULTIAGENT_ALLOW_PARALLEL_WORKERS").as_deref() == Ok("1") - || !name.starts_with("worker-") - { - return Ok(()); - } - for dir in sorted_directories(&cfg.state.join("subagents"))? { - let existing = file_name(&dir)?; - if existing == name || !existing.starts_with("worker-") { - continue; - } - let status = read_trimmed(&dir.join("status")).unwrap_or_else(|| "unknown".into()); - if matches!(status.as_str(), "starting" | "running" | "restoring") - && window_exists(&cfg.session, &existing) - { - return Err(format!("active generic worker already running: existing={existing} status={status}; wait, finalize/kill it, or set MULTIAGENT_ALLOW_PARALLEL_WORKERS=1 only with explicit disjoint ownership")); - } - } - Ok(()) -} - #[allow(clippy::too_many_arguments)] fn build_cli_command( cli: &str, @@ -2578,34 +2838,78 @@ fn secure_agent_write_roots( paths.into_iter().collect() } +struct WorkspaceWriteBoundary { + state: PathBuf, + root: PathBuf, + owned_paths: Vec, + scoped: bool, + active: bool, +} + +impl WorkspaceWriteBoundary { + fn acquire( + state: &Path, + root: &Path, + owned_paths: &[PathBuf], + scoped: bool, + ) -> Result { + prepare_workspace_write_boundary(state, root, owned_paths, scoped)?; + Ok(Self { + state: state.to_path_buf(), + root: root.to_path_buf(), + owned_paths: owned_paths.to_vec(), + scoped, + active: true, + }) + } + + fn close(&mut self) -> Result<(), String> { + if !self.active { + return Ok(()); + } + revoke_workspace_writes(&self.state, &self.root, &self.owned_paths, self.scoped)?; + self.active = false; + Ok(()) + } +} + +impl Drop for WorkspaceWriteBoundary { + fn drop(&mut self) { + let _ = self.close(); + } +} + #[cfg(target_os = "linux")] fn prepare_workspace_write_boundary( state: &Path, root: &Path, owned_paths: &[PathBuf], + scoped: bool, ) -> Result<(), String> { use std::os::unix::fs::PermissionsExt; - let ledger = state.join("launch-authorizations/active-writer-paths"); - if ledger.is_file() { - for line in fs::read_to_string(&ledger) - .map_err(io_error("read prior writer ownership ledger"))? - .lines() - .filter(|line| !line.is_empty()) - { - let path = PathBuf::from(line); - if path.starts_with(root) && path != root && path.exists() { - set_workspace_tree_owner(&path, 0, false)?; + if !scoped { + let ledger = state.join("launch-authorizations/active-writer-paths"); + if ledger.is_file() { + for line in fs::read_to_string(&ledger) + .map_err(io_error("read prior writer ownership ledger"))? + .lines() + .filter(|line| !line.is_empty()) + { + let path = PathBuf::from(line); + if path.starts_with(root) && path != root && path.exists() { + set_workspace_tree_owner(&path, 0, false)?; + } } } + let text = owned_paths + .iter() + .map(|path| format!("{}\n", path.display())) + .collect::(); + atomic_write(&ledger, &text, "active writer ownership ledger")?; + fs::set_permissions(&ledger, fs::Permissions::from_mode(0o600)) + .map_err(io_error("protect writer ownership ledger"))?; } - let text = owned_paths - .iter() - .map(|path| format!("{}\n", path.display())) - .collect::(); - atomic_write(&ledger, &text, "active writer ownership ledger")?; - fs::set_permissions(&ledger, fs::Permissions::from_mode(0o600)) - .map_err(io_error("protect writer ownership ledger"))?; for path in owned_paths { if !path.starts_with(root) || path == root { return Err(format!( @@ -2623,6 +2927,7 @@ fn prepare_workspace_write_boundary( _state: &Path, _root: &Path, _owned_paths: &[PathBuf], + _scoped: bool, ) -> Result<(), String> { Err("filesystem writer ownership requires Linux".into()) } @@ -2632,6 +2937,7 @@ fn revoke_workspace_writes( state: &Path, root: &Path, owned_paths: &[PathBuf], + scoped: bool, ) -> Result<(), String> { use std::os::unix::fs::PermissionsExt; @@ -2640,10 +2946,14 @@ fn revoke_workspace_writes( set_workspace_tree_owner(path, 0, false)?; } } - let ledger = state.join("launch-authorizations/active-writer-paths"); - atomic_write(&ledger, "", "clear writer ownership ledger")?; - fs::set_permissions(&ledger, fs::Permissions::from_mode(0o600)) - .map_err(io_error("protect writer ownership ledger")) + if scoped { + Ok(()) + } else { + let ledger = state.join("launch-authorizations/active-writer-paths"); + atomic_write(&ledger, "", "clear writer ownership ledger")?; + fs::set_permissions(&ledger, fs::Permissions::from_mode(0o600)) + .map_err(io_error("protect writer ownership ledger")) + } } #[cfg(not(target_os = "linux"))] @@ -2651,6 +2961,7 @@ fn revoke_workspace_writes( _state: &Path, _root: &Path, _owned_paths: &[PathBuf], + _scoped: bool, ) -> Result<(), String> { Err("filesystem writer ownership requires Linux".into()) } @@ -3021,11 +3332,12 @@ fn capture_subagent(cfg: &RuntimeConfig, name: &str) -> Result<(), String> { fs::create_dir_all(&dir).map_err(io_error("create subagent directory"))?; match capture_window(&cfg.session, name, 1000) { Ok(capture) => { - atomic_write( - &dir.join("current.txt"), - &format!("{capture}\n"), - "current capture", - )?; + let current = format!("{capture}\n"); + if fs::read_to_string(dir.join("current.txt")).is_ok_and(|previous| previous == current) + { + return Ok(()); + } + atomic_write(&dir.join("current.txt"), ¤t, "current capture")?; append_file( &dir.join("transcript.log"), &format!("\n----- capture {} -----\n{capture}\n", timestamp()), @@ -3047,14 +3359,15 @@ fn capture_subagent(cfg: &RuntimeConfig, name: &str) -> Result<(), String> { if last.is_empty() { String::new() } else { format!("\n----- last-message.txt -----\n{last}") }, if transcript.is_empty() { String::new() } else { format!("\n----- transcript tail -----\n{}", tail_lines(&transcript, 240)) } ); - atomic_write(&dir.join("current.txt"), &recovered, "durable capture")?; - append_file( - &dir.join("transcript.log"), - &format!( - "\n----- durable capture {} -----\n{recovered}\n", - timestamp() - ), - ) + if fs::read_to_string(dir.join("current.txt")) + .is_ok_and(|previous| previous == recovered) + { + return Ok(()); + } + // The transcript already contains the durable process log. Do not + // append this synthetic reconstruction back into it: a watch loop + // would recursively duplicate the same tail on every pass. + atomic_write(&dir.join("current.txt"), &recovered, "durable capture") } } } @@ -3079,6 +3392,23 @@ fn infer_status(cfg: &RuntimeConfig, name: &str) -> String { } } +fn infer_reconcile_status(cfg: &RuntimeConfig, name: &str) -> String { + let dir = cfg.state.join("subagents").join(name); + let current = fs::read_to_string(dir.join("current.txt")).unwrap_or_default(); + let lower = current.to_ascii_lowercase(); + if nonzero_exec_status(&lower) || lower.contains("warning: no last agent message") { + "failed".into() + } else if zero_exec_status(&lower) { + "done".into() + } else if looks_blocked_report(&tail_lines(¤t, 160)) { + "blocked".into() + } else if window_exists(&cfg.session, name) { + "running".into() + } else { + "exited".into() + } +} + fn accepted_report(text: &str) -> bool { text.lines().any(|line| { let value = normalize_report_line(line).to_ascii_lowercase(); @@ -3176,6 +3506,24 @@ fn nonzero_exec_status(text: &str) -> bool { }) } +fn zero_exec_status(text: &str) -> bool { + let markers = [ + "final status: codex exec exited rc=", + "final status: coding agent exited rc=", + ]; + text.lines().any(|line| { + markers.iter().any(|marker| { + line.find(marker).is_some_and(|index| { + line[index + marker.len()..] + .split_whitespace() + .next() + .and_then(|value| value.parse::().ok()) + == Some(0) + }) + }) + }) +} + fn has_recovery_context(dir: &Path) -> bool { file_nonempty(&dir.join("current.txt")) || file_nonempty(&dir.join("transcript.log")) } @@ -3487,17 +3835,6 @@ fn csv_values(raw: &str) -> Vec { values } -fn read_env(path: &Path) -> Result, String> { - let text = fs::read_to_string(path).map_err(io_error("read environment state"))?; - let mut values = BTreeMap::new(); - for line in text.lines() { - if let Some((key, value)) = line.split_once('=') { - values.insert(key.into(), value.into()); - } - } - Ok(values) -} - fn required_env_field<'a>( values: &'a BTreeMap, key: &str, @@ -3629,22 +3966,7 @@ fn append_file(path: &Path, text: &str) -> Result<(), String> { } fn atomic_write(path: &Path, text: &str, label: &str) -> Result<(), String> { - let parent = path - .parent() - .ok_or_else(|| format!("{label} path has no parent: {}", path.display()))?; - fs::create_dir_all(parent).map_err(io_error("create output directory"))?; - let temporary = path.with_file_name(format!( - ".{}.{}.tmp", - path.file_name() - .and_then(|value| value.to_str()) - .unwrap_or("state"), - std::process::id() - )); - let mut file = File::create(&temporary).map_err(io_error("create temporary file"))?; - file.write_all(text.as_bytes()) - .map_err(io_error("write temporary file"))?; - file.sync_all().map_err(io_error("sync temporary file"))?; - fs::rename(temporary, path).map_err(io_error("publish file")) + write_state(path, text).map_err(|error| format!("{label}: {error}")) } #[cfg(unix)] @@ -3662,10 +3984,6 @@ fn set_executable(_path: &Path, _mode: u32) -> Result<(), String> { Ok(()) } -fn timestamp() -> String { - Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) -} - fn io_error(action: &'static str) -> impl Fn(std::io::Error) -> String { move |error| format!("{action}: {error}") } diff --git a/src/snapshot.rs b/src/snapshot.rs index 57aa860..740a42e 100644 --- a/src/snapshot.rs +++ b/src/snapshot.rs @@ -131,6 +131,7 @@ fn untracked_paths(root: &Path) -> Result, String> { .filter(|bytes| !bytes.is_empty()) .map(path_from_git_bytes) .filter(|path| !baseline.contains(&path.to_string_lossy().into_owned())) + .filter(|path| !is_framework_internal_path(path)) .filter(|path| { fs::symlink_metadata(root.join(path)) .map(|metadata| metadata.is_file() || metadata.file_type().is_symlink()) @@ -141,6 +142,14 @@ fn untracked_paths(root: &Path) -> Result, String> { Ok(paths) } +fn is_framework_internal_path(path: &Path) -> bool { + let mut components = path.components(); + matches!( + components.next(), + Some(std::path::Component::Normal(component)) if component == ".multiagent" + ) +} + fn baseline_untracked() -> Result, String> { let Ok(path) = std::env::var("MULTIAGENT_BASELINE_UNTRACKED_FILE") else { return Ok(BTreeSet::new()); @@ -274,4 +283,13 @@ mod tests { assert!(is_test_path("tests/lib.rs")); assert!(is_ignored("vendor/lib.rs")); } + + #[test] + fn recognizes_framework_internal_paths() { + assert!(is_framework_internal_path(Path::new( + ".multiagent/subagents/reviewer/status" + ))); + assert!(!is_framework_internal_path(Path::new("src/.multiagent.rs"))); + assert!(!is_framework_internal_path(Path::new("feature.py"))); + } } diff --git a/src/state.rs b/src/state.rs new file mode 100644 index 0000000..98979e8 --- /dev/null +++ b/src/state.rs @@ -0,0 +1,100 @@ +use chrono::{SecondsFormat, Utc}; +use std::collections::BTreeMap; +use std::fs::{self, File}; +use std::io::Write; +use std::path::Path; + +/// Publish state with one crash-safe filesystem primitive. +/// +/// Domain modules still decide their schemas and locking boundaries; this +/// module owns the shared write mechanics so those guarantees do not drift. +pub fn atomic_write(path: &Path, text: &str) -> Result<(), String> { + atomic_write_bytes(path, text.as_bytes()) +} + +pub fn atomic_write_bytes(path: &Path, bytes: &[u8]) -> Result<(), String> { + let parent = path + .parent() + .ok_or_else(|| format!("state path has no parent: {}", path.display()))?; + fs::create_dir_all(parent) + .map_err(|error| format!("create state directory {}: {error}", parent.display()))?; + let temporary = parent.join(format!( + ".{}.{}.tmp", + path.file_name() + .and_then(|name| name.to_str()) + .unwrap_or("state"), + std::process::id() + )); + let mut file = File::create(&temporary) + .map_err(|error| format!("create temporary state {}: {error}", temporary.display()))?; + file.write_all(bytes) + .map_err(|error| format!("write temporary state {}: {error}", temporary.display()))?; + file.sync_all() + .map_err(|error| format!("sync temporary state {}: {error}", temporary.display()))?; + fs::rename(&temporary, path).map_err(|error| { + format!( + "publish state {} as {}: {error}", + temporary.display(), + path.display() + ) + }) +} + +pub fn read_env(path: &Path) -> Result, String> { + let text = fs::read_to_string(path) + .map_err(|error| format!("read state {}: {error}", path.display()))?; + Ok(parse_env(&text)) +} + +pub fn read_env_optional(path: &Path) -> Result, String> { + if path.is_file() { + read_env(path) + } else { + Ok(BTreeMap::new()) + } +} + +fn parse_env(text: &str) -> BTreeMap { + text.lines() + .filter_map(|line| line.split_once('=')) + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect() +} + +pub fn timestamp() -> String { + Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) +} + +#[cfg(test)] +mod tests { + use super::{atomic_write, parse_env}; + use std::fs; + + #[test] + fn atomic_write_replaces_complete_contents() { + let root = std::env::temp_dir().join(format!( + "multiagent-state-test-{}-{}", + std::process::id(), + std::thread::current().name().unwrap_or("unnamed") + )); + let path = root.join("state.env"); + atomic_write(&path, "phase=implementation\n").expect("initial state"); + atomic_write(&path, "phase=complete\n").expect("replacement state"); + assert_eq!( + fs::read_to_string(path).expect("read state"), + "phase=complete\n" + ); + fs::remove_dir_all(root).expect("remove state test directory"); + } + + #[test] + fn env_values_preserve_additional_equals_signs() { + let values = parse_env("name=worker\ncommand=printf a=b\nignored\n"); + assert_eq!(values.get("name").map(String::as_str), Some("worker")); + assert_eq!( + values.get("command").map(String::as_str), + Some("printf a=b") + ); + assert!(!values.contains_key("ignored")); + } +} diff --git a/src/subagent.rs b/src/subagent.rs index b1fe2f8..612c5a9 100644 --- a/src/subagent.rs +++ b/src/subagent.rs @@ -1,5 +1,8 @@ -use crate::{config, runtime, workflow}; -use chrono::{SecondsFormat, Utc}; +use crate::{ + config, runtime, + state::{atomic_write, read_env, timestamp}, + workflow, +}; use fs2::FileExt; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; @@ -126,6 +129,74 @@ fn assignment_status(args: &[String]) -> Result<(), String> { Ok(()) } +/// Apply the deterministic state cleanup for a terminal coding-agent process. +/// +/// In UID-isolated runs this function is reached only through the typed +/// authority socket. The reconciler observes tmux; the authority supervisor +/// owns the protected assignment and validation state transition. +pub fn supervisor_settle_agent(args: &[String]) -> Result<(), String> { + if args.len() != 2 { + return Err("settle-agent requires NAME STATUS".into()); + } + let name = &args[0]; + validate_name(name)?; + let assignment_state = match args[1].as_str() { + "done" => "done", + "failed" | "killed" => "failed", + other => { + return Err(format!( + "settle-agent status must be done, failed, or killed: {other}" + )) + } + }; + + let state = config::state_dir()?; + let assignment = state.join("assignments").join(name); + if assignment.join("assignment.env").is_file() { + let base = state.join("assignments"); + let _lock = lock_file(&base.join(".lock"), "assignments")?; + atomic_write(&assignment.join("status"), &format!("{assignment_state}\n"))?; + } + + let leases = state.join("validation-leases"); + fs::create_dir_all(&leases).map_err(io_error("create validation leases directory"))?; + let _lock = lock_file(&leases.join(".lock"), "validation leases")?; + let mut stale = 0usize; + for dir in sorted_directories(&leases)? { + let metadata_path = dir.join("lease.env"); + if !metadata_path.is_file() { + continue; + } + let mut metadata = read_env(&metadata_path)?; + if env_value(&metadata, "owner") != name { + continue; + } + let lease_state = fs::read_to_string(dir.join("status")).unwrap_or_default(); + if !matches!(lease_state.trim(), "planned" | "running") { + continue; + } + metadata.insert("updated_at".into(), timestamp()); + write_lease_env(&metadata_path, &metadata)?; + atomic_write(&dir.join("status"), "stale\n")?; + atomic_write( + &dir.join("result.json"), + &format!( + "{}\n", + serde_json::to_string(&json!({ + "reason": "owner-terminal-without-lease-result", + "owner": name, + "owner_status": assignment_state, + })) + .map_err(json_error)? + ), + )?; + write_validation_lease_json(&dir)?; + stale += 1; + } + println!("settled agent\t{name}\t{assignment_state}\tstale_leases={stale}"); + Ok(()) +} + fn assignment_check(args: &[String]) -> Result<(), String> { let name = one_agent("assignment-check", args)?; let dir = require_assignment(name)?; @@ -243,6 +314,7 @@ fn checkpoint_update(args: &[String]) -> Result<(), String> { ("idempotency", idempotency), ("status", status), ("role", role), + ("responsibility", env_value(&metadata, "responsibility")), ("decision_id", env_value(&metadata, "decision_id")), ("plan_id", env_value(&metadata, "plan_id")), ("workflow_id", env_value(&metadata, "workflow_id")), @@ -429,6 +501,7 @@ struct AssignmentOptions { status: String, start_commit: String, role: String, + responsibility: String, decision_id: String, plan_id: String, workflow_id: String, @@ -445,6 +518,7 @@ fn assignment_create(args: &[String]) -> Result<(), String> { options.role )); } + reject_newline("--responsibility", &options.responsibility)?; let root = fs::canonicalize(config::root()?).map_err(io_error("canonicalize MULTIAGENT_ROOT"))?; @@ -527,6 +601,7 @@ fn assignment_create(args: &[String]) -> Result<(), String> { ("subagent_cli", subagent_cli.as_str()), ("verifier_cli", verifier_cli.as_str()), ("role", options.role.as_str()), + ("responsibility", options.responsibility.as_str()), ("decision_id", options.decision_id.as_str()), ("plan_id", options.plan_id.as_str()), ( @@ -902,6 +977,35 @@ fn current_final_diff_sha256() -> Result { Ok(format!("{:x}", digest.finalize())) } +fn current_diff_requires_route_probe() -> Result { + if env::var("MULTIAGENT_REQUIRE_HASH_BOUND_VERIFIER").as_deref() != Ok("1") { + return Ok(false); + } + let root = config::root()?; + if !root.is_dir() { + return Ok(false); + } + let base = env::var("MULTIAGENT_START_HEAD") + .ok() + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| "HEAD".into()); + let diff = crate::snapshot::canonical_diff(&root, &base)?; + let text = String::from_utf8_lossy(&diff); + Ok(text.lines().any(|line| { + let Some(path) = line + .strip_prefix("diff --git a/") + .and_then(|line| line.split_once(" b/").map(|(_, path)| path)) + else { + return false; + }; + path.split('/').any(|component| { + let component = component.to_ascii_lowercase(); + let stem = component.split('.').next().unwrap_or(&component); + matches!(stem, "route" | "routes" | "router" | "routers" | "routing") + }) + })) +} + fn gate_check(args: &[String]) -> Result<(), String> { if !args.is_empty() { return Err("gate-check takes no arguments".into()); @@ -909,6 +1013,7 @@ fn gate_check(args: &[String]) -> Result<(), String> { let state = config::state_dir()?; reconcile_terminal_verifiers(&state)?; let final_hash = current_final_diff_sha256()?; + let route_probe_required = current_diff_requires_route_probe()?; let mut failed = false; for (name, status) in active_verifiers(&state)? { @@ -937,6 +1042,14 @@ fn gate_check(args: &[String]) -> Result<(), String> { println!("reject\tlatest-verifier-final-diff-hash-mismatch\tverifier={name}\texpected={final_hash}\tevidence={}", evidence_path.display()); failed = true; } + if route_probe_required + && !evidence.contains(&format!( + "route-integration-probe-passed: final-diff-sha256={final_hash}" + )) + { + println!("reject\tmissing-route-integration-probe\tverifier={name}\texpected={final_hash}\tevidence={}", evidence_path.display()); + failed = true; + } } _ => {} } @@ -2482,7 +2595,8 @@ fn parse_assignment(args: &[String]) -> Result { match key.as_str() { "--owned" => owned.push(value.clone()), "--assignment-id" | "--branch" | "--status" | "--start-commit" | "--role" - | "--decision-id" | "--plan-id" | "--workflow-id" | "--node-id" | "--depends-on" => { + | "--responsibility" | "--decision-id" | "--plan-id" | "--workflow-id" + | "--node-id" | "--depends-on" => { values.insert(key.clone(), value.clone()); } _ => return Err(format!("unknown assignment-create argument: {key}")), @@ -2513,6 +2627,7 @@ fn parse_assignment(args: &[String]) -> Result { .get("--role") .cloned() .unwrap_or_else(|| "exploitation".into()), + responsibility: values.get("--responsibility").cloned().unwrap_or_default(), decision_id: values.get("--decision-id").cloned().unwrap_or_default(), plan_id: values.get("--plan-id").cloned().unwrap_or_default(), workflow_id: values.get("--workflow-id").cloned().unwrap_or_default(), @@ -2639,18 +2754,6 @@ fn resolve_commit(root: &Path, requested: &str) -> Result { Ok(String::from_utf8_lossy(&output.stdout).trim().to_string()) } -fn read_env(path: &Path) -> Result, String> { - let mut values = BTreeMap::new(); - for line in fs::read_to_string(path) - .map_err(io_error("read assignment"))? - .lines() - { - if let Some((key, value)) = line.split_once('=') { - values.insert(key.into(), value.into()); - } - } - Ok(values) -} fn repeated_options( args: &[String], repeatable: &[&str], @@ -2834,21 +2937,6 @@ fn validate_name(name: &str) -> Result<(), String> { } Ok(()) } -fn atomic_write(path: &Path, text: &str) -> Result<(), String> { - let temporary = path.with_file_name(format!( - ".{}.{}.tmp", - path.file_name().and_then(|v| v.to_str()).unwrap_or("state"), - std::process::id() - )); - let mut file = File::create(&temporary).map_err(io_error("create assignment state"))?; - file.write_all(text.as_bytes()) - .map_err(io_error("write assignment state"))?; - file.sync_all().map_err(io_error("sync assignment state"))?; - fs::rename(temporary, path).map_err(io_error("publish assignment state")) -} -fn timestamp() -> String { - Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) -} fn io_error(action: &'static str) -> impl Fn(std::io::Error) -> String { move |error| format!("{action}: {error}") } diff --git a/src/supervisor.rs b/src/supervisor.rs index 3e2cda2..a0b872f 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -1,9 +1,12 @@ -use crate::config; +use crate::{ + authority::AuthorityRequest, config, role_sandbox, state::read_env as read_env_file, subagent, +}; +use fs2::FileExt; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::collections::BTreeMap; use std::env; -use std::fs; +use std::fs::{self, File, OpenOptions}; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; use std::process::ExitCode; @@ -36,12 +39,6 @@ const CONTROL_DIRECTORIES: &[&str] = &[ "workflows", ]; -#[derive(Deserialize, Serialize)] -struct Request { - command: String, - args: Vec, -} - #[derive(Deserialize, Serialize)] struct Response { code: i32, @@ -53,10 +50,7 @@ pub fn run(args: &[String]) -> Result { match args { [command] if command == "bootstrap-test" => bootstrap_test(), [command] if command == "serve" => serve(&authority_socket(&config::state_dir()?)), - [command] if command == "stop" => proxy_request(Request { - command: "supervisor".into(), - args: vec!["shutdown".into()], - }), + [command] if command == "stop" => proxy_request(AuthorityRequest::shutdown()), [command, rest @ ..] if command == "register-launch" && server_child() => { register_launch(rest, false)?; Ok(ExitCode::SUCCESS) @@ -65,8 +59,19 @@ pub fn run(args: &[String]) -> Result { register_launch(rest, true)?; Ok(ExitCode::SUCCESS) } + [command, rest @ ..] if command == "settle-agent" && server_child() => { + subagent::supervisor_settle_agent(rest)?; + Ok(ExitCode::SUCCESS) + } + [command, rest @ ..] + if command == "settle-agent" + && env::var("MULTIAGENT_UID_SANDBOX").as_deref() != Ok("1") => + { + subagent::supervisor_settle_agent(rest)?; + Ok(ExitCode::SUCCESS) + } [command] if command == "shutdown" && server_child() => Ok(ExitCode::SUCCESS), - _ => Err("usage: multiagent supervisor stop".into()), + _ => Err("usage: multiagent supervisor reconcile|stop".into()), } } @@ -93,52 +98,10 @@ pub fn proxy_if_required(command: &str, args: &[String]) -> Option bool { - match command { - "orchestrator" => args == ["complete"], - "workflow" | "decision" | "dag" => true, - "supervisor" => args.first().is_some_and(|value| { - matches!(value.as_str(), "stop" | "register-launch" | "renew-launch") - }), - "subagent" => args.first().is_some_and(|value| { - matches!( - value.as_str(), - "assignment-create" - | "assignment-show" - | "assignment-status" - | "assignment-check" - | "checkpoint-update" - | "checkpoint-show" - | "finding-create" - | "finding-show" - | "finding-list" - | "finding-dismiss" - | "todo-create" - | "todo-show" - | "todo-list" - | "todo-assign" - | "todo-status" - | "resolution-create" - | "todo-close" - | "validation-lease-acquire" - | "validation-lease-status" - | "validation-lease-show" - | "validation-lease-list" - | "gate-check" - ) - }), - _ => false, - } + AuthorityRequest::from_cli(command, args).map(proxy_request) } #[derive(Clone, Debug)] @@ -152,6 +115,144 @@ pub struct LaunchAuthorization { pub owned_paths: Vec, } +/// A live workspace-writer lease. +/// +/// Kernels with Landlock receive path-scoped leases, so any number of +/// non-overlapping writers can run. Other kernels retain the conservative +/// single-writer lock because Unix ownership alone cannot isolate processes +/// that share the writer UID. +pub struct WriterLease { + file: File, + path: Option, + scoped: bool, +} + +impl WriterLease { + pub fn scoped(&self) -> bool { + self.scoped + } +} + +impl Drop for WriterLease { + fn drop(&mut self) { + if let Some(path) = &self.path { + let _ = fs::remove_file(path); + } + let _ = FileExt::unlock(&self.file); + } +} + +pub fn acquire_writer_lease( + state: &Path, + name: &str, + owned_paths: &[PathBuf], +) -> Result { + if !role_sandbox::supports_scoped_writers() { + let path = state.join("launch-authorizations/.writer.lock"); + let file = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(&path) + .map_err(|error| format!("open serial writer lock {}: {error}", path.display()))?; + file.try_lock_exclusive() + .map_err(|_| "another workspace writer is already active".to_string())?; + return Ok(WriterLease { + file, + path: None, + scoped: false, + }); + } + + acquire_scoped_writer_lease(state, name, owned_paths) +} + +fn acquire_scoped_writer_lease( + state: &Path, + name: &str, + owned_paths: &[PathBuf], +) -> Result { + if owned_paths.is_empty() { + return Err("workspace writer requires at least one owned path".into()); + } + let base = state.join("launch-authorizations/writer-leases"); + fs::create_dir_all(&base).map_err(|error| format!("create writer lease directory: {error}"))?; + let registry_path = base.join(".lock"); + let registry = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(®istry_path) + .map_err(|error| format!("open writer lease registry: {error}"))?; + registry + .lock_exclusive() + .map_err(|error| format!("lock writer lease registry: {error}"))?; + + for entry in fs::read_dir(&base).map_err(|error| format!("read writer leases: {error}"))? { + let path = entry + .map_err(|error| format!("read writer lease entry: {error}"))? + .path(); + if path == registry_path + || path.extension().and_then(|value| value.to_str()) != Some("lease") + { + continue; + } + let existing = match OpenOptions::new().read(true).write(true).open(&path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(format!("open writer lease {}: {error}", path.display())), + }; + if existing.try_lock_exclusive().is_ok() { + fs::remove_file(&path).map_err(|error| { + format!("remove stale writer lease {}: {error}", path.display()) + })?; + continue; + } + let active = fs::read_to_string(&path) + .map_err(|error| format!("read writer lease {}: {error}", path.display()))?; + for requested in owned_paths { + if let Some(conflict) = active + .lines() + .filter(|line| !line.is_empty()) + .map(Path::new) + .find(|current| paths_overlap(requested, current)) + { + return Err(format!( + "active writer path overlap: requested={} existing={}", + requested.display(), + conflict.display() + )); + } + } + } + + let path = base.join(format!("{name}.lease")); + let mut file = OpenOptions::new() + .create_new(true) + .read(true) + .write(true) + .open(&path) + .map_err(|error| format!("create writer lease {}: {error}", path.display()))?; + file.lock_exclusive() + .map_err(|error| format!("lock writer lease {}: {error}", path.display()))?; + for owned in owned_paths { + writeln!(file, "{}", owned.display()) + .map_err(|error| format!("write writer lease {}: {error}", path.display()))?; + } + FileExt::unlock(®istry).map_err(|error| format!("unlock writer lease registry: {error}"))?; + Ok(WriterLease { + file, + path: Some(path), + scoped: true, + }) +} + +fn paths_overlap(left: &Path, right: &Path) -> bool { + left == right || left.starts_with(right) || right.starts_with(left) +} + fn register_launch(args: &[String], renew: bool) -> Result<(), String> { let name = args .first() @@ -479,19 +580,6 @@ fn required_field<'a>(values: &'a BTreeMap, name: &str) -> Resul .ok_or_else(|| format!("launch authorization is missing {name}")) } -fn read_env_file(path: &Path) -> Result, String> { - let mut values = BTreeMap::new(); - for line in fs::read_to_string(path) - .map_err(|error| format!("read launch authorization {}: {error}", path.display()))? - .lines() - { - if let Some((key, value)) = line.split_once('=') { - values.insert(key.into(), value.into()); - } - } - Ok(values) -} - fn valid_name(name: &str) -> bool { !name.is_empty() && !name.starts_with('-') @@ -623,7 +711,7 @@ pub fn validate_runtime_state(_state: &Path) -> Result<(), String> { } #[cfg(unix)] -fn proxy_request(request: Request) -> Result { +fn proxy_request(request: AuthorityRequest) -> Result { let state = config::state_dir()?; let socket = authority_socket(&state); let mut stream = UnixStream::connect(&socket) @@ -648,7 +736,7 @@ fn proxy_request(request: Request) -> Result { } #[cfg(not(unix))] -fn proxy_request(_request: Request) -> Result { +fn proxy_request(_request: AuthorityRequest) -> Result { Err("authority supervisor requires Unix".into()) } @@ -718,7 +806,7 @@ fn serve_connection(stream: &mut UnixStream) -> Result { eprintln!("authority supervisor: read request: {error}"); return Ok(false); } - let request: Request = match serde_json::from_slice(&bytes) { + let request: AuthorityRequest = match serde_json::from_slice(&bytes) { Ok(request) => request, Err(error) => { let _ = write_response( @@ -732,7 +820,7 @@ fn serve_connection(stream: &mut UnixStream) -> Result { return Ok(false); } }; - if request.command == "supervisor" && request.args == ["shutdown"] { + if request.is_shutdown() { let _ = write_response( stream, &Response { @@ -743,18 +831,15 @@ fn serve_connection(stream: &mut UnixStream) -> Result { ); return Ok(true); } - if !proxy_command(&request.command, &request.args) - || !caller_authorized(peer_uid, &request.command, &request.args) - { + if !request.authorized_for(peer_uid) { let _ = write_response( stream, &Response { code: 1, stdout: String::new(), stderr: format!( - "authority supervisor: caller uid {peer_uid} is not authorized for: {} {}\n", - request.command, - request.args.first().map(String::as_str).unwrap_or("") + "authority supervisor: caller uid {peer_uid} is not authorized for: {}\n", + request.display() ), }, ); @@ -768,12 +853,13 @@ fn serve_connection(stream: &mut UnixStream) -> Result { } #[cfg(target_os = "linux")] -fn execute(request: Request) -> Result { +fn execute(request: AuthorityRequest) -> Result { let executable = env::current_exe().map_err(|error| format!("resolve authority executable: {error}"))?; + let (command, args) = request.into_cli(); let output = Command::new(executable) - .arg(&request.command) - .args(&request.args) + .arg(command) + .args(args) .env(SERVER_CHILD_ENV, "1") .stdin(Stdio::null()) .output() @@ -818,46 +904,6 @@ fn peer_uid(stream: &UnixStream) -> Result { Ok(credentials.uid) } -#[cfg(any(target_os = "linux", test))] -fn caller_authorized(uid: u32, command: &str, args: &[String]) -> bool { - if uid == 0 { - return true; - } - let subcommand = args.first().map(String::as_str).unwrap_or(""); - match command { - "workflow" | "decision" | "dag" | "orchestrator" | "supervisor" => { - uid == config::ORCHESTRATOR_UID - } - "subagent" => match subcommand { - "finding-create" => uid == config::READER_UID, - // The orchestrator may request a disposition, but subagent.rs - // authorizes it only from supervisor-sealed reviewer evidence. - "finding-dismiss" | "todo-close" => { - matches!(uid, config::ORCHESTRATOR_UID | config::READER_UID) - } - "resolution-create" => uid == config::WRITER_UID, - "checkpoint-update" | "checkpoint-show" => matches!( - uid, - config::ORCHESTRATOR_UID | config::WRITER_UID | config::READER_UID - ), - "finding-show" - | "finding-list" - | "todo-show" - | "todo-list" - | "validation-lease-show" - | "validation-lease-list" => matches!( - uid, - config::ORCHESTRATOR_UID | config::WRITER_UID | config::READER_UID - ), - "validation-lease-acquire" | "validation-lease-status" => { - matches!(uid, config::WRITER_UID | config::READER_UID) - } - _ => uid == config::ORCHESTRATOR_UID, - }, - _ => false, - } -} - #[cfg(not(target_os = "linux"))] fn serve(_socket: &Path) -> Result { Err("authority supervisor requires Unix".into()) @@ -1048,52 +1094,13 @@ pub fn start(_state: &Path, _executable: &Path) -> Result { #[cfg(test)] mod tests { + use super::acquire_scoped_writer_lease; #[cfg(target_os = "linux")] use super::serve_connection; - use super::{caller_authorized, proxy_command}; - use crate::config; + use std::fs; #[cfg(target_os = "linux")] use std::os::unix::net::UnixStream; - - #[test] - fn typed_api_excludes_runtime_and_arbitrary_execution() { - assert!(proxy_command("workflow", &["status".into()])); - assert!(proxy_command("subagent", &["assignment-create".into()])); - assert!(!proxy_command("agent", &["run".into()])); - assert!(!proxy_command("role-exec", &[])); - assert!(!proxy_command("subagent", &["spawn".into()])); - assert!(!proxy_command("subagent", &["worktree-create".into()])); - assert!(!proxy_command("subagent", &["validation-run".into()])); - } - - #[test] - fn authority_mutations_are_role_typed() { - assert!(caller_authorized( - config::ORCHESTRATOR_UID, - "workflow", - &["transition".into()] - )); - assert!(!caller_authorized( - config::ORCHESTRATOR_UID, - "subagent", - &["finding-create".into()] - )); - assert!(caller_authorized( - config::READER_UID, - "subagent", - &["finding-create".into()] - )); - assert!(caller_authorized( - config::ORCHESTRATOR_UID, - "subagent", - &["todo-close".into()] - )); - assert!(!caller_authorized( - config::WRITER_UID, - "workflow", - &["transition".into()] - )); - } + use std::path::PathBuf; #[cfg(target_os = "linux")] #[test] @@ -1103,4 +1110,30 @@ mod tests { assert!(!serve_connection(&mut server).expect("isolate disconnected client")); } + + #[test] + fn scoped_writer_leases_allow_disjoint_topology_and_reject_overlap() { + let state = std::env::temp_dir().join(format!( + "multiagent-writer-lease-test-{}", + std::process::id() + )); + let first_paths = [PathBuf::from("/workspace/src/api")]; + let second_paths = [PathBuf::from("/workspace/src/storage")]; + let overlap_paths = [PathBuf::from("/workspace/src/api/routes")]; + + let first = acquire_scoped_writer_lease(&state, "first", &first_paths) + .expect("acquire first writer"); + let second = acquire_scoped_writer_lease(&state, "second", &second_paths) + .expect("acquire disjoint writer"); + let error = acquire_scoped_writer_lease(&state, "overlap", &overlap_paths) + .err() + .expect("reject overlapping writer"); + assert!(error.contains("active writer path overlap")); + + drop(first); + acquire_scoped_writer_lease(&state, "replacement", &overlap_paths) + .expect("released paths can be reassigned"); + drop(second); + fs::remove_dir_all(state).expect("remove writer lease test state"); + } } diff --git a/src/workflow.rs b/src/workflow.rs index 926c5e1..f488888 100644 --- a/src/workflow.rs +++ b/src/workflow.rs @@ -1,5 +1,7 @@ -use crate::config; -use chrono::{SecondsFormat, Utc}; +use crate::{ + config, + state::{atomic_write, atomic_write_bytes, read_env_optional, timestamp}, +}; use fs2::FileExt; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet}; @@ -1358,21 +1360,53 @@ fn validate_contract_schema(text: &str, original_task: &str) -> Result<(), Strin } } if requires_embedding_rule { - let positive = rules.iter().any(|rule| rule.contains(" polarity=must ")); - let negative = rules.iter().any(|rule| { + let positive = rules.iter().find_map(|rule| { + (rule.contains(" polarity=must ") && rule.contains(" structure=positive ")).then(|| { + ( + contract_rule_field(rule, "owner"), + contract_rule_field(rule, "member"), + contract_rule_field(rule, "member-type"), + ) + }) + }); + let negative = rules.iter().find_map(|rule| { let statement = contract_rule_statement(rule).to_ascii_lowercase(); - rule.contains(" polarity=must-not ") && statement.contains("embed") + (rule.contains(" polarity=must-not ") + && rule.contains(" structure=negative ") + && statement.contains("embed")) + .then(|| { + ( + contract_rule_field(rule, "owner"), + contract_rule_field(rule, "embedded-type"), + ) + }) }); - if !positive { - return Err("contract scout artifact requires a positive structural rule".into()); - } - if !negative { - return Err("contract scout artifact requires a separate `polarity=must-not` rule covering the requested embedding prohibition".into()); + let Some((positive_owner, member, member_type)) = positive else { + return Err("contract scout artifact requires a machine-readable positive structural rule with `structure=positive owner=OWNER member=FIELD member-type=TYPE`".into()); + }; + let Some((negative_owner, embedded_type)) = negative else { + return Err("contract scout artifact requires a machine-readable negative structural rule with `structure=negative owner=OWNER embedded-type=TYPE` covering the embedding prohibition".into()); + }; + if positive_owner.is_empty() + || member.is_empty() + || member_type.is_empty() + || negative_owner.is_empty() + || embedded_type.is_empty() + || positive_owner != negative_owner + || member_type != embedded_type + { + return Err("contract scout embedding rules must name one matching owner/type pair and a concrete replacement member".into()); } } Ok(()) } +fn contract_rule_field<'a>(rule: &'a str, key: &str) -> &'a str { + rule.split_whitespace() + .find_map(|part| part.strip_prefix(&format!("{key}="))) + .unwrap_or("") +} + fn contract_rule_statement(rule: &str) -> &str { rule.split_once(" statement=") .map(|(_, value)| value) @@ -1411,19 +1445,7 @@ fn read_env(path: &Path, id: &str) -> Result, String> { read_simple_env(path) } fn read_simple_env(path: &Path) -> Result, String> { - let mut out = BTreeMap::new(); - if !path.is_file() { - return Ok(out); - } - for line in fs::read_to_string(path) - .map_err(io_error("read state"))? - .lines() - { - if let Some((k, v)) = line.split_once('=') { - out.insert(k.into(), v.into()); - } - } - Ok(out) + read_env_optional(path) } fn write_env(path: &Path, state: &BTreeMap) -> Result<(), String> { let text = ENV_ORDER @@ -1480,25 +1502,6 @@ fn event(path: &Path, name: &str, detail: &str) -> Result<(), String> { writeln!(file, "{}\t{}\t{}", timestamp(), name, detail) .map_err(io_error("append lifecycle event")) } -fn atomic_write(path: &Path, text: &str) -> Result<(), String> { - atomic_write_bytes(path, text.as_bytes()) -} - -fn atomic_write_bytes(path: &Path, bytes: &[u8]) -> Result<(), String> { - if let Some(parent) = path.parent() { - fs::create_dir_all(parent).map_err(io_error("create state directory"))?; - } - let temp = path.with_file_name(format!( - ".{}.{}.tmp", - path.file_name().and_then(|v| v.to_str()).unwrap_or("state"), - std::process::id() - )); - let mut file = File::create(&temp).map_err(io_error("create temporary state"))?; - file.write_all(bytes) - .map_err(io_error("write temporary state"))?; - file.sync_all().map_err(io_error("sync temporary state"))?; - fs::rename(&temp, path).map_err(io_error("publish state")) -} fn sha256(path: &Path) -> Result { let mut file = File::open(path).map_err(io_error("read implementation context"))?; let mut digest = Sha256::new(); @@ -1530,9 +1533,6 @@ fn active(status: &str) -> bool { fn state_value<'a>(state: &'a BTreeMap, key: &str) -> &'a str { state.get(key).map(String::as_str).unwrap_or("") } -fn timestamp() -> String { - Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true) -} fn valid_id(label: &str, value: &str) -> Result<(), String> { if value.is_empty() || !value @@ -1632,11 +1632,18 @@ contract-rule: id=R1 polarity=must statement=WidgetConfig exposes named fields e contract-rule: id=R2 polarity=must-not statement=Old WidgetConfig names must not remain evidence=task mentions unnecessary embedding\n"; assert!(validate_contract_schema(incomplete, task) .unwrap_err() - .contains("embedding prohibition")); + .contains("machine-readable positive structural rule")); - let complete = format!( - "{incomplete}contract-rule: id=R3 polarity=must-not statement=WidgetConfig must not be anonymously embedded evidence=task\n" - ); + let complete = "contract-artifact: version=1\n\ +contract-rule: id=R1 polarity=must structure=positive owner=Widget member=cfg member-type=WidgetConfig statement=Widget must store WidgetConfig in the named cfg field evidence=source\n\ +contract-rule: id=R2 polarity=must-not structure=negative owner=Widget embedded-type=WidgetConfig statement=Widget must not anonymously embed WidgetConfig evidence=task\n"; assert!(validate_contract_schema(&complete, task).is_ok()); + + let mismatched = "contract-artifact: version=1\n\ +contract-rule: id=R1 polarity=must structure=positive owner=Widget member=cfg member-type=WidgetConfig statement=Widget has a named cfg field evidence=source\n\ +contract-rule: id=R2 polarity=must-not structure=negative owner=Other embedded-type=RouterConfig statement=Other must not embed RouterConfig evidence=task\n"; + assert!(validate_contract_schema(mismatched, task) + .unwrap_err() + .contains("matching owner/type pair")); } } diff --git a/tests/run.sh b/tests/run.sh index a870894..0ee3222 100755 --- a/tests/run.sh +++ b/tests/run.sh @@ -403,6 +403,9 @@ assert_file_contains "$TMPDIR/launch.out" "Dashboard: MULTIAGENT_SESSION=launch- LAUNCH_BOOTSTRAP="$LAUNCH_STATE/orchestrator-bootstrap.sh" assert_file_contains "$MOCK_TMUX_LOG" "$(printf '%q' "$LAUNCH_BOOTSTRAP")" assert_file_contains "$MOCK_TMUX_LOG" "pipe-pane launch-cross-repo:orchestrator cat >> $LAUNCH_STATE/logs/orchestrator.log" +assert_file_contains "$MOCK_TMUX_LOG" "new-window -d launch-cross-repo supervisor" +assert_file_contains "$MOCK_TMUX_LOG" "$MULTIAGENT supervisor reconcile --watch --interval 2" +assert_file_contains "$MOCK_TMUX_LOG" "pipe-pane launch-cross-repo:supervisor cat >> $LAUNCH_STATE/logs/supervisor.log" assert_file_contains "$LAUNCH_BOOTSTRAP" "--cd $LAUNCH_STATE" if [[ "$HOST_KERNEL" == Linux ]]; then assert_file_contains "$LAUNCH_BOOTSTRAP" "$MULTIAGENT role-exec" @@ -703,6 +706,25 @@ printf 'ACCEPTED\nbuild-verification-passed: final-diff-sha256=%s compile_clean= MULTIAGENT_ROOT="$HASH_GATE_ROOT" MULTIAGENT_STATE_DIR="$HASH_GATE_STATE" MULTIAGENT_REQUIRE_HASH_BOUND_VERIFIER=1 \ "$MULTIAGENT" subagent gate-check >"$TMPDIR/gate-verifier-bound-hash.out" assert_file_contains "$TMPDIR/gate-verifier-bound-hash.out" "accepted" +mkdir -p "$HASH_GATE_ROOT/src/routes" +printf 'before route\n' >"$HASH_GATE_ROOT/src/routes/index.js" +git -C "$HASH_GATE_ROOT" add src/routes/index.js +git -C "$HASH_GATE_ROOT" commit -qm route-baseline +printf 'after route\n' >"$HASH_GATE_ROOT/src/routes/index.js" +HASH_GATE_ROUTE_SHA="$("$MULTIAGENT" snapshot --root "$HASH_GATE_ROOT" --format shell | awk '{print $1}')" +printf 'ACCEPTED\nbuild-verification-passed: final-diff-sha256=%s compile_clean=true returncode=0\n' "$HASH_GATE_ROUTE_SHA" >"$HASH_GATE_STATE/subagents/verifier-01-hash/last-message.txt" +if MULTIAGENT_ROOT="$HASH_GATE_ROOT" MULTIAGENT_STATE_DIR="$HASH_GATE_STATE" MULTIAGENT_REQUIRE_HASH_BOUND_VERIFIER=1 \ + "$MULTIAGENT" subagent gate-check >"$TMPDIR/gate-route-missing-probe.out" 2>&1; then + echo "expected a route diff without an integration probe marker to fail gate-check" >&2 + exit 1 +fi +assert_file_contains "$TMPDIR/gate-route-missing-probe.out" $'reject\tmissing-route-integration-probe' +printf 'ACCEPTED\nbuild-verification-passed: final-diff-sha256=%s compile_clean=true returncode=0\nroute-integration-probe-passed: final-diff-sha256=%s command=request-test returncode=0\n' \ + "$HASH_GATE_ROUTE_SHA" "$HASH_GATE_ROUTE_SHA" >"$HASH_GATE_STATE/subagents/verifier-01-hash/last-message.txt" +MULTIAGENT_ROOT="$HASH_GATE_ROOT" MULTIAGENT_STATE_DIR="$HASH_GATE_STATE" MULTIAGENT_REQUIRE_HASH_BOUND_VERIFIER=1 \ + "$MULTIAGENT" subagent gate-check >"$TMPDIR/gate-route-probe-passed.out" +assert_file_contains "$TMPDIR/gate-route-probe-passed.out" "accepted" +git -C "$HASH_GATE_ROOT" restore src/routes/index.js printf 'malicious post-review source\n' >"$HASH_GATE_ROOT/untracked-source.txt" if MULTIAGENT_ROOT="$HASH_GATE_ROOT" MULTIAGENT_STATE_DIR="$HASH_GATE_STATE" MULTIAGENT_REQUIRE_HASH_BOUND_VERIFIER=1 \ "$MULTIAGENT" subagent gate-check >"$TMPDIR/gate-verifier-untracked-bypass.out" 2>&1; then @@ -947,7 +969,8 @@ assert_file_contains "$ROOT/prompts/playbooks/intent-contract.md" "Intent And Co assert_file_contains "$ROOT/prompts/playbooks/intent-contract.md" "proxy/scaffold limitations" assert_file_contains "$ROOT/prompts/playbooks/intent-contract.md" "contract-ledger" assert_file_contains "$ROOT/prompts/playbooks/parallel-execution.md" "Parallel Execution Playbook" -assert_file_contains "$ROOT/prompts/playbooks/parallel-execution.md" "Default to broad safe fan-out" +assert_file_contains "$ROOT/prompts/playbooks/parallel-execution.md" "does not prescribe a worker count" +assert_file_contains "$ROOT/prompts/playbooks/parallel-execution.md" "task-specific responsibility" assert_file_contains "$ROOT/prompts/playbooks/parallel-execution.md" "If one subtree is blocked" assert_file_contains "$ROOT/prompts/playbooks/validation-scheduling.md" "Validation Scheduling Playbook" assert_file_contains "$ROOT/prompts/playbooks/validation-scheduling.md" "Validation Lease" @@ -977,7 +1000,7 @@ assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "gate-check" assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "required-path-outside-owned:" assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "ownership blocker" assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" 'SUBAGENT_CLI="$WORKER_CLI" multiagent subagent spawn' -assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "multiagent subagent wait worker-01-task" +assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "lifecycle supervisor monitors and finalizes" assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "Orchestration Routing Playbook" assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "Contract Scout Workflow" assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "Scope Guard Workflow" @@ -1017,7 +1040,11 @@ assert_file_contains "$ROOT/prompts/verifier.md" "state-space partition audit" assert_file_contains "$ROOT/prompts/verifier.md" "mixed-category, unknown/forward-compatible variant" assert_file_contains "$ROOT/prompts/verifier.md" "state-space-partition-audit:" assert_file_contains "$ROOT/prompts/verifier.md" "behavior-verification-passed:" +assert_file_contains "$ROOT/prompts/verifier.md" "route-integration-probe-passed:" assert_file_contains "$ROOT/prompts/verifier.md" "narrowest visible test file" +assert_file_contains "$ROOT/prompts/verifier.md" "visible-test-replay-passed:" +assert_file_contains "$ROOT/prompts/verifier.md" "exact boundary values" +assert_file_contains "$ROOT/prompts/worker.md" "visible-test-replay-passed:" assert_file_contains "$ROOT/prompts/verifier.md" "Syntax checks, compile-only commands" assert_file_contains "$ROOT/prompts/verifier.md" "command=... returncode=0" assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "partition contract" @@ -1063,6 +1090,7 @@ assert_file_contains "$ROOT/prompts/worker.md" "no-test compile check" assert_file_contains "$ROOT/prompts/worker.md" "declared static type" assert_file_contains "$ROOT/prompts/worker.md" "validation-repair-needed:" assert_file_contains "$ROOT/prompts/worker.md" "multi-value-probe-passed:" +assert_file_contains "$ROOT/prompts/worker.md" "route-integration-probe-passed:" assert_file_contains "$ROOT/prompts/worker.md" "actual-output-count=N" assert_file_contains "$ROOT/prompts/worker.md" "multi-value-probe.txt" assert_file_contains "$ROOT/prompts/worker.md" "source-symbol-map-passed:" @@ -1101,13 +1129,15 @@ assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "Do not spawn w assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "live worker remains no-diff after a planning checkpoint" assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "Scout To Worker Handoff" assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "active generic scout block" -assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "assignment-status NAME failed" +assert_file_contains "$ROOT/prompts/playbooks/agent-spawning.md" "Do not manually reproduce" assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "most one same-owned-path replacement" assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "live worker remains no-diff after a planning checkpoint" assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "active generic scout block" -assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "assignment-status NAME failed" +assert_file_contains "$ROOT/prompts/playbooks/orchestration-routing.md" "lifecycle supervisor settles terminal assignments" assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "source-symbol map contract" assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "source-symbol-map-passed:" +assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "structure=positive owner=OWNER member=FIELD member-type=TYPE" +assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "structure=negative owner=OWNER embedded-type=TYPE" assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "source-owner-ledger:" assert_file_contains "$ROOT/prompts/roles/contract-scout.md" "constructor-dependency contract" assert_file_contains "$ROOT/prompts/roles/build-verifier.md" "build-verification-passed:" @@ -1613,15 +1643,23 @@ if grep -Fq "new-window -d test-session owned-mismatch" "$MOCK_TMUX_LOG"; then fi printf 'Claude prompt ready\n' >"$MOCK_TMUX_CAPTURES/worker-generic-01.txt" -"$MULTIAGENT" subagent spawn worker-generic-01 --instruction "First generic worker" +"$MULTIAGENT" subagent spawn worker-generic-01 --own src/agent.rs \ + --responsibility "adapt agent backend" --instruction "First adaptive worker" assert_file_contains "$MULTIAGENT_STATE_DIR/subagents/worker-generic-01/status" "running" +assert_file_contains "$MULTIAGENT_STATE_DIR/assignments/worker-generic-01/assignment.env" "responsibility=adapt agent backend" printf 'Claude prompt ready\n' >"$MOCK_TMUX_CAPTURES/worker-generic-02.txt" -if "$MULTIAGENT" subagent spawn worker-generic-02 --instruction "Second generic worker" >"$TMPDIR/worker-generic-conflict.out" 2>&1; then - echo "expected generic worker spawn to reject active generic worker" >&2 +"$MULTIAGENT" subagent spawn worker-generic-02 --own src/dag.rs \ + --responsibility "adapt workflow graph" --instruction "Second adaptive worker" +assert_file_contains "$MULTIAGENT_STATE_DIR/subagents/worker-generic-02/status" "running" +assert_file_contains "$MULTIAGENT_STATE_DIR/assignments/worker-generic-02/assignment.env" "responsibility=adapt workflow graph" + +printf 'Claude prompt ready\n' >"$MOCK_TMUX_CAPTURES/worker-generic-overlap.txt" +if "$MULTIAGENT" subagent spawn worker-generic-overlap --own src/agent.rs --instruction "Overlap existing worker" >"$TMPDIR/worker-generic-conflict.out" 2>&1; then + echo "expected overlapping adaptive worker ownership to fail" >&2 cat "$TMPDIR/worker-generic-conflict.out" >&2 exit 1 fi -assert_file_contains "$TMPDIR/worker-generic-conflict.out" "active generic worker already running" +assert_file_contains "$TMPDIR/worker-generic-conflict.out" "active assignment owned-path overlap" printf 'Codex prompt ready\n' >"$MOCK_TMUX_CAPTURES/verifier-01-docs.txt" SUBAGENT_CLI="$VERIFIER_CLI" "$MULTIAGENT" subagent spawn verifier-01-docs --instruction "Review worker-01-docs" @@ -1954,6 +1992,87 @@ restore_all_output="$("$MULTIAGENT" subagent restore-all)" [[ "$restore_all_output" == *"restored subagent-prompt-only"* ]] [[ "$restore_all_output" == *"restore-all complete: restored=1"* ]] +# The Rust lifecycle supervisor owns mechanical terminal cleanup, assignment +# synchronization, abandoned validation leases, and explicitly budgeted +# infrastructure retries. It does not decide how to repair blocked work. +RECON_STATE="$TMPDIR/reconcile-state" +RECON_WINDOWS="$TMPDIR/reconcile-windows" +RECON_CAPTURES="$TMPDIR/reconcile-captures" +RECON_LOG="$TMPDIR/reconcile-tmux.log" +mkdir -p "$RECON_STATE" "$RECON_CAPTURES" +: >"$RECON_WINDOWS" +: >"$RECON_LOG" + +printf 'Claude prompt ready\n' >"$RECON_CAPTURES/reconcile-done.txt" +MOCK_TMUX_WINDOWS="$RECON_WINDOWS" MOCK_TMUX_CAPTURES="$RECON_CAPTURES" \ + MOCK_TMUX_LOG="$RECON_LOG" MULTIAGENT_STATE_DIR="$RECON_STATE" \ + "$MULTIAGENT" subagent spawn reconcile-done --own src/agent.rs \ + --responsibility "exercise lifecycle reconciliation" --instruction "Finish cleanly" >/dev/null +printf 'final status: coding agent exited rc=0\n' >"$RECON_CAPTURES/reconcile-done.txt" +printf 'ACCEPTED\n' >"$RECON_STATE/subagents/reconcile-done/last-message.txt" +MULTIAGENT_STATE_DIR="$RECON_STATE" "$MULTIAGENT" subagent validation-lease-acquire \ + reconcile-lease --owner reconcile-done --target reconcile-target \ + --command "cargo test" >/dev/null + +mkdir -p "$RECON_STATE/subagents/reconcile-blocked" +printf 'running\n' >"$RECON_STATE/subagents/reconcile-blocked/status" +printf 'Blocked: semantic input required\n' >"$RECON_STATE/subagents/reconcile-blocked/current.txt" +printf 'reconcile-blocked\n' >>"$RECON_WINDOWS" +printf 'Blocked: semantic input required\n' >"$RECON_CAPTURES/reconcile-blocked.txt" + +mkdir -p "$RECON_STATE/subagents/reconcile-untrusted-done" +printf 'running\n' >"$RECON_STATE/subagents/reconcile-untrusted-done/status" +printf 'reconcile-untrusted-done\n' >>"$RECON_WINDOWS" +printf 'Final status: completed\n' >"$RECON_CAPTURES/reconcile-untrusted-done.txt" + +mkdir -p "$RECON_STATE/subagents/reconcile-retry" +printf 'missing\n' >"$RECON_STATE/subagents/reconcile-retry/status" +printf 'Previous progress before infrastructure exit\n' >"$RECON_STATE/subagents/reconcile-retry/current.txt" +printf 'Previous transcript before infrastructure exit\n' >"$RECON_STATE/subagents/reconcile-retry/transcript.log" +cat >"$RECON_STATE/subagents/reconcile-retry/meta.env" <"$RECON_CAPTURES/reconcile-retry.txt" + +reconcile_output="$(MOCK_TMUX_WINDOWS="$RECON_WINDOWS" \ + MOCK_TMUX_CAPTURES="$RECON_CAPTURES" MOCK_TMUX_LOG="$RECON_LOG" \ + MULTIAGENT_STATE_DIR="$RECON_STATE" "$MULTIAGENT" supervisor reconcile)" +[[ "$reconcile_output" == *$'reconcile-done\tdone\tfinalized\tterminal-process-cleaned'* ]] +[[ "$reconcile_output" == *$'reconcile-blocked\tblocked\tneeds-decision\twindow-open'* ]] +[[ "$reconcile_output" == *$'reconcile-untrusted-done\trunning\tobserved\twindow-open'* ]] +[[ "$reconcile_output" == *$'reconcile-retry\tmissing\trestored\tbudgeted-infrastructure-retry'* ]] +assert_file_contains "$RECON_STATE/subagents/reconcile-done/status" "done" +assert_file_contains "$RECON_STATE/assignments/reconcile-done/status" "done" +assert_file_contains "$RECON_STATE/validation-leases/reconcile-lease/status" "stale" +assert_file_contains "$RECON_STATE/validation-leases/reconcile-lease/result.json" \ + "owner-terminal-without-lease-result" +assert_file_contains "$RECON_WINDOWS" "reconcile-blocked" +assert_file_contains "$RECON_WINDOWS" "reconcile-untrusted-done" +assert_file_contains "$RECON_WINDOWS" "reconcile-retry" +assert_file_contains "$RECON_STATE/subagents/reconcile-retry/meta.env" "infra_retry_count=1" + +grep -Fvx -- "reconcile-retry" "$RECON_WINDOWS" >"$RECON_WINDOWS.next" || true +mv "$RECON_WINDOWS.next" "$RECON_WINDOWS" +printf 'missing\n' >"$RECON_STATE/subagents/reconcile-retry/status" +printf 'Progress stopped again before completion\n' >"$RECON_STATE/subagents/reconcile-retry/current.txt" +retry_exhausted_output="$(MOCK_TMUX_WINDOWS="$RECON_WINDOWS" \ + MOCK_TMUX_CAPTURES="$RECON_CAPTURES" MOCK_TMUX_LOG="$RECON_LOG" \ + MULTIAGENT_STATE_DIR="$RECON_STATE" "$MULTIAGENT" supervisor reconcile)" +[[ "$retry_exhausted_output" == *$'reconcile-done\tdone\tunchanged\talready-settled'* ]] +[[ "$retry_exhausted_output" == *$'reconcile-retry\tmissing\tneeds-recovery\tclosed-with-recoverable-context'* ]] +if grep -Fqx -- "reconcile-retry" "$RECON_WINDOWS"; then + echo "expected the lifecycle supervisor to honor the exhausted retry budget" >&2 + exit 1 +fi + # Test organizational learning functionality # Test multiagent decision basic functionality @@ -2078,10 +2197,11 @@ mkdir -p "$ORG_ASSIGN_REPO" "$ORG_ASSIGN_STATE" git switch -q -c worker/org-task ) -org_assignment_create_output="$(MULTIAGENT_ROOT="$ORG_ASSIGN_REPO" MULTIAGENT_STATE_DIR="$ORG_ASSIGN_STATE" "$MULTIAGENT" subagent assignment-create worker-org --assignment-id org-001 --branch worker/org-task --owned README.md --role qa --decision-id DEC-001 --plan-id PLAN-A)" +org_assignment_create_output="$(MULTIAGENT_ROOT="$ORG_ASSIGN_REPO" MULTIAGENT_STATE_DIR="$ORG_ASSIGN_STATE" "$MULTIAGENT" subagent assignment-create worker-org --assignment-id org-001 --branch worker/org-task --owned README.md --role qa --responsibility "migrate the storage adapter" --decision-id DEC-001 --plan-id PLAN-A)" [[ "$org_assignment_create_output" == $'assignment created\tworker-org\torg-001\tworker/org-task' ]] assert_file_contains "$ORG_ASSIGN_STATE/assignments/worker-org/assignment.env" "assignment_id=org-001" assert_file_contains "$ORG_ASSIGN_STATE/assignments/worker-org/assignment.env" "role=qa" +assert_file_contains "$ORG_ASSIGN_STATE/assignments/worker-org/assignment.env" "responsibility=migrate the storage adapter" assert_file_contains "$ORG_ASSIGN_STATE/assignments/worker-org/assignment.env" "decision_id=DEC-001" assert_file_contains "$ORG_ASSIGN_STATE/assignments/worker-org/assignment.env" "plan_id=PLAN-A" # Test invalid role rejection @@ -2100,6 +2220,7 @@ checkpoint_org_output="$(MULTIAGENT_ROOT="$ORG_ASSIGN_REPO" MULTIAGENT_STATE_DIR [[ "$checkpoint_org_output" == $'checkpoint updated\tworker-org\trunning' ]] checkpoint_show_org_output="$(MULTIAGENT_ROOT="$ORG_ASSIGN_REPO" MULTIAGENT_STATE_DIR="$ORG_ASSIGN_STATE" "$MULTIAGENT" subagent checkpoint-show worker-org)" [[ "$checkpoint_show_org_output" == *"role=qa"* ]] +[[ "$checkpoint_show_org_output" == *"responsibility=migrate the storage adapter"* ]] [[ "$checkpoint_show_org_output" == *"decision_id=DEC-001"* ]] [[ "$checkpoint_show_org_output" == *"plan_id=PLAN-A"* ]] # Test multiagent status includes organizational metadata columns @@ -2293,15 +2414,12 @@ if MULTIAGENT_STATE_DIR="$DAG_STATE_DIR" "$MULTIAGENT" dag status WF-001 NODE-A fi assert_file_contains "$TMPDIR/invalid-status.out" "invalid status: invalid-status" -# Test role validation - invalid roles should be rejected -if MULTIAGENT_STATE_DIR="$DAG_STATE_DIR" "$MULTIAGENT" dag add-node WF-001 NODE-INVALID-ROLE --agent worker-invalid --assignment-id assign-invalid --role decision --branch worker/invalid --owned file-invalid.txt >"$TMPDIR/invalid-role.out" 2>&1; then - echo "expected invalid role 'decision' to fail" >&2 - cat "$TMPDIR/invalid-role.out" >&2 - exit 1 -fi -assert_file_contains "$TMPDIR/invalid-role.out" "invalid role: decision" +# DAG responsibility is orchestrator-defined rather than a framework enum. +custom_responsibility_output="$(MULTIAGENT_STATE_DIR="$DAG_STATE_DIR" "$MULTIAGENT" dag add-node WF-001 NODE-CUSTOM-RESPONSIBILITY --agent worker-custom --assignment-id assign-custom --responsibility "audit storage compatibility" --branch worker/custom --owned file-custom.txt)" +[[ "$custom_responsibility_output" == *"node added"* ]] +assert_file_contains "$DAG_STATE_DIR/workflows/WF-001/nodes.tsv" $'NODE-CUSTOM-RESPONSIBILITY\tworker-custom\tassign-custom\taudit storage compatibility' -# Test role validation - valid roles should be accepted +# The legacy --role spelling remains a compatibility alias for responsibility. valid_roles=("exploitation" "exploration" "reflection" "architecture" "qa" "verifier" "scout") for i in "${!valid_roles[@]}"; do role="${valid_roles[$i]}" diff --git a/tests/test_migration_contracts.py b/tests/test_migration_contracts.py index 8e3304d..071508c 100644 --- a/tests/test_migration_contracts.py +++ b/tests/test_migration_contracts.py @@ -455,6 +455,33 @@ def test_snapshot_excludes_only_baseline_untracked_files(self): self.assertEqual(payload["changed_paths"], ["src/new.rs"]) self.assertEqual(payload["changed_code_paths"], ["src/new.rs"]) + def test_snapshot_excludes_framework_control_plane_files(self): + internal = self.repo / ".multiagent" / "subagents" / "reviewer" + internal.mkdir(parents=True) + (internal / "status").write_text("missing\n", encoding="utf-8") + (self.repo / "src" / "new.rs").write_text("pub fn added() {}\n", encoding="utf-8") + + result = subprocess.run( + [ + str(MULTIAGENT), + "snapshot", + "--root", + str(self.repo), + "--format", + "json", + ], + cwd=self.repo, + env=self.env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + payload = json.loads(result.stdout) + self.assertEqual(payload["changed_paths"], ["src/new.rs"]) + self.assertNotIn(".multiagent", json.dumps(payload)) + def test_dag_concurrent_node_updates_do_not_lose_rows(self): self.run_cli("dag", "init", "WF-DAG-CONCURRENT", "--title", "Concurrent DAG") processes = [] @@ -471,8 +498,8 @@ def test_dag_concurrent_node_updates_do_not_lose_rows(self): "worker-{:02d}".format(index), "--assignment-id", "A-{:02d}".format(index), - "--role", - "qa", + "--responsibility", + "repair-component-{:02d}".format(index), "--branch", "worker/{:02d}".format(index), "--owned", @@ -496,6 +523,10 @@ def test_dag_concurrent_node_updates_do_not_lose_rows(self): with (dag_dir / "nodes.tsv").open(encoding="utf-8", newline="") as handle: rows = list(csv.DictReader(handle, delimiter="\t")) self.assertEqual(len(rows), 12) + self.assertEqual( + {row["responsibility"] for row in rows}, + {"repair-component-{:02d}".format(index) for index in range(12)}, + ) self.assertEqual( {row["node_id"] for row in rows}, {"NODE-{:02d}".format(index) for index in range(12)}, diff --git a/tests/test_swe_outcomes.py b/tests/test_swe_outcomes.py index b980778..0f0e1e1 100644 --- a/tests/test_swe_outcomes.py +++ b/tests/test_swe_outcomes.py @@ -50,6 +50,22 @@ def _install_evalscope_stubs() -> None: class NativeOutcomeTest(unittest.TestCase): + def test_solver_timeout_reserves_only_orderly_shutdown_by_default(self): + with mock.patch.dict( + evalscope_multiagent_native_runner.os.environ, + {}, + clear=True, + ): + self.assertEqual(evalscope_multiagent_native_runner.solver_internal_timeout(3600), 3420) + + def test_solver_timeout_reserve_remains_configurable(self): + with mock.patch.dict( + evalscope_multiagent_native_runner.os.environ, + {"EVAL_NATIVE_SOLVER_TIMEOUT_RESERVE": "600"}, + clear=True, + ): + self.assertEqual(evalscope_multiagent_native_runner.solver_internal_timeout(3600), 3000) + def test_autonomous_authority_does_not_reopen_explicit_task_behavior(self): root = Path(__file__).resolve().parents[1] reviewer = (root / "prompts/roles/decision-authority-reviewer.md").read_text( @@ -158,7 +174,7 @@ def test_active_workflow_phase_reads_persisted_lifecycle(self): with mock.patch.object(swe_prod_lifecycle, "RUNTIME_ROOT", runtime): self.assertEqual(swe_prod_lifecycle.active_workflow_phase(), "implementation") - def test_incomplete_workflow_is_resumed_before_workspace_handoff(self): + def test_incomplete_workflow_keeps_resuming_before_workspace_handoff(self): completed = SimpleNamespace(returncode=0, stdout="codex-cli 1.0\n", stderr="") with tempfile.TemporaryDirectory() as directory: @@ -184,7 +200,9 @@ def test_incomplete_workflow_is_resumed_before_workspace_handoff(self): "restore_workspace_owner": mock.DEFAULT, "tmux_has_session": mock.Mock(return_value=True), "tmux_has_orchestrator": mock.Mock(return_value=False), - "active_workflow_phase": mock.Mock(side_effect=["implementation", "complete"]), + "active_workflow_phase": mock.Mock( + side_effect=["implementation", "implementation", "implementation", "implementation", "complete"] + ), "materialize_committed_changes": mock.DEFAULT, "mark_untracked_intent_to_add": mock.DEFAULT, } @@ -214,9 +232,10 @@ def test_incomplete_workflow_is_resumed_before_workspace_handoff(self): and str(call.args[0][0]).endswith("launch.sh") ] - self.assertEqual(len(launch_calls), 2) + self.assertEqual(len(launch_calls), 5) self.assertNotIn("--resume", launch_calls[0].args[0]) - self.assertIn("--resume", launch_calls[1].args[0]) + for call in launch_calls[1:]: + self.assertIn("--resume", call.args[0]) def test_shard_problem_statement_uses_relative_sample_id(self): with tempfile.TemporaryDirectory() as directory: @@ -267,6 +286,7 @@ def test_orchestrator_exit_prepares_workspace_for_official_scorer(self): "restore_workspace_owner": mock.DEFAULT, "tmux_has_session": mock.Mock(return_value=True), "tmux_has_orchestrator": mock.Mock(return_value=False), + "active_workflow_phase": mock.Mock(return_value="complete"), "materialize_committed_changes": mock.DEFAULT, "mark_untracked_intent_to_add": mock.DEFAULT, } @@ -314,6 +334,63 @@ def test_orchestrator_exit_prepares_workspace_for_official_scorer(self): str(swe_prod_lifecycle.ROLE_CODEX_HOME_ROOT), ) + def test_incomplete_workflow_at_deadline_is_not_handed_off(self): + completed = SimpleNamespace(returncode=0, stdout="codex-cli 1.0\n", stderr="") + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + prompt = root / "prompt.md" + prompt.write_text("prompt", encoding="utf-8") + lifecycle_patches = { + "require_path": mock.DEFAULT, + "multiagent_command": mock.Mock(return_value=["multiagent"]), + "find_codex_cli": mock.Mock(return_value="/usr/bin/codex"), + "git_head": mock.Mock(return_value="a" * 40), + "list_untracked_files": mock.Mock(return_value=[]), + "run": mock.Mock(return_value=completed), + "write_codex_bridge": mock.DEFAULT, + "write_apply_patch_helper": mock.DEFAULT, + "write_rg_fallback": mock.DEFAULT, + "read_prompt": mock.Mock(return_value="public task"), + "read_task_metadata": mock.Mock(return_value={}), + "make_prompt": mock.Mock(return_value=prompt), + "toolchain_path_prefixes": mock.Mock(return_value=[]), + "ensure_cache_dir": mock.Mock(return_value=str(root)), + "prepare_role_filesystem": mock.DEFAULT, + "restore_workspace_owner": mock.DEFAULT, + "tmux_has_session": mock.Mock(return_value=True), + "tmux_has_orchestrator": mock.Mock(return_value=True), + "active_workflow_phase": mock.Mock(return_value="implementation"), + "materialize_committed_changes": mock.DEFAULT, + "mark_untracked_intent_to_add": mock.DEFAULT, + } + with mock.patch.multiple(swe_prod_lifecycle, **lifecycle_patches): + with mock.patch.object( + swe_prod_lifecycle.shutil, + "which", + side_effect=lambda name: "/usr/bin/tmux" if name == "tmux" else None, + ): + with mock.patch.object( + swe_prod_lifecycle.time, + "monotonic", + side_effect=[0, 1, 61, 61], + ): + with mock.patch.object(swe_prod_lifecycle.time, "sleep"): + with mock.patch.dict( + swe_prod_lifecycle.os.environ, + { + "EVAL_CODEX_AUTH_MODE": "bridge", + "OPENAI_BASE_URL": "http://127.0.0.1:1/v1", + "OPENAI_API_KEY": "test-key", + }, + ): + with self.assertRaisesRegex(RuntimeError, "refusing workspace handoff"): + swe_prod_lifecycle.run_prod_solver(None, root, root, 60) + + swe_prod_lifecycle.restore_workspace_owner.assert_called_once_with(root) + swe_prod_lifecycle.materialize_committed_changes.assert_not_called() + swe_prod_lifecycle.mark_untracked_intent_to_add.assert_not_called() + def test_workspace_handoff_includes_new_source_and_test_files(self): with tempfile.TemporaryDirectory() as directory: repo = Path(directory) @@ -369,6 +446,33 @@ def test_workspace_handoff_excludes_preexisting_image_residue(self): self.assertEqual(exposed, ["new_source.py"]) self.assertNotIn("appendonly.aof", diff) + def test_workspace_handoff_excludes_framework_control_plane_files(self): + with tempfile.TemporaryDirectory() as directory: + repo = Path(directory) + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + internal = repo / ".multiagent" / "subagents" / "reviewer" + internal.mkdir(parents=True) + (internal / "status").write_text("missing\n", encoding="utf-8") + (repo / "feature.py").write_text("fixed = True\n", encoding="utf-8") + + exposed = swe_prod_repository.mark_untracked_intent_to_add(repo) + diff = subprocess.run( + ["git", "diff", "--binary"], + cwd=repo, + check=True, + text=True, + stdout=subprocess.PIPE, + ).stdout + + self.assertEqual(exposed, ["feature.py"]) + self.assertIn("feature.py", diff) + self.assertNotIn(".multiagent", diff) + + def test_framework_path_normalization_remains_python38_compatible(self): + self.assertTrue(swe_prod_repository.is_framework_internal_path("./.multiagent/status")) + source = Path(swe_prod_repository.__file__).read_text(encoding="utf-8") + self.assertNotIn(".removeprefix(", source) + def test_summary_counts_submitted_patch_even_when_official_score_is_zero(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory)