Defensive hardening ports: chain-state CAS guard, snapshot cleanup guard, incident-journal custody

arnold_pipelines/megaplan/chain/spec.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import fcntl
 import hashlib
 import json
 import logging
+import os
 import re
 import stat
 import subprocess
+import tempfile
 import warnings
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -131,12 +134,16 @@ def _state_path_candidates_for(spec_path: Path) -> list[Path]:
 
 def _load_chain_state_file(path: Path) -> ChainState:
     try:
-        raw = json.loads(path.read_text(encoding="utf-8"))
+        encoded = path.read_bytes()
+        raw = json.loads(encoded)
     except json.JSONDecodeError as exc:
         raise CliError("invalid_chain_state", f"chain_state.json is invalid JSON: {exc}") from exc
     if not isinstance(raw, dict):
         raise CliError("invalid_chain_state", "chain_state.json must be an object")
-    return ChainState.from_dict(raw)
+    state = ChainState.from_dict(raw)
+    state._loaded_state_revision = hashlib.sha256(encoded).hexdigest()
+    state._loaded_state_path = str(path.resolve(strict=False))
+    return state
 
 
 def _normalize_stale_current_plan_reference(state: "ChainState") -> "ChainState":
@@ -1399,6 +1406,20 @@ class ChainState:
     schema_version: int = 0
     milestone_boundary_evidence: dict[str, dict[str, Any]] = field(default_factory=dict)
     candidate_invalidation: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
