From 5ad95db67c2b8b30942d686ef9b03428108d74e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vesper=20=F0=9F=8C=99?= Date: Mon, 20 Jul 2026 13:37:57 +0000 Subject: [PATCH 1/2] feat: add provenance-safe actor substitutes Allow each frozen turn to pre-authorize same-role substitute actors and bounded reason codes while preserving legacy primary-only definitions. Keep the scheduled and actual identities distinct across claims, task briefings, runtime evidence, state, and exact Git metadata. Add structured continuation anchors and agent-final statuses without changing the owner decision boundary or final hard stop. Enforce canonical Hermes profile separation, append-only history, and event-specific metadata validation. Cover primary and substitute routes, compatibility, metadata rewriting, profile aliases, continuation mutation, and final-status grammar in the verify gate. --- .github/workflows/ci.yml | 2 +- CONTRIBUTING.md | 2 +- README.md | 3 +- docs/technical-reference.md | 71 +++- examples/demo-common.sh | 2 +- pyproject.toml | 3 + schemas/protocol.schema.json | 67 +++- schemas/runtime-evidence.schema.json | 18 + src/multi_agent_dialogue/adapters/base.py | 8 + src/multi_agent_dialogue/cli.py | 40 +- src/multi_agent_dialogue/config.py | 196 ++++++++- src/multi_agent_dialogue/engine.py | 468 ++++++++++++++++++++-- src/multi_agent_dialogue/evidence.py | 27 ++ src/multi_agent_dialogue/gitops.py | 33 ++ src/multi_agent_dialogue/runner.py | 122 +++++- src/multi_agent_dialogue/unverified.py | 17 +- tests/test_adapters.py | 189 +++++++++ tests/test_cli.py | 77 ++++ tests/test_config.py | 134 +++++++ tests/test_engine.py | 173 +++++++- tests/test_evidence.py | 92 +++++ tests/test_git_transactions.py | 263 +++++++++++- 22 files changed, 1947 insertions(+), 60 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 71ff9df..d7b6351 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,7 +34,7 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install the CLI - run: python -m pip install --disable-pip-version-check -e . + run: python -m pip install --disable-pip-version-check -e '.[test]' - name: Smoke-test the installed command run: madp --help diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 79fc9c4..0268625 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -9,7 +9,7 @@ git clone https://github.com/getaskclaw/multi-agent-dialogue-protocol.git cd multi-agent-dialogue-protocol python3 -m venv .venv . .venv/bin/activate -python -m pip install -e . +python -m pip install -e '.[test]' ``` ## Verify diff --git a/README.md b/README.md index 38d4830..8521e27 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ MADP is useful when you need: - a fixed turn order and hard final stop; - one explicit agent launch at a time; - provider/model/session evidence instead of Markdown identity labels; +- frozen-definition-preauthorized substitute actors for a round when a primary runtime is unavailable, without identity impersonation; - one Git commit for initialization, every accepted turn, and the owner decision; - fail-closed behavior when evidence, process cleanup, state, or Git provenance is wrong. @@ -67,7 +68,7 @@ madp validate DIR --require-git --require-runner-completion madp owner-decide DIR --decision DECISION.md ``` -`madp run` is dry-run by default unless `--launch` is present. One launch processes at most one scheduled turn. Automatic loops are deliberately outside the engine. +`madp run` is dry-run by default unless `--launch` is present. One launch processes at most one scheduled turn. A turn may freeze an ordered `substitute_actor_ids` list and a bounded `substitution_reasons` allowlist beside its primary `actor_id`; `status` shows the legal routes, and the operator explicitly selects a substitute with `--actor ... --substitution-reason ...`. The completed turn records the primary actor, actual runtime actor, and reason. Automatic loops and automatic cooldown classification are deliberately outside the engine. ## Supported adapters diff --git a/docs/technical-reference.md b/docs/technical-reference.md index b6ad64d..8d206a8 100644 --- a/docs/technical-reference.md +++ b/docs/technical-reference.md @@ -23,13 +23,20 @@ madp owner-decide DIR --decision DECISION.md dialogue path that already contains generated files (`state.json`, `definition.json`, `.gitignore`). 2. **`madp status DIR`** — one JSON response naming the current state, - the next scheduled actor and round, the active claim, + the next primary scheduled actor, every frozen-definition-preauthorized substitute + actor for that round, the active claim, `blocked_reason` (why nothing can proceed, or `null`), `next_legal_action` (the one thing that may legally happen next), and `recovered_turns` (how many completed turns carry caller-supplied provenance). 3. **`madp run DIR --actor ACTOR --launch`** — executes exactly one - scheduled turn through its transport adapter and completes it. This + scheduled turn through the selected actor's transport adapter and + completes it. `ACTOR` must be either the primary `actor_id` or one of + that turn's frozen `substitute_actor_ids`. A substitute also requires + `--substitution-reason REASON`, where `REASON` is frozen in that turn's + `substitution_reasons` allowlist. A substitute keeps its own + actor/provider/model/session identity and never impersonates the primary. + This is the **only production-honest completion door**: it is the only path that records `completed_via: runner-launch`. Run it once per scheduled turn (dry-run is the default without `--launch`: @@ -48,6 +55,53 @@ madp owner-decide DIR --decision DECISION.md turn; it is not a required transition. Automatic multi-turn loops are deliberately out of scope; every launch is one explicit turn. +## Primary actors and frozen-definition substitutes + +Every schedule entry still names one primary `actor_id`. It may also name an +ordered `substitute_actor_ids` list. Both are frozen into `definition.json` +at initialization and therefore covered by the definition digest and init +commit. A substitute is allowed only when: + +- its actor definition already exists with explicit transport, provider, and + model constraints; +- its protocol `role` exactly matches the primary actor's role, so failover + cannot turn a challenge into another proposal or otherwise change the seat; +- its ID appears in that exact turn's `substitute_actor_ids` list; +- the operator explicitly selects it with `run --actor ... + --substitution-reason ... --launch`, using a reason code frozen in the + same turn's `substitution_reasons`; and +- its adapter-derived runtime evidence matches its own constraints. + +The claim, adapter-derived runtime evidence, completed-turn state, command +result, task briefing, and Git +provenance record all separate: + +- `scheduled_actor_id` — the primary actor named by the schedule; +- `actor_id` — the actual runtime actor selected for this turn; +- `actor_selection` — `primary` or `substitute`; and +- `substitution_reason` — `null` for the primary, otherwise the selected + frozen reason code. + +This proves preauthorization by the frozen definition, not authenticated owner +approval and not automatic failover. MADP does not authenticate who initialized +the definition; `owner_proof_argv` applies only to the final owner decision. +MADP also does not diagnose a +provider cooldown or choose a substitute on its own. An external scheduler or +human may select an already frozen substitute after verifying the primary is +unavailable. If the active definition did not preapprove a suitable +substitute, do not edit it: create a new bounded continuation with a higher +version and point its evidence roots at the prior instance. + +A bounded continuation may also include a structured `continuation` anchor. +MADP then rechecks the imported artifact's SHA-256, exact publication commit, +original dialogue HEAD, and `start_round` before initialization and on every +state read. Mutable evidence-root paths alone are not continuation proof. + +When `agent_final_statuses` is configured, the final worker artifact must +contain exactly one `Status: ` line from that frozen enum. Those tokens +must not overlap `owner_decisions`; reaching that status moves the dialogue to +`READY_FOR_OWNER` but never creates an owner decision. + ```bash python3 -m unittest discover -s tests # full suite (PYTHONPATH=src, or run scripts/verify.py) python3 scripts/verify.py # compile + tests + schemas + secret scan + git hygiene @@ -62,8 +116,8 @@ top-level CLI. They live, together with `release`, in an explicitly unverified recovery namespace: ```bash -python -m multi_agent_dialogue.unverified claim DIR --actor ACTOR [--revision N] -python -m multi_agent_dialogue.unverified prepare DIR --actor ACTOR --output TASK.md +python -m multi_agent_dialogue.unverified claim DIR --actor ACTOR [--substitution-reason REASON] [--revision N] +python -m multi_agent_dialogue.unverified prepare DIR --actor ACTOR [--substitution-reason REASON] --output TASK.md python -m multi_agent_dialogue.unverified complete DIR --actor ACTOR --turn TURN.md --runtime-evidence EVIDENCE.json python -m multi_agent_dialogue.unverified release DIR --actor ACTOR ``` @@ -305,6 +359,8 @@ happens to see: - **Trailers identify the runtime and the completion door.** Commit messages carry non-secret machine-readable trailers: `Madp-Protocol`, `Madp-Event`, `Madp-Round`, `Madp-Actor`, + `Madp-Scheduled-Actor`, `Madp-Actor-Selection`, + `Madp-Substitution-Reason`, `Madp-Transport`, `Madp-Provider`, `Madp-Model`, `Madp-Session`, `Madp-Completed-Via`, `Madp-Artifact-Sha256`, `Madp-Evidence-Sha256` (turn commits), `Madp-Definition-Digest` @@ -349,7 +405,9 @@ the humans or schedulers around the protocol. - claims are atomic (`O_EXCL` lock + compare-and-swap revision); two writers can never own one turn; -- wrong actor, wrong round, duplicate claim, stale revision → error; +- an actor not frozen as either the primary or an approved substitute for + that exact round, a missing/unapproved substitution reason, wrong round, + duplicate claim, or stale revision → error; - published turns are immutable; any byte change flips the dialogue to `BLOCKED` before the next completion or decision; - missing, malformed, or mismatched runtime evidence blocks completion; @@ -398,7 +456,8 @@ are one local Git commit each. - **Same model ≠ diverse models.** Two sessions of the same provider/model are separate contexts and separate session IDs, but they are *not* model-diverse evidence, and the engine does not claim - otherwise. + otherwise. This remains true when the second session is an explicitly + recorded substitute actor. - **Model audit ≠ provider proof.** The fable audit is the authority for model purity of the observed stream, and the Hermes `state.db` records the billing provider it was configured with; neither diff --git a/examples/demo-common.sh b/examples/demo-common.sh index b2df009..ac3b9bd 100644 --- a/examples/demo-common.sh +++ b/examples/demo-common.sh @@ -52,7 +52,7 @@ madp_example_assert_wrong_actor() { local work_dir="$DIALOGUE/work" local marker="$work_dir/wrong-actor-spawn-marker" local output="$work_dir/wrong-actor-output.txt" - local expected_error="'$wrong_actor' is not the scheduled actor for $round_id; next actor is '$expected_actor'" + local expected_error="'$wrong_actor' is not an allowed actor for $round_id; primary actor is '$expected_actor'" local state_before local state_after diff --git a/pyproject.toml b/pyproject.toml index 9a57646..374de96 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,6 +21,9 @@ classifiers = [ ] dependencies = [] +[project.optional-dependencies] +test = ["jsonschema>=4.23,<5"] + [project.urls] Homepage = "https://github.com/getaskclaw/multi-agent-dialogue-protocol" Repository = "https://github.com/getaskclaw/multi-agent-dialogue-protocol" diff --git a/schemas/protocol.schema.json b/schemas/protocol.schema.json index 02f3b77..1bc3067 100644 --- a/schemas/protocol.schema.json +++ b/schemas/protocol.schema.json @@ -27,11 +27,32 @@ "minItems": 1, "items": {"type": "string", "minLength": 1} }, + "agent_final_statuses": { + "type": "array", + "uniqueItems": true, + "items": {"type": "string", "pattern": "^[A-Z][A-Z0-9_]{0,63}$"}, + "description": "Optional frozen statuses for the final agent turn. They must not overlap owner_decisions. When set, the final artifact must contain exactly one Status: line." + }, "owner_proof_argv": { "type": "array", "items": {"type": "string", "minLength": 1}, "description": "Optional external owner-proof verifier command ({decision_file} is substituted). Without it, owner-decide records caller_identity as 'unverified': nothing authenticates WHO invoked the decision." }, + "continuation": { + "type": "object", + "additionalProperties": false, + "required": ["protocol_id", "round_id", "artifact_path", "artifact_sha256", "published_commit", "original_dialogue_head", "start_round"], + "properties": { + "protocol_id": {"type": "string", "minLength": 1}, + "round_id": {"type": "string", "minLength": 1}, + "artifact_path": {"type": "string", "pattern": "^/"}, + "artifact_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "published_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "original_dialogue_head": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "start_round": {"type": "string", "minLength": 1} + }, + "description": "Optional machine-checked anchor for a bounded continuation. The artifact's live bytes, publication commit, original dialogue head, and first new round are frozen into the definition and rechecked on every state read." + }, "actors": { "type": "array", "minItems": 2, @@ -68,9 +89,53 @@ "type": "object", "required": ["round_id", "actor_id", "purpose", "artifact_kind"], "additionalProperties": false, + "allOf": [ + { + "if": { + "required": ["substitute_actor_ids"], + "properties": {"substitute_actor_ids": {"minItems": 1}} + }, + "then": { + "required": ["substitution_reasons"], + "properties": {"substitution_reasons": {"minItems": 1}} + }, + "else": { + "properties": {"substitution_reasons": {"maxItems": 0}} + } + }, + { + "if": { + "required": ["substitution_reasons"], + "properties": {"substitution_reasons": {"minItems": 1}} + }, + "then": { + "required": ["substitute_actor_ids"], + "properties": {"substitute_actor_ids": {"minItems": 1}} + }, + "else": { + "properties": {"substitute_actor_ids": {"maxItems": 0}} + } + } + ], "properties": { "round_id": {"type": "string", "minLength": 1}, - "actor_id": {"type": "string", "minLength": 1}, + "actor_id": { + "type": "string", + "minLength": 1, + "description": "Primary scheduled actor for this turn." + }, + "substitute_actor_ids": { + "type": "array", + "uniqueItems": true, + "items": {"type": "string", "minLength": 1}, + "description": "Ordered alternative actors preauthorized by the frozen definition. Each substitute must have the same protocol role as the primary actor_id, retains its own actor/provider/model identity, and never impersonates actor_id. Initialization does not authenticate who approved the definition." + }, + "substitution_reasons": { + "type": "array", + "uniqueItems": true, + "items": {"type": "string", "pattern": "^[a-z][a-z0-9_-]{0,63}$"}, + "description": "Frozen reason-code allowlist required whenever substitute_actor_ids is non-empty. A substitute launch must provide one of these codes." + }, "purpose": {"type": "string", "minLength": 1}, "artifact_kind": {"type": "string", "minLength": 1}, "word_limit": {"type": "integer", "minimum": 1} diff --git a/schemas/runtime-evidence.schema.json b/schemas/runtime-evidence.schema.json index 36eb006..e246856 100644 --- a/schemas/runtime-evidence.schema.json +++ b/schemas/runtime-evidence.schema.json @@ -24,6 +24,21 @@ "properties": { "evidence_version": {"const": 1}, "actor_id": {"type": "string", "minLength": 1}, + "scheduled_actor_id": { + "type": "string", + "minLength": 1, + "description": "Primary actor frozen in the turn schedule. Required by engine validation for a substitute turn." + }, + "actor_selection": { + "type": "string", + "enum": ["primary", "substitute"], + "description": "Whether actor_id is the primary scheduled actor or a substitute preauthorized by the frozen definition." + }, + "substitution_reason": { + "type": ["string", "null"], + "pattern": "^[a-z][a-z0-9_-]{0,63}$", + "description": "Null for a primary turn; otherwise the frozen reason code selected at claim time. Required by engine validation for a substitute turn." + }, "round_id": {"type": "string", "minLength": 1}, "adapter": {"type": "string", "minLength": 1}, "transport": { @@ -76,6 +91,9 @@ { "evidence_version": 1, "actor_id": "worker-a", + "scheduled_actor_id": "worker-a", + "actor_selection": "primary", + "substitution_reason": null, "round_id": "R01", "adapter": "command", "transport": "command", diff --git a/src/multi_agent_dialogue/adapters/base.py b/src/multi_agent_dialogue/adapters/base.py index aa1463e..a0b49e1 100644 --- a/src/multi_agent_dialogue/adapters/base.py +++ b/src/multi_agent_dialogue/adapters/base.py @@ -47,6 +47,7 @@ class PrepareContext: task_file: Path turn_file: Path evidence_file: Path + substitution_reason: str | None = None def placeholders(self) -> dict[str, str]: return { @@ -192,6 +193,13 @@ def base_evidence(self, context: PrepareContext, *, provider: str, model: str, return { "evidence_version": EVIDENCE_VERSION, "actor_id": context.actor.actor_id, + "scheduled_actor_id": context.turn.actor_id, + "actor_selection": ( + "primary" + if context.actor.actor_id == context.turn.actor_id + else "substitute" + ), + "substitution_reason": context.substitution_reason, "round_id": context.turn.round_id, "adapter": self.name, "transport": self.transport, diff --git a/src/multi_agent_dialogue/cli.py b/src/multi_agent_dialogue/cli.py index d2fe6d4..af3296e 100644 --- a/src/multi_agent_dialogue/cli.py +++ b/src/multi_agent_dialogue/cli.py @@ -40,6 +40,13 @@ def _next_legal_action(state: dict, turn) -> str: ) if status == engine.STATUS_READY_FOR_OWNER or turn is None: return "owner-decide --decision FILE" + if turn.substitute_actor_ids: + substitutes = "|".join(turn.substitute_actor_ids) + reasons = "|".join(turn.substitution_reasons) + return ( + f"run --actor {turn.actor_id} --launch; or run --actor " + f"<{substitutes}> --substitution-reason <{reasons}> --launch" + ) return f"run --actor {turn.actor_id} --launch" @@ -60,6 +67,10 @@ def _status_payload(dialogue: engine.Dialogue) -> dict: else { "round_id": turn.round_id, "actor_id": turn.actor_id, + "primary_actor_id": turn.actor_id, + "allowed_actor_ids": list(turn.allowed_actor_ids), + "substitute_actor_ids": list(turn.substitute_actor_ids), + "substitution_reasons": list(turn.substitution_reasons), "role": definition.actor(turn.actor_id).role, "transport": definition.actor(turn.actor_id).transport, "purpose": turn.purpose, @@ -107,6 +118,10 @@ def cmd_next(args: argparse.Namespace) -> int: "done": False, "round_id": turn.round_id, "actor_id": turn.actor_id, + "primary_actor_id": turn.actor_id, + "allowed_actor_ids": list(turn.allowed_actor_ids), + "substitute_actor_ids": list(turn.substitute_actor_ids), + "substitution_reasons": list(turn.substitution_reasons), "role": actor.role, "transport": actor.transport, "purpose": turn.purpose, @@ -120,9 +135,22 @@ def cmd_next(args: argparse.Namespace) -> int: def cmd_run(args: argparse.Namespace) -> int: dialogue = engine.Dialogue(args.dialogue) if args.launch: - _emit(runner.launch(dialogue, args.actor, timeout=args.timeout)) + _emit( + runner.launch( + dialogue, + args.actor, + timeout=args.timeout, + substitution_reason=args.substitution_reason, + ) + ) else: - _emit(runner.dry_run(dialogue, args.actor)) + _emit( + runner.dry_run( + dialogue, + args.actor, + substitution_reason=args.substitution_reason, + ) + ) return 0 @@ -178,6 +206,14 @@ def build_parser() -> argparse.ArgumentParser: ) p.add_argument("dialogue", type=Path) p.add_argument("--actor", required=True) + p.add_argument( + "--substitution-reason", + default=None, + help=( + "required for a substitute actor; must be one of the turn's " + "frozen substitution_reasons" + ), + ) group = p.add_mutually_exclusive_group() group.add_argument("--dry-run", action="store_true", help="show the command packet without starting any process (default)") diff --git a/src/multi_agent_dialogue/config.py b/src/multi_agent_dialogue/config.py index ae4e332..d56d082 100644 --- a/src/multi_agent_dialogue/config.py +++ b/src/multi_agent_dialogue/config.py @@ -10,6 +10,8 @@ import hashlib import json +import os +import re from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -17,6 +19,45 @@ KNOWN_TRANSPORTS = ("command", "fable-session", "hermes-cli") DEFAULT_OWNER_DECISIONS = ("APPROVE", "REJECT", "NEED_MORE_EVIDENCE") +SUBSTITUTION_REASON_RE = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") +AGENT_STATUS_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$") +SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +GIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$") + +_ROOT_KEYS = { + "protocol_id", + "version", + "owner", + "source_sha", + "evidence_roots", + "owner_decisions", + "agent_final_statuses", + "owner_proof_argv", + "actors", + "schedule", + "final_round_id", + "continuation", +} +_CONTINUATION_KEYS = { + "protocol_id", + "round_id", + "artifact_path", + "artifact_sha256", + "published_commit", + "original_dialogue_head", + "start_round", +} + + +def static_hermes_home_key(actor: "Actor") -> tuple[bool, str] | None: + """Best-effort profile identity available before a dialogue path exists.""" + if actor.transport != "hermes-cli": + return None + value = actor.settings.get("hermes_home") + if not isinstance(value, str) or not value.strip(): + return None + path = Path(value) + return path.is_absolute(), os.path.normpath(value) _ACTOR_KEYS = { "actor_id", @@ -26,7 +67,15 @@ "expected_model", "settings", } -_TURN_KEYS = {"round_id", "actor_id", "purpose", "artifact_kind", "word_limit"} +_TURN_KEYS = { + "round_id", + "actor_id", + "substitute_actor_ids", + "substitution_reasons", + "purpose", + "artifact_kind", + "word_limit", +} class ConfigError(ValueError): @@ -50,6 +99,26 @@ class TurnSpec: purpose: str artifact_kind: str word_limit: int | None = None + # Ordered alternatives preauthorized by the frozen definition.json. + # ``actor_id`` remains the primary scheduled actor; a substitute never + # inherits that identity and is recorded under its own actor_id. + substitute_actor_ids: tuple[str, ...] = () + substitution_reasons: tuple[str, ...] = () + + @property + def allowed_actor_ids(self) -> tuple[str, ...]: + return (self.actor_id, *self.substitute_actor_ids) + + +@dataclass(frozen=True) +class ContinuationAnchor: + protocol_id: str + round_id: str + artifact_path: str + artifact_sha256: str + published_commit: str + original_dialogue_head: str + start_round: str @dataclass(frozen=True) @@ -63,10 +132,12 @@ class ProtocolDefinition: source_sha: str evidence_roots: tuple[str, ...] owner_decisions: tuple[str, ...] + agent_final_statuses: tuple[str, ...] = () # Optional external owner-proof verifier command. When empty, the # engine records the owner decision as caller-identity "unverified": # nothing authenticates WHO invoked owner-decide. owner_proof_argv: tuple[str, ...] = () + continuation: ContinuationAnchor | None = None raw: dict[str, Any] = field(repr=False, compare=False, default_factory=dict) def actor(self, actor_id: str) -> Actor: @@ -146,6 +217,44 @@ def _parse_turn(raw: Any, position: int, errors: list[str]) -> TurnSpec | None: ) round_id = _require_str(raw, "round_id", where, errors) actor_id = _require_str(raw, "actor_id", where, errors) + substitutes_raw = raw.get("substitute_actor_ids", []) + if not isinstance(substitutes_raw, list) or not all( + isinstance(item, str) and item.strip() for item in substitutes_raw + ): + errors.append( + f"{where}: substitute_actor_ids must be a list of non-empty actor IDs" + ) + substitutes_raw = [] + if actor_id and actor_id in substitutes_raw: + errors.append( + f"{where}: primary actor {actor_id!r} must not be repeated as a substitute" + ) + if len(substitutes_raw) != len(set(substitutes_raw)): + errors.append(f"{where}: duplicate substitute actor IDs are not allowed") + reasons_raw = raw.get("substitution_reasons", []) + if not isinstance(reasons_raw, list) or not all( + isinstance(item, str) and item.strip() for item in reasons_raw + ): + errors.append( + f"{where}: substitution_reasons must be a list of non-empty reason codes" + ) + reasons_raw = [] + if len(reasons_raw) != len(set(reasons_raw)): + errors.append(f"{where}: duplicate substitution reason codes are not allowed") + for reason in reasons_raw: + if not SUBSTITUTION_REASON_RE.fullmatch(reason): + errors.append( + f"{where}: {reason!r} is not a safe reason code; use 1-64 " + "lowercase letters, digits, underscores, or hyphens" + ) + if substitutes_raw and not reasons_raw: + errors.append( + f"{where}: substitution_reasons is required when substitute actors exist" + ) + if reasons_raw and not substitutes_raw: + errors.append( + f"{where}: substitution_reasons requires substitute_actor_ids" + ) purpose = _require_str(raw, "purpose", where, errors) artifact_kind = _require_str(raw, "artifact_kind", where, errors) word_limit = raw.get("word_limit") @@ -158,9 +267,35 @@ def _parse_turn(raw: Any, position: int, errors: list[str]) -> TurnSpec | None: purpose=purpose, artifact_kind=artifact_kind, word_limit=word_limit, + substitute_actor_ids=tuple(substitutes_raw), + substitution_reasons=tuple(reasons_raw), ) +def _parse_continuation(raw: Any, errors: list[str]) -> ContinuationAnchor | None: + if raw is None: + return None + where = "definition.continuation" + if not isinstance(raw, dict): + errors.append(f"{where}: must be an object") + return None + unknown = set(raw) - _CONTINUATION_KEYS + if unknown: + errors.append(f"{where}: unknown keys: {sorted(unknown)}") + values = {key: _require_str(raw, key, where, errors) for key in _CONTINUATION_KEYS} + artifact_path = values["artifact_path"] + if artifact_path and not Path(artifact_path).is_absolute(): + errors.append(f"{where}: artifact_path must be absolute") + if values["artifact_sha256"] and not SHA256_RE.fullmatch( + values["artifact_sha256"] + ): + errors.append(f"{where}: artifact_sha256 must be 64 lowercase hex characters") + for key in ("published_commit", "original_dialogue_head"): + if values[key] and not GIT_SHA_RE.fullmatch(values[key]): + errors.append(f"{where}: {key} must be a full 40-character Git SHA") + return ContinuationAnchor(**values) + + _SECRET_MARKERS = ( "sk-" + "ant-", "AKIA", @@ -179,6 +314,9 @@ def parse_definition(raw: Any) -> ProtocolDefinition: if not isinstance(raw, dict): raise ConfigError("protocol definition must be a JSON object") errors: list[str] = [] + unknown_root = set(raw) - _ROOT_KEYS + if unknown_root: + errors.append(f"definition: unknown keys: {sorted(unknown_root)}") protocol_id = _require_str(raw, "protocol_id", "definition", errors) owner = _require_str(raw, "owner", "definition", errors) @@ -208,6 +346,8 @@ def parse_definition(raw: Any) -> ProtocolDefinition: ) owner_proof_raw = [] + continuation = _parse_continuation(raw.get("continuation"), errors) + decisions_raw = raw.get("owner_decisions", list(DEFAULT_OWNER_DECISIONS)) if ( not isinstance(decisions_raw, list) @@ -217,6 +357,25 @@ def parse_definition(raw: Any) -> ProtocolDefinition: errors.append("definition: owner_decisions must be a non-empty list of strings") decisions_raw = list(DEFAULT_OWNER_DECISIONS) + agent_statuses_raw = raw.get("agent_final_statuses", []) + if not isinstance(agent_statuses_raw, list) or not all( + isinstance(item, str) and AGENT_STATUS_RE.fullmatch(item) + for item in agent_statuses_raw + ): + errors.append( + "definition: agent_final_statuses must be a list of uppercase " + "status tokens" + ) + agent_statuses_raw = [] + if len(agent_statuses_raw) != len(set(agent_statuses_raw)): + errors.append("definition: duplicate agent_final_statuses are not allowed") + overlap = set(agent_statuses_raw) & set(decisions_raw) + if overlap: + errors.append( + "definition: agent_final_statuses must not overlap owner_decisions: " + f"{sorted(overlap)}" + ) + actors_raw = raw.get("actors") actors: list[Actor] = [] if not isinstance(actors_raw, list): @@ -252,7 +411,8 @@ def parse_definition(raw: Any) -> ProtocolDefinition: if turn is not None: schedule.append(turn) seen_rounds: set[str] = set() - actor_ids = {actor.actor_id for actor in actors} + actors_by_id = {actor.actor_id: actor for actor in actors} + actor_ids = set(actors_by_id) for turn in schedule: if turn.round_id in seen_rounds: errors.append(f"definition: duplicate round_id {turn.round_id!r}") @@ -261,6 +421,30 @@ def parse_definition(raw: Any) -> ProtocolDefinition: errors.append( f"definition: schedule references unknown actor {turn.actor_id!r}" ) + for substitute_id in turn.substitute_actor_ids: + if substitute_id not in actor_ids: + errors.append( + "definition: schedule references unknown substitute actor " + f"{substitute_id!r} in {turn.round_id}" + ) + continue + primary = actors_by_id.get(turn.actor_id) + substitute = actors_by_id[substitute_id] + if primary is not None and substitute.role != primary.role: + errors.append( + f"definition: substitute actor {substitute_id!r} role must " + f"match primary actor {turn.actor_id!r} role {primary.role!r} " + f"in {turn.round_id}" + ) + if primary is not None: + primary_home = static_hermes_home_key(primary) + substitute_home = static_hermes_home_key(substitute) + if primary_home is not None and primary_home == substitute_home: + errors.append( + f"definition: substitute actor {substitute_id!r} must use " + f"a distinct hermes_home from primary actor " + f"{turn.actor_id!r} in {turn.round_id}" + ) final_round_id = raw.get("final_round_id") if not isinstance(final_round_id, str) or not final_round_id: @@ -273,6 +457,12 @@ def parse_definition(raw: Any) -> ProtocolDefinition: f"definition: final_round_id {final_round_id!r} must equal the last " f"scheduled round {schedule[-1].round_id!r}" ) + if continuation is not None and schedule: + if continuation.start_round != schedule[0].round_id: + errors.append( + "definition.continuation: start_round must equal the first " + f"scheduled round {schedule[0].round_id!r}" + ) if errors: raise ConfigError("invalid protocol definition:\n- " + "\n- ".join(errors)) @@ -287,7 +477,9 @@ def parse_definition(raw: Any) -> ProtocolDefinition: source_sha=source_sha, evidence_roots=tuple(evidence_roots_raw), owner_decisions=tuple(decisions_raw), + agent_final_statuses=tuple(agent_statuses_raw), owner_proof_argv=tuple(owner_proof_raw), + continuation=continuation, # Snapshot so later caller mutations cannot change the digest. raw=json.loads(canonical_json(raw)), ) diff --git a/src/multi_agent_dialogue/engine.py b/src/multi_agent_dialogue/engine.py index 7f0f9b9..b5be320 100644 --- a/src/multi_agent_dialogue/engine.py +++ b/src/multi_agent_dialogue/engine.py @@ -19,6 +19,7 @@ import json import os +import re import secrets import tempfile from datetime import datetime, timezone @@ -72,6 +73,192 @@ class ProtocolError(RuntimeError): """A dialogue transition is not allowed; nothing was changed.""" +def resolve_actor_selection( + turn: config.TurnSpec, + actor_id: str, + substitution_reason: str | None = None, +) -> tuple[str, str | None]: + """Validate one primary/substitute selection against the frozen turn.""" + if actor_id not in turn.allowed_actor_ids: + raise ProtocolError( + f"{actor_id!r} is not an allowed actor for {turn.round_id}; " + f"primary actor is {turn.actor_id!r}, allowed actors are " + f"{list(turn.allowed_actor_ids)!r}" + ) + reason = substitution_reason.strip() if isinstance(substitution_reason, str) else None + reason = reason or None + if actor_id == turn.actor_id: + if reason is not None: + raise ProtocolError( + f"primary actor {actor_id!r} must not claim a substitution reason" + ) + return "primary", None + if reason is None: + raise ProtocolError( + f"substitute actor {actor_id!r} requires a substitution reason; " + f"allowed reasons are {list(turn.substitution_reasons)!r}" + ) + if reason not in turn.substitution_reasons: + raise ProtocolError( + f"substitution reason {reason!r} is not allowed for {turn.round_id}; " + f"allowed reasons are {list(turn.substitution_reasons)!r}" + ) + return "substitute", reason + + +def selection_record_errors( + turn: config.TurnSpec, + actor_id: object, + record: object, + label: str, +) -> list[str]: + """Validate persisted primary/substitute identity fields. + + Historical primary-only turns may omit the three fields added by the + substitute-actor extension. Once a frozen turn allows substitutes, every + route is explicit: primary and substitute records both carry all fields. + """ + if not isinstance(record, dict): + return [f"{label} is not an object"] + errors: list[str] = [] + if actor_id not in turn.allowed_actor_ids: + return [ + f"{label} actor {actor_id!r} is not allowed for {turn.round_id}; " + f"allowed actors are {list(turn.allowed_actor_ids)!r}" + ] + expected_selection = "primary" if actor_id == turn.actor_id else "substitute" + requires_explicit_identity = bool(turn.substitute_actor_ids) + if requires_explicit_identity or expected_selection == "substitute": + missing = [ + key + for key in ( + "scheduled_actor_id", + "actor_selection", + "substitution_reason", + ) + if key not in record + ] + if missing: + errors.append( + f"{label} explicit actor identity fields are missing: {missing}" + ) + legacy_primary = expected_selection == "primary" and not turn.substitute_actor_ids + scheduled_default = turn.actor_id if legacy_primary else None + selection_default = "primary" if legacy_primary else None + recorded_scheduled = record.get("scheduled_actor_id", scheduled_default) + recorded_selection = record.get("actor_selection", selection_default) + substitution_reason = record.get("substitution_reason") + if recorded_scheduled != turn.actor_id: + errors.append( + f"{label} scheduled_actor_id {recorded_scheduled!r} does not match " + f"frozen primary actor {turn.actor_id!r}" + ) + if recorded_selection != expected_selection: + errors.append( + f"{label} actor_selection {recorded_selection!r} does not match " + f"actual selection {expected_selection!r}" + ) + if expected_selection == "primary" and substitution_reason is not None: + errors.append( + f"{label} primary actor records unexpected substitution_reason " + f"{substitution_reason!r}" + ) + if expected_selection == "substitute" and ( + not isinstance(substitution_reason, str) + or substitution_reason not in turn.substitution_reasons + ): + errors.append( + f"{label} substitute records invalid substitution_reason " + f"{substitution_reason!r}; allowed reasons are " + f"{list(turn.substitution_reasons)!r}" + ) + return errors + + +def hermes_profile_isolation_errors( + definition: config.ProtocolDefinition, + dialogue_directory: Path, + turn: config.TurnSpec | None = None, +) -> list[str]: + """Prove primary/substitute Hermes actors use distinct canonical profiles.""" + errors: list[str] = [] + pairs: list[tuple[config.TurnSpec, str]] = [] + if turn is not None: + pairs.extend((turn, item) for item in turn.substitute_actor_ids) + else: + for spec in definition.schedule: + pairs.extend((spec, item) for item in spec.substitute_actor_ids) + + def canonical_home(actor: config.Actor) -> Path | None: + raw = actor.settings.get("hermes_home") + if not isinstance(raw, str) or not raw.strip(): + return None + path = Path(raw) + if not path.is_absolute(): + path = dialogue_directory / path + return path.resolve(strict=False) + + for spec, substitute_id in pairs: + primary = definition.actor(spec.actor_id) + substitute = definition.actor(substitute_id) + if primary.transport != "hermes-cli" or substitute.transport != "hermes-cli": + continue + primary_home = canonical_home(primary) + substitute_home = canonical_home(substitute) + label = f"turn {spec.round_id} Hermes primary/substitute profile isolation" + if primary_home is None or substitute_home is None: + errors.append(f"{label} requires non-empty hermes_home settings") + elif primary_home == substitute_home: + errors.append( + f"{label} failed: actors {spec.actor_id!r} and {substitute_id!r} " + "resolve to the same HERMES_HOME" + ) + return errors + + +def commit_trailer_errors( + root: Path, + commit: str, + expected: dict[str, str], + label: str, + optional_missing: set[str] | None = None, +) -> list[str]: + """Validate the exact event-specific Madp-* trailer contract.""" + errors: list[str] = [] + expected_folded = { + key.casefold(): (key, value) for key, value in expected.items() + } + optional_folded = {key.casefold() for key in (optional_missing or set())} + try: + pairs = gitops.commit_trailers(root, commit) + except gitops.GitError as exc: + return [f"{label}: cannot read Git trailers: {exc}"] + trailers: dict[str, tuple[str, str]] = {} + for key, value in pairs: + folded = key.casefold() + if not folded.startswith("madp-"): + continue + canonical = expected_folded.get(folded, (None, ""))[0] + if canonical is not None and key != canonical: + errors.append( + f"{label}: non-canonical Git trailer spelling {key!r}; " + f"expected {canonical!r}" + ) + if folded in trailers: + errors.append(f"{label}: duplicate Git trailer {key!r}") + continue + trailers[folded] = (key, value) + for folded in sorted(set(trailers) - set(expected_folded)): + errors.append(f"{label}: unexpected Git trailer {trailers[folded][0]!r}") + for folded, (key, wanted) in expected_folded.items(): + if folded in optional_folded and folded not in trailers: + continue + actual = trailers.get(folded) + if actual is None or actual[1] != wanted: + errors.append(f"{label}: Git trailer {key} does not match committed data") + return errors + + def utc_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") @@ -118,8 +305,56 @@ def atomic_write_json(path: Path, payload: dict) -> None: os.close(dir_fd) +def verify_continuation_anchor(definition: config.ProtocolDefinition) -> None: + """Verify an optional imported-history anchor against bytes and Git.""" + anchor = definition.continuation + if anchor is None: + return + path = Path(anchor.artifact_path) + try: + current = artifacts.read_bytes_nofollow(path, "continuation artifact") + except artifacts.ArtifactError as exc: + raise ProtocolError(str(exc)) from exc + if artifacts.sha256_bytes(current) != anchor.artifact_sha256: + raise ProtocolError( + f"continuation artifact hash mismatch for {anchor.protocol_id}/" + f"{anchor.round_id}" + ) + source_root = gitops.worktree_root(path.parent) + if source_root is None: + raise ProtocolError("continuation artifact is not inside a Git worktree") + try: + rel = gitops.rel_to_root(source_root, path) + except ValueError as exc: + raise ProtocolError("continuation artifact escaped its Git worktree") from exc + if gitops.first_commit_adding(source_root, rel) != anchor.published_commit: + raise ProtocolError( + "continuation published_commit is not the artifact's publication commit" + ) + if not gitops.is_ancestor( + source_root, anchor.published_commit, anchor.original_dialogue_head + ): + raise ProtocolError( + "continuation publication is not an ancestor of original_dialogue_head" + ) + for commit, label in ( + (anchor.published_commit, "published_commit"), + (anchor.original_dialogue_head, "original_dialogue_head"), + ): + if gitops.committed_sha256(source_root, commit, rel) != anchor.artifact_sha256: + raise ProtocolError( + f"continuation artifact bytes do not match {label}" + ) + + def init_dialogue(definition: config.ProtocolDefinition, directory: Path | str) -> "Dialogue": + verify_continuation_anchor(definition) directory = Path(directory) + isolation_errors = hermes_profile_isolation_errors( + definition, directory.resolve(strict=False) + ) + if isolation_errors: + raise ProtocolError("\n".join(isolation_errors)) _reject_symlink(directory, "dialogue directory") if (directory / STATE_FILE).exists() or (directory / DEFINITION_FILE).exists(): raise ProtocolError(f"dialogue already initialized: {directory}") @@ -218,6 +453,7 @@ def state(self) -> dict: raise ProtocolError( "definition digest mismatch: definition.json was modified after init" ) + verify_continuation_anchor(definition) return state def _write_state(self, state: dict) -> dict: @@ -249,7 +485,12 @@ def _require_worker_phase(self, state: dict, definition: config.ProtocolDefiniti # -- claims ---------------------------------------------------------- - def claim(self, actor_id: str, expected_revision: int | None = None) -> dict: + def claim( + self, + actor_id: str, + expected_revision: int | None = None, + substitution_reason: str | None = None, + ) -> dict: definition = self.definition() try: definition.actor(actor_id) @@ -269,14 +510,20 @@ def claim(self, actor_id: str, expected_revision: int | None = None) -> dict: f"at {claim.get('claimed_at')!r}" ) turn = definition.schedule[int(state["turn_index"])] - if turn.actor_id != actor_id: - raise ProtocolError( - f"{actor_id!r} is not the scheduled actor for {turn.round_id}; " - f"next actor is {turn.actor_id!r}" - ) + actor_selection, substitution_reason = resolve_actor_selection( + turn, actor_id, substitution_reason + ) + isolation_errors = hermes_profile_isolation_errors( + definition, self.directory, turn + ) + if isolation_errors: + raise ProtocolError("\n".join(isolation_errors)) claim = { "actor_id": actor_id, + "scheduled_actor_id": turn.actor_id, + "actor_selection": actor_selection, + "substitution_reason": substitution_reason, "round_id": turn.round_id, "nonce": secrets.token_hex(16), "claimed_at": utc_now(), @@ -416,10 +663,17 @@ def complete( f"turn is claimed by {claim.get('actor_id')!r}, not {actor_id!r}" ) turn = definition.schedule[int(state["turn_index"])] - if turn.actor_id != actor_id: - raise ProtocolError( - f"{actor_id!r} is not the scheduled actor for {turn.round_id}" - ) + actor_selection, substitution_reason = resolve_actor_selection( + turn, actor_id, claim.get("substitution_reason") + ) + isolation_errors = hermes_profile_isolation_errors( + definition, self.directory, turn + ) + if isolation_errors: + raise ProtocolError("\n".join(isolation_errors)) + claim_errors = selection_record_errors(turn, actor_id, claim, "active claim") + if claim_errors: + raise ProtocolError("invalid active claim:\n- " + "\n- ".join(claim_errors)) history_errors = self._verify_history(state) if history_errors: @@ -445,7 +699,11 @@ def complete( actor = definition.actor(actor_id) evidence_errors = evidence.validate_evidence( - record, actor=actor, turn=turn, artifact_sha256=artifact_sha + record, + actor=actor, + turn=turn, + artifact_sha256=artifact_sha, + substitution_reason=substitution_reason, ) if evidence_errors: raise ProtocolError( @@ -461,16 +719,35 @@ def complete( "previous turn; each turn needs an independent runtime session" ) - if turn.word_limit is not None: + turn_text: str | None = None + if turn.word_limit is not None or ( + turn.round_id == definition.final_round_id + and definition.agent_final_statuses + ): try: - count = artifacts.word_count(turn_data.decode("utf-8")) + turn_text = turn_data.decode("utf-8") except UnicodeDecodeError as exc: raise ProtocolError(f"turn artifact is not UTF-8: {exc}") from exc + if turn.word_limit is not None: + assert turn_text is not None + count = artifacts.word_count(turn_text) if count > turn.word_limit: raise ProtocolError( f"turn body has {count} words, over the {turn.word_limit} " f"word limit for {turn.round_id}" ) + if turn.round_id == definition.final_round_id and definition.agent_final_statuses: + assert turn_text is not None + statuses = re.findall(r"(?m)^Status:[ \t]*(\S+)[ \t]*$", turn_text) + if len(statuses) != 1: + raise ProtocolError( + "final turn must contain exactly one line 'Status: '" + ) + if statuses[0] not in definition.agent_final_statuses: + raise ProtocolError( + f"final agent status {statuses[0]!r} is not allowed; expected " + f"one of {list(definition.agent_final_statuses)!r}" + ) artifact_rel = f"{TURNS_DIR}/{turn.round_id}-{actor_id}.md" evidence_rel = f"{EVIDENCE_DIR}/{turn.round_id}-{actor_id}.json" @@ -484,6 +761,9 @@ def complete( { "round_id": turn.round_id, "actor_id": actor_id, + "scheduled_actor_id": turn.actor_id, + "actor_selection": actor_selection, + "substitution_reason": substitution_reason, "artifact_file": artifact_rel, "artifact_sha256": published_sha, "evidence_file": evidence_rel, @@ -522,6 +802,9 @@ def complete( "Madp-Event": "turn", "Madp-Round": turn.round_id, "Madp-Actor": actor_id, + "Madp-Scheduled-Actor": turn.actor_id, + "Madp-Actor-Selection": actor_selection, + "Madp-Substitution-Reason": substitution_reason or "none", "Madp-Transport": actor.transport, "Madp-Provider": record["provider"], "Madp-Model": record["model"], @@ -576,8 +859,6 @@ def _run_owner_proof(self, definition: config.ProtocolDefinition, return {"argv": argv, "exit_status": 0} def owner_decide(self, decision_path: Path | str) -> dict: - import re - decision_path = Path(decision_path) definition = self.definition() state = self.state() @@ -683,6 +964,7 @@ def validate( errors.append(str(exc)) if definition is not None and state: + errors.extend(hermes_profile_isolation_errors(definition, self.directory)) errors.extend(self._verify_history(state)) completed = state.get("completed_turns", []) if int(state.get("turn_index", -1)) != len(completed): @@ -703,16 +985,17 @@ def validate( f"completed turn {position} is {record.get('round_id')!r}; " f"schedule requires {spec.round_id!r}" ) - if record.get("actor_id") != spec.actor_id: - errors.append( - f"turn {spec.round_id} was completed by " - f"{record.get('actor_id')!r}, schedule requires {spec.actor_id!r}" + actual_actor_id = record.get("actor_id") + errors.extend( + selection_record_errors( + spec, actual_actor_id, record, f"turn {spec.round_id}" ) + ) evidence_path = self.directory / record.get("evidence_file", "") try: evidence_record = evidence.load_evidence(evidence_path) artifact_sha = record.get("artifact_sha256", "") - actor = definition.actor(record.get("actor_id", "")) + actor = definition.actor(actual_actor_id or "") errors.extend( f"{spec.round_id}: {item}" for item in evidence.validate_evidence( @@ -720,6 +1003,7 @@ def validate( actor=actor, turn=spec, artifact_sha256=artifact_sha, + substitution_reason=record.get("substitution_reason"), ) ) except (evidence.EvidenceError, config.ConfigError) as exc: @@ -748,6 +1032,19 @@ def validate( errors.append("state records an active claim but the claim lock file is missing") if not state.get("claim") and lock_exists: errors.append("claim lock file exists without an active claim") + claim = state.get("claim") + if claim and int(state.get("turn_index", -1)) < len(definition.schedule): + spec = definition.schedule[int(state["turn_index"])] + if claim.get("round_id") != spec.round_id: + errors.append( + f"active claim round {claim.get('round_id')!r} does not match " + f"next round {spec.round_id!r}" + ) + errors.extend( + selection_record_errors( + spec, claim.get("actor_id"), claim, "active claim" + ) + ) provenance = self._git_provenance(require_git, state) if require_git: @@ -849,6 +1146,7 @@ def _prove_commit_history( store its own SHA, so provenance is always derived, never self-reported.""" errors: list[str] = [] + definition = self.definition() def rel(name: str) -> str: return gitops.rel_to_root(root, self.directory / name) @@ -865,9 +1163,21 @@ def rel(name: str) -> str: "state in one init commit" ) provenance["init_commit"] = init_commit + errors.extend( + commit_trailer_errors( + root, + init_commit, + { + "Madp-Protocol": definition.protocol_id, + "Madp-Event": "init", + "Madp-Definition-Digest": definition.digest(), + }, + "init commit", + ) + ) - def committed_state_at(commit: str) -> dict | None: - raw = gitops.committed_bytes(root, commit, state_rel) + def committed_json_at(commit: str, rel_path: str) -> dict | None: + raw = gitops.committed_bytes(root, commit, rel_path) if raw is None: return None try: @@ -876,6 +1186,9 @@ def committed_state_at(commit: str) -> dict | None: return None return loaded if isinstance(loaded, dict) else None + def committed_state_at(commit: str) -> dict | None: + return committed_json_at(commit, state_rel) + turn_commits: list[dict] = [] previous = init_commit for position, record in enumerate(state.get("completed_turns", [])): @@ -929,6 +1242,9 @@ def committed_state_at(commit: str) -> dict | None: for key in ( "round_id", "actor_id", + "scheduled_actor_id", + "actor_selection", + "substitution_reason", "artifact_sha256", "evidence_sha256", "session_id", @@ -957,16 +1273,87 @@ def committed_state_at(commit: str) -> dict | None: f"({record.get('completed_via')!r}); completion " "provenance fails closed" ) + if position < len(definition.schedule): + spec = definition.schedule[position] + errors.extend( + selection_record_errors( + spec, + entry.get("actor_id"), + entry, + f"turn {round_id} committed state", + ) + ) + + committed_evidence = committed_json_at(commit, evidence_rel) + if committed_evidence is None: + errors.append( + f"turn {round_id}: committed runtime evidence is not a JSON object" + ) + if isinstance(entry, dict) and isinstance(committed_evidence, dict): + expected_trailers = { + "Madp-Protocol": definition.protocol_id, + "Madp-Event": "turn", + "Madp-Round": str(entry.get("round_id")), + "Madp-Actor": str(entry.get("actor_id")), + "Madp-Scheduled-Actor": str(entry.get("scheduled_actor_id")), + "Madp-Actor-Selection": str(entry.get("actor_selection")), + "Madp-Substitution-Reason": str( + entry.get("substitution_reason") or "none" + ), + "Madp-Transport": str(committed_evidence.get("transport")), + "Madp-Provider": str(committed_evidence.get("provider")), + "Madp-Model": str(committed_evidence.get("model")), + "Madp-Session": str(committed_evidence.get("session_id")), + "Madp-Completed-Via": str(entry.get("completed_via")), + "Madp-Artifact-Sha256": str(entry.get("artifact_sha256")), + "Madp-Evidence-Sha256": str(entry.get("evidence_sha256")), + } + legacy_optional: set[str] = set() + if position < len(definition.schedule): + spec = definition.schedule[position] + is_substitute = entry.get("actor_id") != spec.actor_id + if not is_substitute and not spec.substitute_actor_ids: + legacy_field_map = { + "Madp-Scheduled-Actor": "scheduled_actor_id", + "Madp-Actor-Selection": "actor_selection", + "Madp-Substitution-Reason": "substitution_reason", + } + legacy_optional = { + trailer + for trailer, state_field in legacy_field_map.items() + if state_field not in entry + } + errors.extend( + commit_trailer_errors( + root, + commit, + expected_trailers, + f"turn {round_id}", + legacy_optional, + ) + ) + else: + # Even malformed committed data must not create an unchecked + # trailer namespace. With no trustworthy values, every Madp-* + # key is unexpected and the structural errors above still name + # the missing state/evidence source. + errors.extend( + commit_trailer_errors(root, commit, {}, f"turn {round_id}") + ) if previous is not None and ( commit == previous or not gitops.is_ancestor(root, previous, commit) ): errors.append( f"turn {round_id}: commit order does not match the schedule" ) + proven_entry = entry if isinstance(entry, dict) else record turn_commits.append( { "round_id": round_id, - "actor_id": record.get("actor_id"), + "actor_id": proven_entry.get("actor_id"), + "scheduled_actor_id": proven_entry.get("scheduled_actor_id"), + "actor_selection": proven_entry.get("actor_selection"), + "substitution_reason": proven_entry.get("substitution_reason"), "commit": commit, # The Git-proven provenance value (from the committed # state at the original turn commit), never the current @@ -1005,6 +1392,41 @@ def committed_state_at(commit: str) -> dict | None: "the owner-decision commit does not record " "OWNER_DECIDED state" ) + committed_decision = (committed_state or {}).get("owner_decision") + if not isinstance(committed_decision, dict): + errors.append( + "the owner-decision commit has no owner_decision record" + ) + errors.extend( + commit_trailer_errors( + root, decision_commit, {}, "owner-decision commit" + ) + ) + else: + if committed_decision != decision: + errors.append( + "the owner-decision commit record does not exactly " + "match the current terminal state" + ) + expected_decision_trailers = { + "Madp-Protocol": definition.protocol_id, + "Madp-Event": "owner-decision", + "Madp-Decision": str(committed_decision.get("decision")), + "Madp-Artifact-Sha256": str( + committed_decision.get("artifact_sha256") + ), + "Madp-Caller-Identity": str( + committed_decision.get("caller_identity") + ), + } + errors.extend( + commit_trailer_errors( + root, + decision_commit, + expected_decision_trailers, + "owner-decision commit", + ) + ) provenance["owner_decision_commit"] = decision_commit mutated = gitops.history_mutations( diff --git a/src/multi_agent_dialogue/evidence.py b/src/multi_agent_dialogue/evidence.py index b73a1d5..71e82d6 100644 --- a/src/multi_agent_dialogue/evidence.py +++ b/src/multi_agent_dialogue/evidence.py @@ -78,6 +78,7 @@ def validate_evidence( actor: Actor, turn: TurnSpec, artifact_sha256: str, + substitution_reason: str | None = None, ) -> list[str]: """Return every reason this evidence fails; empty list means valid.""" errors: list[str] = [] @@ -99,6 +100,32 @@ def validate_evidence( f"evidence actor_id {record['actor_id']!r} does not match " f"claimed actor {actor.actor_id!r}" ) + expected_selection = "primary" if actor.actor_id == turn.actor_id else "substitute" + if turn.substitute_actor_ids or expected_selection == "substitute": + for field in ( + "scheduled_actor_id", + "actor_selection", + "substitution_reason", + ): + if field not in record: + errors.append(f"missing explicit actor evidence field: {field}") + if "scheduled_actor_id" in record and record["scheduled_actor_id"] != turn.actor_id: + errors.append( + f"evidence scheduled_actor_id {record['scheduled_actor_id']!r} does not " + f"match primary actor {turn.actor_id!r}" + ) + if "actor_selection" in record and record["actor_selection"] != expected_selection: + errors.append( + f"evidence actor_selection {record['actor_selection']!r} does not match " + f"actual selection {expected_selection!r}" + ) + if "substitution_reason" in record: + expected_reason = substitution_reason if expected_selection == "substitute" else None + if record["substitution_reason"] != expected_reason: + errors.append( + f"evidence substitution_reason {record['substitution_reason']!r} does " + f"not match claim reason {expected_reason!r}" + ) if record["round_id"] != turn.round_id: errors.append( f"evidence round_id {record['round_id']!r} does not match " diff --git a/src/multi_agent_dialogue/gitops.py b/src/multi_agent_dialogue/gitops.py index fdc10c4..e3f76bc 100644 --- a/src/multi_agent_dialogue/gitops.py +++ b/src/multi_agent_dialogue/gitops.py @@ -19,6 +19,7 @@ from __future__ import annotations import hashlib +import re import subprocess from pathlib import Path @@ -80,6 +81,11 @@ def commit_paths( The commit message carries machine-readable ``Madp-*`` trailers. Values must already be non-secret (round/actor/transport/provider/ model/session/digests are protocol facts by design).""" + for key, value in trailers.items(): + if not re.fullmatch(r"[A-Za-z0-9-]+", key): + raise GitError(f"unsafe Git trailer key: {key!r}") + if not isinstance(value, str) or "\n" in value or "\r" in value: + raise GitError(f"unsafe Git trailer value for {key}") rels = [rel_to_root(root, path) for path in paths] message = subject + "\n\n" + "".join( f"{key}: {value}\n" for key, value in trailers.items() @@ -126,6 +132,33 @@ def committed_sha256(root: Path, commit: str, rel: str) -> str | None: return hashlib.sha256(data).hexdigest() if data is not None else None +def commit_trailers(root: Path, commit: str) -> list[tuple[str, str]]: + """Return parsed trailers from one commit without collapsing duplicates.""" + message = _run(["show", "-s", "--format=%B", commit], cwd=root).stdout + try: + parsed = subprocess.run( + ["git", "interpret-trailers", "--parse"], + cwd=str(root), + input=message, + capture_output=True, + text=True, + check=False, + ) + except OSError as exc: + raise GitError(f"cannot run git interpret-trailers: {exc}") from exc + if parsed.returncode != 0: + raise GitError( + f"git interpret-trailers failed (exit {parsed.returncode})" + ) + result: list[tuple[str, str]] = [] + for line in parsed.stdout.splitlines(): + if ": " not in line: + continue + key, value = line.split(": ", 1) + result.append((key, value)) + return result + + def is_ancestor(root: Path, ancestor: str, descendant: str) -> bool: result = _run( ["merge-base", "--is-ancestor", ancestor, descendant], diff --git a/src/multi_agent_dialogue/runner.py b/src/multi_agent_dialogue/runner.py index 2a6f07b..5a8393f 100644 --- a/src/multi_agent_dialogue/runner.py +++ b/src/multi_agent_dialogue/runner.py @@ -28,8 +28,16 @@ def _context_for( - dialogue: engine.Dialogue, actor_id: str -) -> tuple[config.ProtocolDefinition, config.TurnSpec, adapters.PrepareContext]: + dialogue: engine.Dialogue, + actor_id: str, + substitution_reason: str | None = None, +) -> tuple[ + config.ProtocolDefinition, + config.TurnSpec, + adapters.PrepareContext, + str, + str | None, +]: definition = dialogue.definition() try: actor = definition.actor(actor_id) @@ -41,11 +49,18 @@ def _context_for( f"final turn {definition.final_round_id} is already complete; " "no further worker turns exist" ) - if turn.actor_id != actor_id: - raise engine.ProtocolError( - f"{actor_id!r} is not the scheduled actor for {turn.round_id}; " - f"next actor is {turn.actor_id!r}" - ) + if substitution_reason is None: + claim = dialogue.state().get("claim") or {} + if claim.get("actor_id") == actor_id and claim.get("round_id") == turn.round_id: + substitution_reason = claim.get("substitution_reason") + actor_selection, substitution_reason = engine.resolve_actor_selection( + turn, actor_id, substitution_reason + ) + isolation_errors = engine.hermes_profile_isolation_errors( + definition, dialogue.directory.absolute(), turn + ) + if isolation_errors: + raise engine.ProtocolError("\n".join(isolation_errors)) work_dir = dialogue.directory.absolute() / WORK_DIR / turn.round_id context = adapters.PrepareContext( actor=actor, @@ -55,8 +70,9 @@ def _context_for( task_file=work_dir / "task.md", turn_file=work_dir / "turn.md", evidence_file=work_dir / "evidence.json", + substitution_reason=substitution_reason, ) - return definition, turn, context + return definition, turn, context, actor_selection, substitution_reason def build_task_briefing( @@ -65,9 +81,17 @@ def build_task_briefing( dialogue: engine.Dialogue, context: adapters.PrepareContext, output_contract: tuple[str, ...], + substitution_reason: str | None = None, ) -> str: - actor = definition.actor(turn.actor_id) + # The selected runtime actor may be a frozen-definition substitute. + # It keeps its own actor/provider/model identity; ``turn.actor_id`` is + # always the primary scheduled actor and is never impersonated. + actor = context.actor + actor_selection = "primary" if actor.actor_id == turn.actor_id else "substitute" state = dialogue.state() + if substitution_reason is None: + claim = state.get("claim") or {} + substitution_reason = claim.get("substitution_reason") word_limit = ( f"{turn.word_limit} words" if turn.word_limit is not None else "none" ) @@ -84,10 +108,23 @@ def build_task_briefing( "", f"- round_id: {turn.round_id}", f"- actor_id: {actor.actor_id}", + f"- scheduled_actor_id: {turn.actor_id}", + f"- actor_selection: {actor_selection}", + f"- substitution_reason: {substitution_reason or 'none'}", + f"- allowed_actor_ids: {', '.join(turn.allowed_actor_ids)}", f"- protocol_role: {actor.role}", f"- artifact_kind: {turn.artifact_kind}", f"- word_limit: {word_limit}", ] + if turn.round_id == definition.final_round_id and definition.agent_final_statuses: + lines.append( + "- final agent status: include exactly one `Status: ` line; " + "allowed values: " + ", ".join(definition.agent_final_statuses) + ) + lines.append( + "- agent status is not an owner decision; do not emit any " + f"owner-only token: {', '.join(definition.owner_decisions)}" + ) completed = state.get("completed_turns", []) if completed: # Usable prior-turn context: absolute paths rooted in the dialogue @@ -111,6 +148,14 @@ def build_task_briefing( f"evidence {evidence_path} " f"(sha256 {record['evidence_sha256']})" ) + elif definition.continuation is not None: + anchor = definition.continuation + lines.append( + "- REQUIRED READING: this is a bounded continuation; read the " + f"anchored prior turn {anchor.round_id} at {anchor.artifact_path} " + f"(sha256 {anchor.artifact_sha256}) in full before producing this " + f"{turn.artifact_kind}." + ) else: lines.append( "- prior turns: none — this is the first turn; " @@ -125,6 +170,8 @@ def build_task_briefing( f"- Allowed evidence roots: {', '.join(definition.evidence_roots) or 'none'}.", "- This is exactly one turn; do not continue past it and do not " "claim any other actor's turn.", + "- If actor_selection is substitute, keep your actual actor identity; " + "never write as, claim to be, or impersonate the primary scheduled actor.", "- Runtime identity (provider, model, session) is observed " "externally; a written label never proves it.", "", @@ -136,14 +183,23 @@ def build_task_briefing( return "\n".join(lines) -def dry_run(dialogue: engine.Dialogue, actor_id: str) -> dict: - definition, turn, context = _context_for(dialogue, actor_id) +def dry_run( + dialogue: engine.Dialogue, + actor_id: str, + substitution_reason: str | None = None, +) -> dict: + definition, turn, context, actor_selection, substitution_reason = _context_for( + dialogue, actor_id, substitution_reason + ) packet = adapters.adapter_for(context.actor).prepare(context) return { "dry_run": True, "executed": False, "round_id": turn.round_id, "actor_id": actor_id, + "scheduled_actor_id": turn.actor_id, + "actor_selection": actor_selection, + "substitution_reason": substitution_reason, "transport": context.actor.transport, "work_dir": str(context.work_dir), "packet": packet.as_dict(), @@ -151,14 +207,26 @@ def dry_run(dialogue: engine.Dialogue, actor_id: str) -> dict: } -def prepare(dialogue: engine.Dialogue, actor_id: str, output: Path | str) -> dict: - definition, turn, context = _context_for(dialogue, actor_id) +def prepare( + dialogue: engine.Dialogue, + actor_id: str, + output: Path | str, + substitution_reason: str | None = None, +) -> dict: + definition, turn, context, actor_selection, substitution_reason = _context_for( + dialogue, actor_id, substitution_reason + ) adapter = adapters.adapter_for(context.actor) packet = adapter.prepare(context) output = Path(output) output.parent.mkdir(parents=True, exist_ok=True) briefing = build_task_briefing( - definition, turn, dialogue, context, adapter.output_contract(context) + definition, + turn, + dialogue, + context, + adapter.output_contract(context), + substitution_reason, ) try: artifacts.write_bytes_exclusive( @@ -170,24 +238,39 @@ def prepare(dialogue: engine.Dialogue, actor_id: str, output: Path | str) -> dic "prepared": True, "round_id": turn.round_id, "actor_id": actor_id, + "scheduled_actor_id": turn.actor_id, + "actor_selection": actor_selection, + "substitution_reason": substitution_reason, "task_file": str(output), "packet": packet.as_dict(), } -def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None) -> dict: - definition, turn, context = _context_for(dialogue, actor_id) +def launch( + dialogue: engine.Dialogue, + actor_id: str, + timeout: int | None = None, + substitution_reason: str | None = None, +) -> dict: + definition, turn, context, actor_selection, substitution_reason = _context_for( + dialogue, actor_id, substitution_reason + ) adapter = adapters.adapter_for(context.actor) try: packet = adapter.prepare(context) except adapters.AdapterError as exc: raise engine.ProtocolError(str(exc)) from exc - dialogue.claim(actor_id) + dialogue.claim(actor_id, substitution_reason=substitution_reason) try: context.work_dir.mkdir(parents=True, exist_ok=True) briefing = build_task_briefing( - definition, turn, dialogue, context, adapter.output_contract(context) + definition, + turn, + dialogue, + context, + adapter.output_contract(context), + substitution_reason, ) try: artifacts.write_bytes_exclusive( @@ -236,6 +319,9 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None) "executed": True, "completed_round": turn.round_id, "actor_id": actor_id, + "scheduled_actor_id": turn.actor_id, + "actor_selection": actor_selection, + "substitution_reason": substitution_reason, "status": state["status"], "turn_index": state["turn_index"], "packet": packet.as_dict(), diff --git a/src/multi_agent_dialogue/unverified.py b/src/multi_agent_dialogue/unverified.py index 54d6cdf..482e936 100644 --- a/src/multi_agent_dialogue/unverified.py +++ b/src/multi_agent_dialogue/unverified.py @@ -32,7 +32,11 @@ def cmd_claim(args: argparse.Namespace) -> int: dialogue = engine.Dialogue(args.dialogue) - state = dialogue.claim(args.actor, expected_revision=args.revision) + state = dialogue.claim( + args.actor, + expected_revision=args.revision, + substitution_reason=args.substitution_reason, + ) _emit({"claimed": True, "status": state["status"], "claim": state["claim"], "revision": state["revision"]}) return 0 @@ -40,7 +44,14 @@ def cmd_claim(args: argparse.Namespace) -> int: def cmd_prepare(args: argparse.Namespace) -> int: dialogue = engine.Dialogue(args.dialogue) - _emit(runner.prepare(dialogue, args.actor, args.output)) + _emit( + runner.prepare( + dialogue, + args.actor, + args.output, + substitution_reason=args.substitution_reason, + ) + ) return 0 @@ -91,6 +102,7 @@ def build_parser() -> argparse.ArgumentParser: p = sub.add_parser("claim", help="atomically claim the next turn for an actor") p.add_argument("dialogue", type=Path) p.add_argument("--actor", required=True) + p.add_argument("--substitution-reason", default=None) p.add_argument("--revision", type=int, default=None, help="compare-and-swap: fail if state revision differs") p.set_defaults(func=cmd_claim) @@ -98,6 +110,7 @@ def build_parser() -> argparse.ArgumentParser: p = sub.add_parser("prepare", help="write the non-secret task briefing for a turn") p.add_argument("dialogue", type=Path) p.add_argument("--actor", required=True) + p.add_argument("--substitution-reason", default=None) p.add_argument("--output", required=True, type=Path) p.set_defaults(func=cmd_prepare) diff --git a/tests/test_adapters.py b/tests/test_adapters.py index 884c62e..445a142 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -65,6 +65,113 @@ def test_role_text_never_selects_adapter(self) -> None: self.assertEqual(adapters.adapter_for(actor).name, "command") +class HermesSubstituteIsolationTests(unittest.TestCase): + def definition_with_homes(self, primary_home: Path, substitute_home: Path): + raw = support.two_actor_definition() + raw["actors"][1]["role"] = raw["actors"][0]["role"] + raw["actors"][0].update( + transport="hermes-cli", + settings={"command_name": "hermes", "hermes_home": str(primary_home)}, + ) + raw["actors"][1].update( + transport="hermes-cli", + settings={"command_name": "hermes", "hermes_home": str(substitute_home)}, + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + return config.parse_definition(raw) + + def test_init_rejects_symlinked_profile_alias(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + real_home = base / "profile-real" + real_home.mkdir() + alias_home = base / "profile-alias" + alias_home.symlink_to(real_home, target_is_directory=True) + definition = self.definition_with_homes(real_home, alias_home) + repo = support.init_git_repo(base / "repo") + with self.assertRaises(engine.ProtocolError) as ctx: + engine.init_dialogue(definition, repo / "dialogue") + self.assertIn("same HERMES_HOME", str(ctx.exception)) + + def test_engine_rechecks_alias_drift_at_validation_and_claim(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + primary_home = base / "profile-primary" + substitute_home = base / "profile-substitute" + primary_home.mkdir() + substitute_home.mkdir() + definition = self.definition_with_homes(primary_home, substitute_home) + repo = support.init_git_repo(base / "repo") + dialogue = engine.init_dialogue(definition, repo / "dialogue") + substitute_home.rmdir() + substitute_home.symlink_to(primary_home, target_is_directory=True) + report = dialogue.validate() + self.assertFalse(report["ok"]) + self.assertTrue( + any("same HERMES_HOME" in item for item in report["errors"]), + report["errors"], + ) + with self.assertRaises(engine.ProtocolError): + dialogue.claim( + "worker-b", substitution_reason="provider_cooldown" + ) + + def test_engine_rechecks_alias_drift_before_direct_completion(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + primary_home = base / "profile-primary" + substitute_home = base / "profile-substitute" + primary_home.mkdir() + substitute_home.mkdir() + definition = self.definition_with_homes(primary_home, substitute_home) + repo = support.init_git_repo(base / "repo") + dialogue = engine.init_dialogue(definition, repo / "dialogue") + dialogue.claim("worker-b", substitution_reason="provider_cooldown") + substitute_home.rmdir() + substitute_home.symlink_to(primary_home, target_is_directory=True) + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.complete( + "worker-b", base / "turn.md", base / "evidence.json" + ) + self.assertIn("same HERMES_HOME", str(ctx.exception)) + + def test_engine_rechecks_alias_drift_before_primary_claim(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + primary_home = base / "profile-primary" + substitute_home = base / "profile-substitute" + primary_home.mkdir() + substitute_home.mkdir() + definition = self.definition_with_homes(primary_home, substitute_home) + repo = support.init_git_repo(base / "repo") + dialogue = engine.init_dialogue(definition, repo / "dialogue") + substitute_home.rmdir() + substitute_home.symlink_to(primary_home, target_is_directory=True) + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.claim("worker-a") + self.assertIn("same HERMES_HOME", str(ctx.exception)) + + def test_engine_rechecks_alias_drift_before_primary_completion(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + primary_home = base / "profile-primary" + substitute_home = base / "profile-substitute" + primary_home.mkdir() + substitute_home.mkdir() + definition = self.definition_with_homes(primary_home, substitute_home) + repo = support.init_git_repo(base / "repo") + dialogue = engine.init_dialogue(definition, repo / "dialogue") + dialogue.claim("worker-a") + substitute_home.rmdir() + substitute_home.symlink_to(primary_home, target_is_directory=True) + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.complete( + "worker-a", base / "turn.md", base / "evidence.json" + ) + self.assertIn("same HERMES_HOME", str(ctx.exception)) + + class CommandAdapterTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() @@ -237,6 +344,88 @@ def test_dry_run_rejects_wrong_actor(self) -> None: class LaunchTests(RunnerTestCase): + def substitute_dialogue(self) -> engine.Dialogue: + raw = support.two_actor_definition() + for actor in raw["actors"]: + actor["settings"] = support.command_worker_settings( + self.marker, actor["expected_provider"], actor["expected_model"] + ) + raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "fake-provider-c", + "expected_model": "fake-model-c", + "settings": support.command_worker_settings( + self.marker, "fake-provider-c", "fake-model-c" + ), + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + definition = config.parse_definition(raw) + return engine.init_dialogue(definition, self.base / "dialogue-substitute") + + def test_launch_preapproved_substitute_preserves_actual_identity(self) -> None: + dialogue = self.substitute_dialogue() + dry = runner.dry_run( + dialogue, "worker-c", substitution_reason="provider_cooldown" + ) + self.assertEqual(dry["actor_id"], "worker-c") + self.assertEqual(dry["scheduled_actor_id"], "worker-a") + self.assertEqual(dry["actor_selection"], "substitute") + self.assertEqual(dry["substitution_reason"], "provider_cooldown") + prepared_path = self.base / "substitute-task.md" + prepared = runner.prepare( + dialogue, + "worker-c", + prepared_path, + substitution_reason="provider_cooldown", + ) + self.assertEqual(prepared["substitution_reason"], "provider_cooldown") + self.assertIn( + "substitution_reason: provider_cooldown", + prepared_path.read_text(encoding="utf-8"), + ) + + result = runner.launch( + dialogue, "worker-c", substitution_reason="provider_cooldown" + ) + self.assertEqual(result["actor_id"], "worker-c") + self.assertEqual(result["scheduled_actor_id"], "worker-a") + self.assertEqual(result["actor_selection"], "substitute") + record = dialogue.state()["completed_turns"][0] + self.assertEqual(record["actor_id"], "worker-c") + self.assertEqual(record["scheduled_actor_id"], "worker-a") + self.assertEqual(record["actor_selection"], "substitute") + self.assertEqual(record["substitution_reason"], "provider_cooldown") + self.assertEqual(record["artifact_file"], "turns/R01-worker-c.md") + runtime_record = json.loads( + (dialogue.directory / record["evidence_file"]).read_text(encoding="utf-8") + ) + self.assertEqual(runtime_record["actor_id"], "worker-c") + self.assertEqual(runtime_record["scheduled_actor_id"], "worker-a") + self.assertEqual(runtime_record["actor_selection"], "substitute") + self.assertEqual( + runtime_record["substitution_reason"], "provider_cooldown" + ) + task = dialogue.directory / "work" / "R01" / "task.md" + briefing = task.read_text(encoding="utf-8") + self.assertIn("actor_id: worker-c", briefing) + self.assertIn("scheduled_actor_id: worker-a", briefing) + self.assertIn("actor_selection: substitute", briefing) + self.assertIn("substitution_reason: provider_cooldown", briefing) + self.assertIn("never write as, claim to be, or impersonate", briefing) + report = dialogue.validate( + require_git=True, require_runner_completion=True + ) + self.assertTrue(report["ok"], report["errors"]) + proven = report["provenance"]["turn_commits"][0] + self.assertEqual(proven["actor_id"], "worker-c") + self.assertEqual(proven["scheduled_actor_id"], "worker-a") + self.assertEqual(proven["actor_selection"], "substitute") + def test_launch_executes_exactly_one_turn(self) -> None: result = runner.launch(self.dialogue, "worker-a") self.assertTrue(result["executed"]) diff --git a/tests/test_cli.py b/tests/test_cli.py index ce1797b..be54e4f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -124,6 +124,23 @@ def test_wrong_actor_claim_fails(self) -> None: class RunTests(CliTestCase): + def enable_r01_substitute(self) -> None: + self.raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "fake-provider-c", + "expected_model": "fake-model-c", + "settings": support.command_worker_settings( + self.marker, "fake-provider-c", "fake-model-c" + ), + } + ) + self.raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + self.raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.definition_path.write_text(json.dumps(self.raw), encoding="utf-8") + def test_run_defaults_to_dry_run_no_process(self) -> None: self.init() before = (self.dialogue_dir / "state.json").read_text(encoding="utf-8") @@ -154,6 +171,66 @@ def test_run_launch_wrong_actor_fails(self) -> None: self.assertNotEqual(result.returncode, 0) self.assertEqual(self.spawn_count(), 0) + def test_substitute_cli_requires_and_records_frozen_reason(self) -> None: + self.enable_r01_substitute() + self.init() + missing = run_cli( + "run", str(self.dialogue_dir), "--actor", "worker-c", "--launch" + ) + self.assertNotEqual(missing.returncode, 0) + self.assertIn("substitution reason", missing.stderr) + self.assertEqual(self.spawn_count(), 0) + + result = run_cli( + "run", + str(self.dialogue_dir), + "--actor", + "worker-c", + "--substitution-reason", + "provider_cooldown", + "--launch", + ) + self.assertEqual(result.returncode, 0, result.stderr) + payload = json.loads(result.stdout) + self.assertEqual(payload["actor_id"], "worker-c") + self.assertEqual(payload["scheduled_actor_id"], "worker-a") + self.assertEqual(payload["substitution_reason"], "provider_cooldown") + status = json.loads(run_cli("status", str(self.dialogue_dir)).stdout) + record = status["completed_turns"][0] + self.assertEqual(record["actor_selection"], "substitute") + self.assertEqual(record["substitution_reason"], "provider_cooldown") + + def test_primary_reason_and_post_final_substitute_fail_without_spawn(self) -> None: + self.enable_r01_substitute() + self.raw["schedule"] = self.raw["schedule"][:1] + self.raw["final_round_id"] = "R01" + self.definition_path.write_text(json.dumps(self.raw), encoding="utf-8") + self.init() + + primary_reason = run_cli( + "run", str(self.dialogue_dir), "--actor", "worker-a", + "--substitution-reason", "provider_cooldown", "--launch", + ) + self.assertNotEqual(primary_reason.returncode, 0) + self.assertEqual(self.spawn_count(), 0) + + accepted = run_cli( + "run", str(self.dialogue_dir), "--actor", "worker-c", + "--substitution-reason", "provider_cooldown", "--launch", + ) + self.assertEqual(accepted.returncode, 0, accepted.stderr) + self.assertEqual(self.spawn_count(), 1) + before = (self.dialogue_dir / "state.json").read_text(encoding="utf-8") + post_final = run_cli( + "run", str(self.dialogue_dir), "--actor", "worker-c", + "--substitution-reason", "provider_cooldown", "--launch", + ) + self.assertNotEqual(post_final.returncode, 0) + self.assertEqual(self.spawn_count(), 1) + self.assertEqual( + (self.dialogue_dir / "state.json").read_text(encoding="utf-8"), before + ) + def test_run_rejects_conflicting_flags(self) -> None: self.init() result = run_cli( diff --git a/tests/test_config.py b/tests/test_config.py index f83d437..d72a57f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -7,6 +7,7 @@ import unittest from pathlib import Path +import jsonschema import support from multi_agent_dialogue import config @@ -32,6 +33,30 @@ def test_parses_three_actor_definition(self) -> None: ["lead", "critic", "scribe"], ) + def test_parses_owner_preapproved_substitute_actors(self) -> None: + raw = support.two_actor_definition() + raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "prov-c", + "expected_model": "model-c", + "settings": { + "argv": ["fake-worker"], + "identity_verifier_argv": ["fake-verifier"], + }, + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + definition = config.parse_definition(raw) + turn = definition.schedule[0] + self.assertEqual(turn.actor_id, "worker-a") + self.assertEqual(turn.substitute_actor_ids, ("worker-c",)) + self.assertEqual(turn.allowed_actor_ids, ("worker-a", "worker-c")) + self.assertEqual(turn.substitution_reasons, ("provider_cooldown",)) + def test_actor_lookup_by_id(self) -> None: definition = config.parse_definition(support.two_actor_definition()) actor = definition.actor("worker-b") @@ -63,6 +88,22 @@ def test_digest_changes_when_content_changes(self) -> None: definition_b = config.parse_definition(raw) self.assertNotEqual(definition_a.digest(), definition_b.digest()) + def test_parses_structured_continuation_anchor(self) -> None: + raw = support.two_actor_definition() + raw["continuation"] = { + "protocol_id": "prior-dialogue", + "round_id": "R00", + "artifact_path": "/tmp/prior/R00.md", + "artifact_sha256": "a" * 64, + "published_commit": "b" * 40, + "original_dialogue_head": "c" * 40, + "start_round": "R01", + } + definition = config.parse_definition(raw) + self.assertIsNotNone(definition.continuation) + assert definition.continuation is not None + self.assertEqual(definition.continuation.start_round, "R01") + class InvalidDefinitionTests(unittest.TestCase): def assert_rejected(self, raw: dict, fragment: str) -> None: @@ -87,6 +128,59 @@ def test_rejects_unknown_scheduled_actor(self) -> None: raw["schedule"][1]["actor_id"] = "ghost" self.assert_rejected(raw, "unknown actor") + def test_rejects_unknown_substitute_actor(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitute_actor_ids"] = ["ghost"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.assert_rejected(raw, "unknown substitute actor") + + def test_rejects_primary_actor_repeated_as_substitute(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitute_actor_ids"] = ["worker-a"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.assert_rejected(raw, "primary actor") + + def test_rejects_duplicate_substitute_actor(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b", "worker-b"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.assert_rejected(raw, "duplicate substitute actor") + + def test_rejects_substitute_with_different_protocol_role(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.assert_rejected(raw, "role must match primary actor") + + def test_rejects_substitute_using_same_hermes_home(self) -> None: + raw = support.two_actor_definition() + raw["actors"][1]["role"] = raw["actors"][0]["role"] + for actor in raw["actors"]: + actor["transport"] = "hermes-cli" + actor["settings"] = { + "command_name": "hermes", + "hermes_home": "/profiles/shared", + } + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.assert_rejected(raw, "distinct hermes_home") + + def test_rejects_substitutes_without_frozen_reason_codes(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + self.assert_rejected(raw, "substitution_reasons is required") + + def test_rejects_reason_codes_without_substitutes(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.assert_rejected(raw, "requires substitute_actor_ids") + + def test_rejects_unsafe_substitution_reason_code(self) -> None: + raw = support.two_actor_definition() + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + raw["schedule"][0]["substitution_reasons"] = ["provider\ncooldown"] + self.assert_rejected(raw, "safe reason code") + def test_rejects_duplicate_round_ids(self) -> None: raw = support.two_actor_definition() raw["schedule"][1]["round_id"] = "R01" @@ -112,6 +206,19 @@ def test_rejects_unbounded_schedule_marker(self) -> None: raw["schedule"][0]["repeat"] = "forever" self.assert_rejected(raw, "unbounded") + def test_rejects_continuation_with_wrong_start_round(self) -> None: + raw = support.two_actor_definition() + raw["continuation"] = { + "protocol_id": "prior-dialogue", + "round_id": "R00", + "artifact_path": "/tmp/prior/R00.md", + "artifact_sha256": "a" * 64, + "published_commit": "b" * 40, + "original_dialogue_head": "c" * 40, + "start_round": "R02", + } + self.assert_rejected(raw, "start_round") + def test_rejects_owner_who_is_also_actor(self) -> None: raw = support.two_actor_definition() raw["owner"] = "worker-a" @@ -151,6 +258,11 @@ def test_rejects_owner_decisions_empty(self) -> None: raw["owner_decisions"] = [] self.assert_rejected(raw, "owner_decisions") + def test_rejects_agent_status_that_claims_owner_decision(self) -> None: + raw = support.two_actor_definition() + raw["agent_final_statuses"] = ["READY_FOR_OWNER", "APPROVE"] + self.assert_rejected(raw, "must not overlap owner_decisions") + def test_load_rejects_invalid_json_file(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "broken.json" @@ -189,6 +301,28 @@ def test_schema_examples_pass_python_validator(self) -> None: definition = config.parse_definition(example) self.assertGreaterEqual(len(definition.actors), 2) + def test_schema_enforces_substitute_reason_pairing(self) -> None: + schema = json.loads( + (support.REPO_ROOT / "schemas" / "protocol.schema.json").read_text( + encoding="utf-8" + ) + ) + base = support.two_actor_definition() + base["actors"][1]["role"] = base["actors"][0]["role"] + substitutes_only = json.loads(json.dumps(base)) + substitutes_only["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + reasons_only = json.loads(json.dumps(base)) + reasons_only["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + validator = jsonschema.Draft202012Validator(schema) + for raw in (substitutes_only, reasons_only): + with self.assertRaises(jsonschema.ValidationError): + validator.validate(raw) + + valid = json.loads(json.dumps(base)) + valid["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + valid["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + validator.validate(valid) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_engine.py b/tests/test_engine.py index cf929da..cd9689e 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -26,6 +26,39 @@ def init_dialogue(self) -> engine.Dialogue: class InitTests(EngineTestCase): + def continuation_definition(self) -> tuple[config.ProtocolDefinition, Path]: + artifact = self.root / "prior" / "R00-worker.md" + artifact.parent.mkdir() + artifact.write_text("# R00\n\nPublished prior turn.\n", encoding="utf-8") + support.git(self.root, "add", "prior/R00-worker.md") + support.git(self.root, "commit", "-q", "-m", "publish prior turn") + published = support.git(self.root, "rev-parse", "HEAD").stdout.strip() + marker = self.root / "prior" / "marker.txt" + marker.write_text("original dialogue head\n", encoding="utf-8") + support.git(self.root, "add", "prior/marker.txt") + support.git(self.root, "commit", "-q", "-m", "close prior dialogue") + original_head = support.git(self.root, "rev-parse", "HEAD").stdout.strip() + raw = support.two_actor_definition() + raw["continuation"] = { + "protocol_id": "prior-dialogue", + "round_id": "R00", + "artifact_path": str(artifact), + "artifact_sha256": engine.artifacts.sha256_file(artifact), + "published_commit": published, + "original_dialogue_head": original_head, + "start_round": "R01", + } + return config.parse_definition(raw), artifact + + def test_continuation_anchor_is_rechecked_after_init(self) -> None: + self.definition, artifact = self.continuation_definition() + dialogue = self.init_dialogue() + self.assertEqual(dialogue.state()["status"], "OPEN") + artifact.write_text("tampered prior turn\n", encoding="utf-8") + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.state() + self.assertIn("continuation artifact hash mismatch", str(ctx.exception)) + def test_init_creates_state_and_definition(self) -> None: dialogue = self.init_dialogue() self.assertTrue((self.dialogue_dir / "definition.json").is_file()) @@ -90,6 +123,25 @@ def _force_state(self, **overrides) -> None: class ClaimTests(EngineTestCase): + def enable_substitute_for_r01(self) -> None: + raw = support.two_actor_definition() + raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "prov-c", + "expected_model": "model-c", + "settings": { + "argv": ["fake-worker"], + "identity_verifier_argv": ["fake-verifier"], + }, + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.definition = config.parse_definition(raw) + def test_correct_claim_locks_turn(self) -> None: dialogue = self.init_dialogue() state = dialogue.claim("worker-a") @@ -103,10 +155,77 @@ def test_wrong_actor_cannot_claim(self) -> None: dialogue = self.init_dialogue() with self.assertRaises(engine.ProtocolError) as ctx: dialogue.claim("worker-b") - self.assertIn("not the scheduled actor", str(ctx.exception)) + self.assertIn("not an allowed actor", str(ctx.exception)) self.assertEqual(dialogue.state()["status"], "OPEN") self.assertFalse((self.dialogue_dir / engine.LOCK_FILE).exists()) + def test_preapproved_substitute_can_claim_without_impersonating_primary(self) -> None: + self.enable_substitute_for_r01() + dialogue = self.init_dialogue() + state = dialogue.claim("worker-c", substitution_reason="provider_cooldown") + self.assertEqual(state["claim"]["actor_id"], "worker-c") + self.assertEqual(state["claim"]["scheduled_actor_id"], "worker-a") + self.assertEqual(state["claim"]["actor_selection"], "substitute") + self.assertEqual(state["claim"]["substitution_reason"], "provider_cooldown") + + def test_substitute_requires_frozen_reason_code(self) -> None: + self.enable_substitute_for_r01() + dialogue = self.init_dialogue() + with self.assertRaises(engine.ProtocolError) as missing: + dialogue.claim("worker-c") + self.assertIn("substitution reason", str(missing.exception)) + with self.assertRaises(engine.ProtocolError) as unknown: + dialogue.claim("worker-c", substitution_reason="operator_preference") + self.assertIn("not allowed", str(unknown.exception)) + + def test_primary_actor_rejects_substitution_reason(self) -> None: + dialogue = self.init_dialogue() + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.claim("worker-a", substitution_reason="provider_cooldown") + self.assertIn("primary actor", str(ctx.exception)) + self.assertIsNone(dialogue.state()["claim"]) + + def test_active_substitute_claim_requires_explicit_identity_fields(self) -> None: + self.enable_substitute_for_r01() + dialogue = self.init_dialogue() + dialogue.claim("worker-c", substitution_reason="provider_cooldown") + state = json.loads(dialogue.state_path.read_text(encoding="utf-8")) + for key in ( + "scheduled_actor_id", + "actor_selection", + "substitution_reason", + ): + del state["claim"][key] + dialogue.state_path.write_text(json.dumps(state), encoding="utf-8") + report = dialogue.validate() + self.assertFalse(report["ok"]) + self.assertTrue( + any("explicit actor identity fields" in item for item in report["errors"]), + report["errors"], + ) + + def test_primary_claim_on_substitute_capable_turn_is_explicit(self) -> None: + self.enable_substitute_for_r01() + dialogue = self.init_dialogue() + dialogue.claim("worker-a") + state = json.loads(dialogue.state_path.read_text(encoding="utf-8")) + self.assertEqual(state["claim"]["scheduled_actor_id"], "worker-a") + self.assertEqual(state["claim"]["actor_selection"], "primary") + self.assertIsNone(state["claim"]["substitution_reason"]) + for key in ( + "scheduled_actor_id", + "actor_selection", + "substitution_reason", + ): + del state["claim"][key] + dialogue.state_path.write_text(json.dumps(state), encoding="utf-8") + report = dialogue.validate() + self.assertFalse(report["ok"]) + self.assertTrue( + any("explicit actor identity fields" in item for item in report["errors"]), + report["errors"], + ) + def test_unknown_actor_cannot_claim(self) -> None: dialogue = self.init_dialogue() with self.assertRaises(engine.ProtocolError): @@ -150,6 +269,58 @@ def test_release_requires_claiming_actor(self) -> None: class FinalStopTests(EngineTestCase): + def test_final_agent_status_is_distinct_from_owner_decision(self) -> None: + raw = support.two_actor_definition() + raw["schedule"] = raw["schedule"][:1] + raw["final_round_id"] = "R01" + raw["agent_final_statuses"] = [ + "READY_FOR_OWNER", + "AGENT_NEEDS_MORE_EVIDENCE", + "SPLIT_QUESTION", + ] + self.definition = config.parse_definition(raw) + dialogue = self.init_dialogue() + dialogue.claim("worker-a") + scratch = self.root / "scratch" + scratch.mkdir() + turn_path = scratch / "R01.md" + evidence_path = scratch / "R01.json" + + def write_body(body: str) -> None: + turn_path.write_text(body, encoding="utf-8") + record = support.make_evidence( + actor_id="worker-a", + round_id="R01", + artifact_path=turn_path, + provider="fake-provider-a", + model="fake-model-a", + ) + evidence_path.write_text(json.dumps(record), encoding="utf-8") + + def write_attempt(status: str) -> None: + write_body(f"# R01\n\nFinal analysis.\n\nStatus: {status}\n") + + for malformed in ( + "# R01\n\nFinal analysis.\n\nStatus:\nREADY_FOR_OWNER\n", + "# R01\n\nFinal analysis.\n\nStatus:\n\nREADY_FOR_OWNER\n", + ): + write_body(malformed) + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.complete("worker-a", turn_path, evidence_path) + self.assertIn("exactly one line", str(ctx.exception)) + self.assertEqual(dialogue.state()["status"], "CLAIMED") + + write_attempt("APPROVE") + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.complete("worker-a", turn_path, evidence_path) + self.assertIn("not allowed", str(ctx.exception)) + self.assertEqual(dialogue.state()["status"], "CLAIMED") + + write_attempt("READY_FOR_OWNER") + state = dialogue.complete("worker-a", turn_path, evidence_path) + self.assertEqual(state["status"], "READY_FOR_OWNER") + self.assertIsNone(state["owner_decision"]) + def _finish_all_turns(self) -> None: path = self.dialogue_dir / "state.json" state = json.loads(path.read_text(encoding="utf-8")) diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 71c62b0..87c5f54 100644 --- a/tests/test_evidence.py +++ b/tests/test_evidence.py @@ -44,6 +44,98 @@ def check(self, record: dict) -> list[str]: def test_valid_evidence_passes(self) -> None: self.assertEqual(self.check(self.valid_evidence()), []) + def test_substitute_evidence_requires_selection_provenance(self) -> None: + raw = support.two_actor_definition() + raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "fake-provider-c", + "expected_model": "fake-model-c", + "settings": {}, + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + definition = config.parse_definition(raw) + actor = definition.actor("worker-c") + turn = definition.schedule[0] + record = support.make_evidence( + actor_id="worker-c", + round_id="R01", + artifact_path=self.artifact, + provider="fake-provider-c", + model="fake-model-c", + ) + missing = evidence.validate_evidence( + record, + actor=actor, + turn=turn, + artifact_sha256=evidence.sha256_file(self.artifact), + substitution_reason="provider_cooldown", + ) + self.assertTrue(any("scheduled_actor_id" in item for item in missing)) + record.update( + { + "scheduled_actor_id": "worker-a", + "actor_selection": "substitute", + "substitution_reason": "provider_cooldown", + } + ) + self.assertEqual( + evidence.validate_evidence( + record, + actor=actor, + turn=turn, + artifact_sha256=evidence.sha256_file(self.artifact), + substitution_reason="provider_cooldown", + ), + [], + ) + + def test_primary_evidence_is_explicit_when_turn_allows_substitute(self) -> None: + raw = support.two_actor_definition() + raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "fake-provider-c", + "expected_model": "fake-model-c", + "settings": {}, + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + definition = config.parse_definition(raw) + actor = definition.actor("worker-a") + turn = definition.schedule[0] + record = self.valid_evidence() + missing = evidence.validate_evidence( + record, + actor=actor, + turn=turn, + artifact_sha256=evidence.sha256_file(self.artifact), + ) + self.assertTrue(any("scheduled_actor_id" in item for item in missing)) + record.update( + { + "scheduled_actor_id": "worker-a", + "actor_selection": "primary", + "substitution_reason": None, + } + ) + self.assertEqual( + evidence.validate_evidence( + record, + actor=actor, + turn=turn, + artifact_sha256=evidence.sha256_file(self.artifact), + ), + [], + ) + def test_missing_field_rejected(self) -> None: record = self.valid_evidence() del record["session_id"] diff --git a/tests/test_git_transactions.py b/tests/test_git_transactions.py index 5d65752..6a5b07c 100644 --- a/tests/test_git_transactions.py +++ b/tests/test_git_transactions.py @@ -31,7 +31,7 @@ import support -from multi_agent_dialogue import config, engine, runner +from multi_agent_dialogue import config, engine, gitops, runner SHA_HEX = frozenset("0123456789abcdef") @@ -193,6 +193,18 @@ def test_init_ignore_rules_cover_scratch_lock_and_temp(self) -> None: class TurnCommitTests(GitDialogueTestCase): + def test_commit_api_rejects_trailer_value_injection(self) -> None: + path = self.repo / "safe.txt" + path.write_text("safe\n", encoding="utf-8") + with self.assertRaises(gitops.GitError): + gitops.commit_paths( + self.repo, + [path], + "unsafe trailer fixture", + {"Madp-Actor": "worker-a\nMadp-Event: owner-decision"}, + ) + self.assertEqual(commit_count(self.repo), 0) + def test_complete_creates_exactly_one_turn_commit(self) -> None: dialogue = self.init_dialogue() before = commit_count(self.repo) @@ -394,6 +406,125 @@ def test_validate_proves_commit_provenance_and_exposes_shas(self) -> None: self.assertEqual(len(shas), len(set(shas)), "each transition is its own commit") self.assertEqual(provenance["owner_decision_commit"], head(self.repo)) + def test_validate_rejects_amended_owner_decision_trailers(self) -> None: + dialogue = self.init_dialogue() + self.finish_dialogue(dialogue) + dialogue.owner_decide(self.decision_file()) + message = commit_message(self.repo) + message = message.replace("Madp-Decision: APPROVE", "Madp-Decision: REJECT") + message = message.replace( + "Madp-Caller-Identity: unverified", + "Madp-Caller-Identity: externally-verified", + ) + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any( + "owner-decision commit: Git trailer Madp-Decision" in item + for item in report["errors"] + ), + report["errors"], + ) + self.assertTrue( + any( + "owner-decision commit: Git trailer Madp-Caller-Identity" in item + for item in report["errors"] + ), + report["errors"], + ) + + def test_validate_rejects_mixed_case_owner_decision_trailers(self) -> None: + dialogue = self.init_dialogue() + self.finish_dialogue(dialogue) + dialogue.owner_decide(self.decision_file()) + message = commit_message(self.repo).replace( + "Madp-Event: owner-decision", + "Madp-Event: owner-decision\nmadp-event: turn\nMADP-Round: R99", + ) + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any("duplicate Git trailer 'madp-event'" in item for item in report["errors"]), + report["errors"], + ) + self.assertTrue( + any("unexpected Git trailer 'MADP-Round'" in item for item in report["errors"]), + report["errors"], + ) + + def test_validate_rejects_unexpected_init_trailer(self) -> None: + dialogue = self.init_dialogue() + init_message = commit_message(self.repo).replace( + "Madp-Event: init", + "Madp-Event: init\nmadp-event: owner-decision\nMADP-Decision: APPROVE", + ) + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + init_message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any( + "duplicate Git trailer 'madp-event'" in item + for item in report["errors"] + ), + report["errors"], + ) + self.assertTrue( + any( + "init commit: unexpected Git trailer 'MADP-Decision'" in item + for item in report["errors"] + ), + report["errors"], + ) + + def test_validate_rejects_owner_trailer_on_turn(self) -> None: + dialogue = self.init_dialogue() + self.complete_turn(dialogue, "worker-a", "R01") + turn_message = commit_message(self.repo) + "\nMadp-Decision: APPROVE\n" + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + turn_message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any( + "turn R01: unexpected Git trailer 'Madp-Decision'" in item + for item in report["errors"] + ), + report["errors"], + ) + def test_validate_rejects_committed_artifact_tampering(self) -> None: import hashlib @@ -433,6 +564,136 @@ def test_validate_rejects_committed_artifact_tampering(self) -> None: report["errors"], ) + def test_validate_rejects_contradictory_turn_trailers(self) -> None: + dialogue = self.init_dialogue() + self.complete_turn(dialogue, "worker-a", "R01") + message = commit_message(self.repo).replace( + "Madp-Actor: worker-a", "Madp-Actor: worker-b" + ) + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any("Git trailer Madp-Actor" in error for error in report["errors"]), + report["errors"], + ) + + def test_validate_rejects_deleted_identity_trailers_on_new_primary_turn(self) -> None: + dialogue = self.init_dialogue() + self.complete_turn(dialogue, "worker-a", "R01") + removed = { + "Madp-Scheduled-Actor", + "Madp-Actor-Selection", + "Madp-Substitution-Reason", + } + message = "\n".join( + line + for line in commit_message(self.repo).splitlines() + if line.partition(":")[0] not in removed + ) + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any("Git trailer Madp-Scheduled-Actor" in item for item in report["errors"]), + report["errors"], + ) + + def test_validate_rejects_mixed_case_turn_trailers(self) -> None: + dialogue = self.init_dialogue() + self.complete_turn(dialogue, "worker-a", "R01") + message = commit_message(self.repo).replace( + "Madp-Event: turn", + "Madp-Event: turn\nmadp-event: owner-decision\nMADP-Decision: APPROVE", + ) + support.git( + self.repo, + "commit", + "--amend", + "--quiet", + "--no-verify", + "-m", + message, + ) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any("duplicate Git trailer 'madp-event'" in item for item in report["errors"]), + report["errors"], + ) + self.assertTrue( + any("unexpected Git trailer 'MADP-Decision'" in item for item in report["errors"]), + report["errors"], + ) + + def test_validate_rejects_substitute_identity_deleted_from_original_commit( + self, + ) -> None: + raw = two_round_definition() + raw["actors"][1]["role"] = raw["actors"][0]["role"] + raw["schedule"][0]["substitute_actor_ids"] = ["worker-b"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + self.definition = config.parse_definition(raw) + dialogue = self.init_dialogue() + dialogue.claim("worker-b", substitution_reason="provider_cooldown") + turn_path, evidence_path = self.turn_inputs("worker-b", "R01") + evidence_record = json.loads(evidence_path.read_text(encoding="utf-8")) + evidence_record.update( + { + "scheduled_actor_id": "worker-a", + "actor_selection": "substitute", + "substitution_reason": "provider_cooldown", + } + ) + evidence_path.write_text(json.dumps(evidence_record), encoding="utf-8") + dialogue.complete("worker-b", turn_path, evidence_path) + + state_path = self.dialogue_dir / "state.json" + valid_state = json.loads(state_path.read_text(encoding="utf-8")) + tampered = json.loads(json.dumps(valid_state)) + entry = tampered["completed_turns"][0] + for key in ("scheduled_actor_id", "actor_selection", "substitution_reason"): + entry.pop(key) + engine.atomic_write_json(state_path, tampered) + support.git(self.repo, "add", "-f", "--", f"{self.rel}/state.json") + support.git(self.repo, "commit", "--amend", "--quiet", "--no-edit") + amended_turn = head(self.repo) + + valid_state["last_commit"] = amended_turn + if "commit" in valid_state["completed_turns"][0]: + valid_state["completed_turns"][0]["commit"] = amended_turn + engine.atomic_write_json(state_path, valid_state) + support.git(self.repo, "add", "-f", "--", f"{self.rel}/state.json") + support.git(self.repo, "commit", "-q", "-m", "restore current state only") + + self.assertTrue(dialogue.validate()["ok"]) + report = dialogue.validate(require_git=True) + self.assertFalse(report["ok"]) + self.assertTrue( + any( + "turn R01 committed state" in error + and "identity fields" in error + for error in report["errors"] + ), + report["errors"], + ) + def test_validate_rejects_deleted_historical_artifact(self) -> None: dialogue = self.init_dialogue() self.finish_dialogue(dialogue) From 225471e36d7384f1f1ca291f10d4560084b1d1d1 Mon Sep 17 00:00:00 2001 From: Vesper Date: Thu, 20 Aug 2026 08:01:12 +0000 Subject: [PATCH 2/2] fix: review hardening for the revived substitutes branch - 'none' is reserved at definition parse: it is the null sentinel in Madp-Substitution-Reason trailers, so it can never be a reason code; - continuation-anchor Git facts (content-addressed, immutable) are verified once per Dialogue instance instead of spawning several git subprocesses per state() read; the live artifact bytes are still re-hashed on EVERY read, so the tamper guard is unchanged; - the final-status line grammar tolerates CRLF artifacts; - a whitespace-only substitution_reason is rejected explicitly instead of collapsing to None and passing as a primary claim. --- src/multi_agent_dialogue/config.py | 7 ++++ src/multi_agent_dialogue/engine.py | 43 +++++++++++++++++++--- tests/test_config.py | 21 +++++++++++ tests/test_engine.py | 58 ++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 4 deletions(-) diff --git a/src/multi_agent_dialogue/config.py b/src/multi_agent_dialogue/config.py index 8736b54..435bb52 100644 --- a/src/multi_agent_dialogue/config.py +++ b/src/multi_agent_dialogue/config.py @@ -293,6 +293,13 @@ def _parse_turn(raw: Any, position: int, errors: list[str]) -> TurnSpec | None: f"{where}: {reason!r} is not a safe reason code; use 1-64 " "lowercase letters, digits, underscores, or hyphens" ) + elif reason == "none": + # "none" is the null sentinel written to Madp-Substitution- + # Reason trailers; allowing it as a real code would collide. + errors.append( + f"{where}: 'none' is reserved as the null substitution " + "sentinel and cannot be a reason code" + ) if substitutes_raw and not reasons_raw: errors.append( f"{where}: substitution_reasons is required when substitute actors exist" diff --git a/src/multi_agent_dialogue/engine.py b/src/multi_agent_dialogue/engine.py index ff2acf6..c63abac 100644 --- a/src/multi_agent_dialogue/engine.py +++ b/src/multi_agent_dialogue/engine.py @@ -85,8 +85,15 @@ def resolve_actor_selection( f"primary actor is {turn.actor_id!r}, allowed actors are " f"{list(turn.allowed_actor_ids)!r}" ) - reason = substitution_reason.strip() if isinstance(substitution_reason, str) else None - reason = reason or None + if substitution_reason is not None: + if not isinstance(substitution_reason, str) or not substitution_reason.strip(): + raise ProtocolError( + "substitution_reason must be a non-empty reason code when " + f"given, got {substitution_reason!r}" + ) + reason: str | None = substitution_reason.strip() + else: + reason = None if actor_id == turn.actor_id: if reason is not None: raise ProtocolError( @@ -310,6 +317,11 @@ def verify_continuation_anchor(definition: config.ProtocolDefinition) -> None: anchor = definition.continuation if anchor is None: return + _verify_continuation_artifact_bytes(anchor) + _verify_continuation_git_facts(anchor) + + +def _verify_continuation_artifact_bytes(anchor) -> None: path = Path(anchor.artifact_path) try: current = artifacts.read_bytes_nofollow(path, "continuation artifact") @@ -320,6 +332,10 @@ def verify_continuation_anchor(definition: config.ProtocolDefinition) -> None: f"continuation artifact hash mismatch for {anchor.protocol_id}/" f"{anchor.round_id}" ) + + +def _verify_continuation_git_facts(anchor) -> None: + path = Path(anchor.artifact_path) source_root = gitops.worktree_root(path.parent) if source_root is None: raise ProtocolError("continuation artifact is not inside a Git worktree") @@ -393,6 +409,10 @@ def init_dialogue(definition: config.ProtocolDefinition, directory: Path | str) } atomic_write_json(directory / STATE_FILE, state) dialogue = Dialogue(directory) + if definition.continuation is not None: + # init just verified the anchor's Git facts above; do not spawn + # the same subprocesses again on the first state() read. + dialogue._continuation_git_verified = True try: gitops.commit_paths( root, @@ -420,6 +440,10 @@ def __init__(self, directory: Path | str) -> None: _reject_symlink(self.directory, "dialogue directory") if not self.directory.is_dir(): raise ProtocolError(f"not a dialogue directory: {self.directory}") + # Content-addressed Git facts of a continuation anchor are + # verified once per Dialogue instance (see state()); the artifact + # bytes themselves are re-hashed on every read. + self._continuation_git_verified = False # -- loading --------------------------------------------------------- @@ -453,7 +477,18 @@ def state(self) -> dict: raise ProtocolError( "definition digest mismatch: definition.json was modified after init" ) - verify_continuation_anchor(definition) + anchor = definition.continuation + if anchor is not None: + # The live artifact bytes are re-read and re-hashed on every + # state read (no subprocess, tamper-evident). The Git-side + # facts are content-addressed and verified once per Dialogue + # instance; spawning several git subprocesses per read is + # pure cost on validate/report paths that call state() + # repeatedly. + _verify_continuation_artifact_bytes(anchor) + if not self._continuation_git_verified: + _verify_continuation_git_facts(anchor) + self._continuation_git_verified = True return state def _write_state(self, state: dict) -> dict: @@ -739,7 +774,7 @@ def complete( ) if turn.round_id == definition.final_round_id and definition.agent_final_statuses: assert turn_text is not None - statuses = re.findall(r"(?m)^Status:[ \t]*(\S+)[ \t]*$", turn_text) + statuses = re.findall(r"(?m)^Status:[ \t]*(\S+)[ \t]*\r?$", turn_text) if len(statuses) != 1: raise ProtocolError( "final turn must contain exactly one line 'Status: '" diff --git a/tests/test_config.py b/tests/test_config.py index 4464261..3ca6ea9 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -152,6 +152,27 @@ def test_rejects_substitute_with_different_protocol_role(self) -> None: raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] self.assert_rejected(raw, "role must match primary actor") + def test_none_reason_code_is_reserved(self) -> None: + # "none" is the null sentinel written to Madp-Substitution-Reason + # trailers; as a real code it would collide with it. + raw = support.two_actor_definition() + raw["actors"].append( + { + "actor_id": "worker-c", + "role": "proposer", + "transport": "command", + "expected_provider": "prov-c", + "expected_model": "model-c", + "settings": { + "argv": ["fake-worker"], + "identity_verifier_argv": ["fake-verifier"], + }, + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["worker-c"] + raw["schedule"][0]["substitution_reasons"] = ["none"] + self.assert_rejected(raw, "reserved") + def test_rejects_substitute_using_same_hermes_home(self) -> None: raw = support.two_actor_definition() raw["actors"][1]["role"] = raw["actors"][0]["role"] diff --git a/tests/test_engine.py b/tests/test_engine.py index cd9689e..e138d70 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -7,6 +7,7 @@ import tempfile import unittest from pathlib import Path +from unittest import mock import support @@ -59,6 +60,30 @@ def test_continuation_anchor_is_rechecked_after_init(self) -> None: dialogue.state() self.assertIn("continuation artifact hash mismatch", str(ctx.exception)) + def test_continuation_git_facts_verified_once_per_instance(self) -> None: + # The live artifact bytes are re-hashed on every state read; the + # content-addressed Git facts are verified once per Dialogue + # instance, so validate/report paths do not spawn git per read. + self.definition, artifact = self.continuation_definition() + dialogue = self.init_dialogue() + with mock.patch.object( + engine.gitops, + "first_commit_adding", + wraps=engine.gitops.first_commit_adding, + ) as spy: + dialogue.state() + dialogue.state() + self.assertEqual(spy.call_count, 0, "init already verified them") + fresh = engine.Dialogue(self.dialogue_dir) + with mock.patch.object( + engine.gitops, + "first_commit_adding", + wraps=engine.gitops.first_commit_adding, + ) as spy: + fresh.state() + fresh.state() + self.assertEqual(spy.call_count, 1, "cached after the first read") + def test_init_creates_state_and_definition(self) -> None: dialogue = self.init_dialogue() self.assertTrue((self.dialogue_dir / "definition.json").is_file()) @@ -151,6 +176,15 @@ def test_correct_claim_locks_turn(self) -> None: self.assertEqual(state["revision"], 1) self.assertTrue((self.dialogue_dir / engine.LOCK_FILE).exists()) + def test_whitespace_only_substitution_reason_is_rejected(self) -> None: + # " " must not collapse to None and silently pass as a primary + # claim; a non-empty value is required when the flag is given. + dialogue = self.init_dialogue() + with self.assertRaises(engine.ProtocolError) as ctx: + dialogue.claim("worker-a", substitution_reason=" ") + self.assertIn("non-empty", str(ctx.exception)) + self.assertEqual(dialogue.state()["status"], "OPEN") + def test_wrong_actor_cannot_claim(self) -> None: dialogue = self.init_dialogue() with self.assertRaises(engine.ProtocolError) as ctx: @@ -321,6 +355,30 @@ def write_attempt(status: str) -> None: self.assertEqual(state["status"], "READY_FOR_OWNER") self.assertIsNone(state["owner_decision"]) + def test_crlf_status_line_is_accepted(self) -> None: + raw = support.two_actor_definition() + raw["schedule"] = raw["schedule"][:1] + raw["final_round_id"] = "R01" + raw["agent_final_statuses"] = ["READY_FOR_OWNER"] + self.definition = config.parse_definition(raw) + dialogue = self.init_dialogue() + dialogue.claim("worker-a") + turn_path = self.root / "R01-crlf.md" + turn_path.write_bytes( + b"# R01\r\n\r\nFinal analysis.\r\n\r\nStatus: READY_FOR_OWNER\r\n" + ) + record = support.make_evidence( + actor_id="worker-a", + round_id="R01", + artifact_path=turn_path, + provider="fake-provider-a", + model="fake-model-a", + ) + evidence_path = self.root / "R01-crlf.json" + evidence_path.write_text(json.dumps(record), encoding="utf-8") + state = dialogue.complete("worker-a", turn_path, evidence_path) + self.assertEqual(state["status"], "READY_FOR_OWNER") + def _finish_all_turns(self) -> None: path = self.dialogue_dir / "state.json" state = json.loads(path.read_text(encoding="utf-8"))