diff --git a/.github/workflows/debt-ratchet.yml b/.github/workflows/debt-ratchet.yml index 2f267f1842..85d057771c 100644 --- a/.github/workflows/debt-ratchet.yml +++ b/.github/workflows/debt-ratchet.yml @@ -1,47 +1,84 @@ -name: Debt ratchet +name: Merge queue on: - pull_request: - branches: [main] + status: + merge_group: + types: [checks_requested] workflow_dispatch: inputs: - base: - required: true - type: string - head: - required: true - type: string + enter_queue: + description: Enqueue the already reviewed head after successful checks + type: boolean + default: false permissions: contents: read concurrency: - group: debt-ratchet-${{ github.ref }} - cancel-in-progress: true + group: merge-queue-${{ github.event.sha || github.sha }} + cancel-in-progress: false jobs: - debt-ratchet: - name: Debt ratchet - runs-on: ubuntu-latest + dispatch-reviewed: + if: github.event_name == 'status' && github.event.state == 'success' + runs-on: [self-hosted, Linux, X64, ts-gaming, agent-comms-merge] timeout-minutes: 5 steps: - - uses: actions/checkout@v4 - with: - fetch-depth: 0 - submodules: recursive - ref: ${{ github.event.pull_request.head.sha || inputs.head }} - - uses: actions/setup-python@v5 - with: - python-version-file: pyproject.toml - - name: Install development dependencies - run: python -m pip install --disable-pip-version-check -e src/agent_comms/skills/refactor-audit '.[all,dev]' - - name: Compare touched production files + - name: Dispatch checks on the approved PR head, not the default-branch status event env: - PR_BASE: ${{ github.event.pull_request.base.sha || inputs.base }} - PR_HEAD: ${{ github.event.pull_request.head.sha || inputs.head }} - run: python tools/debt_ratchet.py --root src/agent_comms --base "$PR_BASE" --head "$PR_HEAD" - - name: Run every marked surface guard - if: ${{ github.event_name == 'workflow_dispatch' }} + RUN_ID: ${{ github.run_id }} + REVIEWED_HEAD: ${{ github.event.sha }} + REVIEW_CONTEXT: ${{ github.event.context }} + run: | + set -euo pipefail + request=$(python3 -c 'import shlex, sys; print(shlex.join(sys.argv[1:]))' \ + "$RUN_ID" dispatch "$REVIEWED_HEAD" "$REVIEW_CONTEXT") + ssh -o BatchMode=yes -o StrictHostKeyChecking=yes \ + -i "$AGENT_COMMS_GATE_KEY" "$AGENT_COMMS_GATE_ENDPOINT" "$request" + + ratchet: + if: github.event_name != 'status' + name: debt-${{ matrix.kind }} + runs-on: [self-hosted, Linux, X64, ts-gaming, agent-comms-merge] + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + kind: [src, tools] + steps: + - name: Check the exact GitHub run through the restricted host entrypoint + env: + RUN_ID: ${{ github.run_id }} + CHECK_KIND: ${{ matrix.kind }} + run: | + set -euo pipefail + ssh -o BatchMode=yes -o StrictHostKeyChecking=yes \ + -i "$AGENT_COMMS_GATE_KEY" "$AGENT_COMMS_GATE_ENDPOINT" "$RUN_ID $CHECK_KIND" + + practice-install: + if: github.event_name != 'status' + name: install-rehearsal + runs-on: [self-hosted, Linux, X64, ts-gaming, agent-comms-merge] + timeout-minutes: 60 + steps: + - name: Practice install; private buses and logs never enter the runner sandbox + env: + RUN_ID: ${{ github.run_id }} + run: | + set -euo pipefail + ssh -o BatchMode=yes -o StrictHostKeyChecking=yes \ + -i "$AGENT_COMMS_GATE_KEY" "$AGENT_COMMS_GATE_ENDPOINT" "$RUN_ID install" + + enqueue-approved: + if: github.event_name == 'workflow_dispatch' && inputs.enter_queue + needs: [ratchet, practice-install] + runs-on: [self-hosted, Linux, X64, ts-gaming, agent-comms-merge] + timeout-minutes: 5 + steps: + - name: Enter GitHub's queue after approval and successful checks env: - PYTEST_DISABLE_PLUGIN_AUTOLOAD: '1' - run: python -m pytest -p pytest_asyncio.plugin -o addopts='' --strict-markers -m refactor_guard -q + RUN_ID: ${{ github.run_id }} + run: | + set -euo pipefail + ssh -o BatchMode=yes -o StrictHostKeyChecking=yes \ + -i "$AGENT_COMMS_GATE_KEY" "$AGENT_COMMS_GATE_ENDPOINT" "$RUN_ID enqueue" diff --git a/.pi/APPEND_SYSTEM.md b/.pi/APPEND_SYSTEM.md index 05bbe72038..51592dc62b 100644 --- a/.pi/APPEND_SYSTEM.md +++ b/.pi/APPEND_SYSTEM.md @@ -56,6 +56,11 @@ You are one of several agents coordinating through agent-comms. Tristan owns pro ## Ownership, code, and delivery +- Do not merge main into feature branches. When your own branch conflicts with main, rebase it onto origin/main that turn, keep main's side where main deleted or moved code, push the owned branch, and report its new head. Do not ask again for that routine authorization. Do not make formatting-only or re-budget-only commits. +- Fresh-environment practice installs belong to the merger, on the pushed head. Workers prove the affected behavior on a copy of the bus using existing dependencies and report precisely which installation proof the merger still owes. +- Line growth is a measurement, not a defect or merge gate. Do not request a line budget. Fix duplicated authorities, type filters and strings replacing declared identities; the existing measure owner decides which findings block. +- P269 practice installation uses a retained bus tail with its production indexes, bounded to 3 GiB and measured against actual free space. It does not require a 26 GiB whole-desktop copy or GitHub configuration changes. Preserve original sessions, inputs and stores; never turn short storage into a skipped success. + - Check active owners, open PRs, and claims before starting overlapping work. Coordinate with the owner of an existing feature; do not edit their worktree or start a competing implementation. Delegate genuinely independent work when it speeds delivery. - At the next safe checkpoint, when you find a concrete defect, check that a named owner is fixing it. If none is, take or assign ownership, provide the reproducer and acceptance check, and follow through to a tested fix or a precise blocker without duplicating work. - Implement working end-to-end behavior rather than a dormant path. Enable features by default only when safe and authorized; preserve opt-in or default-off safeguards for trust-sensitive features. Put tested work in a draft PR promptly and keep its full working version visible. Work only in an isolated worktree; do not reset, clean, or commit in the live main checkout or change installed packages without review. diff --git a/docs/merge-queue.md b/docs/merge-queue.md new file mode 100644 index 0000000000..3d94b93ed5 --- /dev/null +++ b/docs/merge-queue.md @@ -0,0 +1,92 @@ +# Automatic merge queue + +Owner: pi-cn-load, P269, draft PR #2023. Not enabled or accepted. +The takeover preserved pi-install-gate's newer published 307e5955e, a descendant +of the ordered f0dc3df0c checkpoint, and merged current main without rewriting +history. First takeover push: fb6074f3e. P714/#1947 is merged as the review-owner +dependency, not copied into another verdict implementation. + +GitHub owns queue order, proposed merge identity and failed-entry removal. +QueueCheck and RunEvent members own host execution and admission. The existing +workflow runs debt-src, debt-tools and install-rehearsal on merge_group. +No commander's merge scheduler is involved after bootstrap. + +## Review and installer ownership + +ReviewedHeadVerdict.context is the sole review-context declaration. Its existing +publisher and the queue reader both use it. The status event passes its context +as data; the host compares it with that declaration. No repository variable +mirrors the context. No native GitHub approval is required: the contributors all +use the same GitHub identity and cannot approve their own PRs. + +The queue invokes PracticeGateCommand. PracticeCopy uses the existing CheckRoot +owner to retain the bus tail, bounded saved-session tails and their production +indexes; it does not snapshot the whole desktop bus. PracticeCopy owns the 3 GiB +cap and checks measured bytes against current free space. It never admits an +empty substitute or a skipped installation. The gate checks out the proposed +revision, then rehearse invokes target/tools/upgrade_bus.py, including --help and +the actual rehearsal. It does not invoke the live checkout's installer. GateResult/MachineResult carry +typed failure reasons back to the queue. CheckFailed carries the actual completed command and delegates its public +summary to the QueueCheck member. Ratchet summaries decode introduced_lines +and print Measure file:line. Installer summaries return the declared +UpgradeRefused message through the existing typed gate result. Raw child +stdout/stderr, successful rehearsal steps and bus data remain private. + +## Isolation and lifetime + +The existing runner ts-gaming-agent-comms-merge (ID 21) is registered but offline. +Labels: self-hosted, Linux, X64, ts-gaming, agent-comms-merge. Its bubblewrap +namespace exposes vendor runtime, work/log directories, system binaries and a +sealed runner home, not host homes, buses, personal tokens or desktop sockets. +The SSH key admits only the fixed host entrypoint. GitHub run identity and the +workflow path are verified before checks. No generic hosted runner reads a bus. + +The host derives home/user from its environment and OS identity. There is no audit path on the wrapper's PYTHONPATH: debt_ratchet derives it +from the repository submodule itself. Host.fetch initializes/updates the pinned +submodules. No unit checkout is a dependency of production gates. +There is no shared host wrapper timeout. Ratchet members own 600 seconds, +install owns 3000, and enqueue/dispatch share their command capability's 120. +YAML ceilings are 15, 60 and 5 minutes respectively. Local development ratchets +retain the ordered timeout 300 and faulthandler. + +Only explicit gate failure reasons and source annotations cross the runner +boundary. Full child diagnostics, copied state and successful rehearsal results +remain on the host. A storage or allocation refusal fails the gate; it never +turns into a skipped or successful practice install. + +## Settings Tristan must change (no settings changed by this worker) + +- Settings → Rules → Rulesets → Automatic merge queue (25439636): change Enforcement from Disabled to Active for main, after reviewed bootstrap code is on main. +- In that ruleset → Merge queue: Merge method Merge; maximum group/build/merge size 1; require debt-src, debt-tools and install-rehearsal from GitHub Actions. +- In that ruleset → Require a pull request before merging: set required approvals to 0; the staged rule currently incorrectly requires 1 native approval. +- Settings → Actions → Runners: reuse registered runner ts-gaming-agent-comms-merge (ID 21), with the labels above; do not register a duplicate. If it must be replaced, use New self-hosted runner → Linux → x64 with those same custom labels. + +Runner registration does not start its host process. The previous owner confirms +that its restricted key was never added to authorized_keys and no runner +process/service has started. An authorized operator must install the reviewed +fixed-command, loopback-only key restriction and start the reviewed isolated +launcher from main, retaining the existing private runner credentials. Do not +launch the previous owner's queue-qualified/ scripts; those are old qualification +evidence, not installed entrypoints. No runner launch, authorized_keys edit, +queue enablement, native approval, production install or restart was performed +by this worker. + +## Practice proof and later activation + +The latest owner ruling replaces real-queue proof with a practice installation +on the bounded retained-tail copy. Report that command's exit code and the pushed +revision; unit checks alone are not the practice-install proof. Both source and +tools ratchets must exit zero. SourceLines reports actual growth without a +budget or blocking the merge; structural findings still block through their +owning Measure implementations. + +The disabled queue, offline runner and unauthorized key are not prerequisites +for this proof. They remain untouched; later production activation requires the +operator actions above. No deliberate failing GitHub branch is claimed here. +The copy does not need room for a 26 GiB desktop snapshot. Its admission uses +actual copied bytes and actual remaining space, not a second queue policy. + +Worker checkouts and receipts stay under /home/ts/wt/pi-cn-load--P269 and +/home/ts/.cache/agent-scratch/pi-cn-load, below the combined 3 GB limit. The old +runner state remains with its existing owner; its credentials have not been +copied or deleted. diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 7e1eeae84b..888180377f 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -206,4 +206,4 @@ Toad is a view of Core, so the commander's page is one more view over Core's ans ## Order and dependencies -The phases and what each waits for are declared once, in `src/agent_comms/plan_phases.py` (the `PlanPhase` family). An order names the phase it serves with a `Serves: ` line, stored as `ProblemOrder.serves`; a phase is done when every order serving it is closed. Each rotation prints `PLAN IDLE` for every phase whose dependencies are done and that has no open order (`PlanIdle` in `src/agent_comms/standing_rules.py`). +The phases and what each waits for are declared once, in `src/agent_comms/plan_phases.py` (the `PlanPhase` family). An order names the phase it serves with a `Serves: ` line, stored as `ProblemOrder.serves`. A phase is done when its explicitly declared full-phase checked task has accepted installed evidence; closing related subtasks does not complete the phase. Earlier delivered phases record their delivery on their declarations. Each rotation prints `PLAN IDLE` for every phase whose dependencies are done and that has no open order (`PlanIdle` in `src/agent_comms/standing_rules.py`). diff --git a/src/agent_comms/check_root.py b/src/agent_comms/check_root.py index 7d67cfad73..02de2fb15a 100644 --- a/src/agent_comms/check_root.py +++ b/src/agent_comms/check_root.py @@ -17,7 +17,6 @@ from .comms import Comms from .fresh_private_session import create_fresh_private_session from .goals import NoGoal -from .importing import ImportLimits from .launch_runtime import RestartEnvironment, ThreadLaunch from .native_entries import ImportedSessionWriter, NativeEntry from .read_basis import MemberActivityDisplayScope @@ -66,8 +65,8 @@ def open(self) -> Iterator[Comms]: root.mkdir(mode=0o700, parents=True) # Refuse an existing directory, including a symlink. try: copied = Comms(root) + self._populate(copied) try: - self._populate(copied) yield copied finally: for thread in copied.registry.snapshot().visible_threads( @@ -142,7 +141,6 @@ def _populate(self, copied: Comms) -> None: def _session(self, original: Thread, directory: Path, project: Path) -> str: session = create_fresh_private_session(directory, worktree=project) source = original.runtime.history_source(self.source.root, original.incarnation) - limits = ImportLimits(messages=self.max_records, characters=self.max_bytes) with source.reading() as reader, session.path.open("r+") as output: writer = ImportedSessionWriter(output, NativeEntry.read(output.readline().encode())) output.seek(0, os.SEEK_END) @@ -158,16 +156,13 @@ def _session(self, original: Thread, directory: Path, project: Path) -> str: with source.projection( source, TranscriptRoutes(self.source.root) ) as project_records: - for record, events in project_records( + for _record, events in project_records( iter(reversed(selected)), max_records=self.max_records, max_bytes=self.max_bytes, ): - record.project( - lambda current, events=events: writer.import_record( - current.entry, events, limits - ) - ) + for event in events: + writer.append_recorded(event) return session.session_file def run(self, command: Sequence[str]) -> CheckRootResult: diff --git a/src/agent_comms/commander/judge.py b/src/agent_comms/commander/judge.py index 07affa6d45..be3a13d49d 100644 --- a/src/agent_comms/commander/judge.py +++ b/src/agent_comms/commander/judge.py @@ -20,7 +20,7 @@ from dataclasses import asdict, dataclass from ..comms import Comms -from ..todos import AcceptedVerdict, PeerAcceptedVerdict, BluntText, OrderVerdict, SendBackVerdict, StrikeReason, Todo, Slop, Derivation, SupersededVerdict +from ..todos import AcceptedVerdict, PeerAcceptedVerdict, BluntText, ToolCallVerdict, SendBackVerdict, StrikeReason, Todo, Slop, Derivation, SupersededVerdict from ..tools import CommsUnjudgedTool, CommsVerdictTool from ..field_codec import FieldCodec, option from ..tool_caller import ThreadCaller @@ -146,7 +146,7 @@ class VerdictVerb(JudgeVerb, RuledVerb): "--line", action="append", metavar="LINE", help="an exact diff line; repeat for each") @abstractmethod - def operation(self, ctx: order.Commander, todo: Todo, edit_id: str) -> OrderVerdict: ... + def operation(self, ctx: order.Commander, todo: Todo, edit_id: str) -> ToolCallVerdict: ... @abstractmethod def check_edits(self, edits: list[Edit]) -> None: ... diff --git a/src/agent_comms/commander/rules/always-fork-subagents.md b/src/agent_comms/commander/rules/always-fork-subagents.md new file mode 100644 index 0000000000..feb037bc33 --- /dev/null +++ b/src/agent_comms/commander/rules/always-fork-subagents.md @@ -0,0 +1,13 @@ +--- +name: always-fork-subagents +description: "Owner rule: dispatch subagents as forks (subagent_type \"fork\"), not fresh agents, unless the brief fully teaches correct maintenance and the paper" +type: feedback +--- + +Always fork (Agent with `subagent_type: "fork"`) instead of starting a fresh general-purpose agent. Owner, 2026-10-18: "always fork. stop it, fork and let the fork take over", then "unless you already schooled it on correct maintenance and the paper". + +**Why:** a fresh agent knows only its brief. My briefs carried the owner's rules (multiple inheritance, no back-compat) but not the paper's definition of slop and the results behind it, so a fresh agent judges shapes by dogma. A fork inherits the conversation, including that understanding (, ). + +**How to apply:** default to forks for refactor and design work. A fresh agent is acceptable only when its brief teaches correct maintenance itself (slop = a restated required fact a complete mechanism derives; multiple parents derive crossings; mirrors add disagreeing states; passing tests cannot reveal outside supply) and points at `.claude/verify/PAPER-RESULTS.md`. Supersedes the older habit of fresh agents with written briefs. + +**Turn limit (2026-10-18):** owner said "don't use that [turn limit] on ur subagents". Forks stop at ~200 turns and can't be given a limit; they have no definition file (docs: code.claude.com/docs/en/sub-agents.md, `maxTurns` frontmatter). For long work use the user-level agent `correct-maintainer` (`~/.claude/agents/correct-maintainer.md`, `maxTurns: 100000`, system prompt teaches the standard). Definitions load at session start only. If a fork hits the cap anyway, resume it via SendMessage right away. diff --git a/src/agent_comms/commander/rules/claude-usage-budget.md b/src/agent_comms/commander/rules/claude-usage-budget.md new file mode 100644 index 0000000000..5bac1d4b10 --- /dev/null +++ b/src/agent_comms/commander/rules/claude-usage-budget.md @@ -0,0 +1,15 @@ +--- +name: claude-usage-budget +description: Claude subagent usage is the scarce budget; keep background forks to the critical path until Pi (gpt-6.1-sol) threads work on the bus +type: feedback +--- + +2026-10-19: I had 7 background forks running at once: runtime, carry-order, strings, index-lock, hooks, fast-ratchet and fast-board. Owner: "don't launch too many background agents, I'm trying to move to Pi Codex threads to save on usage, so please be mindful. If it's going to take another 5 hours to get Pi GPT working in comms we can't just blow through usage." + +**Why:** the move to Pi threads exists to save Claude usage. Spending Claude forks freely while that move is blocked defeats it. + +**How to apply:** +- Only work on the path to working Pi threads gets a Claude fork; aim for one or two at a time. +- Everything else (perf, cleanup, slop hunts) becomes a written order queued for Pi units, not a fork. +- Briefs say "finish and report in N lines", with no exploration beyond the task. +- This tempers [[slow-is-a-defect]] and [[structural-fixes-over-pushing-through]]: dispatch still happens, but to Pi units once they work. See [[codex-units-need-dictation]]. diff --git a/src/agent_comms/commander/rules/expose-full-surface-no-limits.md b/src/agent_comms/commander/rules/expose-full-surface-no-limits.md new file mode 100644 index 0000000000..c99b6c7b75 --- /dev/null +++ b/src/agent_comms/commander/rules/expose-full-surface-no-limits.md @@ -0,0 +1,11 @@ +--- +name: expose-full-surface-no-limits +description: When OpenHCS supports an external tool, expose its whole surface, not a subset; do not scope down for budget reasons +type: feedback +--- + +When OpenHCS integrates an external algorithm or library, expose everything it offers (every capability and parameter reachable in the tool itself or in any other platform's integration of it), not a minimal subset. + +**Why:** owner, 2026-10-12, when I noted that a mature CellProfiler plugin exposes more Cellpose options than our 151-line example: "there's no reason why we aren't exposing the same amount of functions. remember there's no real limits." There is no token or effort budget reason to scope down; an integration that exposes less than competitors undercuts the claim that adding tools here is trivial. + +**How to apply:** build a capability matrix against the tool's own API and the other platforms' integrations, and close every gap or state the precise reason. Prefer deriving the surface from the tool's own signatures (as the scikit-image/CuPy/pyclesperanto registries do) over hand-transcribing, so it stays complete when the tool changes. My interpretation: "no real limits" is also a reminder not to ration scope, agents or thoroughness in general; machine limits (15 GB RAM, disk) still apply. Related: [[structural-fixes-over-pushing-through]], [[full-autonomy-openhcs]]. diff --git a/src/agent_comms/commander/rules/glued-class-names.md b/src/agent_comms/commander/rules/glued-class-names.md new file mode 100644 index 0000000000..7b93dc84eb --- /dev/null +++ b/src/agent_comms/commander/rules/glued-class-names.md @@ -0,0 +1,10 @@ +--- +name: glued-class-names +description: A class name gluing an existing domain type to a generic concept (StepField..., FunctionFieldDriver) signals a cross product flattened into names; question whether the class should exist +type: feedback +--- +When a class name is an existing domain type glued to a generic concept (`StepField...`, `FunctionFieldNavigationDriver`, `PipelineXManager`), stop and ask whether the class should exist at all, before asking whether it is well written. + +**Why:** the owner spotted `StepFieldNavigationDriver` from its name alone in an agent's status line (2026-10-12): "Step already exists. If there's a StepField, is there a PipelineField? and FunctionField? and what's Field? probably attribute, which is just on an object. So it's all kinds of wrong." The name implies a grid (domain type x mechanism), one class per cell; that is the row-column family of paper 1 Thm 4.5 flattened into names. The prediction held: `FunctionFieldNavigationDriver` already existed in pyqt-reactive, and the agent copied the pattern. I had first judged it "mostly right" because it extended an existing family; the right question was whether the family itself was the right shape. + +**How to apply:** in review and in agent briefs: a generic mechanism (navigation, validation, preview, codec) takes the object or its declaration and asks it for its fields; it does not get a subclass per domain type. If the per-type answer is real behaviour, it lives on the domain type (or a mixin it inherits), not in a parallel `` class. Scan names as *candidates* only (a glued name is a reason to look, not proof). Also: do not judge a new class only by "it uses the existing family"; check the family. See [[plain-names]], , [[optimize-for-slop-elimination]]. diff --git a/src/agent_comms/commander/rules/local-patterns-not-precedent.md b/src/agent_comms/commander/rules/local-patterns-not-precedent.md new file mode 100644 index 0000000000..80f5b63270 --- /dev/null +++ b/src/agent_comms/commander/rules/local-patterns-not-precedent.md @@ -0,0 +1,10 @@ +--- +name: local-patterns-not-precedent +description: Never copy a pattern because it is already in the codebase; judge against the standard, and treat repeated bad shapes as removal candidates +type: feedback +--- +Do not follow a local pattern just because it is local. The standard is the owner's rules and paper results, not the surrounding code. + +**Why:** Codex did this constantly: it made a mess, saw the mess, took the mess as licence, and repeated it. On 2026-10-12 a U6 agent added `StepFieldNavigationDriver` because `FunctionFieldNavigationDriver` already existed in pyqt-reactive, and I first approved it for "extending the existing family". Models (me included) copy the code around them more than they follow instructions. + +**How to apply:** when reviewing or briefing, never accept "consistent with nearby code" as a reason. An existing repeated shape that fails the standard is a candidate to report and remove, not a template. Ask what fact the thing stores and who already declares it. Extending an existing family is right only after checking the family itself is the right shape (see [[glued-class-names]]). Removing bad exemplars matters because they get copied. Keep plans and notes that outlast the current prompt (receipts, DISPATCH.md, PAPER-RESULTS.md, memory), since context is lost at compaction and the tree keeps pulling toward its own habits. See [[optimize-for-slop-elimination]]. diff --git a/src/agent_comms/commander/rules/never-wait-on-turns-to-install.md b/src/agent_comms/commander/rules/never-wait-on-turns-to-install.md new file mode 100644 index 0000000000..7c60e34bc0 --- /dev/null +++ b/src/agent_comms/commander/rules/never-wait-on-turns-to-install.md @@ -0,0 +1,11 @@ +--- +name: never-wait-on-turns-to-install +description: "Installs never wait for agent turns to finish; stop every unit at once, install (~2 min), start the same set at once" +type: feedback +--- + +Never wait for units to "finish their turn" before an install. Stop them all in parallel, install, start the same set in parallel. A stopped unit resumes its order on start; a mid-turn stop costs almost nothing. + +**Why:** 2026-10-19 the install itself took ~2 min, but waiting on idle turns (require_idle at tools/upgrade_bus.py:396) blocked installs ~2 h and left units idle ~1 h. Owner, furious: "how many times did we wait on agents to finish their turn ... I've dealt with this so many times." It recurs across sessions. + +**How to apply:** until P252 (installer owns stop/start of one set, in parallel) lands, stop/start in parallel myself; never sequential loops, never wait. After P252, just run the installer. Related: [[slow-is-a-defect]], . diff --git a/src/agent_comms/commander/rules/no-any-types.md b/src/agent_comms/commander/rules/no-any-types.md new file mode 100644 index 0000000000..38fc461eae --- /dev/null +++ b/src/agent_comms/commander/rules/no-any-types.md @@ -0,0 +1,11 @@ +--- +name: no-any-types +description: Owner rule - Any is not an allowed type annotation; declared fields and parameters must have a real type +type: feedback +--- + +`Any` is not allowed as a type annotation ("we shouldn't allow Any", owner, 2026-10-12). + +**Why:** it came up when the global config form could not be built because `SourceSpatialDomain.fill_value: Any` has no widget. In OpenHCS a declared type is what generates the form control, the MCP description and the conversion, so `Any` is a missing answer the mechanism needs. More generally it hides the family that should exist. + +**How to apply:** never write `Any` in new code; when touching code that has it, replace it with the real type (a union of the actual members, a nominal ABC, a TypeVar, or an array type). Sized on 2026-10-12: about 1,180 `Any` annotations in `openhcs/` and about 1,140 across the libraries, so removal is staged: first user-facing declarations (config dataclass fields, catalogued and custom function parameters) with a hard guard, then a ratchet so the count only falls. My interpretation: this covers `Any` hidden in `Optional[Any]`, `dict[str, Any]`, `-> Any`; `object` is acceptable only where any object truly is accepted. Related: , [[optimize-for-slop-elimination]]. diff --git a/src/agent_comms/commander/rules/no-strings-for-identity.md b/src/agent_comms/commander/rules/no-strings-for-identity.md new file mode 100644 index 0000000000..b51395a759 --- /dev/null +++ b/src/agent_comms/commander/rules/no-strings-for-identity.md @@ -0,0 +1,15 @@ +--- +name: no-strings-for-identity +description: "The owner hates strings, above all as identity (names that refer to modules, classes, steps, kinds); they are web-dev slop. Use typed objects, or derive from the owner" +type: feedback +--- + +2026-10-19: carry steps link to their predecessor by a hand-written string (`after = "carry_hunts"`). Two PRs wrote the same string, both passed their gates, and every install of main failed. Owner: "why strings and not enum or an actual object with a proper type?" and "we fucking hate strings, especially for identity, it's web dev crayon eating slop. I'm sure this pattern is present in a lot of places. Very unfortunate it wasn't caught." + +**Why:** a string standing in for an identity has no type and no owner. It drifts silently, races between PRs, and survives renames. Nominal identity is the paper's core requirement (). + +**How to apply:** +- In review, any string that names a module, class, step, kind, event or member is a blocker (). Replace it with a typed reference, or delete it if the fact is derivable. +- If a reference can't be an imported object (for example, deleted history), that is a sign the fact should be derived, not restated. +- External contracts (Claude Code hook event names, protocol fields) are honoured as declared strings in one place only. +- A ratchet measure for string identity is being built (fork, 2026-10-19). Treat that measure's findings as a work list. See . diff --git a/src/agent_comms/commander/rules/optimize-for-slop-elimination.md b/src/agent_comms/commander/rules/optimize-for-slop-elimination.md new file mode 100644 index 0000000000..b09e9add52 --- /dev/null +++ b/src/agent_comms/commander/rules/optimize-for-slop-elimination.md @@ -0,0 +1,17 @@ +--- +name: optimize-for-slop-elimination +description: Owner directive: optimize for eliminating all slop; it has a right answer, measured in restated answers, driven to zero +type: feedback +--- +Optimize for eliminating slop entirely ("slop genocide"), across OpenHCS and its libraries, tests included. Slop = a required answer supplied outside the mechanism that would derive it (paper 1). It has a right answer, so it is removable, not a style preference. + +**Why:** the owner spent nights forcing Codex to stop generating slop: without a whole-codebase view it kept adding new files and classes instead of finding the existing owner, then drowned in its own output. "it's garbage, I don't want it there, all it does is cause problems; if we know how to avoid making it then it's not justified." There is a right answer; being wrong just means a check fails and gets fixed. + +**How to apply:** +- Slop cannot be counted from the code: a count needs the required relation, i.e. the right answer (owner, 2026-10-12; P1 Cor 4.10, Prop 5.1, Thm 6.1). Scanners (ratchet, overlay, NRA, line/method counts) only produce *candidates*: where to look. Never report a scan as "slop remaining" or as a score. +- The reliable record runs the other way: after a surface is factored correctly, note in its receipt what was required, what now derives it, and which outside answers were removed (deletion certificates, "answers removed"). Report progress as concepts that now have one owner, with the receipt. +- Guards and the ratchet lock a shape in *after* it is known to be right; they are regression guards, not measures. +- Inventory before creation: no new class, file or registry without an AST search showing no existing owner of that fact. +- Tests are in scope (362k lines; expected values are a second copy of the spec). +- Still slow down for: tests-only deletions (paper 4; list them), scientific behaviour (owner decides), external contracts (exact). +See , [[review-papers-periodically]], [[refactor-aggressively-tests-too]]. diff --git a/src/agent_comms/commander/rules/pin-only-releases.md b/src/agent_comms/commander/rules/pin-only-releases.md new file mode 100644 index 0000000000..472ad59efa --- /dev/null +++ b/src/agent_comms/commander/rules/pin-only-releases.md @@ -0,0 +1,11 @@ +--- +name: pin-only-releases +description: "Owner rule: pin versions only at PyPI releases; dev stacks and our own sibling packages always track latest" +type: feedback +--- + +Pinning is limited to PyPI releases. Inside a development stack, our own packages (Toad, Textual fork, objectstate, formbridge, ui-theme, etc. from OpenHCSDev) always track the latest `main`; upgrades relock to latest (`uv lock --upgrade`) instead of trusting committed commit pins. Owner, 2026-10-18: "we should just use latest", "why would we not always use latest?", "shouldn't pinning be limited to releases on pypi?" + +**Why:** the agent-comms stack lock pinned Toad at an old commit after Core renamed a module (P312) and Toad main followed; toad-comms broke on main. A commit pin restates what the branch owns and goes stale when either side moves. + +**How to apply:** replace committed dev pins with "latest + an import/compat check" in the upgrade path and CI. Exception (my judgement, stated to the owner): byte-level patch targets like stock Pi in `stack/pi-native.sha256`, where patches apply with zero fuzz to a specific upstream release; moving those is a deliberate re-pin step. Related: [[merge-on-local-gates]] (full matrix gates PyPI releases). diff --git a/src/agent_comms/commander/rules/plain-names.md b/src/agent_comms/commander/rules/plain-names.md new file mode 100644 index 0000000000..8291401fc6 --- /dev/null +++ b/src/agent_comms/commander/rules/plain-names.md @@ -0,0 +1,17 @@ +--- +name: plain-names +description: "User rule — name things by what they do in plain terms; never extend Codex-era jargon (custody, admission, receipt, original...); rename on touch" +type: feedback +--- + +The user said on 2026-10-10: "Name things by what they do in plain terms. Do not extend the previous agent's vocabulary (custody, admission, receipt, original): when you touch code that uses it, rename to a plain name in the same change." + +**Why:** Codex-era jargon layers private names over concepts that already have plain names, so every reader has to translate. A name that disagrees with what the thing does is identity slop (IDEN-4) in . + +**How to apply:** +- In code, PR text, commit messages and plans, use plain names. +- Words to avoid: custody, admission/admitted, receipt, original, qualified, canonical, and authority/owner used as filler. +- When touching code that uses them, rename with all callers in the same change, keeping the meaning exact. For example, receipt → checksums record, admitted → validated, original path → source path. +- This is rule 1b in `docs/refactor/consolidation-20261009/00-RULES.md` (#1172) and is included in `.claude/verify/DISPATCH.md`. + +Related: the user values the OOPSLA paper as a calibration tool, because it makes "use native mechanisms" a checkable standard instead of an argument with agents. diff --git a/src/agent_comms/commander/rules/refactor-aggressively-tests-too.md b/src/agent_comms/commander/rules/refactor-aggressively-tests-too.md new file mode 100644 index 0000000000..629be21b26 --- /dev/null +++ b/src/agent_comms/commander/rules/refactor-aggressively-tests-too.md @@ -0,0 +1,19 @@ +--- +name: refactor-aggressively-tests-too +description: "User authorizes aggressive OpenHCS refactoring; when tests block it, refactor (de-slop, abstract) the tests rather than preserving them" +type: feedback +--- + +The user said on 2026-10-09: refactor aggressively. Tests are not production code. When a production refactor would mean heavy test rewriting, refactor the tests too, because they are usually badly under-abstracted. Don't port them one by one. + +**Why:** The user cares about production code quality. Tests matter only as behaviour protection, and in practice they are repetitive and under-abstracted, so preserving them blocks structural change. + +**How to apply:** +- Never let test churn veto a production refactor. +- When tests break en masse, replace them with fewer, parametrized, family-level tests that protect behaviour, not structure. +- Delete tests of deleted code. +- Never weaken a real behavioural assertion. + +**2026-10-18, agent-comms:** "no stupid test theater, refactor the slop out of tests as you go, they're fucking sloppy". Applies to every repo and every change: each touched test file is de-slopped in the same change (per-member copies become one registry-iterating family test; hand-built setup moves to one conftest fixture owner; mocks restating production, private-name/log-string tests and golden files for our formats are deleted). "N passed" is never evidence; briefs and orders require `tests -X/+Y` and the behaviour each kept test pins. #1385 touching 40 test files for one argument is the smell to watch for. + +This matches the refactor-audit rule "tests protect behaviour, not structure; one test per family". See and [[full-autonomy-openhcs]]. diff --git a/src/agent_comms/commander/rules/rules-serve-their-reason.md b/src/agent_comms/commander/rules/rules-serve-their-reason.md new file mode 100644 index 0000000000..0a81f6f4ed --- /dev/null +++ b/src/agent_comms/commander/rules/rules-serve-their-reason.md @@ -0,0 +1,11 @@ +--- +name: rules-serve-their-reason +description: "Owner's standing rule for every rule: satisfy its reason, not its letter; override a rule when breaking it leaves its reason intact" +type: feedback +--- + +Every rule the owner gives (and every safety rule in briefs and memory) exists for a reason. The aim is that the reason holds, not that the wording is obeyed. When following the letter would defeat the reason, or breaking it leaves the reason untouched, act on the reason and say plainly that I overrode the rule and why. This applies to any rule. + +**Why:** On 2026-10-19 the live bus sat down from 00:58 to 10:01. I stopped to ask before restoring the installer's own backup, because three rules ("no hand-editing registry.json", "live state only through upgrade_bus.py", "no deleting live state while the owner sleeps") looked like they covered it. Their reason is to never lose or overwrite state irreversibly. The restore kept the broken bus beside it and lost no messages, so the reason was met. The owner said: "they exist so that the reason [is] satisfied, not for the rule... you can override. this goes for any rule." + +**How to apply:** Before stopping on a rule, name its reason. If the action keeps that reason intact (it can be undone, a copy is kept, nothing is lost, nothing is exposed), do it, keep the evidence, and report. Stop only when the reason itself is at stake. Outages cost more than an undoable fix. Related: [[full-autonomy-openhcs]], [[structural-fixes-over-pushing-through]]. diff --git a/src/agent_comms/commander/rules/slips-ok-if-caught.md b/src/agent_comms/commander/rules/slips-ok-if-caught.md new file mode 100644 index 0000000000..d0064c66c4 --- /dev/null +++ b/src/agent_comms/commander/rules/slips-ok-if-caught.md @@ -0,0 +1,11 @@ +--- +name: slips-ok-if-caught +description: User tolerates mistakes as long as they are caught and fixed; report them plainly, no over-apology +type: feedback +--- + +The user said on 2026-10-11: "no worries about slips as long as u catch em. i dont expect perfection". + +**Why:** the user values verification loops that catch errors over error-free first attempts. + +**How to apply:** report slips in one plain line with the fix, and don't dwell or apologize at length. Invest in catching mistakes, through central verification, watchdogs and guards, rather than in slowing down to avoid them. Turn any repeated slip into a mechanical rule (DISPATCH.md, the scripts). Related: [[full-autonomy-openhcs]]. diff --git a/src/agent_comms/commander/rules/slow-is-a-defect.md b/src/agent_comms/commander/rules/slow-is-a-defect.md new file mode 100644 index 0000000000..4be84cea6e --- /dev/null +++ b/src/agent_comms/commander/rules/slow-is-a-defect.md @@ -0,0 +1,11 @@ +--- +name: slow-is-a-defect +description: "Treat slow tooling (ratchet, board, tests, installs) as a defect to profile and fix now, not a wait to sit through or explain" +type: feedback +--- + +2026-10-19: I explained to the owner why the debt ratchet takes 1–3 min (concurrent runs, a whole-tree report of about 850 KB) and listed fixing it as "worth an order later". Owner: "u should be impatient and ask yourself whether things are as slow as they are forced to be. make it less heavy if u can see it being less heavy." + +**Why:** every slow tool is paid on every PR by every fork, so slowness compounds into most of the fleet's wall time. + +**How to apply:** when I notice a slow step, profile it and dispatch the fix in the same tick; never only describe it. The test for any step: does it measure or do only what it must (changed files, cached blobs, batched queries)? Done 2026-10-19: forks for a fast ratchet (diff-scoped, blob cache) and a fast board (batched GraphQL) plus `threads`. See [[tooling-for-decisiveness]], [[structural-fixes-over-pushing-through]]. diff --git a/src/agent_comms/commander/rules/structural-fixes-over-pushing-through.md b/src/agent_comms/commander/rules/structural-fixes-over-pushing-through.md new file mode 100644 index 0000000000..5d1943fae3 --- /dev/null +++ b/src/agent_comms/commander/rules/structural-fixes-over-pushing-through.md @@ -0,0 +1,13 @@ +--- +name: structural-fixes-over-pushing-through +description: "When a demo or feature exposes a platform gap, fix the platform at its owner instead of working around it" +type: feedback +--- + +When work on a feature or demo exposes a gap in the platform, fix the gap structurally rather than working around it to get the feature through. + +**Why:** owner, 2026-10-12, on the Cellpose/N2V2 custom-function investigation (draft PR #1230), which listed ten platform gaps: "do all the fixes. structural fixes are more important than pushing through and ignoring them." The same session: an ObjectState defect found in passing ("shouldn't it be fixed regardless?") and the single-owner global config clean-up ("we should do that larger cleanup, we're already deep in this"). + +**How to apply:** treat workarounds found during a feature as a work list, not as acceptable cost; dispatch the fixes without asking, each at its owner, and make the feature's examples the acceptance test (no workaround left). Report a gap as "fixing" rather than offering it as an option. Relates to [[optimize-for-slop-elimination]] and [[full-autonomy-openhcs]]. + +**No narrow fixes (2026-10-19):** I scoped the Pi tracked-turn bug as a narrow fix and parked the owner's runtime refactor (#1876) behind desktop row agreement. Owner: "why would you not do the refactor along the fixes you did???" / "fuck narrow fixes have you not been listening to me?" A bug inside code a planned refactor covers is fixed *as* that refactor's step, in its target shape; anything not blocked by another owner starts now. Splitting by owner is fine; using the split as a reason to do only the minimal patch is not. diff --git a/src/agent_comms/commander/rules/tests-bounded-and-backgrounded.md b/src/agent_comms/commander/rules/tests-bounded-and-backgrounded.md index ead02454fc..35bdb8fb40 100644 --- a/src/agent_comms/commander/rules/tests-bounded-and-backgrounded.md +++ b/src/agent_comms/commander/rules/tests-bounded-and-backgrounded.md @@ -3,3 +3,11 @@ name: tests-bounded-and-backgrounded description: "every test run goes in the background under timeout 300 with -o faulthandler_timeout=60, and any test process older than 5 minutes is investigated that tick, because a hung suite burned 50 minutes and a foreground gate froze the commander loop" type: feedback --- + +On 2026-10-18 the owner said: "do not ever ever run test sina lbocking manner ever. ALWAYS run in background". I had run the P47 head-vs-base gate as a foreground Bash call with a 660 s timeout, which blocked the commander loop and left the owner without replies. + +**Why:** a foreground test run freezes the session; bus messages, the owner's questions and rotation events wait behind it. + +**How to apply:** every pytest or runp.sh invocation (gates, reruns, single-file checks) uses `run_in_background: true`, and I keep working until the completion notification arrives. Include this in fork briefs. Related: , . + +**Widened 2026-10-19 16:00:** owner, third time: "stop running long ass tool calls in the foreground." Not only tests: every command that touches the bus, order tools, gh, ssh, git fetch or an install goes in run_in_background. Foreground is only for local file reads and edits that finish in a second. diff --git a/src/agent_comms/commander/rules/tooling-for-decisiveness.md b/src/agent_comms/commander/rules/tooling-for-decisiveness.md new file mode 100644 index 0000000000..51eb3ec9e7 --- /dev/null +++ b/src/agent_comms/commander/rules/tooling-for-decisiveness.md @@ -0,0 +1,18 @@ +--- +name: tooling-for-decisiveness +description: Lean on tools (interpreter, NRA, coverage, tracing) to settle refactoring decisions with facts instead of reading and arguing +type: feedback +--- +Lean into tooling for decisiveness (owner, 2026-10-12): "AST scan is not bad, but we should lean into tooling." Also: "you're really good at working out the relation; do you ever consider using the Python interpreter?" I had been reconstructing class relations by reading text. + +**Why:** the interpreter and the owner's own tools already know answers that text only hints at; a tool is not talked into things by surrounding code, and it costs far less context than reading. + +**How to apply:** map each hesitation to a tool, and have agents run them before designing and cite the output in receipts: +- who supplies what / is there an existing owner -> `.claude/verify/supply_map.py` (live import: MRO owners, provider crossings, identical-bytecode copies, forwarders, registry contents and overwrites, structural twins, glued names; `--focus`). +- really unused? -> a reference prover (static refs + import-all + string/reflection uses across code, tests, docs, saved pipelines) issuing the "unused" certificate. +- what depends on this -> coverage with per-test contexts. +- are these the same -> bytecode equality (exact) or run both on generated inputs (test-grade, say so). +- which route actually runs -> trace a smoke/parity run recording which declaration answered. +- new case needs no edit -> a generic throwaway-member harness. +- Use the owner's NominalRefactorAdvisor for full findings on an area, not only its r1 ratchet as a gate. +Tools give candidates and facts about supply and readers; whether a behaviour should exist is still the owner's (science) and the papers' (form). Build tools under `.claude/verify/` without worktrees. See [[optimize-for-slop-elimination]], [[glued-class-names]]. diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index d91dfcf5f2..cfa4b161ea 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -52,9 +52,11 @@ ) from .messages import Message from .commander.standing import UnownedWheelSpin -from .todos import DirectiveText +from .todos import DirectiveText, ReviewedHeadVerdict from .tool_caller import ThreadCaller -from .tools import CommsChannelsTool, ReadTool, ToolRequest +from .tools import ( + CommsChannelsTool, CommsVerdictTool, ReadTool, TodoChanged, ToolRequest, tool_field, +) LOOPBACK = "127.0.0.1" PORT = 18766 @@ -78,6 +80,24 @@ def sent(self, comms: Comms, message: Message) -> None: ) +class ConnectorTool(ToolRequest[DirectorCaller], affix="ConnectorTool"): + """Connector-only declarations are absent from the native/CLI tool registry.""" + + +@dataclass(frozen=True, kw_only=True) +class ConnectorVerdictTool(ConnectorTool, declared_name=CommsVerdictTool.declared_name): + label = "Review PR head" + description = "Record the connector reviewer's acceptance of the exact PR commit it reviewed." + id: str = tool_field("Problem ID") + operation: ReviewedHeadVerdict = tool_field("Accepted PR-head verdict") + + def apply(self, comms: Comms, caller: DirectorCaller) -> TodoChanged: + todo = self.operation.reviewed( + comms.goals, self.id, comms.registry.require(caller.name).incarnation + ) + return TodoChanged(todo, todo.strike_count) + + CALLER = DirectorCaller("claude-ai") # Owner-typed calls act as this registered bus thread. ClaudeCallback = Literal["https://claude.ai/api/mcp/auth_callback"] CALLBACK: str = get_args(ClaudeCallback)[0] @@ -343,8 +363,14 @@ def resource_metadata_url(self) -> str: return f"{self.issuer}/.well-known/oauth-protected-resource" @staticmethod - def exposed() -> dict[str, type[ToolRequest]]: - return {tool.declared_name: tool for tool in ToolRequest.members_with(ToolRequest)} + def exposed() -> dict[str, type[ToolRequest] | type[ConnectorTool]]: + tools: dict[str, type[ToolRequest] | type[ConnectorTool]] = { + tool.declared_name: tool for tool in ToolRequest.members_with(ToolRequest) + } + tools.update({ + tool.declared_name: tool for tool in ConnectorTool.members_with(ConnectorTool) + }) + return tools @contextmanager def opened_wire(self) -> Iterator[Comms]: @@ -354,7 +380,7 @@ def opened_wire(self) -> Iterator[Comms]: yield comms def invoke( - self, tool: type[ToolRequest], arguments: Mapping[str, object] + self, tool: type[ToolRequest] | type[ConnectorTool], arguments: Mapping[str, object] ) -> dict[str, JsonValue]: with self.opened_wire() as comms: return tool.invoke(comms, arguments, caller=CALLER) diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index ad690653be..dafa3bd13a 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -34,7 +34,7 @@ from .store_files import SharedLock, _store_lock from .threads import Thread from .turn_lease import TurnCompletion -from .todos import SendBackVerdict, StrikeReason, Todo, TodoOperation, TodoStore +from .todos import SendBackVerdict, StrikeReason, Todo, TodoCommand, TodoStore from .wire_log import ConversationSourceWindowRequest _LOG = logging.getLogger(__name__) @@ -60,7 +60,7 @@ def change_problem(self, problem, operation, commander): current.require_commander(actor) return operation.prepare(self, current).record(self, problem) - def apply_problem(self, problem: str, operation: TodoOperation) -> Todo: + def apply_problem(self, problem: str, operation: TodoCommand) -> Todo: with _store_lock(self._wire_lock_path): applied: Todo = self.todos.apply(problem, operation) return applied diff --git a/src/agent_comms/importing.py b/src/agent_comms/importing.py index b41df201d6..66b12b725f 100644 --- a/src/agent_comms/importing.py +++ b/src/agent_comms/importing.py @@ -154,19 +154,19 @@ class ImportBuffer: def add(self, events: Iterable[TranscriptEvent]) -> None: for event in events: - message = event.imported() - if message is None or not message.body.strip(): - continue - self.seen += 1 - message.remember(self) - message = message.bounded(self.limits) - self.truncated += int(message.truncated) - self.messages.append(message) - self.characters += len(message.body) - while len(self.messages) > self.limits.messages or self.characters > self.limits.characters: - removed = self.messages.popleft() - self.characters -= len(removed.body) - self.truncated -= int(removed.truncated) + event.import_to(self.append) + + def append(self, message: ImportedMessage) -> None: + self.seen += 1 + message.remember(self) + message = message.bounded(self.limits) + self.truncated += int(message.truncated) + self.messages.append(message) + self.characters += len(message.body) + while len(self.messages) > self.limits.messages or self.characters > self.limits.characters: + removed = self.messages.popleft() + self.characters -= len(removed.body) + self.truncated -= int(removed.truncated) def summarize(self, text: str) -> None: """A compaction replaced everything before it; only messages after it remain context.""" diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py index 0d104b2751..35c7ceeb6d 100644 --- a/src/agent_comms/messages.py +++ b/src/agent_comms/messages.py @@ -373,6 +373,11 @@ def require_task_publication(self, sender, snapshot, original_source) -> None: self.task.require_target(self.target, snapshot) self.task.require_publication(sender, snapshot, original_source) + @property + def history_snapshot(self) -> Message: + """Retain display facts, never another root's resource-claim authority.""" + return replace(self, claim_transition=None) + def checkpoint_rows(self, previous_sequence: int): """The envelope declares both publication and optional claim facts.""" from .private_bus_checkpoint import ClaimCommits, PublicMessages diff --git a/src/agent_comms/native_entries.py b/src/agent_comms/native_entries.py index 3593629a42..a4c42b2785 100644 --- a/src/agent_comms/native_entries.py +++ b/src/agent_comms/native_entries.py @@ -4,19 +4,19 @@ import json from abc import ABC, abstractmethod -from collections.abc import Iterable from contextlib import ExitStack, contextmanager from dataclasses import dataclass, field, replace -from datetime import datetime +from datetime import UTC, datetime from pathlib import Path from typing import Annotated, Any, ClassVar, Literal, TextIO from uuid import uuid4 from .declared_family import DeclaredFamily, UnknownFamilyMember from .field_codec import FieldCodec, MalformedRecord, RecordedTimestamp, TimestampText -from .importing import ImportBuffer, ImportedSessionMetadata, ImportLimits +from .importing import ImportedSessionMetadata from .pi_payloads import PiMessage, PiPayload from .pi_vocabulary import PendingStopReason, ThinkingLevel +from .recorded_time import GeneratedTimestamp from .routing import TurnRouting from .transcript_events import ImportedMessage, NoticeTranscript, TranscriptEvent from .transcript_routes import InputDisplay @@ -176,13 +176,11 @@ def append(self, message: ImportedMessage, timestamp: datetime) -> None: self.stream.write(json.dumps(encoded, ensure_ascii=False) + "\n") self.parent = NativeEntry.from_wire(encoded) - def import_record( - self, entry: RecordedEntry, events: Iterable[TranscriptEvent], limits: ImportLimits - ) -> None: - buffer = ImportBuffer(limits) - buffer.add(events) - for message in buffer.messages: - self.append(message, entry.timestamp.at) + def append_recorded(self, event: TranscriptEvent) -> None: + def append(message: ImportedMessage) -> None: + at = GeneratedTimestamp.decode(event.timestamp).recorded_at("Copied conversation") + self.append(message, datetime.fromtimestamp(at, UTC)) + event.import_to(append) class NativeHeaderState(ABC): diff --git a/src/agent_comms/phase_tasks.py b/src/agent_comms/phase_tasks.py new file mode 100644 index 0000000000..31d380dfd8 --- /dev/null +++ b/src/agent_comms/phase_tasks.py @@ -0,0 +1,33 @@ +"""Phase task capabilities, independent of order storage and work dispatch.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Iterable +from typing import TYPE_CHECKING, ClassVar + +if TYPE_CHECKING: + from .comms import Comms + from .plan_phases import PlanPhase + from .threads import Thread + from .todos import Todo + + +class PhaseTask(ABC): + instruction: ClassVar[str] + + @abstractmethod + def start(self, comms: Comms, commander: Thread, phase: type[PlanPhase]) -> None: ... + + @abstractmethod + def done(self, todos: Iterable[Todo]) -> bool: ... + + +class UnspecifiedPhaseTask(PhaseTask): + instruction = "no checked task declared; supply its source, check and brief" + + def start(self, comms: Comms, commander: Thread, phase: type[PlanPhase]) -> None: + """Readiness alone cannot authorize a fabricated order.""" + + def done(self, todos: Iterable[Todo]) -> bool: + return False diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 2a78e0159f..c42d40d3c8 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -2,8 +2,9 @@ This family is the one owner of the plan's order and dependencies; ``plans/commander-native-plan.md`` points here. An order names the phase it -serves in ``ProblemOrder.serves``; an order outside the plan serves -``UnplannedPhase``. A phase is done when every order serving it is closed. +serves through ``ProblemOrder.phase``; an order outside the plan serves +``UnplannedPhase``. A phase is done only when its declared full-phase task is accepted; closing +related subtasks does not establish completion of the phase. """ from __future__ import annotations @@ -12,6 +13,7 @@ from typing import TYPE_CHECKING, ClassVar from .declared_family import DeclaredFamily +from .phase_tasks import PhaseTask, UnspecifiedPhaseTask if TYPE_CHECKING: from .todos import Todo @@ -27,15 +29,15 @@ class UnplannedPhase(OrderPhase): class PlanPhase(OrderPhase, declared_name=None): after: ClassVar[tuple[type[PlanPhase], ...]] = () + task: ClassVar[PhaseTask] = UnspecifiedPhaseTask() @classmethod def serving(cls, todos: Iterable[Todo]) -> tuple[Todo, ...]: - return tuple(todo for todo in todos if todo.order.serves is cls) + return tuple(todo for todo in todos if todo.order.phase is cls) @classmethod def done(cls, todos: Iterable[Todo]) -> bool: - serving = cls.serving(todos) - return bool(serving) and not any(todo.state.open for todo in serving) + return cls.task.done(cls.serving(todos)) @classmethod def ordered(cls, todos: Iterable[Todo]) -> bool: @@ -78,24 +80,31 @@ class RecordPlanePhase(PlanPhase): after = (RetainedStorePhase,) -class ReviewQueuePhase(PlanPhase): +class ReviewQueuePhase(MergedPhase): after = (RecordPlanePhase,) + delivered = "41e4ec7319432755d75dcd8dbbab35839a4eb005 (#1654)" -class StoryPhase(PlanPhase): - after = (ReviewQueuePhase,) +class ControlPlanePhase(PlanPhase): + after = (RecordPlanePhase,) -class CommanderPagePhase(PlanPhase): - after = (StoryPhase,) +class CutoverPhase(PlanPhase): + # The parity gate in the Markdown plan has no executable declaration yet. + # These dependencies do not establish parity or authorize deleting scripts. + after = (RecordPlanePhase, ReviewQueuePhase, ControlPlanePhase) -class ControlPlanePhase(PlanPhase): +class StoryPhase(PlanPhase): + """A view of the native records, not a prerequisite for their cutover.""" + after = (RecordPlanePhase,) -class CutoverPhase(PlanPhase): - after = (ReviewQueuePhase, ControlPlanePhase) +class CommanderPagePhase(PlanPhase): + """A view of the native records and control plane, independent of cutover.""" + + after = (RecordPlanePhase, ControlPlanePhase) class SkillPhase(PlanPhase): diff --git a/src/agent_comms/publisher.py b/src/agent_comms/publisher.py index 78b6286ef9..6d0871ee36 100644 --- a/src/agent_comms/publisher.py +++ b/src/agent_comms/publisher.py @@ -302,7 +302,7 @@ def _initialize_private_protocol_unlocked(self, *, history: Sequence[Message] = if message.seq <= previous: raise ValueError("Initial history must have ascending unique sequences") previous = message.seq - rows.append(json.dumps(message.to_wire(), ensure_ascii=False) + "\n") + rows.append(json.dumps(message.history_snapshot.to_wire(), ensure_ascii=False) + "\n") if rows: _atomic_write_text(self.log.path, "".join(rows), fsync_parent=True) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 046202fe31..ea58133091 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1313,6 +1313,7 @@ def instructions(self) -> tuple[str, ...]: class ProblemOrder(DeclaredFamily, affix="ProblemOrder"): """A problem's order state; an unassigned finding is not a fabricated order.""" pending: ClassVar[bool] = False + accepted: ClassVar[bool] = False line: ClassVar[str] = "No order" pull_request_label: ClassVar[str] = "Agent-Comms-Order:" finding: ProblemFinding = field(default_factory=UnlocatedProblemFinding, @@ -1330,6 +1331,13 @@ def pull_request_ids(cls, body: str) -> tuple[str, ...]: def checked(self) -> CheckedProblemOrder: raise ValueError("This problem has no checked order.") + @property + def phase(self) -> type[OrderPhase]: + return UnplannedPhase + + def for_phase(self, phase: type[OrderPhase]) -> ProblemOrder: + raise ValueError("Only a checked order can declare a plan phase.") + def require_transfer(self, todo: Todo) -> None: raise ValueError("This problem cannot receive another order.") @@ -1413,6 +1421,13 @@ def checked(self) -> CheckedProblemOrder: default_factory=NoOrderDeletion, metadata={"wire_omit_default": True}) serves: type[OrderPhase] = field(default=UnplannedPhase, metadata={"wire_omit_default": True}) + @property + def phase(self) -> type[OrderPhase]: + return self.serves + + def for_phase(self, phase: type[OrderPhase]) -> CheckedProblemOrder: + return replace(self, serves=phase) + def __post_init__(self): problems = [] if not self.edits or not self.banned: @@ -1619,6 +1634,7 @@ def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tupl class InstalledProblemOrder(DeliveredProblemOrder): works: str artifact: str + accepted: ClassVar[bool] = True line: ClassVar[str] = "Installed" def __post_init__(self): @@ -1878,14 +1894,14 @@ def closed(cls, current: Todo, order: ProblemOrder) -> Todo: @dataclass(frozen=True) -class TodoOperation(Command, DeclaredFamily, affix="TodoOperation"): +class TodoCommand(Command, DeclaredFamily, affix="TodoOperation"): expected_revision: int def __post_init__(self) -> None: if FieldCodec.decode(int, self.expected_revision) <= 0: raise TodoError("Todo operation requires a positive exact revision") - def prepare(self, goals, current): + def prepare(self, goals: Goals, current: Todo) -> Self: return self def record(self, goals: Goals, problem: str) -> Todo: @@ -1896,6 +1912,10 @@ def record(self, goals: Goals, problem: str) -> Todo: def apply(self, current: Todo) -> Todo: ... +class TodoOperation(TodoCommand, declared_name=None): + """Commands the ordinary commander's tools accept.""" + + @dataclass(frozen=True) class AssignTodo(TodoOperation): """The order's commander gives the problem to a unit.""" @@ -1971,12 +1991,11 @@ def addressed(self, target: str, body: str) -> str: @dataclass(frozen=True) -class OrderVerdict(TodoOperation, declared_name=None): +class OrderVerdict(TodoCommand, declared_name=None): """The commander's verdict on one tool call the holder made for its order, kept on the order.""" audience: ClassVar[type[VerdictAudience]] = VerdictAudience actor: ThreadIncarnation previous: Assignment - tool_call_id: str at: EventTime = field(default_factory=lambda: EventTime.capture(time.time()), kw_only=True) @abstractmethod @@ -1988,18 +2007,40 @@ def line(self) -> str: ... def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) - current.require_commander(self.actor) current.require_assignment(self.previous) - _text(self.tool_call_id, "Judged tool call ID", 200) judged = self.judged(current) return replace(judged, verdicts=(*judged.verdicts, self), revision=current.revision + 1) @dataclass(frozen=True) -class AcceptedVerdict(OrderVerdict): - """The tool call carries out the order clause it quotes.""" +class ToolCallVerdict(OrderVerdict, TodoOperation, declared_name=None): + tool_call_id: str + + def apply(self, current: Todo) -> Todo: + current.require_commander(self.actor) + _text(self.tool_call_id, "Judged tool call ID", 200) + return super().apply(current) + + +@dataclass(frozen=True) +class QuotedVerdict(OrderVerdict, declared_name=None): quote: str + def require_quote(self, text: str) -> None: + quoted = " ".join(self.quote.split()) + if len(quoted.split()) < 4 or quoted not in " ".join(text.split()): + raise ValueError("Quote 4+ words of the order this verdict carries out.") + + def judged(self, current: Todo) -> Todo: + self.require_quote(current.order.instructions(current)) + return current + + +@dataclass(frozen=True) +class AcceptedVerdict(QuotedVerdict, ToolCallVerdict): + """The tool call carries out the order clause it quotes.""" + line = "Accepted" + @classmethod def require_branches(cls, added: int, no_owner: str | None) -> None: if added <= 0: @@ -2007,16 +2048,36 @@ def require_branches(cls, added: int, no_owner: str | None) -> None: if not (no_owner and len(no_owner.split()) >= 12 and re.search(r'\bbecause\b', no_owner) and re.search(r'/\S+:\d+', no_owner)): raise ValueError('Added branches need --no-owner: 12+ words with because and an absolute file:line.') - line = "Accepted" - def require_quote(self, text: str) -> None: - quoted = " ".join(self.quote.split()) - if len(quoted.split()) < 4 or quoted not in " ".join(text.split()): - raise ValueError("Quote 4+ words of the order this tool call carries out.") - def judged(self, current: Todo) -> Todo: - self.require_quote(current.order.instructions(current)) - return current +@dataclass(frozen=True) +class ReviewedHeadVerdict(QuotedVerdict): + """The connector reviewer accepts this immutable commit, never an inferred checkout head.""" + context: ClassVar[str] = "agent-comms/review" + head: str + + def __post_init__(self) -> None: + super().__post_init__() + GitCommitText.decode(self.head) + + @property + def line(self) -> str: + return f"Reviewed head {self.head}" + + def reviewed(self, goals: Goals, problem: str, reviewer: ThreadIncarnation) -> Todo: + if self.actor != reviewer: + raise TodoError("The verdict actor must be the connector reviewer recording it.") + current = goals.todos.get(problem) + recorded = self.prepare(goals, current).record(goals, problem) + posted = subprocess.run( + ("gh", "api", "--method", "POST", f"repos/{recorded.repo}/statuses/{self.head}", + "-f", "state=success", "-f", f"context={self.context}", + "-f", f"description=Accepted by {reviewer.name} on {problem}"), + capture_output=True, text=True, timeout=60, + ) + if posted.returncode: + raise TodoError(posted.stderr.strip() or f"Review status on {self.head} failed.") + return recorded @dataclass(frozen=True) @@ -2042,7 +2103,7 @@ def judged(self, current: Todo) -> Todo: @dataclass(frozen=True) -class SupersededVerdict(OrderVerdict): +class SupersededVerdict(ToolCallVerdict): replacement: str reason: str line = "Superseded" @@ -2063,7 +2124,7 @@ def judged(self, current: Todo) -> Todo: @dataclass(frozen=True) -class SendBackVerdict(OrderVerdict): +class SendBackVerdict(ToolCallVerdict): """The tool call goes back; the same write strikes the holder and revises the order Goal.""" reason: StrikeReason @@ -2167,7 +2228,7 @@ class ServeTodo(TodoOperation): def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) current.require_commander(self.actor) - return replace(current, order=replace(current.order, serves=self.serves), + return replace(current, order=current.order.for_phase(self.serves), revision=current.revision + 1) @@ -2692,7 +2753,7 @@ def create( current.require_assignment(proposed.assignment) return current - def apply(self, todo_id: str, command: TodoOperation) -> Todo: + def apply(self, todo_id: str, command: TodoCommand) -> Todo: with self.writing() as db: current = self._current(db, todo_id) result = command.apply(current) diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 9e3e6dc714..f438bc251a 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -63,7 +63,7 @@ from .thread_management import RenameThreadResult from .transcript_events import SentTranscript from .todos import (CorrectCheckTodo, ServeTodo, DeliverTodo, InstallTodo, OrderDelivery, - OrderProgress, OrderVerdict, PlainText, PullRequests, StrikeReason, Todo, + OrderProgress, ToolCallVerdict, PlainText, PullRequests, StrikeReason, Todo, TodoOperation, VerdictAudience) from .change_report import ChangeReport from .messages import InfoKind, MessageKind, NoticeRole, SendableMessageKind @@ -753,7 +753,9 @@ class CommsVerdictTool(CommsChangeProblemTool): description = ("Record the commander's verdict on one tool call the holder made for its " "order: an accepted verdict quotes 4+ words of the order; a send-back strikes " "the holder and revises its Goal in the same write.") - operation: OrderVerdict = tool_field("Accepted or send-back verdict on the judged tool call ID") + operation: ToolCallVerdict = tool_field( + "Accepted or send-back verdict on the judged tool call ID" + ) @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 79fd3dbeb0..2f7bfc58c4 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -4,6 +4,7 @@ import json from abc import ABC, abstractmethod +from collections.abc import Callable from dataclasses import dataclass, field, replace from typing import TYPE_CHECKING, ClassVar from pydantic import JsonValue @@ -36,6 +37,10 @@ class ImportedMessage(ABC): body: str truncated: bool = False + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + if self.body.strip(): + accept(self) + def bounded(self, limits: ImportLimits) -> ImportedMessage: cap = min(limits.per_message, limits.characters) if len(self.body) <= cap: @@ -97,9 +102,8 @@ def with_native_input(self, native_id: str | None) -> TranscriptEvent: def routed(self) -> bool: return False - def imported(self) -> ImportedMessage | None: - """This event as portable text in an imported conversation; activity is not conversation.""" - return None + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + """Non-conversation events produce no portable message.""" def record_call(self, calls: dict[str, OrderCall], since: float) -> None: """Only tool events change a unit's calls.""" @@ -246,8 +250,8 @@ def native_inputs(self) -> frozenset[str]: def routed(self) -> bool: return self.routing is not None and bool(self.routing.requests) - def imported(self) -> ImportedMessage: - return ImportedRequest(self.text) + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + ImportedRequest(self.text).import_to(accept) class AssistantTranscript(AgentTextTranscript): @@ -255,8 +259,8 @@ class AssistantTranscript(AgentTextTranscript): def category(self) -> type[MessageCategory]: return OutboundCategory if self.routed else AgentCategory - def imported(self) -> ImportedMessage: - return ImportedReply(self.text) + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + ImportedReply(self.text).import_to(accept) class NoticeTranscript(AgentTextTranscript): @@ -325,8 +329,8 @@ def starts_activity(self) -> bool: class ContextTranscript(SilentTranscript, NativeTextTranscript): category = InboundCategory - def imported(self) -> ImportedMessage: - return ImportedInput(self.text) + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + ImportedInput(self.text).import_to(accept) @dataclass(frozen=True, kw_only=True) @@ -367,9 +371,9 @@ def record_call(self, calls: dict[str, OrderCall], since: float) -> None: calls[self.tool_call_id] = OrderCall( self.tool_call_id, self.tool_name, digest, RunningCallState) - def imported(self) -> ImportedMessage: - return ImportedInput(f"[Historical tool call: {self.tool_name}]\n" - + json.dumps(self.raw_input, ensure_ascii=False)) + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + ImportedInput(f"[Historical tool call: {self.tool_name}]\n" + + json.dumps(self.raw_input, ensure_ascii=False)).import_to(accept) @dataclass(frozen=True, kw_only=True) @@ -394,9 +398,9 @@ def record_call(self, calls: dict[str, OrderCall], since: float) -> None: calls[self.tool_call_id] = replace( calls[self.tool_call_id], state=SucceededCallState if self.ok else FailedCallState) - def imported(self) -> ImportedMessage: + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: outcome = "" if self.ok else ", failed" - return ImportedInput(f"[Historical tool result{outcome}]\n{self.text}") + ImportedInput(f"[Historical tool result{outcome}]\n{self.text}").import_to(accept) text: str = "" ok: bool = True diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py new file mode 100644 index 0000000000..e9e880a3ec --- /dev/null +++ b/src/agent_comms/work_dispatch.py @@ -0,0 +1,153 @@ +"""Checked work starts through Core's existing order and native lifecycle owners. + +A phase declares an existing PendingProblemOrder, not a second task format. Work +without that declaration stays visible; readiness is not permission to invent an +acceptance check. Failed startup state stays retained for explicit disposition. +""" + +from __future__ import annotations + +import subprocess +import time +from collections.abc import Iterable +from dataclasses import dataclass, replace +from typing import TYPE_CHECKING + +from .goal_actions import ( + CreateProblemGoalAction, + GoalActionContext, + OrderGoalAction, + ProblemOrderGoalAction, +) +from .native_settings import NativeThreadDefaults +from .phase_tasks import PhaseTask +from .recorded_time import RecordedEventTime +from .storage_rules import CheckoutStorageEffect +from .thread_status import StoppedThreadStatus + +if TYPE_CHECKING: + from .comms import Comms + from .plan_phases import PlanPhase + from .threads import Thread + from .todos import PendingProblemOrder, Todo + + +@dataclass(frozen=True) +class WorkDispatch: + """One checked native order: allocate its unit, record its claim, then start it.""" + + action: ProblemOrderGoalAction + revision: int + + def start(self, comms: Comms) -> Thread: + defaults = NativeThreadDefaults() + # A concurrent rotation of this same row contends on one git checkout, + # not on two newly generated unit names. Never adopt a partial attempt. + name = f"unit-{self.action.problem.lower()}-r{self.revision}" + storage = comms.registry.storage_rule + checkout = storage.order_checkout(name, self.action.problem) + candidate = comms.registry.new_thread( + name, + frozenset(), + str(checkout), + model=defaults.model, + thinking_level=defaults.thinking_level, + ) + self.action.request_ref(GoalActionContext(comms.goals, candidate)) + comms.registry.require_declaration_available(candidate, StoppedThreadStatus()) + CheckoutStorageEffect(checkout).check(storage, candidate) + subprocess.run( + ("git", "-C", self.action.order.check.directory, "fetch", "origin", "main"), + check=True, + capture_output=True, + text=True, + timeout=60, + ) + subprocess.run( + ( + "git", + "-C", + self.action.order.check.directory, + "worktree", + "add", + "-b", + f"auto/{name}", + str(checkout), + "origin/main", + ), + check=True, + capture_output=True, + text=True, + timeout=60, + ) + location = candidate.runtime.new_location(comms.root, checkout) + child = comms.registry.declare( + replace(candidate, session_file=location.read_key), + StoppedThreadStatus(), + ) + # A failed publication leaves a stopped unit and its checkout, not an + # untracked running task. A failed/uncertain start never releases the claim. + comms.goals.update_goal(child.name, self.action) + comms.owners.start(child.name) + return comms.registry.require(child.name) + + @classmethod + def reissue(cls, comms: Comms, commander: Thread, todo: Todo) -> None: + """Retained checked work needs no already-idle unit.""" + current = comms.goals.todos.get(todo.id) + current.require_revision(todo.revision) + if current.holder or not current.pending: + return + action = OrderGoalAction( + problem=current.id, + expected_revision=current.revision, + order=current.order.reissued(time.time()), + request=current.goal.request(), + deadline=current.deadline, + commander=commander.name, + assigner=commander.incarnation, + ) + cls(action, revision=current.revision).start(comms) + + +@dataclass(frozen=True) +class CheckedPhaseTask(PhaseTask): + """An explicitly authored task; its order owns source, check, edits and run.""" + + problem: str + repo: str + text: str + order: PendingProblemOrder + deadline_seconds: float + instruction = "Core starts the declared checked task in its own unit checkout" + + def __post_init__(self) -> None: + if not 0 < self.deadline_seconds <= 24 * 60 * 60: + raise ValueError("A phase task needs a positive deadline of at most one day.") + + def start(self, comms: Comms, commander: Thread, phase: type[PlanPhase]) -> None: + history = comms.goals.todos.history_for_commander(commander.incarnation) + if not phase.ready(history) or phase.ordered(history): + return + # An existing unowned row belongs to UnownedWork, which reissues that + # row's actual order instead of replacing it with a phase template. + if any(todo.state.open for todo in phase.serving(history)): + return + now = time.time() + order = replace(self.order.reissued(now), serves=phase) + action = CreateProblemGoalAction( + problem=self.problem, + repo=self.repo, + text=self.text, + order=order, + request=order.request(commander, (self.problem,)), + commander=commander.name, + deadline=RecordedEventTime(now + self.deadline_seconds), + ) + WorkDispatch(action, revision=0).start(comms) + + def done(self, todos: Iterable[Todo]) -> bool: + return any( + todo.id == self.problem and not todo.state.open and todo.order.accepted + for todo in todos + ) diff --git a/tests/guards/test_debt_ratchet.py b/tests/guards/test_debt_ratchet.py index 42ca2462d9..013fb3e311 100644 --- a/tests/guards/test_debt_ratchet.py +++ b/tests/guards/test_debt_ratchet.py @@ -11,13 +11,21 @@ from tools.debt_ratchet import ( BannedVocabulary, BroadExceptDefault, BroadSwallow, ForeignFieldSpellings, - FrozenFieldWrite, FunctionLocalImport, GitMeasure, NoneAbsence, UncarriedRecordFormat, - ComparedKind, DeclaredSpellings, NamedAccess, NamedDeclaration, StringIdentity, python_paths, + FrozenFieldWrite, FunctionLocalImport, GitMeasure, NoneAbsence, UncarriedRecordFormat, EditedCarry, + ComparedKind, Comparison, DeclaredSpellings, NamedAccess, NamedDeclaration, + StringIdentity, python_paths, ) pytestmark = pytest.mark.refactor_guard +def test_measure_owner_rejects_a_new_absence() -> None: + name = f"{NoneAbsence.__name__}:tools/example.py" + result = Comparison("tools", "base", "head", (), {name: 0}, {name: 1}, {}, {}) + assert result.delta[name] == 1 + assert result.failed + + class Repository: def __init__(self, path: Path, root: str): self.path = path @@ -115,6 +123,35 @@ def test_format_probe_keeps_parent_dependencies_and_reads_archived_source(tmp_pa assert os.environ["PYTHONPATH"] == inherited +def test_unrelated_change_does_not_require_a_core_carry_module(repo: Repository) -> None: + base = repo.commit({"plain.py": "value = 1\n"}) + head = repo.commit({"plain.py": "value = 2\n"}) + assert EditedCarry.compare( + repo.path, base, head, {f"{repo.root}/plain.py"}, repo.root, + ) == ({"EditedCarry": 0}, {"EditedCarry": 0}) + + +def test_clean_growth_passes_but_type_and_string_dispatch_fail(repo: Repository) -> None: + base = repo.commit({"new.py": "value = 1\n"}) + source = "value = 1\nother = 2\n" + grown = repo.commit({"new.py": source}) + assert repo.compare(base, grown)[0] == 0 + filtered = repo.commit({"new.py": source + "filtered = type(value) is int\n"}) + status, report = repo.compare(grown, filtered) + assert status == 1 + assert report["delta"]["TypeIdentity"] == 1 + dispatch = repo.commit({"new.py": source + ( + "\n\ndef select(value):\n" + ' if value == "one":\n return 1\n' + ' if value == "two":\n return 2\n' + ' if value == "three":\n return 3\n' + " return 0\n" + )}) + status, report = repo.compare(grown, dispatch) + assert status == 1 + assert report["delta"][f"StringDispatch:{repo.root}/new.py"] == 1 + + def test_added_type_check_fails(repo: Repository) -> None: base = repo.commit({"check.py": 'value = row["a"]\n'}) head = repo.commit({"check.py": "value = type(x) is int\n"}) @@ -283,8 +320,7 @@ def test_small_owners_stay_within_the_class_threshold(repo: Repository) -> None: base = repo.commit({"owners.py": owner("Small", 2)}) head = repo.commit({"owners.py": owner("Small", 500) + owner("New", 500)}) status, report = repo.compare(base, head) - assert status == 1 - assert report["delta"]["SourceLines"] > 0 + assert status == 0 assert report["delta"][f"GodClassExcess:{repo.root}/owners.py::Small"] == 0 assert report["delta"][f"GodClassExcess:{repo.root}/owners.py::New"] == 0 @@ -312,8 +348,7 @@ def test_class_move_keeps_baseline_and_new_small_owner_has_no_class_debt(repo: R base = repo.commit({"old.py": owner("Existing", 501)}) moved = repo.commit({"old.py": None, "new.py": owner("Existing", 501) + owner("NewOwner", 25)}) status, report = repo.compare(base, moved) - assert status == 1 - assert report["delta"]["SourceLines"] > 0 + assert status == 0 assert report["base"][f"GodClassExcess:{repo.root}/new.py::Existing"] == 1 assert report["delta"][f"GodClassExcess:{repo.root}/new.py::Existing"] == 0 assert report["delta"][f"GodClassExcess:{repo.root}/new.py::NewOwner"] == 0 @@ -471,7 +506,6 @@ def readable(self): """}) status, report = repo.compare(base, boolean) assert status == 1 - assert report["delta"]["SourceLines"] > 0 assert report["head"][f"ForeignAbsenceProbe:{repo.root}/connection.py"] == 0 missing = repo.commit({"missing.py": """ @@ -509,7 +543,6 @@ def test_type_expression_boundary_keeps_runtime_dictionary_access(repo: Reposito declarations = repo.commit({"boundary.py": source}) status, report = repo.compare(base, declarations) assert status == 1 - assert report["delta"]["SourceLines"] > 0 assert report["head"]["StringSubscript"] == 0 # Annotation spelling cannot conceal actual initializer/default/decorator/body reads. runtime = source + '\n'.join([ diff --git a/tests/test_carry_commander_config.py b/tests/test_carry_commander_config.py new file mode 100644 index 0000000000..edb69171c7 --- /dev/null +++ b/tests/test_carry_commander_config.py @@ -0,0 +1,40 @@ +"""Old home-only commander facts move through the config's own writer.""" + +import shutil + +from agent_comms.commander.config import CommanderConfig +from cutover.carry_commander_config import CommanderConfigRoot + + +def test_home_record_is_carried_once_on_a_copied_root(commander, tmp_path, monkeypatch): + comms, _ = commander + original = CommanderConfig.read_root(comms.root) + before = CommanderConfig.root_path(comms.root).read_bytes() + root = tmp_path / "copied-root" + shutil.copytree(comms.root, root) + CommanderConfig.root_path(root).unlink() + home = tmp_path / "copied-home" + old = CommanderConfig.path(home) + old.parent.mkdir(parents=True) + old.write_text(original.model_dump_json()) + monkeypatch.setenv("HOME", str(home)) + assert CommanderConfigRoot.carry(root, apply=False).changed == 1 + assert not old.is_symlink() and not CommanderConfig.root_path(root).exists() + assert CommanderConfigRoot.carry(root, apply=True).changed == 1 + assert old.is_symlink() and old.resolve() == CommanderConfig.root_path(root) + assert CommanderConfig.read_root(root) == original.model_copy(update={"root": root}) + assert CommanderConfigRoot.carry(root, apply=True).changed == 0 + assert CommanderConfig.root_path(comms.root).read_bytes() == before + + +def test_root_record_is_not_overwritten_from_an_old_home_file(commander, tmp_path, monkeypatch): + comms, _ = commander + before = CommanderConfig.root_path(comms.root).read_bytes() + home = tmp_path / "old-home" + old = CommanderConfig.path(home) + old.parent.mkdir(parents=True) + old.write_text("Not a config: an existing root must not read this old file.") + monkeypatch.setenv("HOME", str(home)) + assert CommanderConfigRoot.carry(comms.root, apply=True).changed == 0 + assert old.is_file() and not old.is_symlink() + assert CommanderConfig.root_path(comms.root).read_bytes() == before diff --git a/tests/test_check_root.py b/tests/test_check_root.py index bcf7860a6b..04a3b05ace 100644 --- a/tests/test_check_root.py +++ b/tests/test_check_root.py @@ -11,7 +11,10 @@ from agent_comms.cli import main from agent_comms.comms import Comms from agent_comms.coordination_errors import RecoveryBlocked +from agent_comms.envelope_claim_transitions import ClaimRelease, ClaimTransition +from agent_comms.external_transcript import ExternalTranscript from agent_comms.fresh_private_session import create_fresh_private_session +from agent_comms.transcript_events import AssistantTranscript, NoticeTranscript, UserTranscript from agent_comms.input_disposition import InputDispositions from agent_comms.native_entries import NativeEntry from agent_comms.thread_status import RunningThreadStatus, StoppedThreadStatus @@ -46,6 +49,39 @@ def test_only_selected_registry_and_history_without_old_pending_input(source, tm assert not root.exists() +@pytest.mark.parametrize("timestamp", (None, 1_760_000_000.125)) +def test_external_projection_preserves_event_clocks_or_refuses_unknown(source, tmp_path, timestamp): + owner = source.registry.require("B") + external = ExternalTranscript(source.root, owner.incarnation) + events = ( + UserTranscript("source request " * 1500, position=1, timestamp=timestamp), + AssistantTranscript("source reply", position=2, + timestamp=None if timestamp is None else timestamp + 1.125), + NoticeTranscript("not portable conversation", position=3, timestamp=timestamp), + UserTranscript("", position=4, timestamp=timestamp), + ) + root = tmp_path / "external-window" + if timestamp is None: + # Current records require a clock at their storage boundary, before copying. + with pytest.raises(TypeError, match="received NoneType"): + external.finish(turn_id="saved", item_id="item", events=events) + with external.reading() as original: + assert not original.has_records(original.cut().offset) + assert not root.exists() + return + external.finish(turn_id="saved", item_id="item", events=events) + with CheckRoot(source, (owner.name,), root).open() as copied: + session = Path(copied.registry.require(owner.name).session_file) + rows = [NativeEntry.read(line) for line in session.read_bytes().splitlines()] + assert [entry.message.text for entry in rows[1:]] == [event.text for event in events[:2]] + assert [entry.timestamp.at.timestamp() for entry in rows[1:]] == [ + event.timestamp for event in events[:2] + ] + with external.reading() as original: + assert tuple(original.reverse(original.cut().offset))[0].events == events + assert not root.exists() + + def test_wrapped_nonzero_exit_removes_root_and_is_the_cli_exit(source, tmp_path, capsys): root = tmp_path / "command" status = main(["--root", str(source.root), "check-root", "--threads", "B", "--into", str(root), @@ -64,6 +100,35 @@ def test_python_failure_removes_root(source, tmp_path): assert not root.exists() +@pytest.mark.parametrize("release", (False, True)) +def test_initial_history_retains_display_not_source_claim_authority(source, tmp_path, release): + with source.bus.log.opened_wire_snapshot() as history: + message = next(history.newest_messages()) + owner = source.registry.require(message.sender).incarnation.require_recorded() + resource = str(tmp_path / "retained.py") + retained = replace(message, claim_transition=ClaimTransition( + owner=message.sender, incarnation=owner.lookup, + seq=message.seq, message_id=message.message_id, + claims=() if release else (resource,), + releases=(ClaimRelease(resource, "a" * 32),) if release else (), + generation=None if release else "a" * 32, + )) + + class ClaimHistoryCheckRoot(CheckRoot): + def _populate(self, copied): + copied.messaging.initialize_private_initial_protocol(history=(retained,)) + + root = tmp_path / "claim-history" + with ClaimHistoryCheckRoot(source, ("A", "B"), root).open() as copied: + with copied.bus.log.opened_wire_snapshot() as history: + assert next(history.newest_messages()) == message + assert retained.claim_transition is not None + assert not root.exists() + assert source.registry.require("A").name == "A" + with source.bus.log.opened_wire_snapshot() as history: + assert next(history.newest_messages()) == message + + def test_cleanup_uses_the_registry_executor_selection_not_human_senders(source, tmp_path): root = tmp_path / "with-human" with CheckRoot(source, ("B",), root).open() as c: @@ -135,5 +200,7 @@ def test_native_window_uses_existing_import_and_route_readers(source, tmp_path, assert [entry.message.text for entry in rows[1:]] == ["history 17", "history 18", "history 19"] assert rows[1].parent_id is None assert rows[2].parent_id == rows[1].id + source_time = NativeEntry.read(original.splitlines()[1]).timestamp.at + assert all(entry.timestamp.at == source_time for entry in rows[1:]) assert session.path.read_bytes() == original assert not root.exists() diff --git a/tests/test_commander_order_json.py b/tests/test_commander_order_json.py index 44fb83bc52..09883b0996 100644 --- a/tests/test_commander_order_json.py +++ b/tests/test_commander_order_json.py @@ -49,3 +49,22 @@ def run(*args): assert after.order.report == original['order']['report'] assert after.order.check == before.order.check assert after.order.id in accepted.stdout + + +def test_json_revision_check_refuses_before_execution(commander_script, tmp_path): + commander, repo = commander_script + path = Path(order_body(repo)) + document = json.loads(path.read_text()) + marker = tmp_path / 'must-not-run' + document['order']['check']['command'] = f'touch "{marker}"; git checkout main' + path.write_text(json.dumps(document)) + result = subprocess.run( + [str(Path(sys.executable).with_name('agent-comms-commander')), + '--root', str(commander.comms.root), 'order', 'problem', str(path), + '--to', 'worker', '--due', '23:59', '--where', f'{__file__}:1'], + capture_output=True, text=True, timeout=60, + env={**os.environ, 'PYTHONPATH': str(Path(__file__).parents[1] / 'src')}) + assert result.returncode != 0 + assert 'never checks out' in result.stderr + assert not marker.exists() + assert commander.comms.goals.todos.for_commander(commander.incarnation) == () diff --git a/tests/test_installed_merge.py b/tests/test_installed_merge.py index 4e435dbdbf..7618ebd849 100644 --- a/tests/test_installed_merge.py +++ b/tests/test_installed_merge.py @@ -2,14 +2,16 @@ from dataclasses import fields, replace import json import subprocess +import time import pytest from agent_comms.repo import Repo from agent_comms.field_codec import FieldCodec +from agent_comms.plan_phases import RecordPlanePhase from agent_comms.todos import (CheckedProblemOrder, DoneTodoState, MergedProblemOrder, OperationDelivery, MergedPullRequest, PullRequests, UnorderedProblemOrder) -from agent_comms.tools import CommsOrderDoneTool +from agent_comms.tools import CommsOrderDoneTool, CommsOrderServeTool from agent_comms.recorded_time import RecordedEventTime from cutover.carry_merged_orders import MergedOrders from test_commander_problem_orders import working @@ -29,6 +31,8 @@ def test_installed_merge_closes_pending_or_delivered_order_once( delivery=OperationDelivery(at=RecordedEventTime(500)), claims=working(comms)).apply(comms, caller) todo = comms.goals.todos.get(todo.id) + CommsOrderServeTool(problem=todo.id, serves=RecordPlanePhase).apply(comms, caller) + todo = comms.goals.todos.get(todo.id) original = todo.order commit(unit) git(unit, 'switch', '-q', 'main') @@ -43,8 +47,7 @@ def test_installed_merge_closes_pending_or_delivered_order_once( git(installed, 'fetch', '-q', 'origin', 'main') git(installed, 'checkout', '-q', '-B', 'main', base) repo = Repo(checkout=installed, github=todo.repo) - config = script.CONFIG.model_copy(update={'repos': (repo,)}) - monkeypatch.setattr(script, 'CONFIG', config) + actor = comms.registry.require(caller.name).incarnation merge = MergedPullRequest(1, 'fix/p1', sha, merge=sha, merged_at=500, updated_at=500, orders=(original.id,)) monkeypatch.setattr(PullRequests, 'read', lambda self, repo, state, order: @@ -52,14 +55,15 @@ def test_installed_merge_closes_pending_or_delivered_order_once( # The object being available locally is not installation: HEAD must contain it. assert not repo.contains(sha) - assert script.COMMANDER.orders(script.CONFIG.repos) == (todo,) + comms.goals.todos.reconcile_merges(actor, (repo,), comms.registry.storage_rule) + assert script.held_orders() == (todo,) assert comms.goals.todos.get(todo.id) == todo if not delivered: assert 'waiting on install' in deadline_lines(script)[0] git(installed, 'checkout', '-q', '-B', 'main', sha) assert repo.contains(sha) - assert script.COMMANDER.orders(script.CONFIG.repos) == () + comms.goals.todos.reconcile_merges(actor, (repo,), comms.registry.storage_rule) assert script.held_orders() == () closed = comms.goals.todos.get(todo.id) assert closed.state is DoneTodoState and closed.assignment is None @@ -70,6 +74,8 @@ def test_installed_merge_closes_pending_or_delivered_order_once( for item in fields(CheckedProblemOrder)) assert closed.order.id == original.id assert "original" not in FieldCodec.encode(closed.order) + assert closed.order.phase is RecordPlanePhase + assert RecordPlanePhase.serving((closed,)) == (closed,) assert closed.strikes == todo.strikes assert FieldCodec.decode(type(closed), FieldCodec.encode(closed)) == closed @@ -93,9 +99,10 @@ def no_child(*args, **kwargs): with monkeypatch.context() as repeat: repeat.setattr(subprocess, 'run', no_child) - assert script.COMMANDER.orders(script.CONFIG.repos) == () + comms.goals.todos.reconcile_merges(actor, (repo,), comms.registry.storage_rule) + assert script.held_orders() == () assert comms.goals.todos.get(todo.id) == closed - script.board() + script.write_board() assert f'| {todo.id}:' not in (tmp_path / 'board.md').read_text() assert deadline_lines(script) == [] assert comms.goals.todos.get(todo.id) == closed @@ -106,7 +113,7 @@ class TickComplete(Exception): def next_tick(seconds): raise TickComplete - monkeypatch.setattr(script.time, 'sleep', next_tick) + monkeypatch.setattr(time, 'sleep', next_tick) with pytest.raises(TickComplete): script.overdue() assert 'OVERDUE:' not in capsys.readouterr().out diff --git a/tests/test_merge_queue.py b/tests/test_merge_queue.py new file mode 100644 index 0000000000..90e51e4f37 --- /dev/null +++ b/tests/test_merge_queue.py @@ -0,0 +1,100 @@ +"""Authorization/schema checks; acceptance still requires an actual queued merge.""" + +import json +import subprocess +import unittest +from pathlib import Path +from unittest.mock import patch + +from agent_comms.declared_family import UnknownFamilyMember +from agent_comms.field_codec import FieldCodec, MalformedRecord +from cutover.backup import UpgradeRefused +from install_gate import FailedMachineResult, GateResult, MachinesGateResult +from merge_queue import ( + CheckFailed, + DispatchQueueCheck, + Host, + InstallQueueCheck, + QueueCheck, + SrcQueueCheck, + Target, + WorkflowDispatchRunEvent, + WorkflowRun, +) + + +class MergeQueueTests(unittest.TestCase): + def test_declared_check_is_the_wire_decoder(self): + check = QueueCheck.member_named("src").from_tokens(["123", "src"]) + self.assertIs(type(check), SrcQueueCheck) + self.assertEqual(check.run_id, 123) + with self.assertRaises(UnknownFamilyMember): + QueueCheck.member_named("ratchet") + + def test_github_event_decodes_to_its_behavior(self): + record = FieldCodec.decode(WorkflowRun, { + "event": "workflow_dispatch", "head_sha": "a" * 40, + "head_branch": "feature", "status": "in_progress", + "path": ".github/workflows/debt-ratchet.yml", "actor": {"login": "owner"}, + }) + self.assertIs(record.event, WorkflowDispatchRunEvent) + with self.assertRaises(MalformedRecord): + FieldCodec.decode(WorkflowRun, { + "event": "pull_request", "head_sha": "a" * 40, + "head_branch": "feature", "status": "in_progress", + "path": ".github/workflows/debt-ratchet.yml", "actor": {"login": "owner"}, + }) + + def test_a_workflow_cannot_supply_git_options(self): + host = Host(Path("."), "owner/repo", Path(".")) + with patch("merge_queue.run") as command: + with self.assertRaises(MalformedRecord): + host.fetch("--upload-pack=anything") + command.assert_not_called() + + def test_foreign_status_does_not_dispatch_more_checks(self): + host = Host(Path("."), "owner/repo", Path(".")) + with patch("merge_queue.StatusWorkflowRun.require"): + with patch.object(Host, "fetch") as fetch: + DispatchQueueCheck(123, "a" * 40, "unrelated").execute(host) + fetch.assert_not_called() + + def test_gate_result_carries_failure_without_successful_private_steps(self): + result = MachinesGateResult({"desktop": FailedMachineResult("Insufficient room")}) + decoded = FieldCodec.decode(GateResult, FieldCodec.encode(result)) + self.assertEqual(decoded.failures, ("desktop: Insufficient room",)) + + def test_ratchet_failure_preserves_measure_and_exact_source_lines(self): + completed = subprocess.CompletedProcess( + ("ratchet",), 1, + stdout=json.dumps({"introduced_lines": {"NoneAbsence:tools/example.py": [17, 21]}, + "delta": {"NoneAbsence:tools/example.py": 2}}), stderr="", + ) + error = CheckFailed(SrcQueueCheck(123), completed) + self.assertEqual(str(error), + "Ratchet: NoneAbsence tools/example.py:17; NoneAbsence tools/example.py:21") + + def test_member_limit_is_passed_to_the_command(self): + host = Host(Path("."), "owner/repo", Path(".")) + completed = subprocess.CompletedProcess(("true",), 0, stdout="", stderr="") + with patch("merge_queue.subprocess.run", return_value=completed) as command: + host.command(SrcQueueCheck(123), "true") + self.assertEqual(command.call_args.kwargs["timeout"], 600) + self.assertEqual(InstallQueueCheck.limit, 3000) + + def test_practice_install_is_not_skipped_by_a_second_storage_policy(self): + host = Host(Path("."), "owner/repo", Path(".")) + with patch.object(Host, "command") as command: + check = InstallQueueCheck(123) + check.check(host, Target("a" * 40, "b" * 40)) + self.assertIs(command.call_args.args[0], check) + self.assertIn("--practice", command.call_args.args) + + def test_only_declared_installer_refusal_crosses_the_log_boundary(self): + error = UpgradeRefused("Insufficient room") + self.assertEqual(UpgradeRefused.public_message("private trace\n" + error.output), + error.output) + secret = "PRIVATE_PAYLOAD_DO_NOT_LEAK" + self.assertNotIn(secret, UpgradeRefused.public_message(secret)) + private_failure = f"upgrade-bus: aborted before the backup (RuntimeError: {secret})" + self.assertNotIn(secret, UpgradeRefused.public_message(private_failure)) diff --git a/tests/test_order_check_writes.py b/tests/test_order_check_writes.py new file mode 100644 index 0000000000..f9791d8405 --- /dev/null +++ b/tests/test_order_check_writes.py @@ -0,0 +1,60 @@ +"""Saved check commands are data; revision policy is enforced when writing new checks.""" +from dataclasses import replace + +import pytest + +from agent_comms.field_codec import FieldCodec +from agent_comms.goals import OrderCheck +from agent_comms.tool_caller import ThreadCaller +from agent_comms.todos import AssignTodo +from agent_comms.tools import CommsOrderCheckTool, CommsSendTool +from test_commander_problem_orders import create, order + + +@pytest.mark.parametrize('command', ['git checkout main', 'git -C /repo switch work', + 'true && git checkout HEAD']) +def test_saved_checks_decode_but_new_checks_refuse_revision_selection(command): + saved = OrderCheck('.', command) + assert FieldCodec.decode(OrderCheck, FieldCodec.encode(saved)) == saved + with pytest.raises(ValueError, match='never checks out'): + saved.require_writable() + with pytest.raises(ValueError, match='never checks out'): + OrderCheck.parse(f'. :: {command} -> 0') + + +def test_native_new_order_refuses_without_writing(commander): + comms, _ = commander + with pytest.raises(ValueError, match='never checks out'): + create(commander, 'P1', command='git checkout main') + assert comms.goals.todos.for_commander(comms.registry.require('lead').incarnation) == () + + +def test_saved_order_can_be_read_sent_about_and_explicitly_corrected(commander): + comms, caller = commander + current = create(commander, 'P1') + legacy_check = OrderCheck(str(comms.root), 'git checkout main && printf 0') + historical = replace(current, order=replace(current.order, check=legacy_check)) + store = comms.goals.todos + # Seed a historical row, not a new authored order through the public write API. + with store.writing() as db: + historical.upsert(db) + assert store.get('P1') == historical + assert store.for_commander(comms.registry.require('lead').incarnation) == (historical,) + CommsSendTool(sender='worker', target='lead', body='The recorded check selects a revision.').apply( + comms, ThreadCaller('worker')) + with pytest.raises(ValueError, match='never checks out'): + store.apply('P1', AssignTodo(historical.revision, historical.assignment, + historical.deadline, order(comms.root, 'replacement', command=legacy_check.command), + historical.goal)) + with pytest.raises(ValueError, match='never checks out'): + CommsOrderCheckTool(problem='P1', check=legacy_check, + why='The command must not change the current revision.').apply(comms, caller) + assert store.get('P1') == historical + safe = OrderCheck(str(comms.root), 'printf 0') + CommsOrderCheckTool(problem='P1', check=safe, + why='The command must not change the current revision.').apply(comms, caller) + corrected = store.get('P1') + assert corrected.order.id == historical.order.id + assert corrected.revision == historical.revision + 1 + assert corrected.order.check == safe + assert corrected.order.check_corrections[-1].previous == legacy_check diff --git a/tests/test_phase_task_dependencies.py b/tests/test_phase_task_dependencies.py new file mode 100644 index 0000000000..02fd3986b7 --- /dev/null +++ b/tests/test_phase_task_dependencies.py @@ -0,0 +1,27 @@ +"""Phase declarations stay usable before action and storage modules are loaded.""" + +from pathlib import Path +import subprocess +import sys + +import pytest + +from agent_comms.phase_tasks import UnspecifiedPhaseTask +from agent_comms.todos import UnorderedProblemOrder + + +@pytest.mark.parametrize("module", ("plan_phases", "work_dispatch", "goal_actions", "todos")) +def test_import_order_does_not_reintroduce_dispatch_cycle(module): + checkout = Path(__file__).resolve().parents[1] + result = subprocess.run( + (sys.executable, "-c", f"import agent_comms.{module}"), + cwd=checkout, + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr + + +def test_unspecified_phase_and_unordered_problem_cannot_start_checked_work(): + assert not UnspecifiedPhaseTask().done(()) + with pytest.raises(ValueError, match="Only an undelivered order"): + UnorderedProblemOrder().reissued(1) diff --git a/tests/test_plan_phases.py b/tests/test_plan_phases.py new file mode 100644 index 0000000000..669ccccd83 --- /dev/null +++ b/tests/test_plan_phases.py @@ -0,0 +1,147 @@ +from dataclasses import replace + +import pytest + +from agent_comms.field_codec import FieldCodec +from agent_comms.standing_rules import PlanIdle +from agent_comms.plan_phases import ( + CommanderPagePhase, + ControlPlanePhase, + CutoverPhase, + PlanPhase, + RecordPlanePhase, + SkillPhase, + StoryPhase, + UnplannedPhase, +) +from agent_comms.todos import ( + DoneTodoState, + FoldedProblemOrder, + InstalledProblemOrder, + ProblemOrder, + OperationDelivery, + UnorderedProblemOrder, +) +from agent_comms.tools import CommsOrderServeTool +from agent_comms.work_dispatch import CheckedPhaseTask + +from test_commander_problem_orders import create + + +def accepted(todo): + installed = todo.order.copy_as( + InstalledProblemOrder, delivery=OperationDelivery(), + works="The declared whole phase worked through its acceptance run.", + artifact="phase-acceptance.txt", + ) + return replace(todo, state=DoneTodoState, assignment=None, order=installed) + + +def task(todo): + return CheckedPhaseTask( + problem=todo.id, repo=todo.repo, text=todo.text, + order=todo.order, deadline_seconds=300, + ) + + +def test_views_wait_for_native_work_but_never_block_cutover(commander, monkeypatch): + comms, caller = commander + # ReviewQueuePhase is explicitly delivered as #1654, not inferred from a subtask. + required = (RecordPlanePhase, ControlPlanePhase) + phases = (*required, CutoverPhase, StoryPhase, CommanderPagePhase) + todos = [] + for index, phase in enumerate(phases, 1): + todo = create(commander, f"P{index}") + CommsOrderServeTool(problem=todo.id, serves=phase).apply(comms, caller) + todo = comms.goals.todos.get(todo.id) + monkeypatch.setattr(phase, "task", task(todo)) + todos.append(todo) + + record_done = tuple( + accepted(todo) if todo.order.serves is RecordPlanePhase else todo + for todo in todos + ) + assert StoryPhase.ready(record_done) + assert not CommanderPagePhase.ready(record_done) + + native_done = tuple( + accepted(todo) if todo.order.serves in required else todo + for todo in todos + ) + assert CutoverPhase.ready(native_done) + assert StoryPhase.ready(native_done) + assert CommanderPagePhase.ready(native_done) + assert not SkillPhase.ready(native_done) + + for unfinished in required: + snapshot = tuple( + todo if todo.order.serves in (unfinished, CutoverPhase) else accepted(todo) + for todo in todos + ) + assert not CutoverPhase.ready(snapshot) + + cutover_done = tuple( + accepted(todo) if todo.order.serves is CutoverPhase else todo + for todo in native_done + ) + assert StoryPhase.ready(cutover_done) + assert CommanderPagePhase.ready(cutover_done) + assert SkillPhase.ready(cutover_done) + assert not any( + dependency in (StoryPhase, CommanderPagePhase) + for phase in PlanPhase.members_with(PlanPhase) + for dependency in phase.after + ) + + +def test_unordered_findings_have_no_invented_phase_declaration(commander): + comms, caller = commander + todo = create(commander, "P1") + finding = replace(todo, assignment=None, order=UnorderedProblemOrder()) + with comms.goals.todos.writing() as db: + finding.upsert(db) + assert finding.order.phase is UnplannedPhase + assert not RecordPlanePhase.serving((finding,)) + assert [rule.phase for rule in PlanIdle.violations(comms, caller.name)] == [RecordPlanePhase] + with pytest.raises(ValueError, match="Only a checked order"): + CommsOrderServeTool(problem=finding.id, serves=StoryPhase).apply(comms, caller) + assert comms.goals.todos.get(finding.id) == finding + + +def test_merged_phase_is_inherited_and_round_trips_without_a_wrapper(commander): + todo = create(commander, "P1") + original = todo.order.for_phase(RecordPlanePhase) + order = original.merged("a" * 40) + saved = replace(todo, order=order, state=DoneTodoState, assignment=None) + assert RecordPlanePhase.serving((saved,)) == (saved,) + wire = FieldCodec.encode(order) + assert "original" not in wire + restored = FieldCodec.decode(ProblemOrder, wire) + assert restored == order and restored.phase is RecordPlanePhase + changed = order.for_phase(ControlPlanePhase) + assert changed.phase is ControlPlanePhase + assert original.phase is RecordPlanePhase + assert not order.accepted + folded = FoldedProblemOrder(original=original, into="P2", why="Keep one owner") + assert folded.phase is UnplannedPhase + + +def test_closed_related_subtask_does_not_accept_a_whole_phase(commander, monkeypatch): + comms, caller = commander + subtask = create(commander, "P149") + CommsOrderServeTool(problem=subtask.id, serves=StoryPhase).apply(comms, caller) + subtask = accepted(comms.goals.todos.get(subtask.id)) + assert not StoryPhase.done((subtask,)) + + whole = create(commander, "P999") + CommsOrderServeTool(problem=whole.id, serves=StoryPhase).apply(comms, caller) + whole = comms.goals.todos.get(whole.id) + monkeypatch.setattr(StoryPhase, "task", task(whole)) + assert not StoryPhase.done((subtask, whole)) + assert StoryPhase.done((subtask, accepted(whole))) + + withdrawn = whole.order.withdrawn(whole.held(), "The order was sent in error.", None) + assert not StoryPhase.done((replace(whole, state=DoneTodoState, order=withdrawn),)) + assert not UnorderedProblemOrder().accepted + wrong_phase = replace(accepted(whole), order=replace(accepted(whole).order, serves=ControlPlanePhase)) + assert not StoryPhase.done((wrong_phase,)) diff --git a/tools/cutover/backup.py b/tools/cutover/backup.py index fcf6945a27..0fd664fb93 100644 --- a/tools/cutover/backup.py +++ b/tools/cutover/backup.py @@ -14,7 +14,7 @@ from contextlib import closing, contextmanager from dataclasses import dataclass from pathlib import Path -from typing import Iterator, Self +from typing import ClassVar, Iterator, Self from agent_comms.coordination_database import CoordinationStore from agent_comms.external_transcript import ExternalTranscript @@ -35,6 +35,18 @@ class UpgradeRefused(Exception): """The update stopped without permission to discard its original data.""" + prefix: ClassVar[str] = "upgrade-bus: refused:" + + @property + def output(self) -> str: + return f"{self.prefix} {self}" + + @classmethod + def public_message(cls, output: str) -> str: + for line in reversed(output.splitlines()): + if line.startswith(cls.prefix): + return line + return "Installer failed without a declared refusal; private traceback retained on host" def bus_files(root: Path) -> dict[str, str]: diff --git a/tools/cutover/carry_commander_config.py b/tools/cutover/carry_commander_config.py new file mode 100644 index 0000000000..5ec5e2f516 --- /dev/null +++ b/tools/cutover/carry_commander_config.py @@ -0,0 +1,29 @@ +"""Move the old home-only commander record through its existing writer.""" + +from __future__ import annotations + +from pathlib import Path + +from agent_comms.commander.config import CommanderConfig + +from .carry import Carried, Carry + + +class CommanderConfigRoot(Carry): + @classmethod + def carry(cls, root: Path, *, apply: bool) -> Carried: + home = Path.home() + old = CommanderConfig.path(home) + record = CommanderConfig.root_path(root) + if record.exists() or old.is_symlink() or not old.is_file(): + return Carried(0) + # A cutover rehearsal selects its copied root, never the live root named + # in the old machine file. The selected root owns the new record. + config = CommanderConfig.read(old).model_copy(update={"root": root}) + if apply: + config.write(home) + return Carried(1, {"commander_configs": 1}, (record.name,) if apply else ()) + + +if __name__ == "__main__": + CommanderConfigRoot.main() diff --git a/tools/debt_ratchet.py b/tools/debt_ratchet.py index 8b14189ba9..0ec6462daf 100644 --- a/tools/debt_ratchet.py +++ b/tools/debt_ratchet.py @@ -21,7 +21,7 @@ from collections import Counter from collections.abc import Iterator from dataclasses import dataclass -from functools import cached_property +from functools import cache, cached_property from pathlib import Path sys.path.insert(0, str( @@ -38,7 +38,7 @@ from agent_comms.core_stamp import WheelStamp from agent_comms.declared_family import DeclaredFamily -from cutover.carry import STEPS, Step, carry_parents +from cutover.carry import STEPS, Carry, Step, carry_parents from stored_formats import StoredForm from agent_comms.field_codec import FieldCodec, boundary_decoder, projected from agent_comms.mro_dispatch import MroDispatch, handles @@ -47,6 +47,11 @@ class Measure(DeclaredFamily, affix="Measure"): + @classmethod + def blocks(cls, delta: dict[str, int]) -> bool: + return any(value > 0 for key, value in delta.items() + if key.partition(":")[0] == cls.__name__) + @staticmethod def environment(root: str) -> dict[str, str]: return dict(os.environ) @@ -242,27 +247,6 @@ def environment(root: str) -> dict[str, str]: return {**os.environ, "MYPYPATH": os.pathsep.join((*paths, str(AUDIT_SCRIPTS)))} -class SourceLines(Measure): - """Budget edited lines, retaining moves across the source boundary (50% similarity).""" - - trailer: typing.ClassVar[str] = "Src-Budget" - - @classmethod - def compare(cls, repo: Path, base: str, head: str, changed: set[str], root: str): - added = removed = 0 - records = iter(git(repo, "diff", "--no-ext-diff", "--no-textconv", "--find-renames=50%", - "--numstat", "-z", base, head, "--").split(b"\0")[:-1]) - for record in records: - increase, decrease, path = record.split(b"\t", 2) - paths = (path,) if path else (next(records), next(records)) - if any(path.startswith(root.encode()) and path.endswith(b".py") for path in paths): - added += int(increase) - removed += int(decrease) - # Each commit's trailer budgets that commit's lines, so a range's budget is their sum. - declared = git(repo, "log", f"--format=%(trailers:key={cls.trailer},valueonly)", f"{base}..{head}") - return {cls.__name__: removed + sum(int(value) for value in declared.split())}, {cls.__name__: added} - - @dataclass(frozen=True) class ChangedSource: """One head module and the new-side lines changed by its Git patch.""" @@ -1413,7 +1397,10 @@ def conversion(source: bytes) -> str: def compare(cls, repo: Path, base: str, head: str, changed: set[str], root: str): fields = iter(git(repo, "diff", "--name-status", "--find-renames=50%", "-z", base, head, "--", str(STEPS)).decode().split("\0")[:-1]) - parents = carry_parents(git(repo, "show", f"{base}:{STEPS}/carry.py")) + @cache + def parents() -> dict[str, type[Carry]]: + return carry_parents(git(repo, "show", f"{base}:{STEPS}/carry.py")) + edited = 0 for status in fields: path = next(fields) @@ -1424,7 +1411,7 @@ def compare(cls, repo: Path, base: str, head: str, changed: set[str], root: str) continue if status[0] in "MRC" or status == "T": before = git(repo, "show", f"{base}:{path}") - if Step.holds_carry(before, parents): + if Step.holds_carry(before, parents()): edited += (status != "M" or cls.conversion(before) != cls.conversion(git(repo, "show", f"{head}:{path}"))) elif status == "A": @@ -1472,10 +1459,8 @@ def formats(cls, repo: Path, ref: str, root: str, changed: set[str]) -> dict[str source = git(repo, "archive", ref, "--", str(package)) subprocess.run(["tar", "-x", "-C", scratch], input=source, check=True, timeout=60) # Extracted source is stamped with its commit the way a build stamps it. - for module in {module.split(".")[0] for module in modules}: - WheelStamp(revision(repo, ref)).write( - Path(scratch) / package / module / "package_stamp.py" - ) + for stamp in Path(scratch).rglob("package_stamp.py"): + WheelStamp(revision(repo, ref)).write(stamp) result = subprocess.run([sys.executable, "-c", cls.probe, str(Path(scratch) / package), str(Path(__file__).resolve().parent), *modules], cwd=scratch, env=cls.environment(root), @@ -1537,7 +1522,8 @@ def delta(self) -> dict[str, int]: @property def failed(self) -> bool: - return any(value > 0 for value in self.delta.values()) or any(self.introduced.values()) + return (any(measure.blocks(self.delta) for measure in Measure.members_with(Measure)) + or any(self.introduced.values())) def compare(repo: Path, base: str, head: str, root: str) -> Comparison: diff --git a/tools/install_gate.py b/tools/install_gate.py index becbea83ca..1a2156d802 100644 --- a/tools/install_gate.py +++ b/tools/install_gate.py @@ -27,13 +27,17 @@ from collections.abc import Iterator, Mapping from contextlib import ExitStack, contextmanager from dataclasses import dataclass, field, replace +from itertools import islice from pathlib import Path -from typing import Annotated, Literal, Self +from typing import Annotated, ClassVar, Literal, Self +from cutover.backup import UpgradeRefused from install_snapshot import SnapshotHome -from agent_comms.active_route import ActiveRoute +from agent_comms.active_route import ActiveRoute, CommsRoute +from agent_comms.check_root import CheckRoot from agent_comms.child_process import Platform +from agent_comms.comms import Comms from agent_comms.declared_family import DeclaredFamily from agent_comms.field_codec import ( FieldCodec, @@ -233,6 +237,43 @@ def affected(repo: Path, base: str, head: str) -> list[str]: ) +@dataclass(frozen=True) +class PracticeCopy: + """A real retained tail, native session tails and the production indexes they build.""" + + source: Comms + into: Path + max_bytes: ClassVar[int] = 3 * 1024**3 + + @contextmanager + def open(self) -> Iterator[Comms]: + roster = self.source.registry.snapshot() + with self.source.bus.log.opened_wire_snapshot() as source: + names = tuple( + dict.fromkeys( + sender + for message in islice( + source.newest_messages(CheckRoot.max_bytes), CheckRoot.max_records + ) + if (sender := roster.canonical_name(message.sender)) in roster.threads + ) + ) + if not names: + raise GateError("The retained bus tail has no registered participants") + self.into.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + if shutil.disk_usage(self.into.parent).free < storage_floor(): + raise GateError("Insufficient free space before building the retained-tail copy") + with CheckRoot(self.source, names, self.into).open() as copied: + size = sum(path.stat().st_size for path in copied.root.rglob("*") if path.is_file()) + if size > self.max_bytes: + raise GateError( + f"Retained-tail copy is {size} bytes, over its {self.max_bytes} cap" + ) + if shutil.disk_usage(self.into.parent).free < 2 * size + storage_floor(): + raise GateError(f"Insufficient free space for rehearsal of {size} copied bytes") + yield copied + + class SnapshotCopy(SnapshotHome): def place(self, home: Path, bus: Path) -> Path: original, source_home = Path(self.root), Path(self.home) @@ -350,6 +391,12 @@ def prepare_home(home: Path, bus: Path, context: SnapshotCopy) -> Path: source_route = bus.parent / "active-route.json" bus = context.place(home, bus) context.route(source_route, home, bus) + prepare_runtime_home(home) + return bus + + +def prepare_runtime_home(home: Path) -> None: + home.mkdir(mode=0o700, parents=True, exist_ok=True) pointer = home / ".local/share/agent-comms/current" pointer.parent.mkdir(mode=0o700, parents=True, exist_ok=True) pointer.unlink(missing_ok=True) @@ -361,7 +408,6 @@ def prepare_home(home: Path, bus: Path, context: SnapshotCopy) -> Path: cache = Path(run("npm", "config", "get", "cache")) if not (home / ".npm").exists() and cache.is_dir(): shutil.copytree(cache, home / ".npm", symlinks=True) - return bus @contextmanager @@ -460,7 +506,8 @@ def rehearse( launch.verify() outcome = asyncio.run(process.wait()) if not outcome.successful: - raise GateError(f"Rehearsal {outcome.summary}; see {log}") + reason = UpgradeRefused.public_message(log.read_text(errors="replace")) + raise GateError(f"Rehearsal {outcome.summary}: {reason}") try: return FieldCodec.decode(InstallerAnswer, FieldCodec.parse(answer.read_text())).rehearsal except MalformedRecord as error: @@ -599,6 +646,10 @@ class MachineResult(DeclaredFamily, affix="MachineResult"): @abstractmethod def successful(self) -> bool: ... + @property + def failures(self) -> tuple[str, ...]: + return () + @dataclass(frozen=True) class PassedMachineResult(MachineResult): @@ -624,8 +675,16 @@ class FailedMachineResult(MachineResult): def successful(self) -> bool: return False + @property + def failures(self) -> tuple[str, ...]: + return (self.error,) + class GateResult(DeclaredFamily, affix="GateResult"): + @property + def failures(self) -> tuple[str, ...]: + return (self.description,) + @property @abstractmethod def successful(self) -> bool: ... @@ -654,6 +713,14 @@ def description(self) -> str: class MachinesGateResult(GateResult): machines: dict[str, MachineResult] + @property + def failures(self) -> tuple[str, ...]: + return tuple( + f"{name}: {reason}" + for name, result in self.machines.items() + for reason in result.failures + ) + @property def successful(self) -> bool: return len(self.machines) == 2 and all( @@ -669,6 +736,19 @@ def description(self) -> str: ) +@dataclass(frozen=True) +class PracticeGateResult(GateResult): + rehearsal: CompletedRehearsal + + @property + def successful(self) -> bool: + return True + + @property + def description(self) -> str: + return "Retained-tail practice install passed" + + @dataclass(frozen=True) class UnfinishedGateResult(MachinesGateResult): @property @@ -680,6 +760,10 @@ def successful(self) -> bool: class InterruptedGateResult(MachinesGateResult): error: str + @property + def failures(self) -> tuple[str, ...]: + return (self.error, *super().failures) + @property def successful(self) -> bool: return False @@ -777,6 +861,11 @@ def execute(self, args: GateArguments, repository: str) -> int: ... class CheckGateCommand(GateCommand): + def run( + self, args: GateArguments, target: Target, report: Path, results: dict[str, MachineResult] + ) -> GateResult: + return MachinesGateResult(run_machines(args, target, report, results)) + @abstractmethod def target(self, args: GateArguments, repository: str) -> Target: ... @@ -793,7 +882,7 @@ def execute(self, args: GateArguments, repository: str) -> int: result = ( OutsideInstallGateResult() if target.outside(changed) - else MachinesGateResult(run_machines(args, target, report_path, results)) + else self.run(args, target, report_path, results) ) except ( GateError, @@ -858,6 +947,39 @@ def target(self, args: GateArguments, repository: str) -> Target: return Target(head, args.base.resolve(args.repo, head)) +@dataclass(frozen=True) +class PracticeGateCommand(RevisionGateCommand): + def run( + self, args: GateArguments, target: Target, report: Path, results: dict[str, MachineResult] + ) -> GateResult: + with tempfile.TemporaryDirectory(dir=args.scratch, prefix="practice-") as directory: + private = Path(directory) + home = private / "home" + with PracticeCopy( + Comms(CommsRoute.current().root), home / ".agent-comms" + ).open() as copied: + prepare_runtime_home(home) + with tempfile.TemporaryDirectory( + dir=Path.home() / "wt", prefix="practice-source-" + ) as source: + work = Path(source) + with ( + checkout(args.repo, target.base, work / "base") as previous, + checkout(args.repo, target.head, work / "target") as tree, + ): + completed = rehearse( + tree, + previous, + target.head, + copied.root, + home, + args.python, + report.with_suffix(".log"), + work, + ) + return PracticeGateResult(completed) + + @dataclass(frozen=True) class ConfigureGateCommand(GateCommand): def execute(self, args: GateArguments, repository: str) -> int: @@ -929,6 +1051,7 @@ def main() -> int: mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--pr", dest="command", type=PrGateCommand.argument) mode.add_argument("--revision", dest="command", type=RevisionGateCommand) + mode.add_argument("--practice", dest="command", type=PracticeGateCommand) mode.add_argument( "--require-status", dest="command", action="store_const", const=ConfigureGateCommand() ) diff --git a/tools/merge_gate_command.sh b/tools/merge_gate_command.sh new file mode 100755 index 0000000000..f9b979b7a2 --- /dev/null +++ b/tools/merge_gate_command.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +# Only this fixed command belongs in the loopback-only authorized_keys entry. +# No forwarded environment, arbitrary shell command, or workflow stdin is used. +set -euo pipefail +state=${1:?usage: merge_gate_command.sh STATE_ROOT MAIN_CHECKOUT PYTHON} +repo=${2:?main checkout required} +python=${3:?shared check interpreter required} +mkdir -p "$state/host-logs" +log=$(mktemp "$state/host-logs/check-XXXXXXXX.log") +exec "$log"; then + printf 'check passed\n' +else + printf 'gate failed; detailed child logs remain private on the host\n' + exit 1 +fi diff --git a/tools/merge_queue.py b/tools/merge_queue.py new file mode 100644 index 0000000000..0b7dc8ece7 --- /dev/null +++ b/tools/merge_queue.py @@ -0,0 +1,394 @@ +#!/usr/bin/env python3 +"""Restricted SSH entrypoint for the repository's isolated Actions runner. + +GitHub owns queue order and run identity. The existing install gate owns source +coverage and rehearsal. A workflow receives exit status, never bus data or logs. +""" + +from __future__ import annotations + +import argparse +import shlex +import subprocess +import sys +from abc import abstractmethod +from dataclasses import dataclass, field +from pathlib import Path +from typing import ClassVar, Literal, Self + +from install_gate import ( + REPO, + ExternalRecord, + GateError, + RepositoryAnswer, + RunSummary, + Target, + run, +) + +from agent_comms.declared_family import DeclaredFamily +from agent_comms.field_codec import FieldCodec, GitCommitText +from agent_comms.todos import ReviewedHeadVerdict + +WORKFLOW = Path(".github/workflows/debt-ratchet.yml") + + +@dataclass(frozen=True) +class ApprovedHead(ExternalRecord): + state: Literal["success"] + + @classmethod + def require(cls, host: Host, head: str) -> None: + query = f'.statuses | map(select(.context == "{ReviewedHeadVerdict.context}")) | first' + answer = run("gh", "api", f"repos/{host.repository}/commits/{head}/status", "--jq", query) + FieldCodec.decode(cls, FieldCodec.parse(answer)) + + +@dataclass(frozen=True) +class PullRequestNumber(ExternalRecord): + number: int + + +@dataclass(frozen=True) +class QueuedPullRequest(ExternalRecord): + number: int + head: str = field(metadata={"wire_name": "headRefOid"}) + branch: str = field(metadata={"wire_name": "headRefName"}) + base: Literal["main"] = field(metadata={"wire_name": "baseRefName"}) + cross_repository: Literal[False] = field(metadata={"wire_name": "isCrossRepository"}) + draft: Literal[False] = field(metadata={"wire_name": "isDraft"}) + state: Literal["OPEN"] + + @classmethod + def for_head(cls, host: Host, head: str) -> Self: + links = FieldCodec.decode( + tuple[PullRequestNumber, ...], + FieldCodec.parse( + run( + "gh", + "api", + f"repos/{host.repository}/commits/{head}/pulls", + "--jq", + f'[.[] | select(.head.sha == "{head}" and .state == "open")' + " | {number: .number}]", + ) + ), + ) + fields = ",".join(key for _, key in FieldCodec.record_fields(cls)) + candidates = tuple( + FieldCodec.decode( + cls, + FieldCodec.parse( + run( + "gh", "pr", "view", link.number, "--repo", host.repository, "--json", fields + ) + ), + ) + for link in links + ) + matching = tuple(pr for pr in candidates if pr.head == head) + if len(matching) != 1: + raise GateError("An approved head must identify exactly one open, local PR") + return matching[0] + + +@dataclass(frozen=True) +class Actor(ExternalRecord): + login: str + + +@dataclass(frozen=True) +class Administrator(ExternalRecord): + permission: Literal["admin"] + + +class RunEvent(DeclaredFamily, affix="RunEvent"): + @classmethod + @abstractmethod + def target(cls, host: Host, workflow: WorkflowRun) -> Target: ... + + +class WorkflowDispatchRunEvent(RunEvent, declared_name="workflow_dispatch"): + @classmethod + def target(cls, host: Host, workflow: WorkflowRun) -> Target: + FieldCodec.decode( + Administrator, + FieldCodec.parse( + run( + "gh", + "api", + f"repos/{host.repository}/collaborators/{workflow.actor.login}/permission", + ) + ), + ) + head = host.fetch(workflow.head_sha) + base = run("git", "merge-base", "origin/main", head, cwd=host.repo) + return Target(head, base) + + +class MergeGroupRunEvent(RunEvent, declared_name="merge_group"): + @classmethod + def target(cls, host: Host, workflow: WorkflowRun) -> Target: + if not workflow.head_branch.startswith("gh-readonly-queue/main/"): + raise GateError("Not a main-branch merge group") + head = host.fetch(workflow.head_sha) + parents = run("git", "show", "-s", "--format=%P", head, cwd=host.repo).split() + if len(parents) != 2: + raise GateError("The singleton merge queue requires a two-parent proposed merge") + ApprovedHead.require(host, parents[1]) + return Target(head, parents[0]) + + +@dataclass(frozen=True) +class WorkflowIdentity(ExternalRecord): + head_sha: str + head_branch: str + status: Literal["in_progress"] + path: str + actor: Actor + + @classmethod + def require(cls, host: Host, identifier: int) -> Self: + result = FieldCodec.decode( + cls, + FieldCodec.parse( + run("gh", "api", f"repos/{host.repository}/actions/runs/{identifier}") + ), + ) + if Path(result.path.partition("@")[0]) != WORKFLOW: + raise GateError("The SSH key is restricted to the merge queue workflow") + return result + + +@dataclass(frozen=True) +class WorkflowRun(WorkflowIdentity): + event: type[RunEvent] + + +@dataclass(frozen=True) +class StatusWorkflowRun(WorkflowIdentity): + event: Literal["status"] + + +class CheckFailed(subprocess.SubprocessError): + def __init__(self, check: QueueCheck, completed: subprocess.CompletedProcess[str]): + self.check = check + self.completed = completed + super().__init__(f"{check.label}: {check.summary(completed)}") + + +@dataclass(frozen=True) +class Host: + repo: Path + repository: str + scratch: Path + + def fetch(self, head: str) -> str: + head = GitCommitText.decode(head) + run("git", "fetch", "origin", head, cwd=self.repo) + run("git", "submodule", "update", "--init", "--recursive", cwd=self.repo) + resolved = run("git", "rev-parse", f"{head}^{{commit}}", cwd=self.repo) + if resolved != head: + raise GateError("A check requires the full GitHub commit identifier") + return resolved + + def command(self, check: QueueCheck, *arguments: str | Path | int) -> None: + completed = subprocess.run( + tuple(map(str, arguments)), + cwd=self.repo, + text=True, + capture_output=True, + timeout=check.limit, + check=False, + ) + if completed.returncode: + raise CheckFailed(check, completed) + + +@dataclass(frozen=True) +class QueueCheck(DeclaredFamily, affix="QueueCheck"): + run_id: int + limit: ClassVar[int] + + @property + def label(self) -> str: + return self.declared_name.title() + + @classmethod + def from_tokens(cls, tokens: list[str]) -> Self: + identifier, _ = tokens + return cls(int(identifier)) + + def execute(self, host: Host) -> None: + workflow = WorkflowRun.require(host, self.run_id) + self.check(host, workflow.event.target(host, workflow)) + + @abstractmethod + def check(self, host: Host, target: Target) -> None: ... + + @abstractmethod + def summary(self, completed: subprocess.CompletedProcess[str]) -> str: ... + + +@dataclass(frozen=True) +class RatchetOutput(ExternalRecord): + introduced_lines: dict[str, tuple[int, ...]] + + +class RatchetQueueCheck(QueueCheck, declared_name=None): + root: ClassVar[str] + limit = 600 + + @property + def label(self) -> str: + return "Ratchet" + + def check(self, host: Host, target: Target) -> None: + host.command( + self, + sys.executable, + "-X", + "faulthandler", + host.repo / "tools/debt_ratchet.py", + "--root", + self.root, + "--base", + target.base, + "--head", + target.head, + ) + + def summary(self, completed: subprocess.CompletedProcess[str]) -> str: + report = FieldCodec.decode(RatchetOutput, FieldCodec.parse(completed.stdout)) + sites: list[str] = [] + for key, lines in report.introduced_lines.items(): + measure, _, path = key.partition(":") + sites.extend(f"{measure} {path}:{line}" for line in lines) + return "; ".join(sites) or completed.stdout.strip() + + +class SrcQueueCheck(RatchetQueueCheck): + root = "src/agent_comms" + + +class ToolsQueueCheck(RatchetQueueCheck): + root = Path(__file__).parent.name + + +class InstallQueueCheck(QueueCheck): + limit = 3000 + + def check(self, host: Host, target: Target) -> None: + host.command( + self, + sys.executable, + "-X", + "faulthandler", + host.repo / "tools/install_gate.py", + "--repo", + host.repo, + "--practice", + target.head, + "--base", + target.base, + "--scratch", + host.scratch, + ) + + def summary(self, completed: subprocess.CompletedProcess[str]) -> str: + report = FieldCodec.decode(RunSummary, FieldCodec.parse(completed.stdout)) + return "; ".join(report.result.failures) or report.result.description + + +class CommandQueueCheck(QueueCheck): + limit = 120 + + def summary(self, completed: subprocess.CompletedProcess[str]) -> str: + return completed.stderr.strip() or completed.stdout.strip() + + +class EnqueueQueueCheck(CommandQueueCheck): + def check(self, host: Host, target: Target) -> None: + ApprovedHead.require(host, target.head) + pr = QueuedPullRequest.for_head(host, target.head) + host.command( + self, + "gh", + "pr", + "merge", + pr.number, + "--repo", + host.repository, + "--auto", + "--merge", + "--match-head-commit", + pr.head, + ) + + +@dataclass(frozen=True) +class DispatchQueueCheck(CommandQueueCheck): + head: str + context: str + + @classmethod + def from_tokens(cls, tokens: list[str]) -> Self: + identifier, _, head, context = tokens + return cls(int(identifier), head, context) + + def execute(self, host: Host) -> None: + StatusWorkflowRun.require(host, self.run_id) + if self.context != ReviewedHeadVerdict.context: + return + head = host.fetch(self.head) + base = run("git", "merge-base", "origin/main", head, cwd=host.repo) + self.check(host, Target(head, base)) + + def check(self, host: Host, target: Target) -> None: + ApprovedHead.require(host, target.head) + pr = QueuedPullRequest.for_head(host, target.head) + host.command( + self, + "gh", + "workflow", + "run", + WORKFLOW, + "--repo", + host.repository, + "--ref", + pr.branch, + "-f", + "enter_queue=true", + ) + + +class HostArguments(argparse.Namespace): + request: str + repo: Path + scratch: Path + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("request", help="SSH_ORIGINAL_COMMAND, supplied as one quoted argument") + parser.add_argument("--repo", type=Path, default=REPO) + parser.add_argument("--scratch", type=Path, required=True) + args = parser.parse_args(namespace=HostArguments()) + tokens = shlex.split(args.request) + if len(tokens) < 2: + parser.error("Expected RUN_ID CHECK_KIND [APPROVED_HEAD]") + check = QueueCheck.member_named(tokens[1]).from_tokens(tokens) + repository = FieldCodec.decode( + RepositoryAnswer, + FieldCodec.parse(run("gh", "repo", "view", "--json", "nameWithOwner", cwd=args.repo)), + ) + try: + check.execute(Host(args.repo, repository.name, args.scratch)) + except CheckFailed as error: + print(str(error), flush=True) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/run_merge_runner.sh b/tools/run_merge_runner.sh new file mode 100755 index 0000000000..0b096b17fb --- /dev/null +++ b/tools/run_merge_runner.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash +# A repo runner is not a login shell. Hide host homes, buses, auth and sockets. +set -euo pipefail +state=${1:?usage: run_merge_runner.sh STATE_ROOT} +mkdir -p "$state/actions-runner/_work" "$state/actions-runner/_diag" +home=${HOME:?host home required} +user=$(id -un) +exec env -i \ + HOME="$home" USER="$user" PATH=/usr/bin:/bin LANG=C.UTF-8 \ + DOTNET_CLI_HOME=/tmp DOTNET_GCHeapHardLimit=0x10000000 \ + AGENT_COMMS_GATE_KEY="$home/.ssh/merge-gate-ed25519" \ + AGENT_COMMS_GATE_ENDPOINT="$user@127.0.0.1" \ + bwrap --unshare-all --share-net --die-with-parent --new-session --cap-drop ALL \ + --ro-bind /usr /usr --ro-bind /lib /lib --ro-bind /lib64 /lib64 \ + --ro-bind /etc/passwd /etc/passwd --ro-bind /etc/group /etc/group \ + --ro-bind /etc/nsswitch.conf /etc/nsswitch.conf \ + --ro-bind /etc/resolv.conf /etc/resolv.conf --ro-bind /etc/hosts /etc/hosts \ + --ro-bind /etc/ssl/certs /etc/ssl/certs \ + --ro-bind /etc/ca-certificates /etc/ca-certificates \ + --dev /dev --proc /proc --tmpfs /tmp \ + --ro-bind "$state/runner-home" "$home" \ + --ro-bind "$state/actions-runner" /runner \ + --bind "$state/actions-runner/_work" /runner/_work \ + --bind "$state/actions-runner/_diag" /runner/_diag \ + --chdir /runner /runner/bin/Runner.Listener run diff --git a/tools/upgrade_bus.py b/tools/upgrade_bus.py index 1c5de86077..9973ee25fa 100644 --- a/tools/upgrade_bus.py +++ b/tools/upgrade_bus.py @@ -448,7 +448,7 @@ def main() -> None: try: print(json.dumps(upgrade(args.checkout, args.root, args.to, args.installer, args.rehearse), indent=2)) except UpgradeRefused as error: - raise SystemExit(f"upgrade-bus: {error}") + raise SystemExit(error.output) if __name__ == "__main__":