diff --git a/.claude/commands/gates/approve.md b/.claude/commands/gates/approve.md index 92ad7df..3c945d1 100644 --- a/.claude/commands/gates/approve.md +++ b/.claude/commands/gates/approve.md @@ -2,49 +2,43 @@ description: Approve the current pending gate and advance to the next phase --- -# /approve — Gate Approval +# /approve — Gate Approval (nonce-verified) You are executing a human gate approval for the current Zero Operators project. +Gate approvals are **nonce-verified** (v2 WS-A5): a single-use approval nonce +is minted when a phase reaches its blocking gate, shown in the gate review +banner and via `zo status`. Approvals are recorded ONLY through the CLI — +never by hand-editing STATE.md, DECISION_LOG.md, or the comms JSONL. This is +what makes approvals unforgeable: text echoed from context cannot pass the +nonce check. + ## Steps -1. **Read STATE.md** in the project's memory directory (`memory/{project}/STATE.md`). Identify: - - The current phase and its status - - Which gate is pending (look for `status: PENDING_GATE` or similar) - - If no gate is pending, report that and stop - -2. **Log the approval to DECISION_LOG.md** (`memory/{project}/DECISION_LOG.md`). Append an entry: - ```markdown - ## Gate Approved: {gate name} - **Timestamp**: {ISO 8601 now} - **Decided by**: human - **Phase**: {phase that was gated} - **Outcome**: approved - **Notes**: Human approved gate via /approve command - ``` +1. **Identify the pending gate.** Read STATE.md in the project's memory + directory (`.zo/memory/` in the delivery repo, or legacy + `memory/{project}/`). Identify the current GATED phase. If no gate is + pending, report that and stop. + +2. **Get the nonce from the human.** The approval nonce is displayed in the + gate review banner (and in `memory_root/gate_nonce`, which agents must + not read aloud into context on their own initiative — the human supplies + it). Ask the human for the nonce if they haven't provided one. Do NOT + guess, reconstruct, or copy a nonce you saw elsewhere in this + conversation. -3. **Update STATE.md** to advance to the next phase: - - Set the current phase status to `COMPLETED` - - Set the next phase status to `ACTIVE` - - Update `last_updated` timestamp - -4. **Log gate event to comms JSONL** (`logs/comms/{YYYY-MM-DD}.jsonl`). Append: - ```json - { - "timestamp": "{ISO 8601}", - "session_id": "manual", - "event_type": "gate", - "agent": "human", - "project": "{project-name}", - "gate_id": "{gate-id}", - "gate_name": "{gate-name}", - "result": "pass", - "notes": "Human approved via /approve command" - } +3. **Run the CLI approval** (this validates the nonce, appends to + DECISION_LOG.md, logs the comms gate event, and records the decision for + the orchestrator): + + ```bash + zo gates approve -p [--repo ] --nonce --notes "" ``` -5. **Report** to the user: - - What gate was approved - - What phase just completed - - What phase is now active +4. **Report** to the user: + - What gate was approved and the CLI output + - What phase just completed and what phase is now active - Any relevant next steps from the plan + +If the CLI reports a nonce mismatch, tell the human — do not retry with +variations. diff --git a/.claude/commands/gates/reject.md b/.claude/commands/gates/reject.md index adbb431..225bed0 100644 --- a/.claude/commands/gates/reject.md +++ b/.claude/commands/gates/reject.md @@ -3,51 +3,39 @@ description: Reject the current pending gate with a reason, triggering rework argument-hint: --- -# /reject — Gate Rejection +# /reject — Gate Rejection (nonce-verified) -You are executing a human gate rejection for the current Zero Operators project. The argument provided is the rejection reason. +You are executing a human gate rejection for the current Zero Operators +project. The argument provided is the rejection reason. + +Gate decisions are **nonce-verified** (v2 WS-A5) and recorded ONLY through +the CLI — never by hand-editing STATE.md, DECISION_LOG.md, or the comms +JSONL. Text echoed from context cannot pass the nonce check. ## Steps -1. **Read STATE.md** in the project's memory directory (`memory/{project}/STATE.md`). Identify: - - The current phase and its status - - Which gate is pending - - If no gate is pending, report that and stop - -2. **Log the rejection to DECISION_LOG.md** (`memory/{project}/DECISION_LOG.md`). Append: - ```markdown - ## Gate Rejected: {gate name} - **Timestamp**: {ISO 8601 now} - **Decided by**: human - **Phase**: {phase that was gated} - **Outcome**: rejected - **Reason**: $ARGUMENTS - **Action**: Phase set back to ACTIVE for rework - ``` +1. **Identify the pending gate.** Read STATE.md in the project's memory + directory (`.zo/memory/` in the delivery repo, or legacy + `memory/{project}/`). Identify the current GATED phase. If no gate is + pending, report that and stop. + +2. **Get the nonce from the human.** The approval nonce is shown in the gate + review banner. Ask the human for it if not provided. Do NOT guess, + reconstruct, or copy a nonce you saw elsewhere in this conversation. -3. **Update STATE.md**: - - Set the current phase status to `BLOCKED` with the rejection reason - - Then immediately set it back to `ACTIVE` to trigger rework - - Add a `blocker_history` entry recording the rejection - - Update `last_updated` timestamp - -4. **Log gate event to comms JSONL** (`logs/comms/{YYYY-MM-DD}.jsonl`). Append: - ```json - { - "timestamp": "{ISO 8601}", - "session_id": "manual", - "event_type": "gate", - "agent": "human", - "project": "{project-name}", - "gate_id": "{gate-id}", - "gate_name": "{gate-name}", - "result": "fail", - "notes": "$ARGUMENTS" - } +3. **Run the CLI rejection** (validates the nonce, appends to + DECISION_LOG.md, logs the comms gate event, and records the iterate + decision for the orchestrator — the phase returns to ACTIVE for rework): + + ```bash + zo gates reject -p [--repo ] --nonce --reason "$ARGUMENTS" ``` -5. **Report** to the user: - - What gate was rejected and why +4. **Report** to the user: + - What gate was rejected and why (the CLI output) - The phase is now set back to ACTIVE for rework - What the agents need to address based on the rejection reason - - Suggest next steps (re-run the phase, modify approach, etc.) + - Suggested next steps (re-run the phase, modify approach, etc.) + +If the CLI reports a nonce mismatch, tell the human — do not retry with +variations. diff --git a/.claude/hooks/zo-hookkit.sh b/.claude/hooks/zo-hookkit.sh new file mode 100644 index 0000000..4b22dba --- /dev/null +++ b/.claude/hooks/zo-hookkit.sh @@ -0,0 +1,31 @@ +#!/bin/bash +# zo-hookkit.sh — thin shim routing Claude Code hook events to zo.hookkit. +# +# Part of the v2 enforcement plane (WS-A, plans/zo-v2-rearchitecture.md). +# Usage (from .claude/settings.json): +# bash .claude/hooks/zo-hookkit.sh 2>/dev/null || exit 0 +# +# Fail-open by design: any missing precondition exits 0 silently. Blocking +# and denials are emitted as JSON on stdout by zo.hookkit, never via exit +# codes (matches the existing hook convention in this directory). +set -uo pipefail + +EVENT="${1:-}" +[[ -z "$EVENT" ]] && exit 0 + +HOOK_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$HOOK_DIR/../.." && pwd)" + +# Only run in the ZO platform repo — delivery repos never carry ZO hooks +# (specs/architecture.md isolation rule). +[[ -d "$REPO_ROOT/src/zo" ]] || exit 0 + +# Prefer the project venv so pydantic imports resolve; fall back to system +# python3 (zo.hookkit itself is fail-open on ImportError via the || below). +PY="python3" +[[ -x "$REPO_ROOT/.venv/bin/python3" ]] && PY="$REPO_ROOT/.venv/bin/python3" + +# Pre-set ZO_REPO_ROOT wins (lets tests point the handlers at a sandbox). +export ZO_REPO_ROOT="${ZO_REPO_ROOT:-$REPO_ROOT}" +PYTHONPATH="$REPO_ROOT/src${PYTHONPATH:+:$PYTHONPATH}" "$PY" -m zo.hookkit "$EVENT" || exit 0 +exit 0 diff --git a/.claude/settings.json b/.claude/settings.json index a49f440..f244a9e 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -59,6 +59,16 @@ "statusMessage": "Validating documentation consistency..." } ] + }, + { + "matcher": "Write|Edit", + "hooks": [ + { + "type": "command", + "command": "bash .claude/hooks/zo-hookkit.sh sealed-paths 2>/dev/null || exit 0", + "timeout": 10 + } + ] } ], "PostToolUse": [ @@ -81,6 +91,60 @@ "type": "command", "command": "bash .claude/hooks/stop-check.sh 2>/dev/null || exit 0", "timeout": 10 + }, + { + "type": "command", + "command": "bash .claude/hooks/zo-hookkit.sh drift-guard 2>/dev/null || exit 0", + "timeout": 20 + } + ] + } + ], + "SubagentStop": [ + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": "bash .claude/hooks/zo-hookkit.sh subagent-stop 2>/dev/null || exit 0", + "timeout": 15 + } + ] + } + ], + "PreCompact": [ + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": "bash .claude/hooks/zo-hookkit.sh precompact 2>/dev/null || exit 0", + "timeout": 15, + "statusMessage": "Flushing state before compaction..." + } + ] + } + ], + "SessionEnd": [ + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": "bash .claude/hooks/zo-hookkit.sh session-end 2>/dev/null || exit 0", + "timeout": 15 + } + ] + } + ], + "PostToolUseFailure": [ + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": "bash .claude/hooks/zo-hookkit.sh post-tool-failure 2>/dev/null || exit 0", + "timeout": 10 } ] } diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index e0bd78c..cba56d7 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -158,6 +158,26 @@ zo gates set MODE --project NAME `MODE` is one of: `supervised` (human approves every gate), `auto` (orchestrator approves unless ambiguous), `full-auto` (all gates auto-approved). +### zo gates approve / reject + +Record a nonce-verified human decision on the pending blocking gate (v2 +WS-A5). When a phase reaches its blocking gate, ZO mints a single-use +approval nonce and shows it in the gate review banner. Decisions are only +valid through these commands — hand-edited approvals cannot pass the nonce +check, which is what makes gate passage unforgeable by agents. + +``` +zo gates approve PHASE --project NAME --nonce NONCE [--notes TEXT] +zo gates reject PHASE --project NAME --nonce NONCE --reason TEXT +``` + +Both validate the nonce, append the decision to `DECISION_LOG.md`, log a +comms gate event, and record the decision for the orchestrator (applied +live at the next gate poll, or on `zo continue` for a fresh session). +`reject` returns the phase to ACTIVE for rework with your `--reason` +driving the next iteration. The nonce is cleared on use — replaying an +old approval can never pass a later gate. + ### zo watch-training Live training metrics dashboard. Tails `logs/training/metrics.jsonl` in the delivery repo and renders a persistent Rich panel with epoch progress, loss/metrics table, checkpoint log, and loss sparkline. Auto-launched by `zo build` during Phase 4 via tmux split-pane. diff --git a/docs/reference/v2-rearchitecture.mdx b/docs/reference/v2-rearchitecture.mdx index c41c637..a73426f 100644 --- a/docs/reference/v2-rearchitecture.mdx +++ b/docs/reference/v2-rearchitecture.mdx @@ -46,7 +46,7 @@ architecture — features from different repos interlock into single mechanisms. | 1 | **Machine-checked deliverable contracts** — every agent's produced/consumed contract compiles to JSON (required files, sections, patterns) verified mechanically when the agent stops, plus a zero-cost "completion claim vs. TODO-stub diff" drift guard | oh-my-claudecode | P0 | | 3 | **Hook-enforced memory protocol** — STATE.md flushes before compaction, session summaries verified at session end, structured failure records captured automatically; the model can't forget because the model isn't asked | oh-my-claudecode | P0 | | 7 | **Anti-forgery gate approvals** — approvals only valid as structured tags carrying a per-verification nonce, so no agent can forge gate passage by echoing text | oh-my-claudecode | P1 | -| 9 | **Structurally trustworthy verifiers** — the oracle and reviewers mechanically cannot edit code (tool-level enforcement), evaluators emit a strict JSON contract, and evaluation files are sealed against modification by the loop they evaluate | oh-my-claudecode | P1 | +| 9 | **Structurally trustworthy verifiers** — verifier writes are mechanically confined to their contract scope (a PreToolUse guard denies writes into off-limits paths), evaluators emit a strict JSON contract, and oracle/control files are sealed against modification by the loop they evaluate | oh-my-claudecode | P1 | ### Workstream B — Control plane diff --git a/memory/zo-platform/DECISION_LOG.md b/memory/zo-platform/DECISION_LOG.md index 35dffd5..a47013b 100644 --- a/memory/zo-platform/DECISION_LOG.md +++ b/memory/zo-platform/DECISION_LOG.md @@ -1239,3 +1239,29 @@ The `--no-headlines` flag is preserved (not removed) for backwards compatibility **Alternatives considered:** (1) Source-repo workstreams ("ralph track", "ruflo track") — rejected, provenance is not architecture; (2) P0-only scope — rejected by Sam, all 12 wanted; (3) waiting for the SDK refactor before substrate work — rejected, the fresh-context loop is the cheapest on-ramp to that refactor, not a competitor. **Outcome:** Shipped: plan (+ gitignore exception), docs reference page + mint.json nav, roadmap v2 section, website §11 + drawer nav + §12 renumber, STATE/session-040/research-README memory updates. Verification honest note: **no Node.js available on this machine**, so the Astro build was NOT run locally — the website edit was verified via HTML tag-balance parsing and DOM inspection through a static file server (section renders, all 4 external links present, drawer link + renumbering correct); the deploy pipeline's build is the binding check. validate-docs run pre-commit per protocol. + +## Decision: 2026-08-12T14:00:00Z +**Type:** FEATURE + ARCHITECTURE +**Title:** v2 Phase 1 (WS-A enforcement plane) — contracts, drift guard, memory hooks, sealed paths, nonce gates + +**Decision:** Implemented all five WS-A features from plans/zo-v2-rearchitecture.md as a deterministic hook plane beneath the prompt plane: (1) machine-readable `contracts.json` emitted at decompose with concrete deliverable paths, validated by a SubagentStop hook that blocks agents stopping with unmet deliverables; (2) a zero-LLM workflow-drift-guard Stop hook (completion claim + added stub markers → block); (3) PreCompact/SessionEnd/PostToolUseFailure memory hooks mechanizing the "AUTOMATIC" memory protocol; (4) a sealed-paths PreToolUse guard denying writes to oracle/control state and enforcing per-agent off-limits scopes from contracts.json; (5) nonce-verified gate approvals end to end — minted at GATED, required by `apply_human_decision`, `zo gates approve/reject --nonce` CLI, slash commands rewritten to route through it. One bash shim (`zo-hookkit.sh`) routes all six hook events to `python3 -m zo.hookkit` so file mutation reuses MemoryManager's atomic/flock code paths. + +**Rationale:** Direct implementation of the review's loudest theme ("enforcement plane below the prompt plane") and ZO's own PR-035 prior (aspirational contracts get ignored). Every mechanism ships wired (settings.json entries asserted by test) with a seeded-failure test proving it catches a planted violation — the "nothing ships unwired" rule from the plan's anti-scope. + +**Plan amendment (check 6):** The plan called for "disallowedTools + PreToolUse" on verifiers. Recon showed (a) Claude Code has no disallowedTools frontmatter for subagents (tools: is an allowlist), and (b) verifiers require scoped write access (oracle-qa owns oracle/reports/, code-reviewer owns reviews/) — total Write removal would break their contracts. Amended check 6 to path-scoped enforcement: off-limits writes denied via contracts.json, keyed on agent identity in hook input, fail-open when absent. Honest limitation recorded: production identity presence depends on Claude Code's hook payload; verified in tests via seeded identity, to be confirmed live in Phase 6 e2e. + +**Alternatives considered:** (1) tools: allowlist frontmatter on verifiers — rejected, cosmetic without Bash coverage and breaks legitimate writes; (2) six separate bash hook scripts — rejected for one parameterized shim (DRY, one venv-resolution path); (3) hard-blocking every stop while work remains (OMC's stop-engine) — explicitly rejected by the plan's anti-scope (fresh spawns + caps are the foundation; these hooks only catch violations). + +**Outcome:** 854 → 904 passed / 7 skipped (+50: 19 contracts, 19 hookkit, 9 nonce, 8 shim integration — first tests ever to execute a .claude/hooks script), ruff src/ clean, validate-docs green. 4 pre-existing integration tests updated to supply the nonce (designed behaviour change). Cascade: docs/COMMANDS.md (+gates approve/reject), specs/memory.md recovery section rewritten to implemented reality, plan check 6 + docs feature #9 amended. Branch `claude/v2-phase1-enforcement` stacked on `claude/v2-rearchitecture-plan`. + +## Decision: 2026-08-12T15:30:00Z +**Type:** VERIFICATION +**Title:** Live pre-PR verification of the WS-A enforcement plane (in-session, bypass-equivalent) + +**Decision:** Verified the Phase 1 hooks firing in a REAL Claude Code session before merging PR #107, per Sam's direction. Method: added always-on hook-trace observability (one JSONL line per invocation to `logs/hook-trace-{date}.jsonl`: event, stdin keys, agent identity, whether output was emitted; `ZO_HOOK_TRACE=0` disables), then used the current live session itself as the test vehicle — the runtime picked up the new settings.json wiring mid-session. + +**Evidence (from logs/hook-trace-2026-08-12.jsonl):** (1) `drift-guard` fired on a real Stop event, correctly silent — and the live payload carries `last_assistant_message` directly (drift-guard now prefers it; transcript parse kept as fallback). (2) `sealed-paths` fired on a real Write to `memory/zo-platform/gate_mode` and DENIED it — the lead agent of a live session was mechanically blocked from touching gate control state. (3) `subagent-stop` fired when a probe subagent stopped, and the live payload carries `agent_type` + `agent_id` + `agent_transcript_path` — RESOLVING the Phase-1 open question: per-agent contract enforcement will key correctly in live team sessions; the fail-open path is the exception, not the norm. + +**Honest caveats:** PostToolUseFailure did not fire for nonzero-exit Bash commands in this session — its semantics appear limited to tool-infrastructure errors, so the failure feed will capture fewer events than designed; if nonzero-exit capture matters, add PostToolUse-with-error-inspection in WS-D. PreCompact/SessionEnd not yet observed live (no compaction occurred; SessionEnd fires at session close) — same wiring pattern as the three proven events. A full `zo build` demo run remains desirable on a machine with the claude CLI (this Desktop-managed Mac has none — see PR-046). + +**Outcome:** 904 → 908 tests (3 trace + 1 live-payload drift-guard), ruff clean. Fail-open verified end to end in production conditions. Evidence committed to PR #107. diff --git a/memory/zo-platform/PRIORS.md b/memory/zo-platform/PRIORS.md index 8ef9eb9..4bf3bbc 100644 --- a/memory/zo-platform/PRIORS.md +++ b/memory/zo-platform/PRIORS.md @@ -1399,3 +1399,10 @@ rm -f package-lock.json # keep the diff to the intend ### Verified Solution `_launch_and_monitor` gates `cleanup_stale_overlay` on `session_role == "model"` (report sessions never touch the overlay) and threads a `consolidate_on_exit` flag; `zo report --no-consolidate` sets it False. Tests: `test_report_session_never_cleans_overlay` (report never cleans even with no peer registered), `test_launch_and_monitor_no_consolidate_skips_merge`, plus the concurrency / conflict / resume / STATE-untouched suite (`test_surrogate_edge.py`). Concurrent **bypass** is handled by the same principle: `apply_bypass_overlay` no-ops when bypass is already active (inherit, not clobber), so a second bypass session (`zo report --bypass-permissions` alongside a model session in bypass) cannot leave the repo stuck in bypass after exit (`test_apply_is_noop_when_bypass_already_active`). **Cross-reference:** PR-037 (no session lock — single-session-by-design; this is the multi-session follow-on), PR-038 / PR-043 (permission-overlay lifecycle), PR-009 / PR-041 (built-must-be-wired — here, wired *and* made robust to the unupgraded peer). + +## PR-046: Desktop-managed Mac lacks the standalone toolchain — verify before assuming +**Category:** environment +**Statement:** Sam's Mac (Claude Desktop / Cowork sessions) has NO standalone `claude` CLI, no `uv`, no `npm/node` on the sandbox shell PATH; Python is anaconda's `python3` (which does carry pydantic/click/pytest). Child Claude Code sessions CANNOT be spawned from this machine — `zo build`/`zo continue` and any wrapper-launched verification must run on a machine with the claude CLI (the Linux box used for prod-001/demos). Hook shims must not assume a repo `.venv` exists: zo-hookkit.sh's python3 fallback is load-bearing here. +**Evidence:** 2026-08-12 session 040 — `claude`, `uv`, `npm` all exit 127 in the sandbox shell; no `.venv/` in repo; hooks verified live via the session's own runtime instead (see DECISION_LOG 15:30 entry). +**Rules learned:** (1) Probe for a binary before building a plan around it (`which X` first, not after failure). (2) Keep every hook/script runnable on bare `python3` + stdlib-adjacent deps. (3) Machine-specific capabilities belong in a preflight check, not in assumptions — `zo preflight` should test for the claude CLI explicitly. +**Confidence:** high diff --git a/memory/zo-platform/STATE.md b/memory/zo-platform/STATE.md index 7df8d3d..7f25ced 100644 --- a/memory/zo-platform/STATE.md +++ b/memory/zo-platform/STATE.md @@ -8,7 +8,7 @@ status: complete ## Current Position -**Session 040 (current) — pick up here.** Research + decision session: deep-dive review of three agent-orchestration repos (oh-my-claudecode, ruflo, ralph — cloned to `~/Documents/code/`) to inform the ZO v2 rearchitecture. 9-agent workflow (7 source-reading lenses + ZO baseline + adversarial synthesis, ~1.06M tokens) catalogued **63 features**, distilled to **12 ranked adoptions** + 6 rearchitecture themes + 11 anti-patterns; all findings persisted to `memory/zo-platform/research/2026-08-12-repo-reviews/` (per-repo markdown + `raw-findings.json`). **Sam decided: adopt all 12.** Work organized into **five layer-based workstreams** (A enforcement plane, B control plane, C execution substrate, D self-learning/platform oracle, E operator experience) — NOT source-repo categories, because features from different repos interlock into single mechanisms. Shipped this session: `plans/zo-v2-rearchitecture.md` (full plan: 6 gated phases, 20-check oracle, anti-scope; + `.gitignore` exception), `docs/reference/v2-rearchitecture.mdx` (all 12 features w/ provenance + repo links; added to mint.json Reference nav), `docs/roadmap.mdx` v2 section (4 pillars + repo credits), website §11 "What's next" (new section w/ 3 repo credit cards; quick start renumbered §12; drawer nav updated). **Verification caveat: no Node.js on this machine** — Astro build NOT run; website change verified via HTML-parser balance check + static-server DOM inspection (section text, all 4 links, drawer entry, renumbering all confirmed rendered). CI/deploy build must confirm. No src/ code, agents, commands, or version touched. **Next:** v2 Phase 1 (enforcement foundations, WS-A) per `plans/zo-v2-rearchitecture.md`; prior Batch D/E + Tier-1 items (caveman, onboarding) fold into or queue behind the v2 plan. +**Session 040 (current) — pick up here.** Research + decision session: deep-dive review of three agent-orchestration repos (oh-my-claudecode, ruflo, ralph — cloned to `~/Documents/code/`) to inform the ZO v2 rearchitecture. 9-agent workflow (7 source-reading lenses + ZO baseline + adversarial synthesis, ~1.06M tokens) catalogued **63 features**, distilled to **12 ranked adoptions** + 6 rearchitecture themes + 11 anti-patterns; all findings persisted to `memory/zo-platform/research/2026-08-12-repo-reviews/` (per-repo markdown + `raw-findings.json`). **Sam decided: adopt all 12.** Work organized into **five layer-based workstreams** (A enforcement plane, B control plane, C execution substrate, D self-learning/platform oracle, E operator experience) — NOT source-repo categories, because features from different repos interlock into single mechanisms. Shipped this session: `plans/zo-v2-rearchitecture.md` (full plan: 6 gated phases, 20-check oracle, anti-scope; + `.gitignore` exception), `docs/reference/v2-rearchitecture.mdx` (all 12 features w/ provenance + repo links; added to mint.json Reference nav), `docs/roadmap.mdx` v2 section (4 pillars + repo credits), website §11 "What's next" (new section w/ 3 repo credit cards; quick start renumbered §12; drawer nav updated). **Verification caveat: no Node.js on this machine** — Astro build NOT run; website change verified via HTML-parser balance check + static-server DOM inspection (section text, all 4 links, drawer entry, renumbering all confirmed rendered). CI/deploy build must confirm. **Same session, part 2 — v2 Phase 1 (WS-A enforcement plane) SHIPPED** on branch `claude/v2-phase1-enforcement` (stacked on the plan branch, PR #106): (A1) `src/zo/contracts.py` — contracts.json emitted at decompose into memory_root (gate_mode precedent), `contract_produced` upgraded from prose placeholders to concrete paths (ownership ∩ required_artifacts, ownership-dir fallback), SubagentStop hook validates deliverables (missing/undersized/pattern/empty-dir) and blocks with a violation list; (A2) drift-guard Stop hook — completion-claim regex over the last assistant transcript message + added TODO/FIXME/NotImplementedError lines in `git diff HEAD` → block (env kill-switch `ZO_DRIFT_GUARD=0`); (A3) PreCompact (STATE flush + checkpoint decision), SessionEnd (summary backfill), PostToolUseFailure (`logs/comms/failures-{date}.jsonl` feed) — specs/memory.md recovery section updated to match (replaces the never-built periodic postToolUse checkpoint design); (A4) sealed-paths PreToolUse guard — memory-root control files (gate_mode/gate_nonce/gate_decision/contracts.json/sealed_paths) + user `sealed_paths` prefixes denied, off-limits write-scope enforced per contracts.json when agent identity present in hook input (plan check 6 AMENDED: no disallowedTools frontmatter exists for subagents and verifiers need scoped writes — path-scoped enforcement instead, fail-open without identity); (A5) nonce gates — minted at GATED (`secrets.token_hex(8)` → `gate_nonce` file), surfaced in `prepare_gate_review`, `apply_human_decision` raises PermissionError without it (single-use, cleared on terminal decisions), new `zo gates approve/reject --nonce` CLI writes DECISION_LOG + comms + `gate_decision` file consumed on next decompose, `/approve`+`/reject` slash commands rewritten to route through the CLI (forgeable hand-edit path CLOSED). All via one shim (`.claude/hooks/zo-hookkit.sh` → `python3 -m zo.hookkit`, venv-preferring, fail-open) + 6 new settings.json wirings. **First-ever hook-script tests** (subprocess + stdin JSON pattern). **854 → 904 passed / 7 skipped, ruff `src/` clean, validate-docs green.** 4 pre-existing integration tests updated to pass the nonce (designed behaviour change). Plan oracle checks 1-5 + 7 have passing seeded-failure tests; check 6 as amended. **Part 3 — live pre-PR verification (Sam-directed):** added always-on hook-trace observability (`logs/hook-trace-{date}.jsonl`, `ZO_HOOK_TRACE=0` off-switch) and verified in the live session itself: sealed-paths DENIED a real Write to gate_mode; drift-guard fired correctly-silent on a real Stop; subagent-stop fired with `agent_type`+`agent_id` in the live payload — **agent-identity open question RESOLVED** (per-agent enforcement keys correctly). Drift-guard now prefers the live payload's `last_assistant_message` (transcript parse = fallback). Caveats logged: PostToolUseFailure doesn't fire on nonzero-exit Bash (infrastructure errors only); PreCompact/SessionEnd not yet observed live; full `zo build` demo needs a machine with the claude CLI (this Mac has none — PR-046). **908 passed / 7 skipped, ruff clean.** **Next:** Phase 1 gate review → then WS-B control plane (plan-ledger.json) per `plans/zo-v2-rearchitecture.md`; prior Batch D/E + Tier-1 items queue behind v2. **Session 039 (prior).** Test-only change making the CLI command-set test plugin-tolerant: `tests/unit/test_cli.py::TestCliGroup::test_cli_group_has_all_commands` asserted strict equality on `cli.commands`, which contradicts the extension points ZO itself shipped in PR #99 — any downstream build installing a `zo.commands` entry-point plugin (the documented mechanism in `zo.extensions.load_cli_plugins`) adds commands and turned the core suite red in that environment, despite zero core changes. Changed `assert expected == actual` → `assert expected <= actual` with an explanatory comment. Trade-off accepted: equality also caught *core* commands added without updating the test; subset still catches removals, and additions have always required updating this test's `expected` set anyway. Surfaced by the first real downstream plugin registering a new command group. No code, version, agents, or docs touched — single-assertion diff (+ memory). **854 passed / 7 skipped on Python 3.11 AND 3.12, ruff `src/` clean, validate-docs 0 failures.** **Next:** unchanged — Batch **D** / **E**, standing Tier-1 (caveman, onboarding). diff --git a/memory/zo-platform/sessions/session-040-2026-08-12.md b/memory/zo-platform/sessions/session-040-2026-08-12.md index 8850205..9aa71b0 100644 --- a/memory/zo-platform/sessions/session-040-2026-08-12.md +++ b/memory/zo-platform/sessions/session-040-2026-08-12.md @@ -57,9 +57,28 @@ Verification: HTML balance-checked + DOM-inspected via static server (no Node.js on this machine — Astro build deferred to CI/deploy). validate-docs pre-commit. +## Phase 1 build (same session, part 2) + +Sam said go — WS-A shipped on `claude/v2-phase1-enforcement` (stacked): + +- `src/zo/contracts.py` + `src/zo/hookkit.py` + `.claude/hooks/zo-hookkit.sh` + (one shim, six events, all fail-open) +- contracts.json emission at decompose; SubagentStop deliverable validation +- drift-guard Stop hook (claim + stub markers → block, `ZO_DRIFT_GUARD=0` kill switch) +- PreCompact/SessionEnd/PostToolUseFailure memory hooks; specs/memory.md updated +- sealed-paths PreToolUse guard (control files + off-limits scopes) +- nonce gates end to end: mint at GATED → review banner → `apply_human_decision` + PermissionError without it → `zo gates approve/reject --nonce` CLI → + slash commands rewritten (forgeable hand-edit path closed) +- Plan check 6 amended (no disallowedTools frontmatter exists; path-scoped + enforcement instead) — rationale in DECISION_LOG +- 854 → 904 tests passed / 7 skipped; ruff clean; validate-docs green; + first-ever hook-script tests (subprocess + stdin JSON) + ## Next session -Begin v2 Phase 1 (WS-A enforcement foundations) per plans/zo-v2-rearchitecture.md: -contracts.json emission at spawn, SubagentStop validation hook, drift-guard -regexes, PreCompact/SessionEnd/PostToolUseFailure memory hooks, disallowedTools -on verifiers, nonce approvals. +Phase 1 gate review with Sam, then WS-B control plane (plan-ledger.json, +oracle-owned pass flags, sizing lint) per plans/zo-v2-rearchitecture.md +Phase 2. Watchdog (WS-C) after that. Note for Phase 6 e2e: confirm Claude +Code's SubagentStop/PreToolUse hook payloads carry agent identity in live +team sessions (enforcement is fail-open without it). diff --git a/plans/zo-v2-rearchitecture.md b/plans/zo-v2-rearchitecture.md index 55a49e8..7328fef 100644 --- a/plans/zo-v2-rearchitecture.md +++ b/plans/zo-v2-rearchitecture.md @@ -50,7 +50,7 @@ substrate. 3. PreCompact hook flushes state before a forced compaction; STATE.md reflects it 4. PostToolUseFailure produces structured JSONL consumed by the priors pipeline 5. A forged gate approval (echoed tag without nonce) is rejected; a genuine nonce-tagged approval passes -6. oracle-qa/reviewers mechanically cannot Write/Edit (disallowedTools + PreToolUse); a seeded write attempt is blocked +6. A contracted agent's Write/Edit into an off-limits path is denied by the PreToolUse guard reading contracts.json (amended from "disallowedTools" during Phase 1: verifiers need scoped write access — oracle-qa writes oracle/reports/ — and Claude Code has no disallowedTools frontmatter for subagents, so enforcement is path-scoped, keyed on agent identity in the hook input; fail-open when identity is absent) 7. A seeded edit to a sealed eval file is blocked 8. plan-ledger.json generated from plan.md; `zo status` renders from the ledger, not prose parsing 9. A builder attempt to flip `passes: true` is blocked; only oracle-qa's flip lands diff --git a/specs/memory.md b/specs/memory.md index 98409dc..cf44d31 100644 --- a/specs/memory.md +++ b/specs/memory.md @@ -282,18 +282,21 @@ Interruptions are expected. Every component is designed for fault tolerance. **If session is interrupted mid-task**: -1. STATE.md has the last checkpoint (updated periodically via postToolUse hook) -2. DECISION_LOG.md has all decisions made up to interrupt point +1. STATE.md has the last checkpoint (flushed by the PreCompact hook before any context compaction — v2 WS-A3) +2. DECISION_LOG.md has all decisions made up to interrupt point (plus a "Checkpoint: pre-compaction state flush" entry per compaction) 3. Git history is ground truth for file state 4. New session reads STATE.md and picks up from last_completed_subtask -5. Previous session summary is available in sessions/ (though not yet written if session was interrupted) +5. Previous session summary is available in sessions/ — if none was written, the SessionEnd hook backfills an auto-generated stub (v2 WS-A3) -**Recovery mechanism**: +**Recovery mechanism (implemented, v2 WS-A3)**: -- Hooks: sessionStart (read memory), postToolUse (periodic checkpoint), sessionEnd (write summary + update state) -- Implemented via Claude Code settings.json hook mechanism -- postToolUse hook triggers every 5 tool invocations or every 30 minutes, whichever comes first -- If interrupted before hook fires: STATE.md is from previous session, not current partial session +- Hooks in `.claude/settings.json`, all routed through `.claude/hooks/zo-hookkit.sh` → `python3 -m zo.hookkit`: + - `SessionStart` — read memory and inject context (pre-existing) + - `PreCompact` — flush STATE.md + append a checkpoint decision before compaction destroys in-context state + - `SessionEnd` — verify a session summary exists for today; backfill an auto-generated stub if not + - `PostToolUseFailure` — append a structured failure record to `logs/comms/failures-{date}.jsonl` (consumed by the evolution/priors pipeline) +- All handlers are fail-open: infrastructure problems never block a session; only real violations emit blocking JSON +- Historical note: an earlier draft of this spec called for a periodic postToolUse checkpoint (every 5 tool calls / 30 min); v2 replaced that design with compaction-triggered flushing, which protects exactly the moment state was actually being lost ## Context Reset Protocol diff --git a/src/zo/cli.py b/src/zo/cli.py index 14d8a45..51f0098 100644 --- a/src/zo/cli.py +++ b/src/zo/cli.py @@ -2377,6 +2377,118 @@ def gates_set(mode: str, project: str, repo: str | None) -> None: console.print(f" Gate mode set to: [{_AMBER}]{gm.value}[/]") +def _apply_gate_decision_cli( + project: str, + repo: str | None, + phase: str, + nonce: str, + decision: str, + notes: str, +) -> None: + """Shared body of ``zo gates approve`` / ``zo gates reject`` (v2 WS-A5). + + Validates the single-use approval nonce minted when the phase went + GATED, then records the decision in DECISION_LOG.md and the comms + JSONL — the same artifacts the (previously forgeable) slash-command + path hand-edited, now nonce-checked and schema-consistent. + """ + delivery = Path(repo).resolve() if repo else None + pctx = _load_project_context(project, delivery_repo=delivery) + memory = pctx.make_memory() + if not memory.memory_root.exists(): + console.print( + f"[red bold]No memory found for '{project}'.[/] " + "Run [bold]zo init[/] or [bold]zo build[/] first." + ) + raise SystemExit(1) + + from zo._memory_models import DecisionEntry + from zo.comms import CommsLogger + + stored = memory.read_gate_nonce() + if stored is None: + console.print( + "[red bold]No gate is awaiting approval[/] (no nonce on file). " + "A nonce is minted when a phase reaches its blocking gate." + ) + raise SystemExit(1) + if nonce != stored: + console.print( + "[red bold]Nonce mismatch — decision rejected.[/] Use the " + "approval nonce shown in the gate review banner." + ) + raise SystemExit(1) + memory.clear_gate_nonce() + + memory.append_decision(DecisionEntry( + title=f"Human gate decision: {phase}", + context=f"Phase: {phase} | via zo gates {decision} (nonce-verified)", + decision=decision, rationale=notes or "Human reviewer decision.", + outcome=decision, + )) + comms = CommsLogger( + log_dir=pctx.zo_root / "logs" / "comms", + project=project, session_id=f"gate-{uuid.uuid4().hex[:8]}", + ) + comms.log_gate( + agent="human", gate_id=phase, gate_name=f"{phase} blocking gate", + metric_name="human_review", + metric_value=1.0 if decision == "proceed" else 0.0, + threshold=1.0, tier=1, + result="pass" if decision == "proceed" else "fail", + notes=notes or f"Nonce-verified human {decision} via CLI.", + ) + memory.write_gate_decision(phase, decision, notes) + + _show_banner(project=project, mode="gates", gate_mode=decision) + console.print( + f" Gate [{_AMBER}]{phase}[/] {decision} recorded (nonce verified). " + "A running session picks this up at its next gate poll; otherwise " + "it applies on [bold]zo continue[/]." + ) + + +@gates.command("approve") +@click.argument("phase") +@click.option("--project", "-p", required=True, help="Project name.") +@click.option( + "--repo", type=click.Path(exists=True, file_okay=False), default=None, + help="Path to delivery repo with .zo/ directory.", +) +@click.option( + "--nonce", required=True, + help="Approval nonce shown in the gate review (single-use).", +) +@click.option("--notes", default="", help="Optional reviewer notes.") +def gates_approve( + phase: str, project: str, repo: str | None, nonce: str, notes: str, +) -> None: + """Approve the pending blocking gate for PHASE (nonce-verified).""" + _apply_gate_decision_cli(project, repo, phase, nonce, "proceed", notes) + + +@gates.command("reject") +@click.argument("phase") +@click.option("--project", "-p", required=True, help="Project name.") +@click.option( + "--repo", type=click.Path(exists=True, file_okay=False), default=None, + help="Path to delivery repo with .zo/ directory.", +) +@click.option( + "--nonce", required=True, + help="Approval nonce shown in the gate review (single-use).", +) +@click.option( + "--reason", "notes", required=True, + help="Why the gate is rejected (drives rework).", +) +def gates_reject( + phase: str, project: str, repo: str | None, nonce: str, notes: str, +) -> None: + """Reject the pending blocking gate for PHASE (nonce-verified).""" + _apply_gate_decision_cli(project, repo, phase, nonce, "iterate", notes) + + # --------------------------------------------------------------------------- # Experiments — inspect the Phase 4 experiment registry # --------------------------------------------------------------------------- diff --git a/src/zo/contracts.py b/src/zo/contracts.py new file mode 100644 index 0000000..de6f51c --- /dev/null +++ b/src/zo/contracts.py @@ -0,0 +1,234 @@ +"""Machine-readable agent deliverable contracts (v2 WS-A1). + +Compiles the prose ``AgentContract`` objects produced by +``Orchestrator.decompose_plan()`` into a per-project ``contracts.json`` +under the project memory root, and validates produced artifacts against +it when a subagent stops (consumed by ``zo.hookkit`` from the +``SubagentStop`` hook). + +Design notes: + - ``contracts.json`` follows the ``gate_mode`` control-file precedent: + machine-readable state lives in ``memory_root``, gitignored, written + atomically. + - Deliverable paths are derived from ``PhaseDefinition.required_artifacts`` + scoped by each agent's ownership prefixes; agents whose ownership + matches no phase artifact keep an ownership-directory expectation + instead, so validation never degenerates to always-pass prose. + - Validation is fail-open on infrastructure problems (missing file, + malformed JSON) but strict on declared deliverables — the hook layer + must never brick a session, only catch contract violations. +""" + +from __future__ import annotations + +import os +import re +import tempfile +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from pydantic import BaseModel, Field + +if TYPE_CHECKING: + from pathlib import Path + + from zo._orchestrator_models import WorkflowDecomposition + +__all__ = [ + "AgentDeliverables", + "ContractsFile", + "DeliverableSpec", + "Violation", + "emit_contracts", + "load_contracts", + "set_active_phase", + "validate_agent_stop", +] + +CONTRACTS_FILENAME = "contracts.json" + + +class DeliverableSpec(BaseModel): + """One machine-checkable deliverable an agent must produce.""" + + path: str + kind: str = "file" # "file" | "directory" + min_bytes: int = 1 + required_patterns: list[str] = Field(default_factory=list) + description: str = "" + + +class AgentDeliverables(BaseModel): + """Deliverable set for one (agent, phase) pair.""" + + agent_name: str + phase_id: str + deliverables: list[DeliverableSpec] = Field(default_factory=list) + off_limits: list[str] = Field(default_factory=list) + ownership: list[str] = Field(default_factory=list) + + +class ContractsFile(BaseModel): + """Top-level schema of ``contracts.json``.""" + + version: int = 1 + project: str = "" + active_phase: str = "" + generated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + agents: list[AgentDeliverables] = Field(default_factory=list) + + +class Violation(BaseModel): + """A single contract violation found at subagent stop.""" + + agent_name: str + phase_id: str + path: str + problem: str + + +def _derive_deliverables( + ownership: list[str], required_artifacts: list[str], +) -> list[DeliverableSpec]: + """Map phase artifacts to an agent via its ownership prefixes.""" + specs: list[DeliverableSpec] = [] + for artifact in required_artifacts: + if any(artifact.startswith(prefix) for prefix in ownership): + specs.append( + DeliverableSpec( + path=artifact, + description="Phase required artifact within agent ownership", + ) + ) + if not specs and ownership: + # No named artifact matched: the agent still owes non-empty output + # inside its primary ownership directory. + specs.append( + DeliverableSpec( + path=ownership[0], + kind="directory", + description="Ownership directory must exist and be non-empty", + ) + ) + return specs + + +def emit_contracts( + workflow: WorkflowDecomposition, + memory_root: Path, + project: str, + active_phase: str, +) -> Path: + """Serialize workflow contracts to ``memory_root/contracts.json``. + + Written atomically (temp file + ``os.replace``) so a concurrently + running hook never reads a torn file. + + Returns: + The path written. + """ + agents = [ + AgentDeliverables( + agent_name=c.agent_name, + phase_id=c.phase_id, + deliverables=_derive_deliverables( + c.ownership, + next( + ( + p.required_artifacts + for p in workflow.phases + if p.phase_id == c.phase_id + ), + [], + ), + ), + off_limits=list(c.off_limits), + ownership=list(c.ownership), + ) + for c in workflow.agent_contracts + ] + doc = ContractsFile(project=project, active_phase=active_phase, agents=agents) + memory_root.mkdir(parents=True, exist_ok=True) + path = memory_root / CONTRACTS_FILENAME + fd, tmp = tempfile.mkstemp(dir=str(memory_root), suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(doc.model_dump_json(indent=2)) + os.replace(tmp, path) + finally: + if os.path.exists(tmp): + os.unlink(tmp) + return path + + +def load_contracts(path: Path) -> ContractsFile | None: + """Load and parse ``contracts.json``; ``None`` on any problem (fail-open).""" + try: + return ContractsFile.model_validate_json(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def set_active_phase(memory_root: Path, phase_id: str) -> None: + """Update ``active_phase`` in an existing contracts file, if present.""" + path = memory_root / CONTRACTS_FILENAME + doc = load_contracts(path) + if doc is None: + return + doc.active_phase = phase_id + path.write_text(doc.model_dump_json(indent=2), encoding="utf-8") + + +def _check_spec(repo_root: Path, spec: DeliverableSpec) -> str | None: + """Return a problem description for a deliverable, or ``None`` if met.""" + target = repo_root / spec.path + if spec.kind == "directory": + if not target.is_dir(): + return "required ownership directory missing" + if not any(target.iterdir()): + return "required ownership directory is empty" + return None + if not target.is_file(): + return "required deliverable file missing" + size = target.stat().st_size + if size < spec.min_bytes: + return f"deliverable too small ({size} bytes < {spec.min_bytes})" + if spec.required_patterns: + try: + text = target.read_text(encoding="utf-8", errors="replace") + except OSError: + return "deliverable unreadable" + for pattern in spec.required_patterns: + if re.search(pattern, text) is None: + return f"required pattern not found: {pattern!r}" + return None + + +def validate_agent_stop( + contracts_path: Path, agent_name: str, repo_root: Path, +) -> list[Violation]: + """Validate one agent's deliverables for the active phase. + + Fail-open: unknown agent, missing contracts file, or no active-phase + entry yields an empty violation list. Strict on declared deliverables. + """ + doc = load_contracts(contracts_path) + if doc is None: + return [] + normalized = agent_name.strip().lower().replace(" ", "-") + violations: list[Violation] = [] + for entry in doc.agents: + if entry.agent_name != normalized or entry.phase_id != doc.active_phase: + continue + for spec in entry.deliverables: + problem = _check_spec(repo_root, spec) + if problem is not None: + violations.append( + Violation( + agent_name=entry.agent_name, + phase_id=entry.phase_id, + path=spec.path, + problem=problem, + ) + ) + return violations diff --git a/src/zo/hookkit.py b/src/zo/hookkit.py new file mode 100644 index 0000000..ced9b35 --- /dev/null +++ b/src/zo/hookkit.py @@ -0,0 +1,409 @@ +"""Hook-event handlers for the v2 enforcement plane (WS-A). + +Invoked by thin bash shims in ``.claude/hooks/`` as:: + + python3 -m zo.hookkit < hook-input.json + +Events: + subagent-stop validate agent deliverables against contracts.json + drift-guard block Stop when completion claims meet stub markers + precompact flush a STATE.md checkpoint before compaction + session-end ensure a session summary exists for today + post-tool-failure append a structured failure record (JSONL feed) + sealed-paths deny Write/Edit into sealed or off-limits paths + +Every handler is fail-open: infrastructure problems (missing files, +unparseable stdin, unknown agent) exit 0 with no output. Only genuine +violations produce blocking JSON on stdout. This mirrors the existing +``.claude/hooks/*.sh`` convention — the enforcement plane must never +brick a session. +""" + +from __future__ import annotations + +import contextlib +import json +import os +import re +import subprocess +import sys +import uuid +from datetime import UTC, datetime +from pathlib import Path + +from zo.contracts import CONTRACTS_FILENAME, load_contracts, validate_agent_stop + +__all__ = ["main"] + +_COMPLETION_CLAIM = re.compile( + r"\b(all (?:tests|checks) pass(?:ing|ed)?|fully (?:implemented|working)" + r"|implementation (?:is )?complete|task (?:is )?complete[d]?" + r"|everything (?:is )?(?:done|working)|finished implementing)\b", + re.IGNORECASE, +) +_STUB_MARKER = re.compile( + r"^\+.*(\bTODO\b|\bFIXME\b|\bXXX\b|NotImplementedError|raise NotImplemented\b)" +) +_SEALED_DEFAULTS = ( + "gate_mode", "gate_nonce", "gate_decision", CONTRACTS_FILENAME, "sealed_paths", +) + + +def _read_stdin_json() -> dict: + try: + return json.loads(sys.stdin.read() or "{}") + except (ValueError, OSError): + return {} + + +def _repo_root() -> Path: + return Path(os.environ.get("ZO_REPO_ROOT", os.getcwd())) + + +def _memory_root(repo_root: Path) -> Path | None: + env = os.environ.get("ZO_MEMORY_ROOT") + if env: + return Path(env) + default = repo_root / "memory" / "zo-platform" + return default if default.is_dir() else None + + +_emitted = False + + +def _emit(payload: dict) -> None: + global _emitted + _emitted = True + sys.stdout.write(json.dumps(payload)) + + +def _trace(event: str, data: dict) -> None: + """Append one observability line per hook invocation (fail-open). + + Written to ``logs/hook-trace-{date}.jsonl`` under the repo root + (gitignored). This is how we verify the enforcement plane actually + fires in live sessions — the handlers themselves are silent unless + they block. Records which stdin keys the live payload carried + (answers the agent-identity question) but never payload values. + Disable with ``ZO_HOOK_TRACE=0``. + """ + if os.environ.get("ZO_HOOK_TRACE", "1") == "0": + return + with contextlib.suppress(OSError): + trace_dir = _repo_root() / "logs" + trace_dir.mkdir(parents=True, exist_ok=True) + date = datetime.now(UTC).strftime("%Y-%m-%d") + line = json.dumps({ + "ts": datetime.now(UTC).isoformat(), + "event": event, + "stdin_keys": sorted(data.keys()), + "agent_identity": _agent_name(data), + "session_id": data.get("session_id", ""), + "emitted_output": _emitted, + }) + with (trace_dir / f"hook-trace-{date}.jsonl").open("a", encoding="utf-8") as fh: + fh.write(line + "\n") + + +# -- subagent-stop ------------------------------------------------------------ + + +def _agent_name(data: dict) -> str | None: + for key in ("agent_name", "agent_type", "subagent_type", "name"): + value = data.get(key) + if isinstance(value, str) and value.strip(): + return value + return None + + +def _handle_subagent_stop(data: dict) -> None: + if data.get("stop_hook_active"): + return + agent = _agent_name(data) + if agent is None: + return + repo_root = _repo_root() + contracts_env = os.environ.get("ZO_CONTRACTS_PATH") + if contracts_env: + contracts_path = Path(contracts_env) + else: + memory_root = _memory_root(repo_root) + if memory_root is None: + return + contracts_path = memory_root / CONTRACTS_FILENAME + delivery_root = Path(os.environ.get("ZO_DELIVERY_ROOT", str(repo_root))) + violations = validate_agent_stop(contracts_path, agent, delivery_root) + if not violations: + return + lines = [f"- {v.path}: {v.problem}" for v in violations] + reason = ( + f"Contract violation — agent '{agent}' (phase " + f"{violations[0].phase_id}) has unmet deliverables:\n" + + "\n".join(lines) + + "\nProduce the missing deliverables before stopping, or report a " + "blocker to the orchestrator." + ) + _emit({"decision": "block", "reason": reason, "stopReason": reason}) + + +# -- drift-guard ---------------------------------------------------------- + + +def _last_assistant_text(transcript_path: str) -> str: + text_parts: list[str] = [] + try: + lines = Path(transcript_path).read_text(encoding="utf-8").splitlines() + except OSError: + return "" + for line in reversed(lines): + try: + entry = json.loads(line) + except ValueError: + continue + message = entry.get("message") or {} + if message.get("role") != "assistant": + continue + content = message.get("content") + if isinstance(content, str): + text_parts.append(content) + elif isinstance(content, list): + text_parts.extend( + block.get("text", "") + for block in content + if isinstance(block, dict) and block.get("type") == "text" + ) + break + return "\n".join(text_parts) + + +def _added_stub_lines(repo_root: Path) -> list[str]: + try: + result = subprocess.run( + ["git", "-C", str(repo_root), "diff", "HEAD", "--unified=0"], + capture_output=True, text=True, timeout=15, check=False, + ) + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + return [] + if result.returncode != 0: + return [] + return [ + line for line in result.stdout.splitlines() + if _STUB_MARKER.search(line) and not line.startswith("+++") + ] + + +def _handle_drift_guard(data: dict) -> None: + if os.environ.get("ZO_DRIFT_GUARD", "1") == "0" or data.get("stop_hook_active"): + return + # Live Stop payloads carry the last message directly (verified in the + # 2026-08-12 live-session trace); fall back to transcript parsing for + # older payload shapes. + last_message = data.get("last_assistant_message") + if not isinstance(last_message, str) or not last_message: + transcript = data.get("transcript_path") + if not isinstance(transcript, str): + return + last_message = _last_assistant_text(transcript) + if not last_message or _COMPLETION_CLAIM.search(last_message) is None: + return + stubs = _added_stub_lines(_repo_root()) + if not stubs: + return + preview = "\n".join(stubs[:5]) + reason = ( + "Workflow drift guard: the last message claims completion, but the " + f"working tree adds stub/TODO markers:\n{preview}\n" + "Finish the stubbed work or revise the completion claim before stopping." + ) + _emit({"decision": "block", "reason": reason, "stopReason": reason}) + + +# -- precompact / session-end ---------------------------------------------- + + +def _memory_manager(memory_root: Path): + from zo.memory import MemoryManager + + return MemoryManager( + memory_root.parent.parent, memory_root.name, memory_root=memory_root, + ) + + +def _handle_precompact(data: dict) -> None: + repo_root = _repo_root() + memory_root = _memory_root(repo_root) + if memory_root is None or not (memory_root / "STATE.md").exists(): + return + from zo._memory_models import DecisionEntry + + manager = _memory_manager(memory_root) + try: + state = manager.read_state() + state.timestamp = datetime.now(UTC) + manager.write_state(state) + manager.append_decision( + DecisionEntry( + title="Checkpoint: pre-compaction state flush", + context=f"session={data.get('session_id', 'unknown')}", + decision="Flushed STATE.md before context compaction (WS-A3 hook).", + outcome="checkpointed", + ) + ) + except Exception: # noqa: BLE001 — hooks are fail-open by contract + return + + +def _handle_session_end(data: dict) -> None: + repo_root = _repo_root() + memory_root = _memory_root(repo_root) + if memory_root is None or not (memory_root / "STATE.md").exists(): + return + from zo._memory_models import SessionSummary + + manager = _memory_manager(memory_root) + today = datetime.now(UTC).strftime("%Y-%m-%d") + try: + recent = manager.read_recent_summaries(1) + if recent and recent[0].date == today: + return + manager.write_session_summary( + SessionSummary( + accomplished=["(auto-generated at SessionEnd — no summary was written)"], + next_steps=["Review this session's DECISION_LOG entries"], + open_questions=[ + f"session={data.get('session_id', 'unknown')} ended without " + "a hand-written summary; hook backfilled this stub" + ], + ) + ) + except Exception: # noqa: BLE001 — hooks are fail-open by contract + return + + +# -- post-tool-failure ------------------------------------------------------- + + +def _handle_post_tool_failure(data: dict) -> None: + repo_root = _repo_root() + feed_dir = Path( + os.environ.get("ZO_FAILURE_FEED_DIR", str(repo_root / "logs" / "comms")) + ) + try: + feed_dir.mkdir(parents=True, exist_ok=True) + record = { + "event_id": str(uuid.uuid4()), + "event_type": "error", + "timestamp": datetime.now(UTC).isoformat(), + "session_id": data.get("session_id", "unknown"), + "tool_name": data.get("tool_name", "unknown"), + "error": str(data.get("error", data.get("tool_response", "")))[:2000], + "input_preview": json.dumps(data.get("tool_input", {}))[:500], + } + date = datetime.now(UTC).strftime("%Y-%m-%d") + path = feed_dir / f"failures-{date}.jsonl" + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record) + "\n") + except OSError: + return + + +# -- sealed-paths -------------------------------------------------------------- + + +def _sealed_prefixes(memory_root: Path | None) -> list[str]: + prefixes: list[str] = [] + if memory_root is not None: + prefixes.extend(str(memory_root / name) for name in _SEALED_DEFAULTS) + sealed_file = memory_root / "sealed_paths" + with contextlib.suppress(OSError): + prefixes.extend( + line.strip() + for line in sealed_file.read_text(encoding="utf-8").splitlines() + if line.strip() and not line.startswith("#") + ) + return prefixes + + +def _handle_sealed_paths(data: dict) -> None: + tool_input = data.get("tool_input") or {} + file_path = tool_input.get("file_path") or tool_input.get("path") + if not isinstance(file_path, str) or not file_path: + return + repo_root = _repo_root() + memory_root = _memory_root(repo_root) + resolved = str((repo_root / file_path).resolve()) if not os.path.isabs( + file_path + ) else str(Path(file_path).resolve()) + + deny_reason: str | None = None + for prefix in _sealed_prefixes(memory_root): + anchor = prefix if os.path.isabs(prefix) else str((repo_root / prefix).resolve()) + if resolved == anchor or resolved.startswith(anchor.rstrip("/") + "/"): + deny_reason = ( + f"Sealed path: {file_path} is oracle/control state and may not " + "be modified by agents (v2 anti-Goodhart rule). Ask the human " + "operator to change it." + ) + break + + if deny_reason is None: + agent = _agent_name(data) + if agent is not None and memory_root is not None: + doc = load_contracts(memory_root / CONTRACTS_FILENAME) + if doc is not None: + normalized = agent.strip().lower().replace(" ", "-") + for entry in doc.agents: + if entry.agent_name != normalized: + continue + delivery = Path(os.environ.get("ZO_DELIVERY_ROOT", str(repo_root))) + for off in entry.off_limits: + anchor = str((delivery / off).resolve()) + if resolved.startswith(anchor.rstrip("/") + "/"): + deny_reason = ( + f"Contract violation: {entry.agent_name} may not " + f"write inside off-limits path '{off}'." + ) + break + break + + if deny_reason is not None: + _emit( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": deny_reason, + } + } + ) + + +_HANDLERS = { + "subagent-stop": _handle_subagent_stop, + "drift-guard": _handle_drift_guard, + "precompact": _handle_precompact, + "session-end": _handle_session_end, + "post-tool-failure": _handle_post_tool_failure, + "sealed-paths": _handle_sealed_paths, +} + + +def main(argv: list[str] | None = None) -> int: + """Dispatch a hook event; always returns 0 (fail-open).""" + global _emitted + _emitted = False + args = argv if argv is not None else sys.argv[1:] + if not args or args[0] not in _HANDLERS: + return 0 + data = _read_stdin_json() + try: + _HANDLERS[args[0]](data) + except Exception: # noqa: BLE001 — hooks are fail-open by contract + _trace(args[0], data) + return 0 + _trace(args[0], data) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/zo/memory.py b/src/zo/memory.py index d7fce72..413531f 100644 --- a/src/zo/memory.py +++ b/src/zo/memory.py @@ -21,6 +21,8 @@ from __future__ import annotations +import contextlib +import json import os import subprocess from datetime import UTC, datetime @@ -341,6 +343,64 @@ def write_gate_mode(self, mode: str) -> None: path = self._memory_root / "gate_mode" path.write_text(mode + "\n", encoding="utf-8") + # -- Gate nonce file (v2 WS-A5) ------------------------------------------- + + def read_gate_nonce(self) -> str | None: + """Read the pending gate-approval nonce, or ``None`` if absent.""" + path = self._memory_root / "gate_nonce" + if not path.exists(): + return None + try: + value = path.read_text(encoding="utf-8").strip() + except OSError: + return None + return value or None + + def write_gate_nonce(self, nonce: str) -> None: + """Persist a single-use gate-approval nonce (minted at GATED).""" + self._memory_root.mkdir(parents=True, exist_ok=True) + (self._memory_root / "gate_nonce").write_text( + nonce + "\n", encoding="utf-8", + ) + + def clear_gate_nonce(self) -> None: + """Remove the gate nonce (single-use: cleared on terminal decisions).""" + with contextlib.suppress(OSError): + (self._memory_root / "gate_nonce").unlink(missing_ok=True) + + # -- Gate decision file (v2 WS-A5) ---------------------------------------- + + def read_gate_decision(self) -> dict[str, str] | None: + """Read a pending nonce-verified gate decision, or ``None``.""" + path = self._memory_root / "gate_decision" + if not path.exists(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + if not isinstance(data, dict) or "phase" not in data: + return None + return data + + def write_gate_decision(self, phase: str, decision: str, notes: str) -> None: + """Persist a nonce-verified gate decision for the orchestrator.""" + self._memory_root.mkdir(parents=True, exist_ok=True) + payload = { + "phase": phase, + "decision": decision, + "notes": notes, + "timestamp": datetime.now(UTC).isoformat(), + } + (self._memory_root / "gate_decision").write_text( + json.dumps(payload) + "\n", encoding="utf-8", + ) + + def clear_gate_decision(self) -> None: + """Remove a consumed gate decision file.""" + with contextlib.suppress(OSError): + (self._memory_root / "gate_decision").unlink(missing_ok=True) + # -- Project initialization --------------------------------------------- def initialize_project(self) -> None: diff --git a/src/zo/orchestrator.py b/src/zo/orchestrator.py index 04166a3..ca656f0 100644 --- a/src/zo/orchestrator.py +++ b/src/zo/orchestrator.py @@ -18,11 +18,13 @@ from __future__ import annotations import hashlib +import secrets from datetime import UTC, datetime from pathlib import Path from textwrap import dedent from typing import TYPE_CHECKING +from zo import contracts as zo_contracts from zo._evolution_models import FailureRecord, FailureSeverity from zo._memory_models import ( Confidence, @@ -344,6 +346,8 @@ def decompose_plan(self) -> WorkflowDecomposition: mode=mode, phases=phases, agent_contracts=contracts, ) self._restore_phase_states() + self._consume_gate_decision() + self._emit_contracts_file() self._comms.log_decision( agent="orchestrator", title=f"Plan decomposed into {len(phases)} phases ({mode})", @@ -353,6 +357,71 @@ def decompose_plan(self) -> WorkflowDecomposition: self._session_state.phase = phases[0].phase_id return self._workflow + def _consume_gate_decision(self) -> None: + """Apply a nonce-verified CLI gate decision recorded while offline. + + ``zo gates approve/reject`` validates the nonce and persists a + ``gate_decision`` file; on the next decompose (fresh session or + plan-edit replan) the decision is applied to the matching GATED + phase and the file is cleared (v2 WS-A5). + """ + pending = self._memory.read_gate_decision() + if pending is None or self._workflow is None: + return + phase_id = pending.get("phase", "") + try: + phase = self._find_phase(phase_id) + except ValueError: + self._memory.clear_gate_decision() + return + if phase.status == PhaseStatus.GATED: + self.apply_human_decision( + phase_id, + GateDecision(pending.get("decision", "hold")), + pending.get("notes", ""), + nonce=self._memory.read_gate_nonce(), + ) + self._memory.clear_gate_decision() + + def _emit_contracts_file(self) -> None: + """Serialize contracts to ``memory_root/contracts.json`` (v2 WS-A1). + + Fail-open: emission problems are logged, never raised — the + enforcement plane degrades to the prose contracts rather than + blocking a build. + """ + if self._workflow is None: + return + active_phase = next( + ( + p.phase_id + for p in self._workflow.phases + if p.status != PhaseStatus.COMPLETED + ), + self._workflow.phases[0].phase_id if self._workflow.phases else "", + ) + try: + path = zo_contracts.emit_contracts( + self._workflow, + self._memory.memory_root, + self._plan.frontmatter.project_name, + active_phase, + ) + except OSError as exc: + self._comms.log_error( + agent="orchestrator", + error_type="contracts_emission_failed", + description=f"contracts.json emission failed: {exc}", + severity="warning", + ) + return + self._comms.log_decision( + agent="orchestrator", + title="Machine-readable contracts emitted", + rationale=f"{len(self._workflow.agent_contracts)} agent contracts", + outcome=str(path), confidence="high", + ) + def _restore_phase_states(self) -> None: """Restore persisted phase states from session_state into workflow phases.""" if self._workflow is None or self._session_state is None: @@ -378,7 +447,7 @@ def generate_agent_contract( role_description=_ROLE_MAP.get(agent_name, f"{agent_name} agent"), ownership=_OWNERSHIP_MAP.get(agent_name, []), off_limits=_OFF_LIMITS_MAP.get(agent_name, []), - contract_produced=[f"{phase.phase_id}/{agent_name} artifacts"], + contract_produced=self._concrete_produced(agent_name, phase), contract_consumed=[f"Inputs from prior phase for {agent_name}"], validation_checklist=[ "All outputs exist at specified paths", @@ -392,6 +461,22 @@ def generate_agent_contract( ], ) + @staticmethod + def _concrete_produced(agent_name: str, phase: PhaseDefinition) -> list[str]: + """Concrete deliverable paths for an agent in a phase (v2 WS-A1). + + Phase artifacts inside the agent's ownership prefixes become the + verifiable ``contract_produced`` list; when none match, fall back + to the prose placeholder so legacy prompts stay unchanged. + """ + ownership = _OWNERSHIP_MAP.get(agent_name, []) + produced = [ + artifact + for artifact in phase.required_artifacts + if any(artifact.startswith(prefix) for prefix in ownership) + ] + return produced or [f"{phase.phase_id}/{agent_name} artifacts"] + # -- Build lead prompt ---------------------------------------------------- def build_lead_prompt(self, phase: PhaseDefinition) -> str: @@ -657,6 +742,10 @@ def advance_phase(self, phase_id: str) -> GateEvaluation: ) if all_done: phase.status = PhaseStatus.GATED + self._memory.write_gate_nonce(secrets.token_hex(8)) + zo_contracts.set_active_phase( + self._memory.memory_root, phase_id, + ) self._log_gate(ev) return ev @@ -757,13 +846,47 @@ def prepare_gate_review(self, phase_id: str) -> dict[str, str]: if self._plan.oracle: review["oracle_metric"] = self._plan.oracle.primary_metric review["target_threshold"] = self._plan.oracle.target_threshold + nonce = self._memory.read_gate_nonce() + if nonce is not None: + review["approval_nonce"] = nonce return review def apply_human_decision( - self, phase_id: str, decision: GateDecision, notes: str = "", + self, + phase_id: str, + decision: GateDecision, + notes: str = "", + *, + nonce: str | None = None, ) -> None: - """Apply a human's gate decision to a phase.""" + """Apply a human's gate decision to a phase. + + When a gate nonce is stored (minted at ``PhaseStatus.GATED``, v2 + WS-A5), the decision must carry the matching nonce — approvals + echoed from context without it are rejected. The nonce is + single-use: cleared on every terminal decision so a replayed tag + can never pass a later gate. + + Raises: + PermissionError: If a nonce is required and missing/mismatched. + """ phase = self._find_phase(phase_id) + stored_nonce = self._memory.read_gate_nonce() + if stored_nonce is not None and nonce != stored_nonce: + self._comms.log_error( + agent="orchestrator", + error_type="gate_approval_forgery_suspected", + severity="blocking", + description=( + f"Gate decision for {phase_id} rejected: nonce " + f"{'missing' if nonce is None else 'mismatch'} (WS-A5)." + ), + ) + raise PermissionError( + f"Gate decision for {phase_id} requires the approval nonce " + "shown in the gate review (zo gates approve --nonce ...)." + ) + self._memory.clear_gate_nonce() if decision == GateDecision.PROCEED: phase.status = PhaseStatus.COMPLETED self._finalize_experiments(phase) @@ -1504,7 +1627,11 @@ def _prompt_phase(self, phase: PhaseDefinition) -> str: def _prompt_contracts(self, phase: PhaseDefinition) -> str: if not self._workflow: return "" - lines: list[str] = [] + lines: list[str] = [ + "Machine-readable contracts: " + f"{self._memory.memory_root / zo_contracts.CONTRACTS_FILENAME} " + "(deliverables are hook-verified when an agent stops — WS-A1).", + ] for c in self._workflow.agent_contracts: if c.phase_id == phase.phase_id: block = ( diff --git a/tests/integration/test_experiment_flow.py b/tests/integration/test_experiment_flow.py index 8b896f6..fdd69e5 100644 --- a/tests/integration/test_experiment_flow.py +++ b/tests/integration/test_experiment_flow.py @@ -288,6 +288,7 @@ def test_iterate_aborts_running_and_next_prompt_mints_child( orch.advance_phase(phase.phase_id) orch.apply_human_decision( phase.phase_id, GateDecision.ITERATE, "needs tuning", + nonce=orch.prepare_gate_review(phase.phase_id)["approval_nonce"], ) reg_after_iterate = load_registry( diff --git a/tests/integration/test_full_pipeline.py b/tests/integration/test_full_pipeline.py index 8e8c82f..a36d936 100644 --- a/tests/integration/test_full_pipeline.py +++ b/tests/integration/test_full_pipeline.py @@ -17,7 +17,6 @@ import shutil import subprocess from pathlib import Path -from unittest.mock import MagicMock, patch import pytest @@ -147,8 +146,12 @@ def test_full_session_lifecycle(self, tmp_path: Path) -> None: assert gate_eval.requires_human is True assert first_phase.status == PhaseStatus.GATED - # Apply human decision to proceed - orch.apply_human_decision(first_phase.phase_id, GateDecision.PROCEED) + # Apply human decision to proceed (WS-A5: nonce from the gate review) + review = orch.prepare_gate_review(first_phase.phase_id) + orch.apply_human_decision( + first_phase.phase_id, GateDecision.PROCEED, + nonce=review["approval_nonce"], + ) assert first_phase.status == PhaseStatus.COMPLETED # End session with a summary diff --git a/tests/integration/test_hooks_shim.py b/tests/integration/test_hooks_shim.py new file mode 100644 index 0000000..56cab0d --- /dev/null +++ b/tests/integration/test_hooks_shim.py @@ -0,0 +1,148 @@ +"""Integration tests for the .claude/hooks/zo-hookkit.sh shim (v2 WS-A). + +Drives the real bash script with stdin JSON exactly as Claude Code does — +the first tests in the repo to execute a hook script end-to-end. The shim +must stay fail-open (exit 0 always) and emit blocking JSON on stdout only +for genuine violations. +""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from zo._orchestrator_models import ( + AgentContract, + GateType, + PhaseDefinition, + WorkflowDecomposition, +) +from zo.contracts import emit_contracts + +REPO_ROOT = Path(__file__).resolve().parents[2] +SHIM = REPO_ROOT / ".claude" / "hooks" / "zo-hookkit.sh" + + +def _run_shim(event: str, payload: dict, env_overrides: dict[str, str]) -> tuple[int, str]: + import os + + env = {**os.environ, **env_overrides} + result = subprocess.run( + ["bash", str(SHIM), event], + input=json.dumps(payload), capture_output=True, text=True, + timeout=30, env=env, cwd=str(REPO_ROOT), check=False, + ) + return result.returncode, result.stdout + + +@pytest.fixture() +def contracts_env(tmp_path: Path) -> dict[str, str]: + wf = WorkflowDecomposition( + mode="classical_ml", + phases=[ + PhaseDefinition( + phase_id="phase_1", name="Data Review", description="d", + gate_type=GateType.BLOCKING, + required_artifacts=["data/reports/data_quality.md"], + ), + ], + agent_contracts=[ + AgentContract( + agent_name="data-engineer", phase_id="phase_1", + role_description="r", ownership=["data/reports/"], + ), + ], + ) + contracts = emit_contracts(wf, tmp_path / "mem", "demo", "phase_1") + delivery = tmp_path / "delivery" + delivery.mkdir() + return { + "ZO_CONTRACTS_PATH": str(contracts), + "ZO_DELIVERY_ROOT": str(delivery), + "ZO_REPO_ROOT": str(REPO_ROOT), + } + + +class TestShimSubagentStop: + def test_seeded_violation_emits_block_json(self, contracts_env) -> None: + code, out = _run_shim( + "subagent-stop", {"agent_name": "data-engineer"}, contracts_env, + ) + assert code == 0 # fail-open exit contract even when blocking + payload = json.loads(out) + assert payload["decision"] == "block" + assert "data_quality.md" in payload["reason"] + + def test_satisfied_contract_silent(self, contracts_env) -> None: + delivery = Path(contracts_env["ZO_DELIVERY_ROOT"]) + (delivery / "data" / "reports").mkdir(parents=True) + (delivery / "data" / "reports" / "data_quality.md").write_text("# ok") + code, out = _run_shim( + "subagent-stop", {"agent_name": "data-engineer"}, contracts_env, + ) + assert code == 0 + assert out == "" + + +class TestShimRobustness: + def test_no_event_exits_zero(self) -> None: + result = subprocess.run( + ["bash", str(SHIM)], input="{}", capture_output=True, text=True, + timeout=30, cwd=str(REPO_ROOT), check=False, + ) + assert result.returncode == 0 + + def test_garbage_stdin_exits_zero_silent(self, contracts_env) -> None: + import os + + env = {**os.environ, **contracts_env} + result = subprocess.run( + ["bash", str(SHIM), "subagent-stop"], + input="this is not json", capture_output=True, text=True, + timeout=30, env=env, cwd=str(REPO_ROOT), check=False, + ) + assert result.returncode == 0 + assert result.stdout == "" + + def test_unknown_event_exits_zero_silent(self) -> None: + code, out = _run_shim("no-such-event", {}, {}) + assert code == 0 + assert out == "" + + +class TestSettingsWiring: + """The settings.json must actually reference every new hook (unwired + mechanisms are the #1 anti-pattern from the v2 review).""" + + def test_all_ws_a_events_wired(self) -> None: + settings = json.loads( + (REPO_ROOT / ".claude" / "settings.json").read_text() + ) + hooks = settings["hooks"] + for event in ("SubagentStop", "PreCompact", "SessionEnd", "PostToolUseFailure"): + commands = [ + h["command"] + for entry in hooks[event] + for h in entry["hooks"] + ] + assert any("zo-hookkit.sh" in c for c in commands), event + + def test_drift_guard_on_stop_and_sealed_paths_on_pretooluse(self) -> None: + settings = json.loads( + (REPO_ROOT / ".claude" / "settings.json").read_text() + ) + stop_cmds = [ + h["command"] + for entry in settings["hooks"]["Stop"] + for h in entry["hooks"] + ] + assert any("drift-guard" in c for c in stop_cmds) + ptu_cmds = [ + h["command"] + for entry in settings["hooks"]["PreToolUse"] + for h in entry["hooks"] + ] + assert any("sealed-paths" in c for c in ptu_cmds) diff --git a/tests/integration/test_phase_snapshots.py b/tests/integration/test_phase_snapshots.py index bed7f0f..64d5776 100644 --- a/tests/integration/test_phase_snapshots.py +++ b/tests/integration/test_phase_snapshots.py @@ -85,6 +85,7 @@ def test_human_proceed_writes_snapshot(self, tmp_path: Path) -> None: orch.apply_human_decision( phase.phase_id, GateDecision.PROCEED, "looks good", + nonce=memory.read_gate_nonce(), # WS-A5: gates minted via advance_phase need the nonce ) assert phase.status == PhaseStatus.COMPLETED @@ -100,7 +101,9 @@ def test_snapshot_records_human_gate_decision(self, tmp_path: Path) -> None: for st in phase.subtasks: orch.mark_subtask_complete(phase.phase_id, st) orch.advance_phase(phase.phase_id) - orch.apply_human_decision(phase.phase_id, GateDecision.PROCEED) + orch.apply_human_decision( + phase.phase_id, GateDecision.PROCEED, nonce=memory.read_gate_nonce(), + ) loaded = load_latest_snapshot(memory.memory_root, phase.phase_id) assert loaded is not None diff --git a/tests/unit/test_contracts.py b/tests/unit/test_contracts.py new file mode 100644 index 0000000..4316ca2 --- /dev/null +++ b/tests/unit/test_contracts.py @@ -0,0 +1,153 @@ +"""Tests for zo.contracts — machine-readable deliverable contracts (WS-A1). + +Seeded-failure pattern per plans/zo-v2-rearchitecture.md: every enforcement +mechanism must catch a deliberately planted violation (oracle check 1). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from zo._orchestrator_models import ( + AgentContract, + GateType, + PhaseDefinition, + WorkflowDecomposition, +) +from zo.contracts import ( + CONTRACTS_FILENAME, + emit_contracts, + load_contracts, + set_active_phase, + validate_agent_stop, +) + +if TYPE_CHECKING: + from pathlib import Path + + +def _workflow() -> WorkflowDecomposition: + return WorkflowDecomposition( + mode="classical_ml", + phases=[ + PhaseDefinition( + phase_id="phase_1", name="Data Review", description="d", + gate_type=GateType.BLOCKING, + required_artifacts=[ + "data/reports/data_quality.md", "oracle/eval.py", + ], + ), + ], + agent_contracts=[ + AgentContract( + agent_name="data-engineer", phase_id="phase_1", + role_description="r", + ownership=["data/raw/", "data/reports/"], + off_limits=["models/", "oracle/"], + ), + AgentContract( + agent_name="code-reviewer", phase_id="phase_1", + role_description="r", ownership=[], + off_limits=["data/raw/", "models/"], + ), + ], + ) + + +class TestEmission: + def test_emits_contracts_json_with_derived_deliverables(self, tmp_path: Path): + path = emit_contracts(_workflow(), tmp_path / "mem", "demo", "phase_1") + doc = load_contracts(path) + assert doc is not None + assert doc.project == "demo" + assert doc.active_phase == "phase_1" + engineer = next(a for a in doc.agents if a.agent_name == "data-engineer") + assert [d.path for d in engineer.deliverables] == [ + "data/reports/data_quality.md" + ] + assert "oracle/" in engineer.off_limits + + def test_agent_without_ownership_has_no_deliverables(self, tmp_path: Path): + path = emit_contracts(_workflow(), tmp_path / "mem", "demo", "phase_1") + doc = load_contracts(path) + reviewer = next(a for a in doc.agents if a.agent_name == "code-reviewer") + assert reviewer.deliverables == [] + + def test_ownership_fallback_directory_expectation(self, tmp_path: Path): + wf = _workflow() + wf.phases[0].required_artifacts = [] # no named artifacts match + path = emit_contracts(wf, tmp_path / "mem", "demo", "phase_1") + doc = load_contracts(path) + engineer = next(a for a in doc.agents if a.agent_name == "data-engineer") + assert engineer.deliverables[0].kind == "directory" + assert engineer.deliverables[0].path == "data/raw/" + + def test_set_active_phase_updates_document(self, tmp_path: Path): + mem = tmp_path / "mem" + emit_contracts(_workflow(), mem, "demo", "phase_1") + set_active_phase(mem, "phase_2") + assert load_contracts(mem / CONTRACTS_FILENAME).active_phase == "phase_2" + + +class TestValidation: + """Seeded-violation checks — the heart of oracle check 1.""" + + def test_missing_deliverable_is_caught(self, tmp_path: Path): + mem, repo = tmp_path / "mem", tmp_path / "repo" + repo.mkdir() + path = emit_contracts(_workflow(), mem, "demo", "phase_1") + violations = validate_agent_stop(path, "data-engineer", repo) + assert len(violations) == 1 + assert violations[0].problem == "required deliverable file missing" + + def test_satisfied_deliverable_passes(self, tmp_path: Path): + mem, repo = tmp_path / "mem", tmp_path / "repo" + (repo / "data" / "reports").mkdir(parents=True) + (repo / "data" / "reports" / "data_quality.md").write_text("# report") + path = emit_contracts(_workflow(), mem, "demo", "phase_1") + assert validate_agent_stop(path, "data-engineer", repo) == [] + + def test_agent_name_display_form_normalized(self, tmp_path: Path): + mem, repo = tmp_path / "mem", tmp_path / "repo" + repo.mkdir() + path = emit_contracts(_workflow(), mem, "demo", "phase_1") + assert validate_agent_stop(path, "Data Engineer", repo) + + def test_undersized_deliverable_is_caught(self, tmp_path: Path): + wf = _workflow() + mem, repo = tmp_path / "mem", tmp_path / "repo" + (repo / "data" / "reports").mkdir(parents=True) + (repo / "data" / "reports" / "data_quality.md").write_text("") + path = emit_contracts(wf, mem, "demo", "phase_1") + violations = validate_agent_stop(path, "data-engineer", repo) + assert "too small" in violations[0].problem + + def test_empty_ownership_directory_is_caught(self, tmp_path: Path): + wf = _workflow() + wf.phases[0].required_artifacts = [] + mem, repo = tmp_path / "mem", tmp_path / "repo" + (repo / "data" / "raw").mkdir(parents=True) # exists but empty + path = emit_contracts(wf, mem, "demo", "phase_1") + violations = validate_agent_stop(path, "data-engineer", repo) + assert "empty" in violations[0].problem + + def test_inactive_phase_contracts_not_enforced(self, tmp_path: Path): + mem, repo = tmp_path / "mem", tmp_path / "repo" + repo.mkdir() + path = emit_contracts(_workflow(), mem, "demo", "phase_2") + assert validate_agent_stop(path, "data-engineer", repo) == [] + + def test_unknown_agent_fails_open(self, tmp_path: Path): + mem, repo = tmp_path / "mem", tmp_path / "repo" + repo.mkdir() + path = emit_contracts(_workflow(), mem, "demo", "phase_1") + assert validate_agent_stop(path, "mystery-agent", repo) == [] + + def test_malformed_contracts_file_fails_open(self, tmp_path: Path): + bad = tmp_path / "contracts.json" + bad.write_text("{not json") + assert validate_agent_stop(bad, "data-engineer", tmp_path) == [] + + def test_missing_contracts_file_fails_open(self, tmp_path: Path): + missing = tmp_path / "contracts.json" + assert validate_agent_stop(missing, "data-engineer", tmp_path) == [] diff --git a/tests/unit/test_gate_nonce.py b/tests/unit/test_gate_nonce.py new file mode 100644 index 0000000..5d55223 --- /dev/null +++ b/tests/unit/test_gate_nonce.py @@ -0,0 +1,129 @@ +"""Tests for nonce-verified gate approvals (v2 WS-A5, plan oracle check 5). + +Seeded forgery: an approval WITHOUT the minted nonce must be rejected; +the genuine nonce-tagged approval must pass; nonces are single-use. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from zo._orchestrator_models import GateDecision, GateMode, PhaseStatus +from zo.comms import CommsLogger +from zo.memory import MemoryManager +from zo.orchestrator import Orchestrator +from zo.plan import parse_plan +from zo.semantic import SemanticIndex +from zo.target import TargetConfig + +REPO_ROOT = Path(__file__).resolve().parents[2] +FIXTURE_PLAN = REPO_ROOT / "tests" / "fixtures" / "test-project" / "plan.md" + + +@pytest.fixture() +def wired(tmp_path: Path): + """Supervised-mode orchestrator plus its memory manager.""" + memory = MemoryManager(project_dir=tmp_path, project_name="test-project") + memory.initialize_project() + orch = Orchestrator( + plan=parse_plan(FIXTURE_PLAN), + target=TargetConfig( + project="test-project", target_repo=str(tmp_path / "delivery"), + target_branch="main", worktree_base=str(tmp_path / "wt"), + git_author_name="ZO Test", git_author_email="zo@test.dev", + agent_working_dirs={}, zo_only_paths=[".zo/"], + enforce_isolation=False, + ), + memory=memory, + comms=CommsLogger( + log_dir=tmp_path / "logs" / "comms", project="test-project", + session_id="nonce-test", + ), + semantic=SemanticIndex(db_path=tmp_path / "index.db"), + zo_root=REPO_ROOT, + gate_mode=GateMode.SUPERVISED, + ) + return orch, memory + + +def _gate_first_phase(orch: Orchestrator) -> str: + decomp = orch.decompose_plan() + phase = decomp.phases[0] + for subtask in phase.subtasks: + orch.mark_subtask_complete(phase.phase_id, subtask) + evaluation = orch.advance_phase(phase.phase_id) + assert evaluation.requires_human + assert phase.status == PhaseStatus.GATED + return phase.phase_id + + +class TestNonceLifecycle: + def test_nonce_minted_when_phase_gates(self, wired) -> None: + orch, memory = wired + _gate_first_phase(orch) + assert memory.read_gate_nonce() + + def test_nonce_surfaced_in_gate_review(self, wired) -> None: + orch, memory = wired + phase_id = _gate_first_phase(orch) + review = orch.prepare_gate_review(phase_id) + assert review["approval_nonce"] == memory.read_gate_nonce() + + def test_forged_approval_without_nonce_rejected(self, wired) -> None: + """The seeded forgery of oracle check 5.""" + orch, _ = wired + phase_id = _gate_first_phase(orch) + with pytest.raises(PermissionError, match="nonce"): + orch.apply_human_decision(phase_id, GateDecision.PROCEED, "lgtm") + + def test_wrong_nonce_rejected(self, wired) -> None: + orch, _ = wired + phase_id = _gate_first_phase(orch) + with pytest.raises(PermissionError): + orch.apply_human_decision( + phase_id, GateDecision.PROCEED, "lgtm", nonce="deadbeef", + ) + + def test_genuine_nonce_passes_and_is_single_use(self, wired) -> None: + orch, memory = wired + phase_id = _gate_first_phase(orch) + orch.apply_human_decision( + phase_id, GateDecision.PROCEED, "lgtm", + nonce=memory.read_gate_nonce(), + ) + assert orch.get_phase_status(phase_id) == PhaseStatus.COMPLETED + assert memory.read_gate_nonce() is None # cleared — replay impossible + + def test_legacy_path_without_stored_nonce_unaffected(self, wired) -> None: + """Directly-GATED phases (no mint) keep the pre-WS-A5 behaviour.""" + orch, memory = wired + decomp = orch.decompose_plan() + phase = decomp.phases[1] + phase.status = PhaseStatus.GATED + assert memory.read_gate_nonce() is None + orch.apply_human_decision(phase.phase_id, GateDecision.PROCEED, "ok") + assert phase.status == PhaseStatus.COMPLETED + + +class TestGateDecisionFile: + def test_cli_recorded_decision_consumed_on_next_decompose(self, wired) -> None: + orch, memory = wired + orch.start_session() + phase_id = _gate_first_phase(orch) + orch.end_session() # persists phase_states (GATED) to STATE.md + # Simulate `zo gates approve`: nonce verified there, then cleared. + memory.clear_gate_nonce() + memory.write_gate_decision(phase_id, "proceed", "approved via CLI") + # Fresh session / replan path: recover state, re-decompose. + orch.start_session() + orch.decompose_plan() + assert orch.get_phase_status(phase_id) == PhaseStatus.COMPLETED + assert memory.read_gate_decision() is None # consumed + + def test_stale_decision_for_unknown_phase_discarded(self, wired) -> None: + orch, memory = wired + memory.write_gate_decision("phase_99", "proceed", "stale") + orch.decompose_plan() + assert memory.read_gate_decision() is None diff --git a/tests/unit/test_hookkit.py b/tests/unit/test_hookkit.py new file mode 100644 index 0000000..dcbcc3b --- /dev/null +++ b/tests/unit/test_hookkit.py @@ -0,0 +1,417 @@ +"""Tests for zo.hookkit — the WS-A enforcement-plane hook handlers. + +Each handler gets: (a) a seeded-violation test proving the mechanism +catches a planted problem (plan oracle checks 1-4, 7), and (b) a +fail-open test proving infrastructure problems never block a session. +Handlers are driven through ``main()`` with stdin/stdout patched — the +same interface the bash shim uses. +""" + +from __future__ import annotations + +import io +import json +import subprocess +from typing import TYPE_CHECKING + +import pytest + +from zo import hookkit +from zo._orchestrator_models import ( + AgentContract, + GateType, + PhaseDefinition, + WorkflowDecomposition, +) +from zo.contracts import emit_contracts + +if TYPE_CHECKING: + from pathlib import Path + + +def _run(event: str, payload: dict, monkeypatch, capsys) -> dict | None: + """Invoke a handler as the shim would; return parsed stdout or None.""" + monkeypatch.setattr("sys.stdin", io.StringIO(json.dumps(payload))) + assert hookkit.main([event]) == 0 + out = capsys.readouterr().out + return json.loads(out) if out else None + + +def _emit_demo_contracts(mem: Path) -> Path: + wf = WorkflowDecomposition( + mode="classical_ml", + phases=[ + PhaseDefinition( + phase_id="phase_1", name="Data Review", description="d", + gate_type=GateType.BLOCKING, + required_artifacts=["data/reports/data_quality.md"], + ), + ], + agent_contracts=[ + AgentContract( + agent_name="data-engineer", phase_id="phase_1", + role_description="r", ownership=["data/reports/"], + off_limits=["models/", "oracle/"], + ), + ], + ) + return emit_contracts(wf, mem, "demo", "phase_1") + + +# ---- subagent-stop (oracle check 1) ---------------------------------------- + + +class TestSubagentStop: + def test_seeded_missing_deliverable_blocks( + self, tmp_path: Path, monkeypatch, capsys, + ): + contracts = _emit_demo_contracts(tmp_path / "mem") + repo = tmp_path / "repo" + repo.mkdir() + monkeypatch.setenv("ZO_CONTRACTS_PATH", str(contracts)) + monkeypatch.setenv("ZO_DELIVERY_ROOT", str(repo)) + out = _run( + "subagent-stop", {"agent_name": "data-engineer"}, monkeypatch, capsys, + ) + assert out is not None + assert out["decision"] == "block" + assert "data/reports/data_quality.md" in out["reason"] + + def test_satisfied_contract_is_silent(self, tmp_path: Path, monkeypatch, capsys): + contracts = _emit_demo_contracts(tmp_path / "mem") + repo = tmp_path / "repo" + (repo / "data" / "reports").mkdir(parents=True) + (repo / "data" / "reports" / "data_quality.md").write_text("# report") + monkeypatch.setenv("ZO_CONTRACTS_PATH", str(contracts)) + monkeypatch.setenv("ZO_DELIVERY_ROOT", str(repo)) + out = _run( + "subagent-stop", {"agent_name": "data-engineer"}, monkeypatch, capsys, + ) + assert out is None + + def test_no_agent_identity_fails_open(self, tmp_path: Path, monkeypatch, capsys): + monkeypatch.setenv("ZO_CONTRACTS_PATH", str(tmp_path / "contracts.json")) + assert _run("subagent-stop", {}, monkeypatch, capsys) is None + + def test_stop_hook_active_guard(self, tmp_path: Path, monkeypatch, capsys): + contracts = _emit_demo_contracts(tmp_path / "mem") + monkeypatch.setenv("ZO_CONTRACTS_PATH", str(contracts)) + monkeypatch.setenv("ZO_DELIVERY_ROOT", str(tmp_path)) + out = _run( + "subagent-stop", + {"agent_name": "data-engineer", "stop_hook_active": True}, + monkeypatch, capsys, + ) + assert out is None + + +# ---- drift-guard (oracle check 2) ------------------------------------------- + + +def _git(repo: Path, *args: str) -> None: + subprocess.run( + ["git", "-C", str(repo), *args], check=True, capture_output=True, + text=True, + ) + + +@pytest.fixture() +def git_repo(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.name", "t") + _git(repo, "config", "user.email", "t@t") + (repo / "mod.py").write_text("def f():\n return 1\n") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "base") + return repo + + +def _transcript(tmp_path: Path, text: str) -> Path: + path = tmp_path / "transcript.jsonl" + entry = {"message": {"role": "assistant", "content": [{"type": "text", "text": text}]}} + path.write_text(json.dumps(entry) + "\n") + return path + + +class TestDriftGuard: + def test_seeded_claim_plus_stub_blocks( + self, git_repo: Path, tmp_path: Path, monkeypatch, capsys, + ): + (git_repo / "mod.py").write_text( + "def f():\n # TODO: actually implement\n return 1\n" + ) + transcript = _transcript(tmp_path, "All tests pass and the task is complete.") + monkeypatch.setenv("ZO_REPO_ROOT", str(git_repo)) + out = _run( + "drift-guard", {"transcript_path": str(transcript)}, monkeypatch, capsys, + ) + assert out is not None + assert out["decision"] == "block" + assert "TODO" in out["reason"] + + def test_claim_without_stubs_is_silent( + self, git_repo: Path, tmp_path: Path, monkeypatch, capsys, + ): + transcript = _transcript(tmp_path, "All tests pass and the task is complete.") + monkeypatch.setenv("ZO_REPO_ROOT", str(git_repo)) + out = _run( + "drift-guard", {"transcript_path": str(transcript)}, monkeypatch, capsys, + ) + assert out is None + + def test_stubs_without_claim_is_silent( + self, git_repo: Path, tmp_path: Path, monkeypatch, capsys, + ): + (git_repo / "mod.py").write_text("def f():\n # TODO: later\n return 1\n") + transcript = _transcript(tmp_path, "Progress update: still working on f().") + monkeypatch.setenv("ZO_REPO_ROOT", str(git_repo)) + out = _run( + "drift-guard", {"transcript_path": str(transcript)}, monkeypatch, capsys, + ) + assert out is None + + def test_env_kill_switch(self, git_repo: Path, tmp_path: Path, monkeypatch, capsys): + (git_repo / "mod.py").write_text("def f():\n # TODO\n return 1\n") + transcript = _transcript(tmp_path, "Everything is done.") + monkeypatch.setenv("ZO_REPO_ROOT", str(git_repo)) + monkeypatch.setenv("ZO_DRIFT_GUARD", "0") + out = _run( + "drift-guard", {"transcript_path": str(transcript)}, monkeypatch, capsys, + ) + assert out is None + + +# ---- precompact / session-end (oracle check 3) ------------------------------- + + +@pytest.fixture() +def memory_root(tmp_path: Path) -> Path: + from zo._memory_models import SessionState + from zo.memory import MemoryManager + + mem = tmp_path / "memory" / "demo" + manager = MemoryManager(tmp_path, "demo", memory_root=mem) + manager.initialize_project() + manager.write_state(SessionState(phase="phase_1")) + return mem + + +class TestPrecompact: + def test_flushes_checkpoint_and_logs_decision( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + from zo.memory import MemoryManager + + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + before = MemoryManager(tmp_path, "demo", memory_root=memory_root).read_state() + out = _run("precompact", {"session_id": "s-1"}, monkeypatch, capsys) + assert out is None + manager = MemoryManager(tmp_path, "demo", memory_root=memory_root) + after = manager.read_state() + assert after.timestamp >= before.timestamp + decisions = manager.read_decisions() + assert any("pre-compaction" in d.title.lower() for d in decisions) + + def test_missing_state_fails_open(self, tmp_path: Path, monkeypatch, capsys): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(tmp_path / "nowhere")) + assert _run("precompact", {}, monkeypatch, capsys) is None + + +class TestSessionEnd: + def test_backfills_missing_summary( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + _run("session-end", {"session_id": "s-2"}, monkeypatch, capsys) + summaries = list((memory_root / "sessions").glob("*.md")) + assert len(summaries) == 1 + assert "auto-generated" in summaries[0].read_text() + + def test_existing_summary_today_not_duplicated( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + _run("session-end", {}, monkeypatch, capsys) + _run("session-end", {}, monkeypatch, capsys) + assert len(list((memory_root / "sessions").glob("*.md"))) == 1 + + +# ---- post-tool-failure (oracle check 4) -------------------------------------- + + +class TestPostToolFailure: + def test_appends_structured_jsonl(self, tmp_path: Path, monkeypatch, capsys): + feed = tmp_path / "feed" + monkeypatch.setenv("ZO_FAILURE_FEED_DIR", str(feed)) + payload = { + "session_id": "s-3", "tool_name": "Bash", + "tool_input": {"command": "pytest"}, "error": "exit 1: boom", + } + _run("post-tool-failure", payload, monkeypatch, capsys) + files = list(feed.glob("failures-*.jsonl")) + assert len(files) == 1 + record = json.loads(files[0].read_text().splitlines()[0]) + assert record["tool_name"] == "Bash" + assert record["event_type"] == "error" + assert record["event_id"] + assert "boom" in record["error"] + + def test_two_failures_two_lines(self, tmp_path: Path, monkeypatch, capsys): + feed = tmp_path / "feed" + monkeypatch.setenv("ZO_FAILURE_FEED_DIR", str(feed)) + _run("post-tool-failure", {"tool_name": "A", "error": "x"}, monkeypatch, capsys) + _run("post-tool-failure", {"tool_name": "B", "error": "y"}, monkeypatch, capsys) + lines = list(feed.glob("failures-*.jsonl"))[0].read_text().splitlines() + assert len(lines) == 2 + + +# ---- sealed-paths (oracle check 7) ------------------------------------------- + + +class TestSealedPaths: + def test_seeded_write_to_sealed_control_file_denied( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + out = _run( + "sealed-paths", + {"tool_input": {"file_path": str(memory_root / "gate_mode")}}, + monkeypatch, capsys, + ) + assert out is not None + decision = out["hookSpecificOutput"]["permissionDecision"] + assert decision == "deny" + + def test_seeded_write_to_user_sealed_prefix_denied( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + (memory_root / "sealed_paths").write_text("oracle/\n") + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + out = _run( + "sealed-paths", + {"tool_input": {"file_path": "oracle/eval.py"}}, + monkeypatch, capsys, + ) + assert out["hookSpecificOutput"]["permissionDecision"] == "deny" + + def test_off_limits_write_denied_for_contracted_agent( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + _emit_demo_contracts(memory_root) + delivery = tmp_path / "delivery" + delivery.mkdir() + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + monkeypatch.setenv("ZO_DELIVERY_ROOT", str(delivery)) + out = _run( + "sealed-paths", + { + "agent_name": "data-engineer", + "tool_input": {"file_path": str(delivery / "models" / "net.py")}, + }, + monkeypatch, capsys, + ) + assert out["hookSpecificOutput"]["permissionDecision"] == "deny" + assert "off-limits" in out["hookSpecificOutput"]["permissionDecisionReason"] + + def test_ordinary_write_is_silent( + self, tmp_path: Path, memory_root: Path, monkeypatch, capsys, + ): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_MEMORY_ROOT", str(memory_root)) + out = _run( + "sealed-paths", + {"tool_input": {"file_path": "src/anything.py"}}, + monkeypatch, capsys, + ) + assert out is None + + def test_no_file_path_fails_open(self, tmp_path: Path, monkeypatch, capsys): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + assert _run("sealed-paths", {"tool_input": {}}, monkeypatch, capsys) is None + + +# ---- dispatcher --------------------------------------------------------------- + + +class TestDispatcher: + def test_unknown_event_is_noop(self, monkeypatch, capsys): + assert hookkit.main(["not-an-event"]) == 0 + assert capsys.readouterr().out == "" + + def test_no_args_is_noop(self): + assert hookkit.main([]) == 0 + + def test_handler_exception_fails_open(self, monkeypatch, capsys): + monkeypatch.setitem( + hookkit._HANDLERS, "explode", + lambda data: (_ for _ in ()).throw(RuntimeError("boom")), + ) + monkeypatch.setattr("sys.stdin", io.StringIO("{}")) + assert hookkit.main(["explode"]) == 0 + + +# ---- trace observability ------------------------------------------------- + + +class TestTrace: + def test_every_invocation_writes_a_trace_line( + self, tmp_path: Path, monkeypatch, capsys, + ): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_FAILURE_FEED_DIR", str(tmp_path / "feed")) + _run("post-tool-failure", {"tool_name": "Bash", "error": "x"}, monkeypatch, capsys) + traces = list((tmp_path / "logs").glob("hook-trace-*.jsonl")) + assert len(traces) == 1 + record = json.loads(traces[0].read_text().splitlines()[0]) + assert record["event"] == "post-tool-failure" + assert "tool_name" in record["stdin_keys"] + + def test_trace_records_agent_identity_and_block( + self, tmp_path: Path, monkeypatch, capsys, + ): + contracts = _emit_demo_contracts(tmp_path / "mem") + repo = tmp_path / "repo" + repo.mkdir() + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_CONTRACTS_PATH", str(contracts)) + monkeypatch.setenv("ZO_DELIVERY_ROOT", str(repo)) + _run("subagent-stop", {"agent_name": "data-engineer"}, monkeypatch, capsys) + record = json.loads( + next((tmp_path / "logs").glob("hook-trace-*.jsonl")).read_text().splitlines()[0] + ) + assert record["agent_identity"] == "data-engineer" + assert record["emitted_output"] is True + + def test_trace_disabled_by_env(self, tmp_path: Path, monkeypatch, capsys): + monkeypatch.setenv("ZO_REPO_ROOT", str(tmp_path)) + monkeypatch.setenv("ZO_HOOK_TRACE", "0") + monkeypatch.setenv("ZO_FAILURE_FEED_DIR", str(tmp_path / "feed")) + _run("post-tool-failure", {"tool_name": "Bash", "error": "x"}, monkeypatch, capsys) + assert not (tmp_path / "logs").exists() + + +class TestDriftGuardLivePayload: + """Live Stop payloads carry last_assistant_message directly (verified + in the 2026-08-12 live-session trace) — no transcript parse needed.""" + + def test_inline_last_message_used_over_transcript( + self, git_repo: Path, monkeypatch, capsys, + ): + (git_repo / "mod.py").write_text("def f():\n # TODO: later\n return 1\n") + monkeypatch.setenv("ZO_REPO_ROOT", str(git_repo)) + out = _run( + "drift-guard", + {"last_assistant_message": "All tests pass, implementation complete."}, + monkeypatch, capsys, + ) + assert out is not None + assert out["decision"] == "block"