diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f26a92f..08fc1aa 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 055d2a0..8ed52a9 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 cea48ed..b54a338 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`: @@ -58,6 +65,53 @@ evidence file is flagged here AND by `validate` independently; real acceptance path in a fresh local dialogue and validates it with the production gate (always local-only). +## 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 @@ -72,8 +126,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 ``` @@ -364,6 +418,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` @@ -408,7 +464,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; @@ -461,7 +519,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 c6ff557..05b1a8d 100644 --- a/schemas/protocol.schema.json +++ b/schemas/protocol.schema.json @@ -27,6 +27,12 @@ "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}, @@ -39,6 +45,21 @@ "items": {"type": "integer", "minimum": 1}, "description": "Closed set of evidence-record versions this dialogue accepts (default: the engine-supported set). A turn whose evidence_version is outside this set fails closed; versions the engine cannot interpret are rejected at definition load even if listed." }, + "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, @@ -81,9 +102,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 2fe84b4..bad01ab 100644 --- a/schemas/runtime-evidence.schema.json +++ b/schemas/runtime-evidence.schema.json @@ -29,6 +29,21 @@ "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 @@ -173,6 +188,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 d524524..09145d7 100644 --- a/src/multi_agent_dialogue/adapters/base.py +++ b/src/multi_agent_dialogue/adapters/base.py @@ -52,6 +52,7 @@ class PrepareContext: # required_capabilities; runner attaches it via dataclasses.replace # so the evidence records exactly what gated — never a re-probe. capability_manifest: dict | None = None + substitution_reason: str | None = None def placeholders(self) -> dict[str, str]: return { @@ -356,6 +357,13 @@ def base_evidence(self, context: PrepareContext, *, provider: str, model: str, record = { "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 d2f1bab..965171d 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 @@ -196,6 +224,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 ff99c95..435bb52 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 @@ -36,6 +38,46 @@ ) 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", + "evidence_versions", + "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", @@ -46,7 +88,15 @@ "settings", "required_capabilities", } -_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): @@ -73,6 +123,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) @@ -90,10 +160,12 @@ class ProtocolDefinition: # guaranteed ⊆ SUPPORTED_EVIDENCE_VERSIONS at parse time. A turn whose # evidence_version is outside this set fails closed. evidence_versions: tuple[int, ...] = SUPPORTED_EVIDENCE_VERSIONS + 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: @@ -191,6 +263,51 @@ 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" + ) + 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" + ) + 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") @@ -203,9 +320,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", @@ -224,6 +367,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) @@ -253,6 +399,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) @@ -291,6 +439,25 @@ def parse_definition(raw: Any) -> ProtocolDefinition: if len(set(versions_raw)) != len(versions_raw): errors.append("definition: evidence_versions contains duplicates") + 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): @@ -326,7 +493,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}") @@ -335,6 +503,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: @@ -347,6 +539,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)) @@ -362,7 +560,9 @@ def parse_definition(raw: Any) -> ProtocolDefinition: evidence_roots=tuple(evidence_roots_raw), owner_decisions=tuple(decisions_raw), evidence_versions=tuple(versions_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 76f1f3b..c63abac 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,199 @@ 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}" + ) + 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( + 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 +312,65 @@ 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 + _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") + 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}" + ) + + +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") + 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}") @@ -158,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, @@ -185,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 --------------------------------------------------------- @@ -218,6 +477,18 @@ def state(self) -> dict: raise ProtocolError( "definition digest mismatch: definition.json was modified after init" ) + 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: @@ -249,7 +520,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 +545,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 +698,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,8 +734,12 @@ 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, accepted_versions=definition.evidence_versions, + substitution_reason=substitution_reason, ) if evidence_errors: raise ProtocolError( @@ -462,16 +755,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]*\r?$", 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" @@ -485,6 +797,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, @@ -523,6 +838,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"], @@ -577,8 +895,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() @@ -684,6 +1000,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): @@ -704,16 +1021,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( @@ -722,6 +1040,7 @@ def validate( turn=spec, artifact_sha256=artifact_sha, accepted_versions=definition.evidence_versions, + substitution_reason=record.get("substitution_reason"), ) ) except (evidence.EvidenceError, config.ConfigError) as exc: @@ -750,6 +1069,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: @@ -811,6 +1143,9 @@ def build_report(self) -> dict: row: dict = { "round_id": round_id, "actor_id": record.get("actor_id"), + "scheduled_actor_id": record.get("scheduled_actor_id"), + "actor_selection": record.get("actor_selection"), + "substitution_reason": record.get("substitution_reason"), "completed_via": record.get("completed_via"), "commit": (commits.get(round_id) or {}).get("commit"), "artifact_file": record.get("artifact_file"), @@ -964,6 +1299,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) @@ -980,9 +1316,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: @@ -991,6 +1339,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", [])): @@ -1044,6 +1395,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", @@ -1072,16 +1426,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 @@ -1120,6 +1545,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 e21bb02..5d01907 100644 --- a/src/multi_agent_dialogue/evidence.py +++ b/src/multi_agent_dialogue/evidence.py @@ -81,6 +81,7 @@ def validate_evidence( turn: TurnSpec, artifact_sha256: str, accepted_versions: tuple[int, ...] | None = None, + substitution_reason: str | None = None, ) -> list[str]: """Return every reason this evidence fails; empty list means valid. @@ -122,6 +123,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 ce18f58..a817409 100644 --- a/src/multi_agent_dialogue/runner.py +++ b/src/multi_agent_dialogue/runner.py @@ -124,8 +124,16 @@ def _finalize_receipt( 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) @@ -137,11 +145,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, @@ -151,8 +166,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( @@ -161,9 +177,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" ) @@ -180,10 +204,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 @@ -207,6 +244,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; " @@ -221,6 +266,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.", "", @@ -232,14 +279,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(), @@ -247,14 +303,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( @@ -266,13 +334,23 @@ 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) @@ -281,7 +359,9 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None) # Capability gate: when the definition requires capabilities, the # engine probes them from the CLI itself and refuses the launch - # before any claim or runtime spawn. + # before any claim or runtime spawn. The probed actor is the SELECTED + # runtime actor — a substitute is gated on its own CLI, never the + # primary's. required = context.actor.required_capabilities if required: try: @@ -302,7 +382,7 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None) # The evidence must record exactly what gated the launch. context = dataclasses.replace(context, capability_manifest=manifest) - dialogue.claim(actor_id) + dialogue.claim(actor_id, substitution_reason=substitution_reason) # Forensics are best-effort: a receipt that cannot be written must # never hold a claim hostage or block a turn the ledger can prove. try: @@ -314,7 +394,12 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None) 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( @@ -382,6 +467,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 b3f1c04..15a20eb 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -131,6 +131,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") @@ -161,6 +178,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 55c0edc..3ca6ea9 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,80 @@ 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_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"] + 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 +227,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 +279,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 +322,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) + class EvidenceVersionsDefinitionTests(unittest.TestCase): """The definition pins the closed set of evidence-record versions it diff --git a/tests/test_engine.py b/tests/test_engine.py index cf929da..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 @@ -26,6 +27,63 @@ 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_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()) @@ -90,6 +148,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") @@ -99,14 +176,90 @@ 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: 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 +303,82 @@ 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 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")) diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 34add74..386ab7c 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) diff --git a/tests/test_real_contracts.py b/tests/test_real_contracts.py index af94fc6..d1ca393 100644 --- a/tests/test_real_contracts.py +++ b/tests/test_real_contracts.py @@ -1528,6 +1528,48 @@ def test_failing_version_probe_fails_the_gate(self) -> None: self.assertFalse(self.marker.exists()) self.assertEqual(dialogue.state()["turn_index"], 0) + def test_substitute_is_gated_on_its_own_cli(self) -> None: + # A substitute actor is capability-gated on ITS OWN probed CLI, + # never on the primary's. + stub = self.base / "hermes-sub-no-source" + stub.write_text( + "#!/bin/sh\n" + 'if [ "$1" = "--version" ]; then echo "stub-hermes 1.0"; exit 0; fi\n' + 'if [ "$1" = "chat" ] && [ "$2" = "--help" ]; then\n' + ' echo "usage: hermes chat [-h] [-q QUERY] [-Q]"\n' + " exit 0\n" + "fi\n" + f'exec "{HERMES}" "$@"\n', + encoding="utf-8", + ) + os.chmod(stub, 0o755) + raw = self.definition_raw() + sub_settings = self.hermes_settings("homes/sub") + sub_settings["command_name"] = str(stub) + raw["actors"].append( + { + "actor_id": "hermes-sub", + "role": "proposer", + "transport": "hermes-cli", + "expected_provider": "nousresearch", + "expected_model": "hermes-4-405b", + "settings": sub_settings, + "required_capabilities": ["one-shot-source-tagging"], + } + ) + raw["schedule"][0]["substitute_actor_ids"] = ["hermes-sub"] + raw["schedule"][0]["substitution_reasons"] = ["provider_cooldown"] + definition = config.parse_definition(raw) + dialogue = engine.init_dialogue(definition, self.base / "dialogue-sub") + with self.assertRaisesRegex(engine.ProtocolError, "capabilities"): + runner.launch( + dialogue, "hermes-sub", + substitution_reason="provider_cooldown", + ) + self.assertFalse(self.marker.exists(), "no runtime process may spawn") + self.assertIsNone(dialogue.state()["claim"]) + self.assertEqual(dialogue.state()["turn_index"], 0) + if __name__ == "__main__": unittest.main()