+    # In-memory CAS custody. These fields are populated only by
+    # _load_chain_state_file() / save_chain_state() and are never serialized.
+    _loaded_state_revision: str | None = field(
+        default=None,
+        init=False,
+        repr=False,
+        compare=False,
+    )
+    _loaded_state_path: str | None = field(
+        default=None,
+        init=False,
+        repr=False,
+        compare=False,
+    )
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -2190,7 +2211,7 @@ def save_chain_state(
     (used by internal rebuild/repair callers that should not create
     duplicate records).
     """
-    state_path = _state_path_for(spec_path)
+    state_path = _state_path_for(spec_path).resolve(strict=False)
     state_path.parent.mkdir(parents=True, exist_ok=True)
     spec_identity = _storage_identity_for_chain_spec(spec_path)
     metadata = dict(state.metadata)
@@ -2213,9 +2234,96 @@ def save_chain_state(
     metadata.setdefault("_m7_projection_first_seen_at", now_utc())
     # ─────────────────────────────────────────────────────────────────────
     state.metadata = metadata
-    tmp = state_path.with_suffix(".tmp")
-    tmp.write_text(json.dumps(state.to_dict(), indent=2) + "\n", encoding="utf-8")
-    tmp.replace(state_path)
+    encoded = (json.dumps(state.to_dict(), indent=2) + "\n").encode("utf-8")
+    lock_path = state_path.with_suffix(state_path.suffix + ".lock")
+    with lock_path.open("a+", encoding="utf-8") as lock:
+        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
+        current_bytes: bytes | None
+        try:
+            current_bytes = state_path.read_bytes()
+        except FileNotFoundError:
+            current_bytes = None
+
+        expected_revision = (
+            state._loaded_state_revision
+            if state._loaded_state_path == str(state_path)
+            else None
+        )
+        if current_bytes is not None:
+            current_revision = hashlib.sha256(current_bytes).hexdigest()
+            if (
+                expected_revision is not None
+                and expected_revision != current_revision
+            ):
+                raise CliError(
+                    "chain_state_stale_write",
+                    "chain state changed after it was loaded; refusing stale writer",
+                )
+            try:
+                current_raw = json.loads(current_bytes)
+            except json.JSONDecodeError as exc:
+                raise CliError(
+                    "invalid_chain_state",
+                    f"chain_state.json is invalid JSON: {exc}",
+                ) from exc
+            if not isinstance(current_raw, dict):
+                raise CliError(
+                    "invalid_chain_state",
+                    "chain_state.json must be an object",
+                )
+            current_state = ChainState.from_dict(current_raw)
+            if state.current_milestone_index < current_state.current_milestone_index:
+                raise CliError(
+                    "chain_state_regression",
+                    "chain state cursor regression refused: "
+                    f"{current_state.current_milestone_index} -> "
+                    f"{state.current_milestone_index}",
+                )
+            current_completed = {
+                str(record.get("label") or ""): str(record.get("plan") or "")
+                for record in current_state.completed
+                if isinstance(record, dict) and str(record.get("label") or "")
+            }
+            candidate_completed = {
+                str(record.get("label") or ""): str(record.get("plan") or "")
+                for record in state.completed
+                if isinstance(record, dict) and str(record.get("label") or "")
+            }
+            lost_or_changed = sorted(
+                label
+                for label, plan in current_completed.items()
+                if candidate_completed.get(label) != plan
+            )
+            if lost_or_changed:
+                raise CliError(
+                    "chain_state_regression",
+                    "chain state completed-set regression refused: "
+                    + ", ".join(lost_or_changed),
+                )
+        elif expected_revision is not None:
+            raise CliError(
+                "chain_state_stale_write",
+                "chain state disappeared after it was loaded; refusing stale writer",
+            )
+
+        fd, tmp_name = tempfile.mkstemp(
+            prefix=state_path.name + ".",
+            dir=state_path.parent,
+        )
+        try:
+            with os.fdopen(fd, "wb") as handle:
+                handle.write(encoded)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(tmp_name, state_path)
+        except BaseException:
+            try:
+                os.unlink(tmp_name)
+            except OSError:
+                pass
+            raise
+        state._loaded_state_revision = hashlib.sha256(encoded).hexdigest()
+        state._loaded_state_path = str(state_path)
 
     # ── M7 projection side-effect ────────────────────────────────────────
     if _record_projection:

arnold_pipelines/megaplan/cloud/wrappers/arnold-meta-repair-loop

@@ -47,7 +47,23 @@ if [[ "${ARNOLD_META_REPAIR_LOOP_SNAPSHOT_ACTIVE:-0}" != "1" \
   exec bash "$meta_repair_loop_snapshot" "$@"
 fi
 ARNOLD_META_REPAIR_LOOP_SNAPSHOT_PATH="${BASH_SOURCE[0]:-$0}"
+meta_repair_loop_snapshot_real="$(readlink -f "$ARNOLD_META_REPAIR_LOOP_SNAPSHOT_PATH" 2>/dev/null || printf '%s' "$ARNOLD_META_REPAIR_LOOP_SNAPSHOT_PATH")"
+meta_repair_loop_origin_real="$(readlink -f "$ARNOLD_META_REPAIR_LOOP_ORIGIN" 2>/dev/null || printf '%s' "$ARNOLD_META_REPAIR_LOOP_ORIGIN")"
+meta_repair_loop_origin_root="$(cd "$(dirname "$meta_repair_loop_origin_real")/../../../.." 2>/dev/null && pwd -P || printf '%s' "")"
+meta_repair_loop_snapshot_root="$(cd "${TMPDIR:-/tmp}" 2>/dev/null && pwd -P || printf '%s' "")"
+if [[ -z "$meta_repair_loop_snapshot_root" \
+  || "$meta_repair_loop_snapshot_real" != "$meta_repair_loop_snapshot_root"/arnold-meta-repair-loop.* \
+  || "$meta_repair_loop_snapshot_real" == "$meta_repair_loop_origin_real" \
+  || ( -n "$meta_repair_loop_origin_root" \
+    && ( "$meta_repair_loop_snapshot_real" == "$meta_repair_loop_origin_root" \
+      || "$meta_repair_loop_snapshot_real" == "$meta_repair_loop_origin_root"/* ) ) ]]; then
+  echo "arnold-meta-repair-loop: refusing unsafe immutable snapshot cleanup path" >&2
+  exit 78
+fi
+ARNOLD_META_REPAIR_LOOP_SNAPSHOT_PATH="$meta_repair_loop_snapshot_real"
+readonly ARNOLD_META_REPAIR_LOOP_SNAPSHOT_PATH
 trap 'rm -f -- "$ARNOLD_META_REPAIR_LOOP_SNAPSHOT_PATH"' EXIT
+unset meta_repair_loop_snapshot_real meta_repair_loop_origin_real meta_repair_loop_origin_root meta_repair_loop_snapshot_root
 
 if [[ $# -lt 1 ]]; then
   echo "usage: arnold-meta-repair-loop <session> [trigger]" >&2

arnold_pipelines/megaplan/cloud/wrappers/arnold-progress-auditor

@@ -100,6 +100,22 @@ if [[ "$progress_auditor_running_snapshot" != "1" ]]; then
   exec bash "$progress_auditor_snapshot" "$@"
 fi
 ARNOLD_PROGRESS_AUDITOR_SNAPSHOT_PATH="$progress_auditor_current"
+progress_auditor_snapshot_real="$(readlink -f "$ARNOLD_PROGRESS_AUDITOR_SNAPSHOT_PATH" 2>/dev/null || printf '%s' "$ARNOLD_PROGRESS_AUDITOR_SNAPSHOT_PATH")"
+progress_auditor_origin_real="$(readlink -f "$ARNOLD_PROGRESS_AUDITOR_ORIGIN" 2>/dev/null || printf '%s' "$ARNOLD_PROGRESS_AUDITOR_ORIGIN")"
+progress_auditor_origin_root="$(cd "$(dirname "$progress_auditor_origin_real")/../../../.." 2>/dev/null && pwd -P || printf '%s' "")"
+progress_auditor_snapshot_root="$(cd "${TMPDIR:-/tmp}" 2>/dev/null && pwd -P || printf '%s' "")"
+if [[ -z "$progress_auditor_snapshot_root" \
+  || "$progress_auditor_snapshot_real" != "$progress_auditor_snapshot_root"/arnold-progress-auditor.* \
+  || "$progress_auditor_snapshot_real" == "$progress_auditor_origin_real" \
+  || ( -n "$progress_auditor_origin_root" \
+    && ( "$progress_auditor_snapshot_real" == "$progress_auditor_origin_root" \
+      || "$progress_auditor_snapshot_real" == "$progress_auditor_origin_root"/* ) ) ]]; then
+  echo "arnold-progress-auditor: refusing unsafe immutable snapshot cleanup path" >&2
+  exit 78
+fi
+ARNOLD_PROGRESS_AUDITOR_SNAPSHOT_PATH="$progress_auditor_snapshot_real"
+readonly ARNOLD_PROGRESS_AUDITOR_SNAPSHOT_PATH
+unset progress_auditor_snapshot_real progress_auditor_origin_real progress_auditor_origin_root progress_auditor_snapshot_root
 
 declare -a progress_auditor_cleanup_paths=()
 register_progress_auditor_cleanup() {

arnold_pipelines/megaplan/cloud/wrappers/arnold-repair-loop

@@ -35,6 +35,25 @@ elif [[ "${ARNOLD_REPAIR_LOOP_SNAPSHOT_ACTIVE:-0}" != "1" \
 fi
 ARNOLD_REPAIR_LOOP_SNAPSHOT_PATH="${BASH_SOURCE[0]:-$0}"
 
+if [[ "${ARNOLD_REPAIR_LOOP_SKIP_SELF_COPY:-0}" != "1" ]]; then
+  repair_loop_snapshot_real="$(readlink -f "$ARNOLD_REPAIR_LOOP_SNAPSHOT_PATH" 2>/dev/null || printf '%s' "$ARNOLD_REPAIR_LOOP_SNAPSHOT_PATH")"
+  repair_loop_origin_real="$(readlink -f "$ARNOLD_REPAIR_LOOP_ORIGIN" 2>/dev/null || printf '%s' "$ARNOLD_REPAIR_LOOP_ORIGIN")"
+  repair_loop_origin_root="$(cd "$(dirname "$repair_loop_origin_real")/../../../.." 2>/dev/null && pwd -P || printf '%s' "")"
+  repair_loop_snapshot_root="$(cd "${TMPDIR:-/tmp}" 2>/dev/null && pwd -P || printf '%s' "")"
+  if [[ -z "$repair_loop_snapshot_root" \
+    || "$repair_loop_snapshot_real" != "$repair_loop_snapshot_root"/arnold-repair-loop.* \
+    || "$repair_loop_snapshot_real" == "$repair_loop_origin_real" \
+    || ( -n "$repair_loop_origin_root" \
+      && ( "$repair_loop_snapshot_real" == "$repair_loop_origin_root" \
+        || "$repair_loop_snapshot_real" == "$repair_loop_origin_root"/* ) ) ]]; then
+    echo "arnold-repair-loop: refusing unsafe immutable snapshot cleanup path" >&2
+    exit 78
+  fi
+  ARNOLD_REPAIR_LOOP_SNAPSHOT_PATH="$repair_loop_snapshot_real"
+  readonly ARNOLD_REPAIR_LOOP_SNAPSHOT_PATH
+  unset repair_loop_snapshot_real repair_loop_origin_real repair_loop_origin_root repair_loop_snapshot_root
+fi
+
 cleanup_repair_loop_snapshot() {
   if [[ "${ARNOLD_REPAIR_LOOP_SKIP_SELF_COPY:-0}" == "1" ]]; then
     # With the self-copy disabled the snapshot path aliases the origin script,

arnold_pipelines/megaplan/cloud/wrappers/arnold-watchdog

@@ -18,7 +18,23 @@ if [[ "${ARNOLD_WATCHDOG_SNAPSHOT_ACTIVE:-0}" != "1" ]]; then
   exec bash "$watchdog_snapshot" "$@"
 fi
 ARNOLD_WATCHDOG_SNAPSHOT_PATH="${BASH_SOURCE[0]:-$0}"
+watchdog_snapshot_real="$(readlink -f "$ARNOLD_WATCHDOG_SNAPSHOT_PATH" 2>/dev/null || printf '%s' "$ARNOLD_WATCHDOG_SNAPSHOT_PATH")"
+watchdog_origin_real="$(readlink -f "$ARNOLD_WATCHDOG_ORIGIN" 2>/dev/null || printf '%s' "$ARNOLD_WATCHDOG_ORIGIN")"
+watchdog_origin_root="$(cd "$(dirname "$watchdog_origin_real")/../../../.." 2>/dev/null && pwd -P || printf '%s' "")"
+watchdog_snapshot_root="$(cd "${TMPDIR:-/tmp}" 2>/dev/null && pwd -P || printf '%s' "")"
+if [[ -z "$watchdog_snapshot_root" \
+  || "$watchdog_snapshot_real" != "$watchdog_snapshot_root"/arnold-watchdog.* \
+  || "$watchdog_snapshot_real" == "$watchdog_origin_real" \
+  || ( -n "$watchdog_origin_root" \
+    && ( "$watchdog_snapshot_real" == "$watchdog_origin_root" \
+      || "$watchdog_snapshot_real" == "$watchdog_origin_root"/* ) ) ]]; then
+  echo "arnold-watchdog: refusing unsafe immutable snapshot cleanup path" >&2
+  exit 78
+fi
+ARNOLD_WATCHDOG_SNAPSHOT_PATH="$watchdog_snapshot_real"
+readonly ARNOLD_WATCHDOG_SNAPSHOT_PATH
 trap 'rm -f -- "$ARNOLD_WATCHDOG_SNAPSHOT_PATH"' EXIT
+unset watchdog_snapshot_real watchdog_origin_real watchdog_origin_root watchdog_snapshot_root
 
 # Caller-provided supervisor runtime controls are an explicit invocation
 # contract (and are used by readiness probes).  The shared hot-env supplies

arnold_pipelines/megaplan/managed_agent.py

@@ -650,6 +650,18 @@ def _bind_repair_claim(manifest: dict[str, Any]) -> None:
         manifest["repair_claim"]["repair_identity"] = repair_identity
 
 
+def _managed_incident_root(manifest: Mapping[str, Any]) -> Path:
+    """Keep cloud control journals outside immutable runtime candidates."""
+
+    links = manifest.get("links")
+    links = links if isinstance(links, Mapping) else {}
+    if str(links.get("cloud_session") or "").strip():
+        return Path(
+            os.environ.get("ARNOLD_INCIDENT_LEDGER_ROOT") or "/workspace"
+        ).resolve(strict=False)
+    return Path(str(manifest.get("project_dir") or ".")).resolve(strict=False)
+
+
 def _emit_attempt(manifest: Mapping[str, Any]) -> tuple[str, str] | None:
     links = manifest.get("links")
     if not isinstance(links, Mapping):
@@ -678,6 +690,7 @@ def _emit_attempt(manifest: Mapping[str, Any]) -> tuple[str, str] | None:
         or run_kind == "automatic_root_cause_repair"
         else "immediate_repair"
     )
+    incident_root = _managed_incident_root(manifest)
     claim_event = incident_bridge.append_managed_repair_claim(
         incident_id=incident_id,
         claim_id=f"managed:{manifest.get('run_id')}",
@@ -692,7 +705,7 @@ def _emit_attempt(manifest: Mapping[str, Any]) -> tuple[str, str] | None:
             else "immediate_repair.repair_attempt"
         ),
         links={"managed_agent": str(manifest.get("manifest_path"))},
-        root=str(manifest.get("project_dir") or "."),
+        root=incident_root,
     )
     claim_event_id = str(
         claim_event.get("event_id")
@@ -709,7 +722,7 @@ def _emit_attempt(manifest: Mapping[str, Any]) -> tuple[str, str] | None:
         problem_id=str(links.get("problem_id") or "") or None,
         parent_event_ids=[claim_event_id] if claim_event_id else [],
         links={"managed_agent": str(manifest.get("manifest_path"))},
-        root=str(manifest.get("project_dir") or "."),
+        root=incident_root,
     )
     if manifest.get("run_kind") in {
         "automatic_meta_repair",

arnold_pipelines/megaplan/pipeline_ids.json

@@ -42,4 +42,4 @@
     }
   ],
   "version": 1
-}
+}

docs/arnold/manifest-identity-report.json

@@ -30,7 +30,7 @@
       }
     },
     {
-      "compiled_manifest_hash": "sha256:74563f60ae604b96822a308178eff6a4e7d308a43f7ecd726e02824cbafbfb96",
+      "compiled_manifest_hash": "sha256:09255e89b8ec7f90612ccae382bf01b4ff39228c47f777c73f2d253042dcf362",
       "disposition": "migrate",
       "example_docs": {
         "exists": true,
@@ -42,14 +42,14 @@
       "pipeline_id": "megaplan",
       "registry_entry": {
         "m6_disposition": "keep",
-        "manifest_hash": "sha256:74563f60ae604b96822a308178eff6a4e7d308a43f7ecd726e02824cbafbfb96",
+        "manifest_hash": "sha256:09255e89b8ec7f90612ccae382bf01b4ff39228c47f777c73f2d253042dcf362",
         "name": "megaplan",
         "seam_ids": [],
         "stable_id": "megaplan.core",
         "typed_contract_capable": false
       },
       "registry_id": "megaplan.core",
-      "registry_manifest_hash": "sha256:74563f60ae604b96822a308178eff6a4e7d308a43f7ecd726e02824cbafbfb96",
+      "registry_manifest_hash": "sha256:09255e89b8ec7f90612ccae382bf01b4ff39228c47f777c73f2d253042dcf362",
       "registry_path": "arnold_pipelines/megaplan/pipeline_ids.json",
       "skill_docs": {
         "exists": true,
@@ -158,7 +158,7 @@
       }
     },
     {
-      "compiled_manifest_hash": "sha256:74563f60ae604b96822a308178eff6a4e7d308a43f7ecd726e02824cbafbfb96",
+      "compiled_manifest_hash": "sha256:09255e89b8ec7f90612ccae382bf01b4ff39228c47f777c73f2d253042dcf362",
       "disposition": "migrate",
       "example_docs": {
         "exists": true,
@@ -170,14 +170,14 @@
       "pipeline_id": "megaplan",
       "registry_entry": {
         "m6_disposition": "keep",
-        "manifest_hash": "sha256:74563f60ae604b96822a308178eff6a4e7d308a43f7ecd726e02824cbafbfb96",
+        "manifest_hash": "sha256:09255e89b8ec7f90612ccae382bf01b4ff39228c47f777c73f2d253042dcf362",
         "name": "planning",
         "seam_ids": [],
         "stable_id": "megaplan.planning",
         "typed_contract_capable": true
       },
       "registry_id": "megaplan.planning",
-      "registry_manifest_hash": "sha256:74563f60ae604b96822a308178eff6a4e7d308a43f7ecd726e02824cbafbfb96",
+      "registry_manifest_hash": "sha256:09255e89b8ec7f90612ccae382bf01b4ff39228c47f777c73f2d253042dcf362",
       "registry_path": "arnold_pipelines/megaplan/pipeline_ids.json",
       "skill_docs": {
         "exists": true,