diff --git a/main.py b/main.py index cecb10612..254b00204 100644 --- a/main.py +++ b/main.py @@ -35,9 +35,10 @@ import atexit import hashlib import json +from functools import wraps from datetime import datetime, timezone from pathlib import Path -from typing import Optional, Dict, Any, Mapping, Sequence +from typing import Optional, Dict, Any, Callable, Mapping, Sequence # Load environment variables for DAEs (API keys, ports, feature flags). # Managed mode builds `.env.managed` from `.env` (last duplicate wins) for @@ -3904,8 +3905,7 @@ def _reddog_queue_stage_rejection_reasons(stage_callable: Any) -> tuple[str, ... def _reddog_record_queue_control_result(repo_root: Path, result: Mapping[str, Any]) -> Dict[str, Any]: """Record and expose one resident queue control-loop result.""" - - recorded = dict(result) + recorded = _reddog_control_lock_annotated_result(result) try: from modules.communication.moltbot_bridge.src.reddog_resident_queue_binding_profile import ( resident_queue_runtime_file_path, @@ -3955,6 +3955,49 @@ def _reddog_record_queue_control_result(repo_root: Path, result: Mapping[str, An return recorded +def _reddog_control_lock_annotated_result(result: Mapping[str, Any]) -> Dict[str, Any]: + from modules.communication.moltbot_bridge.src.reddog_resident_queue_control_lock import ( + resident_queue_control_lock_held, + ) + + recorded = dict(result) + recorded["control_lock_acquired"] = resident_queue_control_lock_held() + return recorded + + +def _with_reddog_queue_control_lock(control_loop: Callable[[Path], bool]) -> Callable[[Path], bool]: + """Decorate every control-loop call with the shared non-blocking lock.""" + + @wraps(control_loop) + def locked(repo_root: Path) -> bool: + from modules.communication.moltbot_bridge.src.reddog_resident_queue_control_lock import ( + acquire_resident_queue_control_lock, + ) + + with acquire_resident_queue_control_lock(repo_root) as lock: + if not lock.acquired: + locked.last_result = _reddog_locked_control_result(lock.reason, lock.path) + return False + return control_loop(repo_root) + + return locked + + +def _reddog_locked_control_result(reason: str, lock_path: Path) -> Dict[str, Any]: + return { + "accepted": False, + "status": "CONTROL_LOOP_LOCKED", + "rounds": 0, + "serial_progress": 0, + "claim_progress": 0, + "receipt_ids": (), + "rejection_reasons": (reason,), + "control_lock_path": str(lock_path), + "control_lock_acquired": False, + } + + +@_with_reddog_queue_control_lock def run_reddog_resident_queue_control_loop_preflight(repo_root: Path) -> bool: """ Drive the resident queue through bounded serial/claim rounds. diff --git a/modules/communication/moltbot_bridge/INTERFACE.md b/modules/communication/moltbot_bridge/INTERFACE.md index 57536ef9e..9c2c6adc0 100644 --- a/modules/communication/moltbot_bridge/INTERFACE.md +++ b/modules/communication/moltbot_bridge/INTERFACE.md @@ -489,6 +489,17 @@ Resident OpenClaw contract: - `tail ` = recent window - `watch|follow since ` = incremental follow with returned `next_cursor` +Resident RedDog live-canary contract: +- Entry point: `python -m modules.communication.moltbot_bridge.src.reddog_resident_live_canary` +- Default mode performs readiness checks only and writes an audit-safe receipt outside the repository. +- Live invocation requires Linux, `--execute`, and exact confirmation token `REDDOG_RESIDENT_LIVE_CANARY_PHASE1`. +- The selected profile is fixed to `signed_0102_bounded_code_fusion_worktree_draft_pr_pattern_memory`. +- The harness delegates to `main.run_reddog_resident_queue_control_loop_preflight`; it does not duplicate queue stages. +- `LIVE_PROOF_COMPLETE` requires a shared-lock-proven, newly persisted `reddog_resident_control_loop_receipt.v1` with accepted PASS, matching repository digest, and positive serial progress; changed pre/post chain revisions; a canonically recomputable chain revision and new persisted final-revision receipt witness; matching queue/slice and work-order/slice/head lineage; accepted verified draft-PR evidence; an external Git worktree present in the repository worktree registry with accepted invoke/create decisions and matching `HEAD`; and PatternMemory admission/record/digest identities recomputed from the canonical SQLite row. +- `--receipt-path` may name only canonical `live_canary_receipt.json` inside the runtime root. Any alternate receipt must resolve outside both repository and runtime roots; reserved runtime artifacts and nested collisions fail before execution. +- `READY_FOR_EXECUTION` does not claim that a live canary ran. Missing authority artifacts, signer socket, Git/GitHub readiness, or OpenRouter key presence returns `BLOCKED` without exposing values. +- The surface has no signer launch, secret resolution, PR-ready, merge, reward, or HoloIndex re-index authority. + ### OpenClaw Supervisor Contract Canonical 0102 lifecycle owner: diff --git a/modules/communication/moltbot_bridge/ModLog.md b/modules/communication/moltbot_bridge/ModLog.md index 8b735dc71..ce96b77a1 100644 --- a/modules/communication/moltbot_bridge/ModLog.md +++ b/modules/communication/moltbot_bridge/ModLog.md @@ -1,5 +1,44 @@ # ModLog - moltbot_bridge +## 2026-07-18: REDDOG_RESIDENT_LIVE_CANARY_PHASE1 + +**WSP Protocol**: WSP 00, 06, 15, 22, 46, 62, 71, 97 +**Phase**: Audit-hardened operator harness; live proof blocked pending operator artifacts +**Agent**: 0102 focused worker under architect review + +**Changes**: + +- Added a Linux-only, explicit-confirmation operator surface for one invocation + of the existing highest guarded resident queue profile. +- Reused the existing control loop and planner; no queue stage, signer, Fusion, + worktree, verifier, draft-PR, or PatternMemory orchestration was duplicated. +- Added outside-repo atomic receipts that distinguish static readiness from a + completed live proof and serialize only secret/reference presence. +- Added a shared OS advisory lock around every main resident control-loop call; + the persisted v1 control receipt records and proves lock ownership. +- Required a matching newly persisted control receipt with exact schema, + accepted PASS, repo digest, positive serial progress, and changed pre/post + chain revisions. +- Required the exact chain envelope, matching queue/slice, a new chain-store + receipt bound to the final revision, and draft/held-out/PatternMemory lineage + for one work order, slice, and candidate head. +- Repaired the chain-store commit witness: its canonical revision normalizes + only the newest receipt witness before hashing, then atomically persists that + revision in both the envelope and receipt for non-circular readback proof. +- Required accepted invoke/result worktree decisions plus an existing external + Git worktree registered by the repository with matching `rev-parse` HEAD. +- Required PatternMemory admission/record/digest identities to recompute from + the canonical SQLite row read back through the production sink adapter. +- Restricted runtime-root receipt output to canonical + `live_canary_receipt.json`; alternate output must be outside both roots. +- Split evidence and lock helpers so every live-canary production file is at + most 675 lines and every function in those files is at most 50 lines. + +**Truth boundary**: Focused tests prove the harness and evidence gates. No live +signer was started and no production draft PR was created by this slice; the +live proof remains blocked until WSL receives the required outside-repo +authority/signer artifacts and an already-running isolated signer. + ## 2026-07-18: REDDOG_HOLOINDEX_QUERY_OWNER_BOUNDARY_POC_PHASE1 **WSP Protocol**: WSP 00, 05, 06, 15, 22, 50, 62, 87, 96, 97 diff --git a/modules/communication/moltbot_bridge/README.md b/modules/communication/moltbot_bridge/README.md index 1c5fb61e2..308bddc4b 100644 --- a/modules/communication/moltbot_bridge/README.md +++ b/modules/communication/moltbot_bridge/README.md @@ -202,6 +202,38 @@ Supervisor repair policy: - exhausted restart budget escalates instead of looping forever - every cycle advances a DAEmon follow cursor so repair decisions stay tied to observed runtime history +### Resident RedDog live canary + +`reddog_resident_live_canary` is the operator admission surface for one run of +the existing highest guarded resident profile. Run it on Linux/WSL first +without `--execute`; readiness never invokes the control loop: + +```bash +python -m modules.communication.moltbot_bridge.src.reddog_resident_live_canary \ + --repo-root /mnt/o/Foundups-Agent \ + --runtime-root /mnt/o/.reddog/resident/Foundups-Agent +``` + +Execution additionally requires `--execute --confirm +REDDOG_RESIDENT_LIVE_CANARY_PHASE1`. The runtime root must be outside the +repository and already contain the authority, permission, execution-valve, +signer config/run-packet, and live signer socket artifacts named by the CLI +receipt. The harness never starts the signer or resolves secret values. + +`READY_FOR_EXECUTION` is only static readiness. Every main control-loop caller +uses one shared OS advisory lock. `LIVE_PROOF_COMPLETE` requires the matching +new v1 control receipt to prove lock ownership, accepted PASS, repository +binding, and positive serial progress; changed pre/post chain revisions; an +exact completed chain envelope with a new final-revision store receipt; +work-order/slice/head lineage; accepted draft-only PR evidence; an external +Git worktree registered by the repository with matching `HEAD`; and +PatternMemory identities recomputed from the canonical SQLite record. The +chain revision is non-circular: the store normalizes the newest receipt witness +for hashing, then persists the same revision in the envelope and that receipt. +It never marks a PR ready or merges it. Inside the runtime root, receipt output +is reserved to canonical `live_canary_receipt.json`; other outputs must be +outside both the runtime root and repository. + PQN runtime examples: - `run pqn simulation` - `status pqn simulation` diff --git a/modules/communication/moltbot_bridge/src/reddog_resident_control_loop_receipt_store.py b/modules/communication/moltbot_bridge/src/reddog_resident_control_loop_receipt_store.py index cb37d3269..3cebf6779 100644 --- a/modules/communication/moltbot_bridge/src/reddog_resident_control_loop_receipt_store.py +++ b/modules/communication/moltbot_bridge/src/reddog_resident_control_loop_receipt_store.py @@ -33,6 +33,7 @@ class ResidentControlLoopReceipt: rejection_reasons: tuple[str, ...] created_at: str repo_root_digest: str + control_lock_acquired: bool no_authority_issued: bool = True no_worker_spawn_performed: bool = True no_shell_command_executed: bool = True @@ -66,6 +67,7 @@ def build_resident_control_loop_receipt( "rejection_reasons": _string_tuple(result.get("rejection_reasons")), "created_at": str(created_at or ""), "repo_root_digest": _digest(str(Path(repo_root).resolve())), + "control_lock_acquired": result.get("control_lock_acquired") is True, "no_authority_issued": True, "no_worker_spawn_performed": True, "no_shell_command_executed": True, diff --git a/modules/communication/moltbot_bridge/src/reddog_resident_live_canary.py b/modules/communication/moltbot_bridge/src/reddog_resident_live_canary.py new file mode 100644 index 000000000..fec13e7bc --- /dev/null +++ b/modules/communication/moltbot_bridge/src/reddog_resident_live_canary.py @@ -0,0 +1,632 @@ +"""Operator harness for one guarded resident RedDog live canary. + +Slice: REDDOG_RESIDENT_LIVE_CANARY_PHASE1 + +This module does not implement queue orchestration. It validates the operator +boundary for the existing highest guarded resident profile and, only after an +explicit confirmation, delegates once to ``main.py``'s bounded resident control +loop. A live proof is reported only when that invocation creates both a new +chain revision and a new control-loop receipt and the existing planner proves +the complete draft-PR-only chain. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import stat +import subprocess +import sys +import tempfile +from contextlib import contextmanager +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Iterator, Mapping, Optional, Sequence + +from modules.communication.moltbot_bridge.src.reddog_resident_queue_binding_profile import ( + PROFILE_SIGNED_0102_BOUNDED_CODE_FUSION_WORKTREE_DRAFT_PR_PATTERN_MEMORY, +) +from modules.communication.moltbot_bridge.src.reddog_resident_live_canary_evidence import ( + CanaryInvocationEvidence, + chain_receipt_ids, + evaluate_live_proof, + read_control_receipts, + select_new_control_receipt, +) + + +LIVE_CANARY_SCHEMA_VERSION = "reddog_resident_live_canary_receipt.v1" +LIVE_CANARY_CONFIRMATION = "REDDOG_RESIDENT_LIVE_CANARY_PHASE1" +LIVE_CANARY_READY = "READY_FOR_EXECUTION" +LIVE_CANARY_BLOCKED = "BLOCKED" +LIVE_CANARY_EXECUTION_FAILED = "EXECUTION_FAILED" +LIVE_CANARY_PROOF_INCOMPLETE = "LIVE_PROOF_INCOMPLETE" +LIVE_CANARY_PROOF_COMPLETE = "LIVE_PROOF_COMPLETE" + +REQUIRED_JSON_ARTIFACTS = ( + "authoritative_work_state.json", + "authority_profile.json", + "execution_valve_env.json", + "permission_snapshots.json", + "principal_authority_records.json", + "signer_service_config.json", + "signer_service_run_packet.json", +) + + +@dataclass(frozen=True) +class LiveCanaryReadinessCheck: + name: str + passed: bool + reason: str + + +@dataclass(frozen=True) +class LiveCanaryReceipt: + schema_version: str + receipt_id: str + created_at: str + status: str + profile: str + execution_requested: bool + execution_confirmed: bool + execution_invoked: bool + ready_for_execution: bool + control_loop_accepted: bool + live_proof_complete: bool + readiness_checks: tuple[LiveCanaryReadinessCheck, ...] + blockers: tuple[str, ...] + control_receipt_id: Optional[str] + previous_chain_revision: Optional[str] + observed_chain_revision: Optional[str] + chain_plan_id: Optional[str] + verified_draft_pr_receipt_id: Optional[str] + verified_draft_pr_url: Optional[str] + pattern_memory_admission_id: Optional[str] + pattern_memory_record_id: Optional[str] + pattern_memory_record_digest: Optional[str] + accepted_stage_count: int + repo_root_digest: str + runtime_root_digest: str + draft_pr_only: bool = True + merge_authority_available: bool = False + no_merge_performed: bool = True + runtime_state_outside_repo: bool = True + isolated_worktree_observed: bool = False + secret_values_serialized: bool = False + + def to_dict(self) -> dict[str, Any]: + payload = asdict(self) + payload["readiness_checks"] = [asdict(item) for item in self.readiness_checks] + payload["blockers"] = list(self.blockers) + return payload + + +ControlLoopRunner = Callable[[Path], Mapping[str, Any]] +CommandResolver = Callable[[str], Optional[str]] +CommandProbe = Callable[[Sequence[str], Path], bool] +SocketProbe = Callable[[Path], bool] + + +@dataclass(frozen=True) +class _CanaryContext: + repo_root: Path + runtime_root: Path + receipt_path: Path + environ: Mapping[str, str] + platform_name: str + + +def run_reddog_resident_live_canary( + *, + repo_root: Path | str, + runtime_root: Path | str, + receipt_path: Path | str | None = None, + execute: bool = False, + confirmation: str = "", + queue_item_id: str = "", + max_rounds: int = 8, + environ: Optional[Mapping[str, str]] = None, + platform_name: Optional[str] = None, + command_resolver: CommandResolver = shutil.which, + command_probe: Optional[CommandProbe] = None, + socket_probe: Optional[SocketProbe] = None, + control_loop_runner: Optional[ControlLoopRunner] = None, + now: Optional[Callable[[], datetime]] = None, +) -> LiveCanaryReceipt: + """Assess readiness and optionally invoke the existing resident control loop.""" + context = _canary_context(repo_root, runtime_root, receipt_path, environ, platform_name) + checks = _readiness_checks( + repo_root=context.repo_root, + runtime_root=context.runtime_root, + receipt_path=context.receipt_path, + environ=context.environ, + platform_name=context.platform_name, + command_resolver=command_resolver, + command_probe=command_probe or _command_succeeds, + socket_probe=socket_probe or _is_unix_socket, + max_rounds=max_rounds, + ) + invocation = _invoke_canary( + context=context, + checks=checks, + execute=execute, + confirmation=confirmation, + queue_item_id=queue_item_id, + max_rounds=max_rounds, + runner=control_loop_runner or _run_existing_control_loop, + ) + timestamp = (now or (lambda: datetime.now(timezone.utc)))().astimezone(timezone.utc).isoformat() + proof = evaluate_live_proof( + repo_root=context.repo_root, + runtime_root=context.runtime_root, + queue_item_id=queue_item_id, + invocation=invocation, + now_iso=timestamp, + ) + receipt = _build_live_canary_receipt(context, checks, invocation, proof, execute, timestamp) + _persist_receipt(context, receipt) + return receipt + + +def _persist_receipt(context: _CanaryContext, receipt: LiveCanaryReceipt) -> None: + _write_json_atomic( + context.receipt_path, + receipt.to_dict(), + repo_root=context.repo_root, + runtime_root=context.runtime_root, + ) + + +def _canary_context( + repo_root: Path | str, + runtime_root: Path | str, + receipt_path: Path | str | None, + environ: Optional[Mapping[str, str]], + platform_name: Optional[str], +) -> _CanaryContext: + root = Path(repo_root).resolve() + runtime = Path(runtime_root).resolve() + target = _validated_receipt_path(root, runtime, receipt_path) + return _CanaryContext( + repo_root=root, + runtime_root=runtime, + receipt_path=target, + environ=dict(os.environ if environ is None else environ), + platform_name=str(platform_name or sys.platform).lower(), + ) + + +def _validated_receipt_path( + repo_root: Path, + runtime_root: Path, + receipt_path: Path | str | None, +) -> Path: + canonical_path = runtime_root / "live_canary_receipt.json" + if canonical_path.is_symlink(): + raise ValueError("receipt_path_reserved_or_collision") + canonical = canonical_path.resolve() + target = Path(receipt_path).resolve() if receipt_path else canonical + if _is_inside(target, repo_root): + raise ValueError("receipt_path_inside_repo") + if _is_inside(target, runtime_root) and target != canonical: + raise ValueError("receipt_path_reserved_or_collision") + return target + + +def _invoke_canary( + *, + context: _CanaryContext, + checks: Sequence[LiveCanaryReadinessCheck], + execute: bool, + confirmation: str, + queue_item_id: str, + max_rounds: int, + runner: ControlLoopRunner, +) -> CanaryInvocationEvidence: + blockers = [check.reason for check in checks if not check.passed] + confirmed = execute and confirmation == LIVE_CANARY_CONFIRMATION + if execute and not confirmed: + blockers.append("explicit_execution_confirmation_missing") + chain_path = context.runtime_root / "resident_queue_chain_results.json" + control_path = context.runtime_root / "resident_queue_control_loop_receipts.jsonl" + pre_chain_state = _read_json_mapping(chain_path) + previous_revision = _text(pre_chain_state.get("revision")) + pre_chain_ids = chain_receipt_ids(pre_chain_state) + pre_control_receipts = read_control_receipts(control_path) + invoked = execute and confirmed and not blockers + result: Mapping[str, Any] = {} + if invoked: + with _temporary_environment( + _canary_environment(context.runtime_root, queue_item_id, max_rounds) + ): + result = _mapping(runner(context.repo_root)) + chain_state = _read_json_mapping(chain_path) + control_receipt_id = _text(result.get("receipt_id")) + control_receipt = select_new_control_receipt( + pre_control_receipts, + read_control_receipts(control_path), + control_receipt_id, + ) + return CanaryInvocationEvidence( + confirmed=confirmed, + invoked=invoked, + blockers=tuple(blockers), + control_result=result, + previous_revision=previous_revision, + observed_revision=_text(chain_state.get("revision")), + control_receipt_id=control_receipt_id, + control_receipt=control_receipt, + pre_chain_receipt_ids=pre_chain_ids, + work_state=_read_json_mapping(context.runtime_root / "authoritative_work_state.json"), + chain_state=chain_state, + ) + + +def _readiness_checks( + *, + repo_root: Path, + runtime_root: Path, + receipt_path: Path, + environ: Mapping[str, str], + platform_name: str, + command_resolver: CommandResolver, + command_probe: CommandProbe, + socket_probe: SocketProbe, + max_rounds: int, +) -> list[LiveCanaryReadinessCheck]: + checks = [ + _check("linux_execution_plane", platform_name.startswith("linux"), "linux_execution_plane_required"), + _check("repo_root", repo_root.is_dir() and (repo_root / ".git").exists(), "repo_root_invalid"), + _check("runtime_root_outside_repo", not _is_inside(runtime_root, repo_root), "runtime_root_inside_repo"), + _check("receipt_outside_repo", not _is_inside(receipt_path, repo_root), "receipt_path_inside_repo"), + _check("max_rounds", isinstance(max_rounds, int) and 1 <= max_rounds <= 8, "invalid_max_rounds"), + _check("git_available", bool(command_resolver("git")), "git_not_available"), + _check("gh_available", bool(command_resolver("gh")), "gh_not_available"), + _check( + "git_worktree_ready", + command_probe(("git", "rev-parse", "--is-inside-work-tree"), repo_root), + "git_worktree_not_ready", + ), + _check( + "gh_authenticated", + command_probe(("gh", "auth", "status"), repo_root), + "gh_not_authenticated", + ), + _check("openrouter_key_reference", bool(str(environ.get("OPENROUTER_API_KEY") or "")), "openrouter_api_key_missing"), + ] + for filename in REQUIRED_JSON_ARTIFACTS: + valid = _valid_json_mapping(runtime_root / filename) + checks.append(_check(f"artifact:{filename}", valid, f"missing_or_malformed:{filename}")) + socket_path = runtime_root / "reddog_signer.sock" + checks.append(_check("signer_socket", socket_probe(socket_path), "signer_socket_not_ready")) + return checks + + +def _canary_environment(runtime_root: Path, queue_item_id: str, max_rounds: int) -> dict[str, str]: + env = _canary_runtime_paths(runtime_root) + env.update(_canary_runtime_modes()) + env.update(_canary_runtime_controls(max_rounds)) + env["REDDOG_WRE_QUEUE_ITEM_ID"] = queue_item_id + return env + + +def _canary_runtime_paths(runtime_root: Path) -> dict[str, str]: + return { + "REDDOG_RESIDENT_QUEUE_BINDING_PROFILE": PROFILE_SIGNED_0102_BOUNDED_CODE_FUSION_WORKTREE_DRAFT_PR_PATTERN_MEMORY, + "REDDOG_RESIDENT_RUNTIME_ROOT": str(runtime_root), + "REDDOG_AUTHORITATIVE_WORK_STATE_PATH": str(runtime_root / "authoritative_work_state.json"), + "REDDOG_RESIDENT_QUEUE_AUTHORITY_PROFILE_PATH": str(runtime_root / "authority_profile.json"), + "REDDOG_RESIDENT_QUEUE_CHAIN_RESULTS_PATH": str(runtime_root / "resident_queue_chain_results.json"), + "REDDOG_EXECUTION_VALVE_ENV_PATH": str(runtime_root / "execution_valve_env.json"), + "REDDOG_AUTHORITY_RUNTIME_STATE_PATH": str(runtime_root / "authority_runtime_state.json"), + "REDDOG_PERMISSION_SNAPSHOTS_PATH": str(runtime_root / "permission_snapshots.json"), + "REDDOG_PRINCIPAL_AUTHORITY_RECORDS_PATH": str(runtime_root / "principal_authority_records.json"), + "REDDOG_SIGNER_SERVICE_CONFIG_PATH": str(runtime_root / "signer_service_config.json"), + "REDDOG_SIGNER_SERVICE_RUN_PACKET_PATH": str(runtime_root / "signer_service_run_packet.json"), + "REDDOG_SIGNER_SOCKET_PATH": str(runtime_root / "reddog_signer.sock"), + "REDDOG_OUTCOME_RATCHET_STORE_PATH": str(runtime_root / "verified_outcomes.jsonl"), + "REDDOG_MODEL_FEEDBACK_LEDGER_STORE_PATH": str(runtime_root / "model_feedback.jsonl"), + "REDDOG_PATTERN_MEMORY_ADMISSION_DB_PATH": str(runtime_root / "pattern_memory.db"), + "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_RECEIPTS_PATH": str(runtime_root / "resident_queue_control_loop_receipts.jsonl"), + "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_LOCK_PATH": str(runtime_root / "resident_queue_control_loop.lock"), + } + + +def _canary_runtime_modes() -> dict[str, str]: + return { + "REDDOG_SIGNATURE_VERIFIER_BACKEND": "ed25519", + "REDDOG_WORK_ORDER_MATERIALIZER_MODE": "authority_profile", + "REDDOG_ARTIFACT_GENERATOR_MODE": "foundups_fusion", + "REDDOG_RESIDENT_QUEUE_WORKTREE_RUNNER_MODE": "real", + "REDDOG_EVIDENCE_COMMAND_RUNNER_MODE": "real", + "REDDOG_DRAFT_PR_RUNNER_MODE": "real", + "REDDOG_RESIDENT_QUEUE_WORKTREE_RUNNER_TIMEOUT_S": "120", + "REDDOG_DRAFT_PR_RUNNER_TIMEOUT_S": "120", + "REDDOG_WORK_ORDERS_PATH": "", + "REDDOG_ARTIFACT_CONTENTS_PATH": "", + "REDDOG_ARTIFACT_GENERATION_REQUEST_PATH": "", + "REDDOG_SIGNER_SERVICE_CONFIG_SUPPLY": "0", + "REDDOG_SIGNER_SERVICE_RUN_PACKET_SUPPLY": "0", + "REDDOG_AUTHORITY_RUNTIME_RESOLVER_ARTIFACT_SUPPLY": "0", + "REDDOG_RESIDENT_QUEUE_NOW_EPOCH": "", + } + + +def _canary_runtime_controls(max_rounds: int) -> dict[str, str]: + return { + "REDDOG_SIGNER_SERVICE_HEALTHCHECK": "1", + "REDDOG_SIGNER_SERVICE_HEALTHCHECK_ENFORCED": "1", + "REDDOG_PILOT_DRYRUN_BINDING": "1", + "REDDOG_ARTIFACT_GENERATION_REQUEST_BINDING": "1", + "REDDOG_SLICE_VERIFIER_REQUEST_BINDING": "1", + "REDDOG_DRAFT_PR_PUBLISH_REQUEST_BINDING": "1", + "REDDOG_OUTCOME_RATCHET_REQUEST_BINDING": "1", + "REDDOG_HELD_OUT_GATE_REQUEST_BINDING": "1", + "REDDOG_PATTERN_MEMORY_ADMISSION_REQUEST_BINDING": "1", + "REDDOG_WORKER_DISPATCH_AGENTDB_WRITER": "1", + "OPENCLAW_SIGNED_WORKER_TASKS_ENABLED": "1", + "OPENCLAW_SIGNED_0102_BOUNDED_CODE_TASKS_ENABLED": "1", + "OPENCLAW_SIGNED_QUEUE_STAGE_TASKS_ENABLED": "1", + "REDDOG_SIGNED_WORKER_QUEUE_LOOP_RUNNER": "1", + "REDDOG_RESIDENT_QUEUE_SERIAL_LOOP": "1", + "REDDOG_RESIDENT_QUEUE_SERIAL_LOOP_ENFORCED": "1", + "REDDOG_RESIDENT_QUEUE_SERIAL_LOOP_MAX_STEPS": "16", + "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP": "1", + "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_ENFORCED": "1", + "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_MAX_ROUNDS": str(max_rounds), + "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_RECEIPT_PERSISTENCE": "1", + "REDDOG_OPENCLAW_SIGNED_WORKER_CLAIM_LOOP": "1", + "REDDOG_OPENCLAW_SIGNED_WORKER_CLAIM_LOOP_ENFORCED": "1", + "OPENCLAW_SIGNED_WORKER_TASK_MAX_CLAIMS": "1", + } + + +def _run_existing_control_loop(repo_root: Path) -> Mapping[str, Any]: + import main + + accepted = main.run_reddog_resident_queue_control_loop_preflight(repo_root) + result = _mapping(getattr(main.run_reddog_resident_queue_control_loop_preflight, "last_result", {})) + if not result: + return {"accepted": bool(accepted), "status": "CONTROL_RESULT_MISSING"} + return result + + +def _build_live_canary_receipt( + context: _CanaryContext, + checks: Sequence[LiveCanaryReadinessCheck], + invocation: CanaryInvocationEvidence, + proof: Mapping[str, Any], + execute: bool, + created_at: str, +) -> LiveCanaryReceipt: + blockers = list(invocation.blockers) + if invocation.invoked: + blockers.extend(str(value) for value in proof.get("blockers", ())) + blockers = list(dict.fromkeys(blockers)) + ready = all(check.passed for check in checks) + control_accepted = invocation.control_result.get("accepted") is True + live_complete = invocation.invoked and proof.get("complete") is True + status = _canary_status(invocation.invoked, control_accepted, live_complete, blockers) + seed = _receipt_seed(created_at, status, invocation, blockers) + return LiveCanaryReceipt( + schema_version=LIVE_CANARY_SCHEMA_VERSION, + receipt_id="reddog_live_canary_" + _digest(seed)[:16], + created_at=created_at, + status=status, + profile=PROFILE_SIGNED_0102_BOUNDED_CODE_FUSION_WORKTREE_DRAFT_PR_PATTERN_MEMORY, + execution_requested=bool(execute), + execution_confirmed=invocation.confirmed, + execution_invoked=invocation.invoked, + ready_for_execution=ready, + control_loop_accepted=control_accepted, + live_proof_complete=live_complete, + readiness_checks=tuple(checks), + blockers=tuple(blockers), + control_receipt_id=invocation.control_receipt_id, + previous_chain_revision=invocation.previous_revision, + observed_chain_revision=invocation.observed_revision, + chain_plan_id=proof.get("plan_id"), + verified_draft_pr_receipt_id=proof.get("draft_pr_receipt_id"), + verified_draft_pr_url=proof.get("draft_pr_url"), + pattern_memory_admission_id=proof.get("pattern_memory_admission_id"), + pattern_memory_record_id=proof.get("pattern_memory_record_id"), + pattern_memory_record_digest=proof.get("pattern_memory_record_digest"), + accepted_stage_count=int(proof.get("accepted_stage_count") or 0), + repo_root_digest=_digest(str(context.repo_root)), + runtime_root_digest=_digest(str(context.runtime_root)), + no_merge_performed=( + not invocation.invoked or proof.get("no_merge_performed") is True + ), + runtime_state_outside_repo=not _is_inside(context.runtime_root, context.repo_root), + isolated_worktree_observed=proof.get("isolated_worktree_observed") is True, + ) + + +def _canary_status( + invoked: bool, + control_accepted: bool, + live_complete: bool, + blockers: Sequence[str], +) -> str: + if live_complete: + return LIVE_CANARY_PROOF_COMPLETE + if invoked and not control_accepted: + return LIVE_CANARY_EXECUTION_FAILED + if invoked: + return LIVE_CANARY_PROOF_INCOMPLETE + return LIVE_CANARY_BLOCKED if blockers else LIVE_CANARY_READY + + +def _receipt_seed( + created_at: str, + status: str, + invocation: CanaryInvocationEvidence, + blockers: Sequence[str], +) -> dict[str, Any]: + return { + "created_at": created_at, + "status": status, + "profile": PROFILE_SIGNED_0102_BOUNDED_CODE_FUSION_WORKTREE_DRAFT_PR_PATTERN_MEMORY, + "execution_invoked": invocation.invoked, + "control_receipt_id": invocation.control_receipt_id, + "observed_chain_revision": invocation.observed_revision, + "blockers": list(blockers), + } + + +@contextmanager +def _temporary_environment(values: Mapping[str, str]) -> Iterator[None]: + previous = {key: os.environ.get(key) for key in values} + os.environ.update(values) + try: + yield + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + +def _write_json_atomic( + path: Path, + payload: Mapping[str, Any], + *, + repo_root: Path, + runtime_root: Path, +) -> None: + if path.is_symlink(): + raise ValueError("receipt_path_reserved_or_collision") + resolved = path.resolve() + if _is_inside(resolved, repo_root): + raise ValueError("receipt_path_inside_repo") + canonical = (runtime_root / "live_canary_receipt.json").resolve() + if _is_inside(resolved, runtime_root) and resolved != canonical: + raise ValueError("receipt_path_reserved_or_collision") + resolved.parent.mkdir(parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(prefix=f".{resolved.name}.", suffix=".tmp", dir=str(resolved.parent)) + try: + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: + json.dump(payload, handle, sort_keys=True, indent=2, ensure_ascii=True) + handle.write("\n") + os.replace(temporary, resolved) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +def _valid_json_mapping(path: Path) -> bool: + return bool(_read_json_mapping(path)) + + +def _read_json_mapping(path: Path) -> Mapping[str, Any]: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return {} + return payload if isinstance(payload, Mapping) else {} + + +def _is_unix_socket(path: Path) -> bool: + try: + return stat.S_ISSOCK(path.stat().st_mode) + except OSError: + return False + + +def _command_succeeds(argv: Sequence[str], cwd: Path) -> bool: + """Run an audit-safe readiness command without returning its output.""" + + try: + completed = subprocess.run( + list(argv), + cwd=str(cwd), + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return False + return completed.returncode == 0 + + +def _check(name: str, passed: bool, failure_reason: str) -> LiveCanaryReadinessCheck: + return LiveCanaryReadinessCheck(name=name, passed=bool(passed), reason="ok" if passed else failure_reason) + + +def _mapping(value: Any) -> Mapping[str, Any]: + if hasattr(value, "to_dict"): + candidate = value.to_dict() + return candidate if isinstance(candidate, Mapping) else {} + return value if isinstance(value, Mapping) else {} + + +def _text(value: Any) -> Optional[str]: + text = str(value or "").strip() + return text or None + + +def _digest(value: Any) -> str: + raw = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, default=str) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _is_inside(child: Path, parent: Path) -> bool: + child_r = child.resolve() + parent_r = parent.resolve() + return child_r == parent_r or parent_r in child_r.parents + + +def build_reddog_resident_live_canary_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="reddog-resident-live-canary", + description="Assess or explicitly execute one highest-profile resident RedDog canary.", + ) + parser.add_argument("--repo-root", required=True) + parser.add_argument("--runtime-root", required=True) + parser.add_argument("--receipt-path") + parser.add_argument("--queue-item-id", default="") + parser.add_argument("--max-rounds", type=int, default=8) + parser.add_argument("--execute", action="store_true") + parser.add_argument("--confirm", default="") + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> int: + args = build_reddog_resident_live_canary_parser().parse_args(list(argv) if argv is not None else None) + receipt = run_reddog_resident_live_canary( + repo_root=args.repo_root, + runtime_root=args.runtime_root, + receipt_path=args.receipt_path, + execute=args.execute, + confirmation=args.confirm, + queue_item_id=args.queue_item_id, + max_rounds=args.max_rounds, + ) + print(json.dumps(receipt.to_dict(), sort_keys=True, separators=(",", ":"), ensure_ascii=True)) + return 0 if receipt.status in {LIVE_CANARY_READY, LIVE_CANARY_PROOF_COMPLETE} else 2 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main(sys.argv[1:])) + + +__all__ = [ + "LIVE_CANARY_BLOCKED", + "LIVE_CANARY_CONFIRMATION", + "LIVE_CANARY_EXECUTION_FAILED", + "LIVE_CANARY_PROOF_COMPLETE", + "LIVE_CANARY_PROOF_INCOMPLETE", + "LIVE_CANARY_READY", + "LiveCanaryReadinessCheck", + "LiveCanaryReceipt", + "build_reddog_resident_live_canary_parser", + "main", + "run_reddog_resident_live_canary", +] diff --git a/modules/communication/moltbot_bridge/src/reddog_resident_live_canary_evidence.py b/modules/communication/moltbot_bridge/src/reddog_resident_live_canary_evidence.py new file mode 100644 index 000000000..9ebe72151 --- /dev/null +++ b/modules/communication/moltbot_bridge/src/reddog_resident_live_canary_evidence.py @@ -0,0 +1,560 @@ +"""Fail-closed evidence validation for the resident RedDog live canary.""" + +from __future__ import annotations + +import hashlib +import json +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping, Optional + +from modules.communication.moltbot_bridge.src.reddog_resident_control_loop_receipt_store import ( + CONTROL_LOOP_RECEIPT_SCHEMA_VERSION, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_chain_results_store import ( + CHAIN_RESULTS_SCHEMA_VERSION, + resident_queue_chain_receipt_id, + resident_queue_chain_snapshot_is_canonical, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_pattern_memory_admission_handler import ( + PATTERN_MEMORY_ADMISSION_STAGE_KEY, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_orchestration_plan import ( + NEXT_QUEUE_CHAIN_COMPLETE, + NEXT_QUEUE_PATTERN_MEMORY_ADMISSION_INVOKE, + RESIDENT_QUEUE_ORCHESTRATION_PLAN_COMPLETE, + RESIDENT_QUEUE_ORCHESTRATION_PLAN_READY, + plan_reddog_resident_queue_orchestration, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_pattern_memory_admission_invoke import ( + QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_ACCEPT, + canonical_pattern_memory_admission_identity, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_verified_draft_pr_publish_invoke import ( + QUEUE_AUTHORIZED_VERIFIED_DRAFT_PR_PUBLISH_INVOKE_ACCEPT, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_worktree_create_invoke import ( + QUEUE_AUTHORIZED_WORKTREE_CREATE_INVOKE_ACCEPT, +) +from modules.communication.moltbot_bridge.src.reddog_wre_worktree_create import ( + WORKTREE_CREATE_ACCEPT, +) +from modules.infrastructure.wre_core.src.reddog_verified_draft_pr_publish import ( + VERIFIED_DRAFT_PR_PUBLISH_ACCEPT, +) +from modules.communication.moltbot_bridge.src.reddog_verified_pattern_memory_sink import ( + build_reddog_verified_pattern_memory_sink, + reddog_verified_pattern_memory_record_digest, +) + + +@dataclass(frozen=True) +class CanaryInvocationEvidence: + confirmed: bool + invoked: bool + blockers: tuple[str, ...] + control_result: Mapping[str, Any] + control_receipt: Mapping[str, Any] + previous_revision: Optional[str] + observed_revision: Optional[str] + control_receipt_id: Optional[str] + pre_chain_receipt_ids: frozenset[str] + work_state: Mapping[str, Any] + chain_state: Mapping[str, Any] + + +def read_control_receipts(path: Path) -> tuple[Mapping[str, Any], ...]: + """Parse valid JSON-object receipt rows; malformed rows cannot prove a run.""" + + if not path.is_file(): + return () + receipts: list[Mapping[str, Any]] = [] + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError: + return () + for line in lines: + try: + payload = json.loads(line) + except (json.JSONDecodeError, TypeError): + continue + if isinstance(payload, Mapping): + receipts.append(payload) + return tuple(receipts) + + +def select_new_control_receipt( + pre_receipts: tuple[Mapping[str, Any], ...], + post_receipts: tuple[Mapping[str, Any], ...], + receipt_id: Optional[str], +) -> Mapping[str, Any]: + """Return exactly one newly persisted receipt matching the runner result.""" + + if not receipt_id: + return {} + pre_ids = {_text(item.get("receipt_id")) for item in pre_receipts} + if receipt_id in pre_ids: + return {} + matches = [ + item for item in post_receipts + if _text(item.get("receipt_id")) == receipt_id + ] + return matches[0] if len(matches) == 1 else {} + + +def chain_receipt_ids(chain_state: Mapping[str, Any]) -> frozenset[str]: + """Return non-empty receipt IDs only from the canonical chain envelope.""" + + if chain_state.get("schema_version") != CHAIN_RESULTS_SCHEMA_VERSION: + return frozenset() + raw = chain_state.get("receipts") + if not isinstance(raw, list): + return frozenset() + return frozenset( + value for item in raw + if isinstance(item, Mapping) and (value := _text(item.get("receipt_id"))) + ) + + +def evaluate_live_proof( + *, + repo_root: Path, + runtime_root: Path, + queue_item_id: str, + invocation: CanaryInvocationEvidence, + now_iso: str, +) -> dict[str, Any]: + """Evaluate one invocation against persisted causal and stage evidence.""" + + stages = _mapping(invocation.chain_state.get("stage_results")) + plan, chain_blockers = _chain_evidence(invocation, queue_item_id, now_iso) + draft = _draft_pr_evidence(stages) + worktree = _worktree_evidence(stages, repo_root) + pattern = _pattern_memory_evidence( + stages, + repo_root, + runtime_root, + plan, + draft, + worktree, + ) + blockers = [ + *_fresh_execution_blockers(invocation, repo_root), + *chain_blockers, + *draft["blockers"], + *pattern["blockers"], + *_lineage_blockers(stages, plan), + *worktree["blockers"], + ] + return { + "complete": not blockers, + "blockers": list(dict.fromkeys(blockers)), + "plan_id": plan.plan_id if plan else None, + "accepted_stage_count": len(plan.accepted_stages) if plan else 0, + "draft_pr_receipt_id": draft["receipt_id"], + "draft_pr_url": draft["url"], + "no_merge_performed": draft["no_merge"], + "isolated_worktree_observed": not worktree["blockers"], + "pattern_memory_admission_id": pattern["admission_id"], + "pattern_memory_record_id": pattern["record_id"], + "pattern_memory_record_digest": pattern["record_digest"], + } + + +def _fresh_execution_blockers( + invocation: CanaryInvocationEvidence, + repo_root: Path, +) -> tuple[str, ...]: + blockers: list[str] = [] + if not invocation.invoked: + blockers.append("control_loop_not_invoked") + if invocation.control_result.get("accepted") is not True: + blockers.append("control_loop_not_accepted") + if invocation.control_result.get("status") != "PASS": + blockers.append("control_loop_status_not_pass") + if ( + not invocation.previous_revision + or not invocation.observed_revision + or invocation.observed_revision == invocation.previous_revision + ): + blockers.append("new_chain_revision_not_observed") + blockers.extend(_control_receipt_blockers(invocation, repo_root)) + return tuple(blockers) + + +def _control_receipt_blockers( + invocation: CanaryInvocationEvidence, + repo_root: Path, +) -> tuple[str, ...]: + receipt = invocation.control_receipt + if not receipt: + return ("new_control_receipt_not_observed",) + blockers: list[str] = [] + if receipt.get("schema_version") != CONTROL_LOOP_RECEIPT_SCHEMA_VERSION: + blockers.append("control_receipt_schema_mismatch") + if _text(receipt.get("receipt_id")) != invocation.control_receipt_id: + blockers.append("control_receipt_id_mismatch") + if receipt.get("accepted") is not True or receipt.get("status") != "PASS": + blockers.append("control_receipt_not_accepted_pass") + if receipt.get("control_lock_acquired") is not True: + blockers.append("control_receipt_shared_lock_missing") + if receipt.get("repo_root_digest") != _digest(str(repo_root.resolve())): + blockers.append("control_receipt_repo_root_mismatch") + progress = receipt.get("serial_progress") + if isinstance(progress, bool) or not isinstance(progress, int) or progress <= 0: + blockers.append("control_receipt_serial_progress_missing") + return tuple(blockers) + + +def _chain_evidence( + invocation: CanaryInvocationEvidence, + queue_item_id: str, + now_iso: str, +) -> tuple[Any, tuple[str, ...]]: + state = invocation.chain_state + if state.get("schema_version") != CHAIN_RESULTS_SCHEMA_VERSION: + return None, ("chain_results_schema_mismatch",) + if not resident_queue_chain_snapshot_is_canonical(state): + return None, ("chain_results_revision_invalid",) + stages = _mapping(state.get("stage_results")) + plan = plan_reddog_resident_queue_orchestration( + invocation.work_state, + chain_results={str(key): _mapping(value) for key, value in stages.items()}, + requested_queue_item_id=queue_item_id or None, + now_iso=now_iso, + ) + blockers: list[str] = [] + if plan.accepted is not True or plan.status != RESIDENT_QUEUE_ORCHESTRATION_PLAN_COMPLETE: + blockers.append("resident_chain_not_complete") + if ( + _text(state.get("queue_item_id")) != _text(plan.selected_queue_item_id) + or _text(state.get("selected_slice")) != _text(plan.selected_slice) + ): + blockers.append("chain_envelope_plan_mismatch") + previous_plan = _previous_chain_plan(invocation, queue_item_id, now_iso) + blockers.extend(_new_chain_receipt_blockers(invocation, previous_plan, plan)) + return plan, tuple(blockers) + + +def _previous_chain_plan( + invocation: CanaryInvocationEvidence, + queue_item_id: str, + now_iso: str, +) -> Any: + stages = dict(_mapping(invocation.chain_state.get("stage_results"))) + stages.pop(PATTERN_MEMORY_ADMISSION_STAGE_KEY, None) + return plan_reddog_resident_queue_orchestration( + invocation.work_state, + chain_results={str(key): _mapping(value) for key, value in stages.items()}, + requested_queue_item_id=queue_item_id or None, + now_iso=now_iso, + ) + + +def _new_chain_receipt_blockers( + invocation: CanaryInvocationEvidence, + previous_plan: Any, + final_plan: Any, +) -> tuple[str, ...]: + state = invocation.chain_state + raw = state.get("receipts") + receipts = [item for item in raw if isinstance(item, Mapping)] if isinstance(raw, list) else [] + new = [ + item for item in receipts + if (receipt_id := _text(item.get("receipt_id"))) + and receipt_id not in invocation.pre_chain_receipt_ids + ] + if not new: + return ("new_chain_store_receipt_not_observed",) + final = receipts[-1] if receipts else {} + final_id = _text(final.get("receipt_id")) + if final_id not in {_text(item.get("receipt_id")) for item in new}: + return ("final_chain_store_receipt_not_new",) + if not _chain_store_receipt_shape(final): + return ("new_chain_store_receipt_malformed",) + queue_id = _text(final_plan.selected_queue_item_id) + selected_slice = _text(final_plan.selected_slice) + if _text(final.get("queue_item_id")) != queue_id or _text(final.get("selected_slice")) != selected_slice: + return ("new_chain_store_receipt_envelope_mismatch",) + if _text(final.get("store_revision")) != invocation.observed_revision: + return ("new_chain_store_receipt_revision_mismatch",) + if not _final_receipt_transition_matches(final, previous_plan, final_plan): + return ("final_chain_store_receipt_transition_mismatch",) + return () + + +def _final_receipt_transition_matches( + receipt: Mapping[str, Any], + previous_plan: Any, + final_plan: Any, +) -> bool: + expected_id = resident_queue_chain_receipt_id( + queue_item_id=str(final_plan.selected_queue_item_id or ""), + selected_slice=str(final_plan.selected_slice or ""), + recorded_stage=PATTERN_MEMORY_ADMISSION_STAGE_KEY, + previous_plan_id=str(previous_plan.plan_id or ""), + next_plan_id=str(final_plan.plan_id or ""), + ) + return ( + previous_plan.accepted is True + and previous_plan.status == RESIDENT_QUEUE_ORCHESTRATION_PLAN_READY + and previous_plan.current_stage == PATTERN_MEMORY_ADMISSION_STAGE_KEY + and previous_plan.next_action == NEXT_QUEUE_PATTERN_MEMORY_ADMISSION_INVOKE + and receipt.get("recorded_stage") == PATTERN_MEMORY_ADMISSION_STAGE_KEY + and receipt.get("previous_plan_id") == previous_plan.plan_id + and receipt.get("next_plan_id") == final_plan.plan_id + and receipt.get("next_action") == NEXT_QUEUE_CHAIN_COMPLETE + and receipt.get("receipt_id") == expected_id + ) + + +def _chain_store_receipt_shape(receipt: Mapping[str, Any]) -> bool: + return ( + _is_sha256_digest(receipt.get("receipt_id")) + and bool(_text(receipt.get("recorded_stage"))) + and bool(_text(receipt.get("previous_plan_id"))) + and bool(_text(receipt.get("next_plan_id"))) + and bool(_text(receipt.get("next_action"))) + and bool(_text(receipt.get("store_revision"))) + ) + + +def _draft_pr_evidence(stages: Mapping[str, Any]) -> dict[str, Any]: + stage = _mapping(stages.get("verified_draft_pr_publish")) + result = _mapping(stage.get("publish_result")) + receipt = _mapping(result.get("receipt")) + url = _text(receipt.get("draft_pr_url")) + receipt_id = _text(receipt.get("receipt_id")) + blockers: list[str] = [] + accepted = ( + stage.get("decision") == QUEUE_AUTHORIZED_VERIFIED_DRAFT_PR_PUBLISH_INVOKE_ACCEPT + and result.get("decision") == VERIFIED_DRAFT_PR_PUBLISH_ACCEPT + and result.get("accepted") is True + and receipt.get("accepted") is True + and bool(receipt_id and url and url.startswith("https://github.com/") and "/pull/" in url) + ) + if not accepted: + blockers.append("accepted_verified_draft_pr_evidence_missing") + no_merge = all(item.get("no_merge_performed") is True for item in (stage, result, receipt)) + no_ready = all(item.get("no_ready_performed") is True for item in (stage, result, receipt)) + if not no_merge: + blockers.append("explicit_no_merge_evidence_missing") + if not no_ready: + blockers.append("explicit_draft_only_evidence_missing") + return { + "blockers": tuple(blockers), + "receipt_id": receipt_id, + "url": url, + "no_merge": no_merge, + "work_order_id": _text(receipt.get("work_order_id")), + "slice_name": _text(receipt.get("slice_name")), + "head": _text(receipt.get("verified_head_sha")), + } + + +def _pattern_memory_evidence( + stages: Mapping[str, Any], + repo_root: Path, + runtime_root: Path, + plan: Any, + draft: Mapping[str, Any], + worktree: Mapping[str, Any], +) -> dict[str, Any]: + stage = _mapping(stages.get("pattern_memory_admission")) + receipt = _mapping(stage.get("receipt")) + admission_id = _text(receipt.get("admission_id")) + record_id = _text(receipt.get("pattern_memory_record_id")) + record_digest = _text(receipt.get("record_digest")) + record = _load_pattern_memory_record(repo_root, runtime_root, record_id) + identity = canonical_pattern_memory_admission_identity(record, record_id) if record else ("", "") + context_matches = _pattern_record_context_matches(record, plan, draft, worktree) + accepted = ( + stage.get("decision") == QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_ACCEPT + and stage.get("pattern_memory_write_performed") is True + and stage.get("no_merge_performed") is True + and bool(admission_id and record_id and _is_sha256_digest(record_digest)) + and identity == (admission_id, record_digest) + and reddog_verified_pattern_memory_record_digest(record) == record_digest + and context_matches + ) + blockers = () if accepted else ("highest_profile_completion_evidence_missing",) + return { + "blockers": blockers, + "admission_id": admission_id, + "record_id": record_id, + "record_digest": record_digest, + } + + +def _pattern_record_context_matches( + record: Mapping[str, Any], + plan: Any, + draft: Mapping[str, Any], + worktree: Mapping[str, Any], +) -> bool: + if plan is None: + return False + work_order_id = _text(record.get("work_order_id")) + selected_slice = _text(record.get("slice_name")) + candidate_head = _text(record.get("candidate_head_sha")) + return bool( + work_order_id + and work_order_id == draft.get("work_order_id") + and selected_slice == _text(plan.selected_slice) + and selected_slice == draft.get("slice_name") + and candidate_head + and candidate_head == draft.get("head") + and candidate_head == worktree.get("head") + ) + + +def _load_pattern_memory_record( + repo_root: Path, + runtime_root: Path, + record_id: Optional[str], +) -> Mapping[str, Any]: + db_path = runtime_root / "pattern_memory.db" + if not record_id or not db_path.is_file(): + return {} + sink = build_reddog_verified_pattern_memory_sink(repo_root=repo_root, db_path=db_path) + record = sink.load_verified_outcome(record_id) if sink is not None else None + return record if isinstance(record, Mapping) else {} + + +def _worktree_evidence(stages: Mapping[str, Any], repo_root: Path) -> dict[str, Any]: + stage = _mapping(stages.get("worktree_create")) + result = _mapping(stage.get("worktree_create_result")) + raw_path = _text(result.get("worktree_path")) + if ( + stage.get("decision") != QUEUE_AUTHORIZED_WORKTREE_CREATE_INVOKE_ACCEPT + or result.get("decision") != WORKTREE_CREATE_ACCEPT + or not raw_path + ): + return {"blockers": ("isolated_worktree_evidence_missing",), "head": None} + path = Path(raw_path) + expected_head = _lineage_head(stages) + observed_head = _registered_worktree_head(repo_root, path) + valid = bool( + path.is_absolute() + and path.is_dir() + and not _is_inside(path, repo_root) + and expected_head + and observed_head == expected_head + ) + blockers = () if valid else ("isolated_worktree_evidence_missing",) + return {"blockers": blockers, "head": observed_head} + + +def _registered_worktree_head(repo_root: Path, worktree_path: Path) -> Optional[str]: + root = worktree_path.resolve() + records = _git_output(repo_root, "worktree", "list", "--porcelain") + registered = _registered_worktree_records(records) + entry = registered.get(str(root)) + if not entry or not _valid_git_worktree(root): + return None + head = _git_output(root, "rev-parse", "HEAD") + return head.strip() if head.strip() == entry else None + + +def _registered_worktree_records(output: str) -> dict[str, str]: + records: dict[str, str] = {} + path = "" + for line in output.splitlines(): + if line.startswith("worktree "): + path = str(Path(line.removeprefix("worktree ")).resolve()) + elif path and line.startswith("HEAD "): + records[path] = line.removeprefix("HEAD ").strip() + elif not line.strip(): + path = "" + return records + + +def _valid_git_worktree(path: Path) -> bool: + inside = _git_output(path, "rev-parse", "--is-inside-work-tree").strip() + top = _git_output(path, "rev-parse", "--show-toplevel").strip() + git_dir = _git_output(path, "rev-parse", "--git-dir").strip() + return inside == "true" and Path(top).resolve() == path and bool(git_dir) + + +def _git_output(cwd: Path, *args: str) -> str: + try: + result = subprocess.run( + ["git", "-C", str(cwd), *args], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return "" + return result.stdout if result.returncode == 0 else "" + + +def _lineage_head(stages: Mapping[str, Any]) -> Optional[str]: + draft = _mapping(_mapping(_mapping(stages.get("verified_draft_pr_publish")).get("publish_result")).get("receipt")) + held = _mapping(_mapping(_mapping(stages.get("held_out_regression_gate")).get("gate_result")).get("receipt")) + draft_head = _text(draft.get("verified_head_sha")) + held_head = _text(held.get("candidate_head_sha")) + return draft_head if draft_head and draft_head == held_head else None + + +def _lineage_blockers(stages: Mapping[str, Any], plan: Any) -> tuple[str, ...]: + if plan is None: + return ("key_receipt_lineage_missing",) + draft = _mapping(_mapping(_mapping(stages.get("verified_draft_pr_publish")).get("publish_result")).get("receipt")) + held = _mapping(_mapping(_mapping(stages.get("held_out_regression_gate")).get("gate_result")).get("receipt")) + pattern = _mapping(_mapping(stages.get("pattern_memory_admission")).get("receipt")) + work_ids = [_text(item.get("work_order_id")) for item in (draft, held, pattern)] + slices = [_text(item.get("slice_name")) for item in (draft, held, pattern)] + draft_head = _text(draft.get("verified_head_sha")) + held_head = _text(held.get("candidate_head_sha")) + if ( + not all(work_ids) + or len(set(work_ids)) != 1 + or not all(slices) + or len(set(slices)) != 1 + or slices[0] != _text(plan.selected_slice) + or not draft_head + or draft_head != held_head + ): + return ("key_receipt_lineage_mismatch",) + return () + + +def _mapping(value: Any) -> Mapping[str, Any]: + if hasattr(value, "to_dict"): + candidate = value.to_dict() + return candidate if isinstance(candidate, Mapping) else {} + return value if isinstance(value, Mapping) else {} + + +def _text(value: Any) -> Optional[str]: + text = str(value or "").strip() + return text or None + + +def _digest(value: Any) -> str: + raw = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, default=str) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _is_sha256_digest(value: Any) -> bool: + text = str(value or "") + suffix = text.removeprefix("sha256:") + return text.startswith("sha256:") and len(suffix) == 64 and all(ch in "0123456789abcdef" for ch in suffix) + + +def _is_inside(child: Path, parent: Path) -> bool: + child_r = child.resolve() + parent_r = parent.resolve() + return child_r == parent_r or parent_r in child_r.parents + + +__all__ = [ + "CanaryInvocationEvidence", + "chain_receipt_ids", + "evaluate_live_proof", + "read_control_receipts", + "select_new_control_receipt", +] diff --git a/modules/communication/moltbot_bridge/src/reddog_resident_queue_chain_results_store.py b/modules/communication/moltbot_bridge/src/reddog_resident_queue_chain_results_store.py index 38ab0d275..ebf88a63b 100644 --- a/modules/communication/moltbot_bridge/src/reddog_resident_queue_chain_results_store.py +++ b/modules/communication/moltbot_bridge/src/reddog_resident_queue_chain_results_store.py @@ -66,9 +66,7 @@ def commit(self, snapshot: Mapping[str, Any], *, expected_revision: Optional[str current_revision = self._state.get("revision") if current_revision != expected_revision: raise RuntimeError("revision_conflict") - committed = json.loads(json.dumps(snapshot, sort_keys=True)) - revision = _digest(committed) - committed["revision"] = revision + committed, revision = _committed_snapshot(snapshot) self._state = committed return revision @@ -88,9 +86,7 @@ def commit(self, snapshot: Mapping[str, Any], *, expected_revision: Optional[str current = self.load() if current.get("revision") != expected_revision: raise RuntimeError("revision_conflict") - committed = json.loads(json.dumps(snapshot, sort_keys=True)) - revision = _digest(committed) - committed["revision"] = revision + committed, revision = _committed_snapshot(snapshot) self.path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp(prefix=f".{self.path.name}.", suffix=".tmp", dir=str(self.path.parent)) try: @@ -182,6 +178,62 @@ def _digest(payload: Any) -> str: return "sha256:" + hashlib.sha256(raw.encode("utf-8")).hexdigest() +def resident_queue_chain_receipt_id( + *, + queue_item_id: str, + selected_slice: str, + recorded_stage: str, + previous_plan_id: str, + next_plan_id: str, +) -> str: + """Return the canonical ID binding one chain transition receipt.""" + + return _digest( + { + "queue_item_id": queue_item_id, + "selected_slice": selected_slice, + "recorded_stage": recorded_stage, + "previous_plan_id": previous_plan_id, + "next_plan_id": next_plan_id, + } + ) + + +def resident_queue_chain_snapshot_revision(snapshot: Mapping[str, Any]) -> str: + """Compute the canonical revision while excluding its final receipt witness.""" + + payload = json.loads(json.dumps(snapshot, sort_keys=True)) + payload.pop("revision", None) + receipts = payload.get("receipts") + if isinstance(receipts, list) and receipts and isinstance(receipts[-1], Mapping): + receipts[-1] = {**receipts[-1], "store_revision": None} + return _digest(payload) + + +def resident_queue_chain_snapshot_is_canonical(snapshot: Mapping[str, Any]) -> bool: + """Verify the envelope revision and newest persisted receipt witness.""" + + revision = str(snapshot.get("revision") or "") + receipts = snapshot.get("receipts") + if not revision or not isinstance(receipts, list) or not receipts: + return False + newest = _mapping(receipts[-1]) + return ( + str(newest.get("store_revision") or "") == revision + and resident_queue_chain_snapshot_revision(snapshot) == revision + ) + + +def _committed_snapshot(snapshot: Mapping[str, Any]) -> tuple[Dict[str, Any], str]: + committed = json.loads(json.dumps(snapshot, sort_keys=True)) + revision = resident_queue_chain_snapshot_revision(committed) + committed["revision"] = revision + receipts = committed.get("receipts") + if isinstance(receipts, list) and receipts and isinstance(receipts[-1], Mapping): + receipts[-1] = {**receipts[-1], "store_revision": revision} + return committed, revision + + def _mapping(value: Any) -> Mapping[str, Any]: if hasattr(value, "to_dict"): candidate = value.to_dict() @@ -296,15 +348,14 @@ def record_resident_queue_stage_result( next_plan=next_plan, ) - receipt_seed = { - "queue_item_id": previous_plan.selected_queue_item_id, - "selected_slice": previous_plan.selected_slice, - "recorded_stage": clean_stage_key, - "previous_plan_id": previous_plan.plan_id, - "next_plan_id": next_plan.plan_id, - } receipt = ResidentQueueChainResultReceipt( - receipt_id=_digest(receipt_seed), + receipt_id=resident_queue_chain_receipt_id( + queue_item_id=str(previous_plan.selected_queue_item_id or ""), + selected_slice=str(previous_plan.selected_slice or ""), + recorded_stage=clean_stage_key, + previous_plan_id=previous_plan.plan_id, + next_plan_id=next_plan.plan_id, + ), queue_item_id=str(previous_plan.selected_queue_item_id or ""), selected_slice=str(previous_plan.selected_slice or ""), recorded_stage=clean_stage_key, @@ -374,4 +425,7 @@ def record_resident_queue_stage_result( "ResidentQueueChainResultRecordResult", "ResidentQueueChainResultsStore", "record_resident_queue_stage_result", + "resident_queue_chain_receipt_id", + "resident_queue_chain_snapshot_is_canonical", + "resident_queue_chain_snapshot_revision", ] diff --git a/modules/communication/moltbot_bridge/src/reddog_resident_queue_control_lock.py b/modules/communication/moltbot_bridge/src/reddog_resident_queue_control_lock.py new file mode 100644 index 000000000..331cbc1c0 --- /dev/null +++ b/modules/communication/moltbot_bridge/src/reddog_resident_queue_control_lock.py @@ -0,0 +1,121 @@ +"""Shared interprocess lock for all resident queue control-loop callers.""" + +from __future__ import annotations + +import hashlib +import os +import tempfile +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from pathlib import Path +from typing import BinaryIO, Iterator, Mapping, Optional + + +CONTROL_LOOP_LOCK_PATH_ENV = "REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_LOCK_PATH" +_CONTROL_LOCK_HELD: ContextVar[bool] = ContextVar("reddog_control_lock_held", default=False) + + +@dataclass(frozen=True) +class ResidentQueueControlLock: + acquired: bool + path: Path + reason: str + + +def resident_queue_control_lock_path( + repo_root: Path | str, + environ: Optional[Mapping[str, str]] = None, +) -> Path: + """Resolve one shared lock path for the repository/runtime pair.""" + + env = os.environ if environ is None else environ + explicit = str(env.get(CONTROL_LOOP_LOCK_PATH_ENV) or "").strip() + if explicit: + return Path(explicit).resolve() + runtime = str(env.get("REDDOG_RESIDENT_RUNTIME_ROOT") or "").strip() + if runtime: + return (Path(runtime).resolve() / "resident_queue_control_loop.lock") + root = Path(repo_root).resolve() + digest = hashlib.sha256(str(root).encode("utf-8")).hexdigest()[:20] + return Path(tempfile.gettempdir()).resolve() / "foundups-reddog-control-locks" / f"{digest}.lock" + + +@contextmanager +def acquire_resident_queue_control_lock( + repo_root: Path | str, + environ: Optional[Mapping[str, str]] = None, +) -> Iterator[ResidentQueueControlLock]: + """Acquire a non-blocking OS advisory lock that releases on process exit.""" + + root = Path(repo_root).resolve() + path = resident_queue_control_lock_path(root, environ) + if _is_inside(path, root): + yield ResidentQueueControlLock(False, path, "control_lock_path_inside_repo") + return + handle: Optional[BinaryIO] = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + handle = path.open("a+b") + _lock_file(handle) + except (OSError, BlockingIOError): + if handle is not None: + handle.close() + yield ResidentQueueControlLock(False, path, "control_loop_already_running") + return + try: + token = _CONTROL_LOCK_HELD.set(True) + yield ResidentQueueControlLock(True, path, "ok") + finally: + _CONTROL_LOCK_HELD.reset(token) + _unlock_file(handle) + handle.close() + + +def resident_queue_control_lock_held() -> bool: + """Return whether this execution context owns the shared control lock.""" + + return _CONTROL_LOCK_HELD.get() + + +def _lock_file(handle: BinaryIO) -> None: + handle.seek(0) + if os.name == "nt": + import msvcrt + + if handle.read(1) == b"": + handle.write(b"0") + handle.flush() + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + return + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + + +def _unlock_file(handle: BinaryIO) -> None: + handle.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + return + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + +def _is_inside(child: Path, parent: Path) -> bool: + child_r = child.resolve() + parent_r = parent.resolve() + return child_r == parent_r or parent_r in child_r.parents + + +__all__ = [ + "CONTROL_LOOP_LOCK_PATH_ENV", + "ResidentQueueControlLock", + "acquire_resident_queue_control_lock", + "resident_queue_control_lock_held", + "resident_queue_control_lock_path", +] diff --git a/modules/communication/moltbot_bridge/src/reddog_verified_pattern_memory_sink.py b/modules/communication/moltbot_bridge/src/reddog_verified_pattern_memory_sink.py index bc1c11bb3..e6809f072 100644 --- a/modules/communication/moltbot_bridge/src/reddog_verified_pattern_memory_sink.py +++ b/modules/communication/moltbot_bridge/src/reddog_verified_pattern_memory_sink.py @@ -70,6 +70,12 @@ def _record_id(record: Mapping[str, Any]) -> str: return "reddog_verified_outcome_" + _digest(record).removeprefix("sha256:")[:16] +def reddog_verified_pattern_memory_record_digest(record: Mapping[str, Any]) -> str: + """Return the canonical digest used by admission receipts.""" + + return _digest(record) + + def _utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() @@ -129,6 +135,34 @@ def store_verified_outcome(self, record: Mapping[str, Any]) -> str: finally: memory.close() + def load_verified_outcome(self, record_id: str) -> Optional[Dict[str, Any]]: + """Read back one canonical RedDog outcome through PatternMemory schema.""" + + if not self.db_path.is_file() or not str(record_id or "").strip(): + return None + memory = PatternMemory(db_path=self.db_path) + try: + cursor = memory.conn.cursor() + cursor.execute( + "SELECT execution_id, agent, success, output_result " + "FROM skill_outcomes WHERE execution_id = ? LIMIT 1", + (record_id,), + ) + row = cursor.fetchone() + finally: + memory.close() + if row is None or row["agent"] != self.agent or row["success"] != 1: + return None + try: + payload = json.loads(row["output_result"]) + except (json.JSONDecodeError, TypeError): + return None + if not isinstance(payload, dict) or _record_id(payload) != record_id: + return None + if payload.get("record_type") != "reddog_verified_recursive_improvement_outcome": + return None + return payload + def build_reddog_verified_pattern_memory_sink( *, @@ -156,4 +190,5 @@ def build_reddog_verified_pattern_memory_sink( "REDDOG_VERIFIED_PATTERN_MEMORY_SINK_READY", "RedDogVerifiedPatternMemorySink", "build_reddog_verified_pattern_memory_sink", + "reddog_verified_pattern_memory_record_digest", ] diff --git a/modules/communication/moltbot_bridge/src/reddog_wre_queue_authorized_pattern_memory_admission_invoke.py b/modules/communication/moltbot_bridge/src/reddog_wre_queue_authorized_pattern_memory_admission_invoke.py index ccafb5d4d..8a2db4221 100644 --- a/modules/communication/moltbot_bridge/src/reddog_wre_queue_authorized_pattern_memory_admission_invoke.py +++ b/modules/communication/moltbot_bridge/src/reddog_wre_queue_authorized_pattern_memory_admission_invoke.py @@ -242,6 +242,16 @@ def _build_receipt( ) +def canonical_pattern_memory_admission_identity( + record: Mapping[str, Any], + record_id: str, +) -> tuple[str, str]: + """Return the admission ID and record digest for a persisted outcome.""" + + receipt = _build_receipt(record=record, record_id=record_id, reasons=[]) + return receipt.admission_id, receipt.record_digest + + def _reject( reasons: Sequence[str], *, @@ -352,6 +362,7 @@ def invoke_reddog_wre_queue_authorized_pattern_memory_admission( __all__ = [ "PatternMemoryAdmissionSink", + "canonical_pattern_memory_admission_identity", "QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_ACCEPT", "QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_REJECT", "QueueAuthorizedPatternMemoryAdmissionInvokeReason", diff --git a/modules/communication/moltbot_bridge/tests/README.md b/modules/communication/moltbot_bridge/tests/README.md index 5a36cf173..6dd0a72cf 100644 --- a/modules/communication/moltbot_bridge/tests/README.md +++ b/modules/communication/moltbot_bridge/tests/README.md @@ -32,3 +32,15 @@ Focused RedDog WRE operational spine: ```powershell python -m pytest modules/communication/moltbot_bridge/tests/test_reddog_wre_operational_spine.py modules/communication/moltbot_bridge/tests/test_reddog_wre_worktree_create.py modules/communication/moltbot_bridge/tests/test_reddog_wre_execution_valve.py modules/communication/moltbot_bridge/tests/test_reddog_wre_executor_dryrun.py modules/communication/moltbot_bridge/tests/test_reddog_work_order_runtime_invocation.py -q ``` + +Focused resident live-canary harness: +```powershell +python -m pytest modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary.py -q +``` + +This suite uses injected readiness/control-loop probes, a temporary local Git +repository with a registered worktree, the atomic chain store and planner, and +a temporary PatternMemory SQLite database. The draft-PR runner remains an +injected no-network test double. One bounded Python subprocess proves +interprocess lock exclusion. It does not start a signer, call OpenRouter, push +a branch, or create a PR. diff --git a/modules/communication/moltbot_bridge/tests/TestModLog.md b/modules/communication/moltbot_bridge/tests/TestModLog.md index f05942d8f..77b9615b9 100644 --- a/modules/communication/moltbot_bridge/tests/TestModLog.md +++ b/modules/communication/moltbot_bridge/tests/TestModLog.md @@ -1,3 +1,44 @@ +## 2026-07-18: REDDOG_RESIDENT_LIVE_CANARY_PHASE1 + +**Files**: `test_reddog_resident_live_canary.py` (NEW) + +**Coverage**: + +- Readiness-only mode never invokes the control loop and never serializes the + supplied secret value. +- Linux, outside-repo state/receipt, required JSON artifacts, signer socket, + Git/GitHub readiness, OpenRouter key presence, and exact execution + confirmation fail closed. +- Real v1 control receipts and exact chain envelopes are used in the positive + proof fixture; false schema, acceptance, status, lock, repository, progress, + revision, envelope, new-receipt, and lineage fixtures fail closed. +- Accepted worktree invoke/create decisions plus an existing external Git + worktree and all durable PatternMemory IDs are required. +- The positive proof now advances the atomic chain store through the real + planner, creates a registered Git worktree, uses production draft/gate + builders, and performs a real PatternMemory SQLite admission/readback. +- Store round-trip tests prove the persisted newest receipt witness equals the + canonically recomputable envelope revision; forged witnesses fail closed. +- Exact terminal-receipt adversaries replace the nonempty stage, previous/final + plan IDs, and stop action; completion now requires the canonical PatternMemory + terminal transition and receipt ID. +- Registered/unregistered worktrees, invalid gitdirs, HEAD mismatches, missing + PatternMemory rows, and forged admission identities are adversarial cases. +- Digest-valid PatternMemory rows with a wrong work order, selected slice, or + candidate HEAD fail direct DB-to-plan/draft/registered-worktree binding. +- Split canonical integration support from the focused test module; WSP 62 + enforces the 675-line ceiling on both test files and all canary production + files, plus the 50-line production-function ceiling. +- Same-process and second-process tests prove the shared non-blocking control + lock prevents a competing main loop from reaching queue stages. +- Reserved runtime receipt collisions and inside-repository paths fail before + execution; canonical and external paths remain accepted. +- AST coverage enforces both WSP 62's 675-line communication file limit and + 50-line production-function limit across the split canary modules. + +**Truth boundary**: Tests use injected probes and do not perform live side +effects. The production live canary remains unexecuted. + ## 2026-07-18: REDDOG_HOLOINDEX_QUERY_OWNER_BOUNDARY_POC_PHASE1 **Files**: test_reddog_holoindex_query_boundary.py (NEW), diff --git a/modules/communication/moltbot_bridge/tests/reddog_resident_live_canary_test_support.py b/modules/communication/moltbot_bridge/tests/reddog_resident_live_canary_test_support.py new file mode 100644 index 000000000..dcf21c943 --- /dev/null +++ b/modules/communication/moltbot_bridge/tests/reddog_resident_live_canary_test_support.py @@ -0,0 +1,314 @@ +"""Canonical local integration support for resident live-canary tests.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +from modules.communication.moltbot_bridge.src.reddog_resident_control_loop_receipt_store import ( + build_resident_control_loop_receipt, +) +from modules.communication.moltbot_bridge.src.reddog_resident_live_canary import ( + LIVE_CANARY_CONFIRMATION, + REQUIRED_JSON_ARTIFACTS, + run_reddog_resident_live_canary, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_chain_results_store import ( + CHAIN_RESULTS_SCHEMA_VERSION, + AtomicJsonResidentQueueChainResultsStore, + record_resident_queue_stage_result, + resident_queue_chain_receipt_id, + resident_queue_chain_snapshot_is_canonical, + resident_queue_chain_snapshot_revision, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_orchestration_plan import ( + NEXT_QUEUE_CHAIN_COMPLETE, + _CHAIN, + plan_reddog_resident_queue_orchestration, +) +from modules.communication.moltbot_bridge.src.reddog_verified_pattern_memory_sink import ( + build_reddog_verified_pattern_memory_sink, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_held_out_regression_gate_invoke import ( + invoke_reddog_wre_queue_authorized_held_out_regression_gate, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_pattern_memory_admission_invoke import ( + QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_ACCEPT, + invoke_reddog_wre_queue_authorized_pattern_memory_admission, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_verified_draft_pr_publish_invoke import ( + invoke_reddog_wre_queue_authorized_verified_draft_pr_publish, +) +from modules.communication.moltbot_bridge.src.reddog_wre_worktree_create import WORKTREE_CREATE_ACCEPT +from modules.communication.moltbot_bridge.tests.test_reddog_resident_queue_serial_loop import _snapshot + + +QUEUE_ID = "queue-1" +SLICE_NAME = "REDDOG_TEST_SLICE_PHASE1" +WORK_ORDER_ID = "work-order-1" +NOW = "2026-07-14T00:00:00+00:00" + + +def _git(cwd: Path, *args: str) -> str: + result = subprocess.run( + ["git", "-C", str(cwd), *args], capture_output=True, text=True, check=True + ) + return result.stdout.strip() + + +def _roots(tmp_path: Path) -> tuple[Path, Path]: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init") + _git(repo, "config", "user.email", "reddog-canary@example.invalid") + _git(repo, "config", "user.name", "RedDog Canary Test") + (repo / "seed.txt").write_text("seed\n", encoding="utf-8") + _git(repo, "add", "seed.txt") + _git(repo, "commit", "-m", "test: seed live canary repo") + runtime = tmp_path / "runtime" + runtime.mkdir() + for filename in REQUIRED_JSON_ARTIFACTS: + (runtime / filename).write_text(json.dumps({"kind": filename}), encoding="utf-8") + return repo, runtime + + +def _kwargs(repo: Path, runtime: Path) -> dict[str, object]: + return { + "repo_root": repo, + "runtime_root": runtime, + "environ": {"OPENROUTER_API_KEY": "must-never-be-serialized"}, + "platform_name": "linux", + "command_resolver": lambda command: f"/usr/bin/{command}", + "command_probe": lambda argv, cwd: True, + "socket_probe": lambda path: path.name == "reddog_signer.sock", + } + + +class _DraftRunner: + def push_branch(self, *, worktree_path: Path, branch_name: str): + return {"ok": True, "returncode": 0, "stdout": "", "stderr": ""} + + def create_draft_pr(self, **_: object) -> str: + return "https://github.com/FOUNDUPS/Foundups-Agent/pull/9999" + + +def _create_registered_worktree(repo: Path, runtime: Path) -> tuple[Path, str]: + isolated = runtime.parent / "isolated-worker" + _git(repo, "worktree", "add", "--detach", str(isolated), "HEAD") + return isolated.resolve(), _git(isolated, "rev-parse", "HEAD") + + +def _draft_stage(isolated: Path, head: str) -> dict[str, object]: + verifier = { + "decision": "QUEUE_AUTHORIZED_SLICE_VERIFIER_INVOKE_ACCEPT", + "verifier_result": { + "decision": "AUTONOMOUS_SLICE_VERIFIER_ACCEPT", + "accepted": True, + "receipt": { + "receipt_id": "wre_slice_verify_canary", + "work_order_id": WORK_ORDER_ID, + "slice_name": SLICE_NAME, + "head_sha": head, + "changed_paths": ["seed.txt"], + }, + }, + } + request = { + "work_order_id": WORK_ORDER_ID, + "pre_publish_branch_head_sha": head, + "branch_name": "feat/reddog-live-canary-test", + "base_branch": "main", + "pr_title": "test: resident live canary evidence", + "pr_body": "Canonical test-only draft receipt.", + "worktree_path": str(isolated), + "draft_pr_only": True, + "mark_ready": False, + "merge": False, + } + return invoke_reddog_wre_queue_authorized_verified_draft_pr_publish( + explicit_queue_authorized_verified_draft_pr_publish_requested=True, + queue_slice_verifier_result=verifier, + publish_request=request, + runner=_DraftRunner(), + ).to_dict() + + +def _held_out_stage(head: str) -> dict[str, object]: + ratchet = { + "decision": "QUEUE_AUTHORIZED_VERIFIED_OUTCOME_RATCHET_INVOKE_ACCEPT", + "ratchet_result": { + "decision": "OUTCOME_RATCHET_RECORDED", + "accepted": True, + "receipt": { + "ratchet_id": "outcome_ratchet_canary", + "work_order_id": WORK_ORDER_ID, + "slice_name": SLICE_NAME, + "verifier_receipt_id": "wre_slice_verify_canary", + "pattern_memory_eligible": True, + }, + }, + } + request = { + "work_order_id": WORK_ORDER_ID, + "slice_name": SLICE_NAME, + "worker_id": "reddog-0102", + "enable_pattern_memory_admission": True, + "improvement_job": {"job_id": "imp_live_canary", "status": "pending", "dry_run": True}, + "verification_result": { + "accepted": True, + "decision": "AUTONOMOUS_SLICE_VERIFIER_ACCEPT", + "receipt": { + "receipt_id": "wre_slice_verify_canary", + "work_order_id": WORK_ORDER_ID, + "slice_name": SLICE_NAME, + "head_sha": head, + }, + }, + "held_out_regression": { + "suite_id": "heldout-live-canary", + "is_held_out": True, + "independent": True, + "generated_by_author": False, + "evidence_author_id": "verifier-0102", + "passed": True, + "test_count": 1, + "failure_count": 0, + "suite_digest": "sha256:" + "1" * 64, + "baseline_digest": "sha256:" + "2" * 64, + "candidate_digest": "sha256:" + "3" * 64, + "candidate_head_sha": head, + }, + "holoindex_evidence": { + "index_gap_detected": False, + "holoindex_freshness_receipt_digest": "sha256:" + "4" * 64, + }, + } + return invoke_reddog_wre_queue_authorized_held_out_regression_gate( + explicit_queue_authorized_held_out_regression_gate_requested=True, + queue_verified_outcome_ratchet_result=ratchet, + held_out_gate_request=request, + ).to_dict() + + +def _stage_results(repo: Path, runtime: Path) -> dict[str, dict[str, object]]: + isolated, head = _create_registered_worktree(repo, runtime) + stages = {stage.key: {stage.status_field: stage.accepted_value} for stage in _CHAIN} + stages["worktree_create"].update( + worktree_create_result={"decision": WORKTREE_CREATE_ACCEPT, "worktree_path": str(isolated)} + ) + stages["verified_draft_pr_publish"] = _draft_stage(isolated, head) + stages["held_out_regression_gate"] = _held_out_stage(head) + stages.pop("pattern_memory_admission") + return stages + + +def _write_pre_state(repo: Path, runtime: Path) -> dict[str, object]: + (runtime / "authoritative_work_state.json").write_text(json.dumps(_snapshot()), encoding="utf-8") + store = AtomicJsonResidentQueueChainResultsStore(runtime / "resident_queue_chain_results.json") + for stage_key, stage_result in _stage_results(repo, runtime).items(): + result = record_resident_queue_stage_result( + work_state_snapshot=_snapshot(), store=store, stage_key=stage_key, + stage_result=stage_result, now_iso=NOW, requested_queue_item_id=QUEUE_ID, + ) + assert result.accepted is True + state = dict(store.load()) + assert resident_queue_chain_snapshot_is_canonical(state) is True + return state + + +def _control_receipt(repo: Path, **changes: object) -> dict[str, object]: + receipt = build_resident_control_loop_receipt( + result={ + "accepted": True, "status": "PASS", "rounds": 1, "serial_progress": 1, + "claim_progress": 0, "control_lock_acquired": True, + "receipt_ids": ("serial-receipt-1",), "rejection_reasons": (), + }, + repo_root=repo, + created_at="2026-07-14T00:00:00Z", + ).to_dict() + receipt.update(changes) + return receipt + + +def _canonicalize_terminal_receipt(chain: dict[str, object]) -> None: + stages = dict(chain["stage_results"]) + final_plan = plan_reddog_resident_queue_orchestration( + _snapshot(), chain_results=stages, requested_queue_item_id=QUEUE_ID, now_iso=NOW + ) + previous_stages = dict(stages) + previous_stages.pop("pattern_memory_admission") + previous_plan = plan_reddog_resident_queue_orchestration( + _snapshot(), chain_results=previous_stages, + requested_queue_item_id=QUEUE_ID, now_iso=NOW, + ) + receipt = chain["receipts"][-1] + receipt.update( + recorded_stage="pattern_memory_admission", + previous_plan_id=previous_plan.plan_id, + next_plan_id=final_plan.plan_id, + next_action=NEXT_QUEUE_CHAIN_COMPLETE, + receipt_id=resident_queue_chain_receipt_id( + queue_item_id=QUEUE_ID, + selected_slice=SLICE_NAME, + recorded_stage="pattern_memory_admission", + previous_plan_id=previous_plan.plan_id, + next_plan_id=final_plan.plan_id, + ), + ) + + +def _runner( + repo: Path, + runtime: Path, + *, + chain_mutator=None, + receipt_changes: dict[str, object] | None = None, + result_receipt_id: str | None = None, + rebind_after_mutation: bool = True, + pattern_db_mutator=None, +): + def run(_: Path) -> dict[str, object]: + store = AtomicJsonResidentQueueChainResultsStore(runtime / "resident_queue_chain_results.json") + chain = dict(store.load()) + held = chain["stage_results"]["held_out_regression_gate"] + sink = build_reddog_verified_pattern_memory_sink(repo_root=repo, db_path=runtime / "pattern_memory.db") + pattern = invoke_reddog_wre_queue_authorized_pattern_memory_admission( + explicit_queue_authorized_pattern_memory_admission_requested=True, + queue_held_out_gate_result=held, + admission_request={"work_order_id": WORK_ORDER_ID}, + sink=sink, + ) + assert pattern.decision == QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_ACCEPT + recorded = record_resident_queue_stage_result( + work_state_snapshot=_snapshot(), store=store, stage_key="pattern_memory_admission", + stage_result=pattern.to_dict(), now_iso=NOW, requested_queue_item_id=QUEUE_ID, + ) + assert recorded.accepted is True + if pattern_db_mutator: + pattern_db_mutator(runtime / "pattern_memory.db", pattern.receipt.pattern_memory_record_id) + chain = dict(store.load()) + if chain_mutator: + chain_mutator(chain) + if rebind_after_mutation and chain.get("schema_version") == CHAIN_RESULTS_SCHEMA_VERSION: + revision = resident_queue_chain_snapshot_revision(chain) + chain["revision"] = revision + if chain.get("receipts"): + chain["receipts"][-1]["store_revision"] = revision + (runtime / "resident_queue_chain_results.json").write_text(json.dumps(chain), encoding="utf-8") + control = _control_receipt(repo, **(receipt_changes or {})) + (runtime / "resident_queue_control_loop_receipts.jsonl").write_text( + json.dumps(control) + "\n", encoding="utf-8" + ) + return {"accepted": True, "status": "PASS", "receipt_id": result_receipt_id or control["receipt_id"]} + + return run + + +def _execute(repo: Path, runtime: Path, **runner_kwargs: object): + _write_pre_state(repo, runtime) + return run_reddog_resident_live_canary( + **_kwargs(repo, runtime), execute=True, confirmation=LIVE_CANARY_CONFIRMATION, + queue_item_id=QUEUE_ID, control_loop_runner=_runner(repo, runtime, **runner_kwargs), + now=lambda: __import__("datetime").datetime.fromisoformat(NOW), + ) diff --git a/modules/communication/moltbot_bridge/tests/test_reddog_main_openclaw_signed_worker_claim_loop_preflight.py b/modules/communication/moltbot_bridge/tests/test_reddog_main_openclaw_signed_worker_claim_loop_preflight.py index 24ce4d9a4..76365bafb 100644 --- a/modules/communication/moltbot_bridge/tests/test_reddog_main_openclaw_signed_worker_claim_loop_preflight.py +++ b/modules/communication/moltbot_bridge/tests/test_reddog_main_openclaw_signed_worker_claim_loop_preflight.py @@ -1159,6 +1159,8 @@ def _serve_signer() -> None: == main.run_reddog_resident_queue_control_loop_preflight.last_result["receipt_id"] ) assert control_receipt["receipt_ids"][0].startswith("signed_worker_task_execution_") + assert control_receipt["control_lock_acquired"] is True + assert main.run_reddog_resident_queue_control_loop_preflight.last_result["control_lock_acquired"] is True assert "REDDOG_WORK_ORDERS_PATH" not in os.environ pending = [ @@ -1469,6 +1471,8 @@ def _serve_signer_cli() -> None: == main.run_reddog_resident_queue_control_loop_preflight.last_result["receipt_id"] ) assert control_receipt["receipt_ids"][0].startswith("signed_worker_task_execution_") + assert control_receipt["control_lock_acquired"] is True + assert main.run_reddog_resident_queue_control_loop_preflight.last_result["control_lock_acquired"] is True assert calls stored = json.loads(chain.read_text(encoding="utf-8")) assert stored["stage_results"]["worker_dispatch_runtime"]["accepted"] is True diff --git a/modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary.py b/modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary.py new file mode 100644 index 000000000..54824012f --- /dev/null +++ b/modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary.py @@ -0,0 +1,351 @@ +from __future__ import annotations + +import ast +import json +import os +import subprocess +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +from modules.communication.moltbot_bridge.src.reddog_resident_control_loop_receipt_store import ( + CONTROL_LOOP_RECEIPT_SCHEMA_VERSION, + build_resident_control_loop_receipt, +) +from modules.communication.moltbot_bridge.src.reddog_resident_live_canary import ( + LIVE_CANARY_BLOCKED, + LIVE_CANARY_CONFIRMATION, + LIVE_CANARY_PROOF_COMPLETE, + LIVE_CANARY_PROOF_INCOMPLETE, + LIVE_CANARY_READY, + REQUIRED_JSON_ARTIFACTS, + run_reddog_resident_live_canary, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_binding_profile import ( + PROFILE_SIGNED_0102_BOUNDED_CODE_FUSION_WORKTREE_DRAFT_PR_PATTERN_MEMORY, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_chain_results_store import ( + CHAIN_RESULTS_SCHEMA_VERSION, + AtomicJsonResidentQueueChainResultsStore, + record_resident_queue_stage_result, + resident_queue_chain_snapshot_is_canonical, + resident_queue_chain_snapshot_revision, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_control_lock import ( + CONTROL_LOOP_LOCK_PATH_ENV, + acquire_resident_queue_control_lock, +) +from modules.communication.moltbot_bridge.src.reddog_resident_queue_orchestration_plan import ( + RESIDENT_QUEUE_ORCHESTRATION_PLAN_COMPLETE, + _CHAIN, + plan_reddog_resident_queue_orchestration, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_pattern_memory_admission_invoke import ( + QUEUE_AUTHORIZED_PATTERN_MEMORY_ADMISSION_INVOKE_ACCEPT, + canonical_pattern_memory_admission_identity, + invoke_reddog_wre_queue_authorized_pattern_memory_admission, +) +from modules.communication.moltbot_bridge.src.reddog_verified_pattern_memory_sink import ( + build_reddog_verified_pattern_memory_sink, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_held_out_regression_gate_invoke import ( + invoke_reddog_wre_queue_authorized_held_out_regression_gate, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_verified_draft_pr_publish_invoke import ( + QUEUE_AUTHORIZED_VERIFIED_DRAFT_PR_PUBLISH_INVOKE_ACCEPT, + invoke_reddog_wre_queue_authorized_verified_draft_pr_publish, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_worktree_create_invoke import ( + QUEUE_AUTHORIZED_WORKTREE_CREATE_INVOKE_ACCEPT, +) +from modules.communication.moltbot_bridge.src.reddog_wre_worktree_create import ( + WORKTREE_CREATE_ACCEPT, +) +from modules.communication.moltbot_bridge.tests.test_reddog_resident_queue_serial_loop import ( + _snapshot, +) +from modules.infrastructure.wre_core.src.reddog_verified_draft_pr_publish import ( + VERIFIED_DRAFT_PR_PUBLISH_ACCEPT, +) +from modules.infrastructure.wre_core.src.pattern_memory import PatternMemory + + +SRC_ROOT = Path(__file__).resolve().parents[1] / "src" +PRODUCTION_PATHS = ( + SRC_ROOT / "reddog_resident_live_canary.py", + SRC_ROOT / "reddog_resident_live_canary_evidence.py", + SRC_ROOT / "reddog_resident_queue_control_lock.py", + SRC_ROOT / "reddog_resident_control_loop_receipt_store.py", +) +COMMUNICATION_TEST_PATHS = ( + Path(__file__), + Path(__file__).with_name("reddog_resident_live_canary_test_support.py"), + Path(__file__).with_name("test_reddog_resident_live_canary_integration.py"), +) +from modules.communication.moltbot_bridge.tests.reddog_resident_live_canary_test_support import ( + NOW, + QUEUE_ID, + SLICE_NAME, + _canonicalize_terminal_receipt, + _control_receipt, + _execute, + _kwargs, + _roots, + _runner, + _write_pre_state, +) + + +def test_readiness_is_non_executing_and_does_not_serialize_secret(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + receipt = run_reddog_resident_live_canary(**_kwargs(repo, runtime)) + + assert receipt.status == LIVE_CANARY_READY + assert receipt.ready_for_execution is True + assert receipt.execution_invoked is False + assert receipt.live_proof_complete is False + serialized = (runtime / "live_canary_receipt.json").read_text(encoding="utf-8") + assert "must-never-be-serialized" not in serialized + assert json.loads(serialized)["secret_values_serialized"] is False + + +def test_windows_plane_is_truthfully_blocked(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + args = _kwargs(repo, runtime) + args["platform_name"] = "win32" + receipt = run_reddog_resident_live_canary(**args) + + assert receipt.status == LIVE_CANARY_BLOCKED + assert "linux_execution_plane_required" in receipt.blockers + + +def test_execute_requires_exact_confirmation_and_never_calls_runner(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + called = False + + def runner(_: Path) -> dict[str, object]: + nonlocal called + called = True + return {"accepted": True} + + receipt = run_reddog_resident_live_canary( + **_kwargs(repo, runtime), execute=True, confirmation="wrong", control_loop_runner=runner + ) + assert receipt.status == LIVE_CANARY_BLOCKED + assert receipt.execution_invoked is False + assert called is False + assert "explicit_execution_confirmation_missing" in receipt.blockers + + +@pytest.mark.parametrize( + ("changes", "blocker"), + [ + ({"schema_version": "wrong"}, "control_receipt_schema_mismatch"), + ({"accepted": False}, "control_receipt_not_accepted_pass"), + ({"status": "WARN"}, "control_receipt_not_accepted_pass"), + ({"control_lock_acquired": False}, "control_receipt_shared_lock_missing"), + ({"repo_root_digest": "wrong"}, "control_receipt_repo_root_mismatch"), + ({"serial_progress": 0}, "control_receipt_serial_progress_missing"), + ], +) +def test_false_control_receipts_cannot_complete_proof( + tmp_path: Path, changes: dict[str, object], blocker: str +) -> None: + repo, runtime = _roots(tmp_path) + receipt = _execute(repo, runtime, receipt_changes=changes) + + assert receipt.status == LIVE_CANARY_PROOF_INCOMPLETE + assert blocker in receipt.blockers + + +def test_runner_result_must_match_one_new_persisted_control_receipt(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + receipt = _execute(repo, runtime, result_receipt_id="different-receipt") + + assert receipt.live_proof_complete is False + assert "new_control_receipt_not_observed" in receipt.blockers + + +def test_preseeded_complete_chain_cannot_be_relabelled_as_new_live_proof(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + _write_pre_state(repo, runtime) + _runner(repo, runtime)(repo) + (runtime / "resident_queue_control_loop_receipts.jsonl").unlink() + control = _control_receipt(repo) + + def runner(_: Path) -> dict[str, object]: + (runtime / "resident_queue_control_loop_receipts.jsonl").write_text( + json.dumps(control) + "\n", encoding="utf-8" + ) + return {"accepted": True, "status": "PASS", "receipt_id": control["receipt_id"]} + + receipt = run_reddog_resident_live_canary( + **_kwargs(repo, runtime), execute=True, confirmation=LIVE_CANARY_CONFIRMATION, + queue_item_id=QUEUE_ID, control_loop_runner=runner, + now=lambda: __import__("datetime").datetime.fromisoformat(NOW), + ) + assert receipt.live_proof_complete is False + assert "new_chain_revision_not_observed" in receipt.blockers + + +def test_live_proof_requires_a_pre_invocation_chain_revision(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + pre = _write_pre_state(repo, runtime) + pre.pop("revision") + (runtime / "resident_queue_chain_results.json").write_text(json.dumps(pre), encoding="utf-8") + receipt = run_reddog_resident_live_canary( + **_kwargs(repo, runtime), execute=True, confirmation=LIVE_CANARY_CONFIRMATION, + queue_item_id=QUEUE_ID, control_loop_runner=_runner(repo, runtime), + now=lambda: __import__("datetime").datetime.fromisoformat(NOW), + ) + assert receipt.live_proof_complete is False + assert "new_chain_revision_not_observed" in receipt.blockers + + +def test_receipt_path_allows_only_canonical_name_inside_runtime(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + canonical = runtime / "live_canary_receipt.json" + receipt = run_reddog_resident_live_canary( + **_kwargs(repo, runtime), receipt_path=canonical + ) + assert receipt.status == LIVE_CANARY_READY + assert canonical.is_file() + + +@pytest.mark.parametrize( + "relative", + [ + "resident_queue_chain_results.json", + "resident_queue_control_loop_receipts.jsonl", + "authoritative_work_state.json", + "nested/live_canary_receipt.json", + ], +) +def test_receipt_path_rejects_runtime_reserved_and_collision_paths( + tmp_path: Path, relative: str +) -> None: + repo, runtime = _roots(tmp_path) + with pytest.raises(ValueError, match="receipt_path_reserved_or_collision"): + run_reddog_resident_live_canary( + **_kwargs(repo, runtime), receipt_path=runtime / relative + ) + + +def test_receipt_path_outside_repo_and_runtime_is_allowed(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + external = tmp_path / "receipts" / "canary.json" + receipt = run_reddog_resident_live_canary( + **_kwargs(repo, runtime), receipt_path=external + ) + assert receipt.status == LIVE_CANARY_READY + assert external.is_file() + + +def test_canonical_receipt_symlink_collision_is_rejected(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + target = runtime / "resident_queue_chain_results.json" + target.write_text("preserve", encoding="utf-8") + canonical = runtime / "live_canary_receipt.json" + try: + canonical.symlink_to(target) + except OSError: + pytest.skip("symlink creation unavailable") + with pytest.raises(ValueError, match="receipt_path_reserved_or_collision"): + run_reddog_resident_live_canary(**_kwargs(repo, runtime)) + assert target.read_text(encoding="utf-8") == "preserve" + + +def test_receipt_path_inside_repo_is_rejected(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + with pytest.raises(ValueError, match="receipt_path_inside_repo"): + run_reddog_resident_live_canary( + **_kwargs(repo, runtime), receipt_path=repo / "receipt.json" + ) + + +def test_shared_control_lock_blocks_competing_main_control_loop(tmp_path: Path) -> None: + import main + + repo, runtime = _roots(tmp_path) + env = {"REDDOG_RESIDENT_QUEUE_CONTROL_LOOP_LOCK_PATH": str(runtime / "control.lock")} + with patch.dict(os.environ, env, clear=True): + with acquire_resident_queue_control_lock(repo) as held: + assert held.acquired is True + with patch.object( + main, "run_reddog_resident_queue_serial_loop_preflight" + ) as serial_loop: + assert main.run_reddog_resident_queue_control_loop_preflight(repo) is False + serial_loop.assert_not_called() + assert main.run_reddog_resident_queue_control_loop_preflight.last_result["status"] == "CONTROL_LOOP_LOCKED" + + +def test_shared_control_lock_excludes_a_competing_process(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + lock_path = runtime / "interprocess.lock" + code = ( + "from modules.communication.moltbot_bridge.src.reddog_resident_queue_control_lock " + "import acquire_resident_queue_control_lock, CONTROL_LOOP_LOCK_PATH_ENV\n" + f"with acquire_resident_queue_control_lock(r'{repo}', " + f"{{CONTROL_LOOP_LOCK_PATH_ENV: r'{lock_path}'}}) as lock:\n" + " print(str(lock.acquired), flush=True)\n" + " input()\n" + ) + child = subprocess.Popen( + [sys.executable, "-c", code], stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, cwd=str(Path(__file__).resolve().parents[4]), + ) + try: + assert child.stdout is not None + assert child.stdout.readline().strip() == "True" + with acquire_resident_queue_control_lock( + repo, {CONTROL_LOOP_LOCK_PATH_ENV: str(lock_path)} + ) as competing: + assert competing.acquired is False + assert competing.reason == "control_loop_already_running" + finally: + if child.stdin is not None: + child.stdin.write("\n") + child.stdin.flush() + child.wait(timeout=10) + + +def test_environment_is_restored_after_control_loop(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + before = os.environ.get("REDDOG_RESIDENT_QUEUE_BINDING_PROFILE") + run_reddog_resident_live_canary( + **_kwargs(repo, runtime), execute=True, confirmation=LIVE_CANARY_CONFIRMATION, + control_loop_runner=lambda _: {"accepted": False, "status": "TEST_STOP"}, + ) + assert os.environ.get("REDDOG_RESIDENT_QUEUE_BINDING_PROFILE") == before + + +def test_actual_resident_chain_schema_and_constants_reach_complete_plan() -> None: + stages = {stage.key: {stage.status_field: stage.accepted_value} for stage in _CHAIN} + plan = plan_reddog_resident_queue_orchestration( + _snapshot(), chain_results=stages, requested_queue_item_id=QUEUE_ID, + now_iso="2026-07-14T00:00:00+00:00", + ) + assert CONTROL_LOOP_RECEIPT_SCHEMA_VERSION == "reddog_resident_control_loop_receipt.v1" + assert CHAIN_RESULTS_SCHEMA_VERSION == "reddog_resident_queue_chain_results.v1" + assert plan.status == RESIDENT_QUEUE_ORCHESTRATION_PLAN_COMPLETE + assert len(plan.accepted_stages) == len(_CHAIN) + 1 + + +def test_live_canary_production_files_and_functions_follow_wsp62() -> None: + oversized_files = { + path.name: len(path.read_text(encoding="utf-8").splitlines()) + for path in (*PRODUCTION_PATHS, *COMMUNICATION_TEST_PATHS) + if len(path.read_text(encoding="utf-8").splitlines()) > 675 + } + oversized_functions: dict[str, int] = {} + for path in PRODUCTION_PATHS: + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.end_lineno: + lines = node.end_lineno - node.lineno + 1 + if lines > 50: + oversized_functions[f"{path.name}:{node.name}"] = lines + assert oversized_files == {} + assert oversized_functions == {} diff --git a/modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary_integration.py b/modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary_integration.py new file mode 100644 index 000000000..7c45af495 --- /dev/null +++ b/modules/communication/moltbot_bridge/tests/test_reddog_resident_live_canary_integration.py @@ -0,0 +1,200 @@ +"""Canonical store, Git, and PatternMemory integration proofs for live canary.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from modules.communication.moltbot_bridge.src.reddog_resident_live_canary import ( + LIVE_CANARY_PROOF_COMPLETE, +) +from modules.communication.moltbot_bridge.src.reddog_verified_pattern_memory_sink import ( + build_reddog_verified_pattern_memory_sink, +) +from modules.communication.moltbot_bridge.src.reddog_wre_queue_authorized_pattern_memory_admission_invoke import ( + canonical_pattern_memory_admission_identity, +) +from modules.communication.moltbot_bridge.tests.reddog_resident_live_canary_test_support import ( + _canonicalize_terminal_receipt, + _execute, + _roots, +) +from modules.infrastructure.wre_core.src.pattern_memory import PatternMemory + + +def test_live_proof_uses_canonical_store_git_and_pattern_memory(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + receipt = _execute(repo, runtime) + + assert receipt.status == LIVE_CANARY_PROOF_COMPLETE + assert receipt.live_proof_complete is True + assert receipt.previous_chain_revision != receipt.observed_chain_revision + assert receipt.verified_draft_pr_receipt_id.startswith("verified_draft_pr_") + assert receipt.pattern_memory_admission_id.startswith("pattern_memory_admission_") + assert receipt.pattern_memory_record_id.startswith("reddog_verified_outcome_") + assert receipt.pattern_memory_record_digest.startswith("sha256:") + assert receipt.isolated_worktree_observed is True + memory = PatternMemory(db_path=runtime / "pattern_memory.db") + try: + row = memory.conn.execute( + "SELECT execution_id FROM skill_outcomes WHERE execution_id = ?", + (receipt.pattern_memory_record_id,), + ).fetchone() + finally: + memory.close() + assert row is not None + + +@pytest.mark.parametrize( + ("mutator", "blocker"), + [ + (lambda chain: chain.update(schema_version="wrong"), "chain_results_schema_mismatch"), + (lambda chain: chain.update(queue_item_id="other"), "chain_envelope_plan_mismatch"), + (lambda chain: chain.update(receipts=chain["receipts"][:-1]), "new_chain_store_receipt_not_observed"), + (lambda chain: chain["receipts"][-1].pop("recorded_stage"), "new_chain_store_receipt_malformed"), + ( + lambda chain: chain["receipts"][-1].update(recorded_stage="other_nonempty_stage"), + "final_chain_store_receipt_transition_mismatch", + ), + ( + lambda chain: chain["receipts"][-1].update(previous_plan_id="sha256:wrong-previous"), + "final_chain_store_receipt_transition_mismatch", + ), + ( + lambda chain: chain["receipts"][-1].update(next_plan_id="sha256:wrong-final"), + "final_chain_store_receipt_transition_mismatch", + ), + ( + lambda chain: chain["receipts"][-1].update(next_action="OTHER_NONEMPTY_ACTION"), + "final_chain_store_receipt_transition_mismatch", + ), + ( + lambda chain: chain["stage_results"]["held_out_regression_gate"]["gate_result"]["receipt"].update(candidate_head_sha="c" * 40), + "key_receipt_lineage_mismatch", + ), + ( + lambda chain: chain["stage_results"]["pattern_memory_admission"]["receipt"].update(work_order_id="other"), + "key_receipt_lineage_mismatch", + ), + ], +) +def test_false_chain_evidence_cannot_complete_proof(tmp_path: Path, mutator, blocker: str) -> None: + repo, runtime = _roots(tmp_path) + receipt = _execute(repo, runtime, chain_mutator=mutator) + assert receipt.live_proof_complete is False + assert blocker in receipt.blockers + + +def test_forged_chain_store_revision_fails_canonical_verification(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + receipt = _execute( + repo, runtime, + chain_mutator=lambda chain: chain["receipts"][-1].update(store_revision="wrong"), + rebind_after_mutation=False, + ) + assert "chain_results_revision_invalid" in receipt.blockers + + +@pytest.mark.parametrize("field", ["admission_id", "pattern_memory_record_id", "record_digest"]) +def test_pattern_memory_receipt_requires_all_durable_ids(tmp_path: Path, field: str) -> None: + repo, runtime = _roots(tmp_path) + + def mutate(chain): + chain["stage_results"]["pattern_memory_admission"]["receipt"].pop(field) + + receipt = _execute(repo, runtime, chain_mutator=mutate) + assert "highest_profile_completion_evidence_missing" in receipt.blockers + + +def test_pattern_memory_record_must_read_back_from_canonical_db(tmp_path: Path) -> None: + repo, runtime = _roots(tmp_path) + + def delete_record(db_path: Path, record_id: str) -> None: + memory = PatternMemory(db_path=db_path) + try: + memory.conn.execute("DELETE FROM skill_outcomes WHERE execution_id = ?", (record_id,)) + memory.conn.commit() + finally: + memory.close() + + receipt = _execute(repo, runtime, pattern_db_mutator=delete_record) + assert "highest_profile_completion_evidence_missing" in receipt.blockers + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("work_order_id", "wrong-but-valid-work-order"), + ("slice_name", "WRONG_BUT_VALID_SLICE"), + ("candidate_head_sha", "f" * 40), + ], +) +def test_digest_valid_db_context_must_match_plan_draft_and_git_head( + tmp_path: Path, field: str, value: str +) -> None: + repo, runtime = _roots(tmp_path) + replacement: dict[str, str] = {} + + def mutate_db(db_path: Path, record_id: str) -> None: + sink = build_reddog_verified_pattern_memory_sink(repo_root=repo, db_path=db_path) + assert sink is not None + record = sink.load_verified_outcome(record_id) + assert record is not None + modified = {**record, field: value} + new_id = sink.store_verified_outcome(modified) + admission_id, digest = canonical_pattern_memory_admission_identity(modified, new_id) + replacement.update( + admission_id=admission_id, + pattern_memory_record_id=new_id, + record_digest=digest, + ) + + def mutate_chain(chain: dict[str, object]) -> None: + chain["stage_results"]["pattern_memory_admission"]["receipt"].update(replacement) + _canonicalize_terminal_receipt(chain) + + receipt = _execute( + repo, runtime, pattern_db_mutator=mutate_db, chain_mutator=mutate_chain + ) + assert "highest_profile_completion_evidence_missing" in receipt.blockers + + +@pytest.mark.parametrize( + "failure", + ["stage", "invoke", "missing", "not_git", "inside", "unregistered", "head"], +) +def test_worktree_proof_requires_registered_git_worktree(tmp_path: Path, failure: str) -> None: + repo, runtime = _roots(tmp_path) + + def mutate(chain): + stage = chain["stage_results"]["worktree_create"] + result = stage["worktree_create_result"] + if failure == "stage": + stage["decision"] = "REJECT" + elif failure == "invoke": + result["decision"] = "REJECT" + elif failure == "missing": + result["worktree_path"] = str(runtime / "absent") + elif failure == "not_git": + Path(result["worktree_path"], ".git").unlink() + elif failure == "inside": + inside = repo / "worker" + inside.mkdir() + (inside / ".git").write_text("gitdir: inside", encoding="utf-8") + result["worktree_path"] = str(inside) + elif failure == "unregistered": + clone = runtime.parent / "unregistered-clone" + subprocess.run( + ["git", "clone", "--no-hardlinks", str(repo), str(clone)], + capture_output=True, text=True, check=True, + ) + result["worktree_path"] = str(clone) + else: + forged = "f" * 40 + chain["stage_results"]["verified_draft_pr_publish"]["publish_result"]["receipt"]["verified_head_sha"] = forged + chain["stage_results"]["held_out_regression_gate"]["gate_result"]["receipt"]["candidate_head_sha"] = forged + + receipt = _execute(repo, runtime, chain_mutator=mutate) + assert "isolated_worktree_evidence_missing" in receipt.blockers diff --git a/modules/communication/moltbot_bridge/tests/test_reddog_resident_queue_chain_results_store.py b/modules/communication/moltbot_bridge/tests/test_reddog_resident_queue_chain_results_store.py index 0e0aed299..cbfe25dcb 100644 --- a/modules/communication/moltbot_bridge/tests/test_reddog_resident_queue_chain_results_store.py +++ b/modules/communication/moltbot_bridge/tests/test_reddog_resident_queue_chain_results_store.py @@ -18,6 +18,8 @@ AtomicJsonResidentQueueChainResultsStore, InMemoryResidentQueueChainResultsStore, record_resident_queue_stage_result, + resident_queue_chain_snapshot_is_canonical, + resident_queue_chain_snapshot_revision, ) from modules.communication.moltbot_bridge.src.reddog_resident_queue_orchestration_plan import ( NEXT_QUEUE_AUTHORITY_RUNTIME_INVOKE, @@ -231,6 +233,10 @@ def test_atomic_json_store_writes_schema_for_bootstrap(tmp_path: Path) -> None: assert result.accepted is True data = json.loads(path.read_text(encoding="utf-8")) assert data["revision"] == result.receipt.store_revision + assert data["receipts"][-1]["store_revision"] == data["revision"] + assert resident_queue_chain_snapshot_revision(data) == data["revision"] + assert resident_queue_chain_snapshot_is_canonical(data) is True + assert store.load() == data assert data["stage_results"]["authority_request"]["status"] == "QUEUE_AUTHORITY_REQUEST_DRYRUN_ACCEPT" assert not list(path.parent.glob("*.tmp")) diff --git a/modules/communication/moltbot_bridge/tests/test_reddog_verified_pattern_memory_sink.py b/modules/communication/moltbot_bridge/tests/test_reddog_verified_pattern_memory_sink.py index 8e96d23d0..19a469ac1 100644 --- a/modules/communication/moltbot_bridge/tests/test_reddog_verified_pattern_memory_sink.py +++ b/modules/communication/moltbot_bridge/tests/test_reddog_verified_pattern_memory_sink.py @@ -92,6 +92,29 @@ def test_sink_stores_verified_outcome_in_outside_repo_pattern_memory_db(tmp_path assert json.loads(row["output_result"])["record_type"] == ( "reddog_verified_recursive_improvement_outcome" ) + assert sink.load_verified_outcome(record_id) == _record() + + +def test_sink_readback_rejects_noncanonical_record_id(tmp_path: Path) -> None: + repo = _repo(tmp_path) + sink = build_reddog_verified_pattern_memory_sink( + repo_root=repo, + db_path=tmp_path / "runtime" / "pattern_memory.db", + ) + assert sink is not None + record_id = sink.store_verified_outcome(_record()) + + memory = PatternMemory(db_path=sink.db_path) + try: + memory.conn.execute( + "UPDATE skill_outcomes SET output_result = ? WHERE execution_id = ?", + (json.dumps(_record(work_order_id="tampered")), record_id), + ) + memory.conn.commit() + finally: + memory.close() + + assert sink.load_verified_outcome(record_id) is None def test_sink_is_idempotent_for_same_verified_outcome_record(tmp_path: Path) -> None: