fix(reddog): confine runtime artifacts and redact DAEmon telemetry

modules/communication/moltbot_bridge/ModLog.md

@@ -39,6 +39,14 @@ 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_RUNTIME_ARTIFACT_PATH_DAEMON_REDACTION_PHASE1
+
+- Confined resident runtime paths outside the repository and beneath the
+  configured runtime root.
+- Hardened resident control receipts with bounded fields, existing-chain
+  validation, locked append, descriptor revalidation, and fsync.
+- Redacted DAEmon fix details before supervisor action reporting.
+
 ## 2026-07-18: REDDOG_HOLOINDEX_QUERY_OWNER_BOUNDARY_POC_PHASE1
 
 **WSP Protocol**: WSP 00, 05, 06, 15, 22, 50, 62, 87, 96, 97

modules/communication/moltbot_bridge/config/openclaw_integration_manifest.json

@@ -204,7 +204,8 @@
       "env_keys": [
         "OPENCLAW_SELF_AUDIT_ENABLED",
         "OPENCLAW_SELF_AUDIT_AUTO_FIX",
-        "OPENCLAW_SELF_AUDIT_INTERVAL_SEC"
+        "OPENCLAW_SELF_AUDIT_INTERVAL_SEC",
+        "OPENCLAW_SELF_AUDIT_RUNTIME_ROOT"
       ],
       "secret_keys": [],
       "external_service": "local_service",

modules/communication/moltbot_bridge/src/openclaw_supervisor.py

@@ -35,6 +35,11 @@
 from pathlib import Path
 from typing import Any, Callable, Deque, Dict, List, Mapping, Optional, Sequence
 
+from modules.infrastructure.shared_utilities.runtime_artifact_safety import (
+    redact_runtime_text,
+    redact_runtime_value,
+)
+
 logger = logging.getLogger(__name__)
 
 READONLY_AUDIT_OPENCLAW_CLAIM_ACCEPT = "READONLY_AUDIT_OPENCLAW_CLAIM_ACCEPT"
@@ -1357,7 +1362,9 @@ def __init__(
         self._stop_event = threading.Event()
         self._broker = broker
         self._observer = observer
-        self._action_reporter = action_reporter or self._build_daemon_reporter()
+        self._action_reporter = self._redacting_reporter(
+            action_reporter or self._build_daemon_reporter()
+        )
         self._self_audit_factory = self_audit_factory
         self._self_audit_loop: Any | None = None
         self._event_cursor = 0
@@ -1571,19 +1578,32 @@ def run_cycle(self, parent_context=None) -> Dict[str, Any]:
             self._remember(observation, plan, action_result, verify)
             self._transition(SupervisorState.IDLE_WATCH, "cycle_complete")
 
-        self.last_cycle = {
+        self.last_cycle = redact_runtime_value({
             "state": self.current_state.value,
             "plan": plan,
             "action_result": action_result,
             "verify": verify,
             "observation": observation,
-        }
+        })
         return self.last_cycle
 
     # ------------------------------------------------------------------ #
     #  Infrastructure helpers                                             #
     # ------------------------------------------------------------------ #
 
+    @staticmethod
+    def _redacting_reporter(
+        reporter: Callable[[str, str, Dict[str, Any]], None],
+    ) -> Callable[[str, str, Dict[str, Any]], None]:
+        def safe_reporter(action_type: str, result: str, details: Dict[str, Any]) -> None:
+            reporter(
+                redact_runtime_text(action_type, max_chars=120).text,
+                redact_runtime_text(result, max_chars=200).text,
+                redact_runtime_value(details),
+            )
+
+        return safe_reporter
+
     def _build_daemon_reporter(self) -> Callable[[str, str, Dict[str, Any]], None]:
         from modules.infrastructure.dae_daemon.src.dae_daemon import get_central_daemon
         from modules.infrastructure.dae_daemon.src.schemas import DAEEventType
@@ -2211,22 +2231,30 @@ def _execute(self, plan: Dict[str, Any]) -> Dict[str, Any]:
 
         elif plan["action"] == "execute_self_audit_fix":
             recommended_fix = plan.get("recommended_fix", "")
+            safe_plan = dict(plan)
+            safe_plan["event_signature"] = redact_runtime_text(
+                safe_plan.get("event_signature"), max_chars=240
+            ).text
             result: Dict[str, Any] = {"ok": False, "error": "no_audit_loop"}
             if self._self_audit_loop and hasattr(self._self_audit_loop, "_apply_policy_fix"):
                 try:
                     success, detail = self._self_audit_loop._apply_policy_fix(recommended_fix)
                     result = {
                         "ok": success,
                         "status": "applied" if success else "fix_failed",
-                        "detail": str(detail)[:500],
+                        "detail": redact_runtime_text(detail, max_chars=500).text,
                     }
                 except Exception as exc:
-                    result = {"ok": False, "status": "fix_error", "error": str(exc)[:500]}
+                    result = {
+                        "ok": False,
+                        "status": "fix_error",
+                        "error": redact_runtime_text(exc, max_chars=500).text,
+                    }
 
             self._action_reporter(
                 "supervisor_execute",
                 result.get("status", result.get("error", "unknown")),
-                {"plan": plan, "result": result},
+                {"plan": safe_plan, "result": result},
             )
             if _rt_start is not None:
                 try:
@@ -2427,16 +2455,20 @@ def _remember(
             if verify.get("ok"):
                 self.metrics.tasks_succeeded += 1
 
+        safe_plan = redact_runtime_value(plan_or_triage)
+        safe_action_result = redact_runtime_value(action_result)
+        safe_verify = redact_runtime_value(verify)
+
         # Report to daemon
         self._action_reporter(
             "supervisor_cycle",
             "recorded",
             {
                 "state": self.current_state.value,
                 "reason": self.last_reason,
-                "plan": plan_or_triage,
-                "action_result": action_result,
-                "verify": verify,
+                "plan": safe_plan,
+                "action_result": safe_action_result,
+                "verify": safe_verify,
                 "git": observation.get("git", {}),
                 "restart_budget": observation.get("restart_budget", {}),
                 "openclaw_follow": observation.get("openclaw_follow", {}),
@@ -2448,21 +2480,21 @@ def _remember(
             try:
                 from modules.infrastructure.wre_core.src.pattern_memory import SkillOutcome
 
-                skill_name = plan_or_triage.get("action", "unknown")
+                skill_name = safe_plan.get("action", "unknown")
                 fidelity = float(verify.get("fidelity", 0.85))
                 outcome = SkillOutcome(
                     execution_id=f"supervisor_{uuid.uuid4().hex[:12]}",
                     skill_name=skill_name,
                     agent="openclaw_supervisor",
                     timestamp=datetime.now().isoformat(),
-                    input_context=json.dumps(plan_or_triage, default=str)[:2000],
-                    output_result=json.dumps(action_result, default=str)[:2000],
+                    input_context=json.dumps(safe_plan, default=str)[:2000],
+                    output_result=json.dumps(safe_action_result, default=str)[:2000],
                     success=bool(verify.get("ok", False)),
                     pattern_fidelity=fidelity,
                     outcome_quality=1.0 if verify.get("ok") else 0.0,
                     execution_time_ms=int(action_result.get("execution_time_ms", 0)),
                     step_count=1,
-                    notes=f"Supervisor cycle: {plan_or_triage.get('reason', '')}",
+                    notes=f"Supervisor cycle: {safe_plan.get('reason', '')}",
                 )
                 self._pattern_memory.store_outcome(outcome)
                 logger.debug(
@@ -2477,7 +2509,7 @@ def _remember(
         self._event_cursor = int(follow.get("next_cursor", self._event_cursor) or self._event_cursor)
 
         # Gateway Continuity Layer: Record breadcrumb with continuity metadata
-        self._record_continuity_breadcrumb(plan_or_triage, action_result, verify)
+        self._record_continuity_breadcrumb(safe_plan, safe_action_result, safe_verify)
 
     def _create_continuity_context(self, cycle_id: str, parent_context=None) -> Any:
         """Create continuity context for a supervisor cycle.

modules/communication/moltbot_bridge/src/reddog_resident_control_loop_receipt_store.py

@@ -16,6 +16,10 @@
 from pathlib import Path
 from typing import Any, Mapping
 
+from modules.infrastructure.shared_utilities.runtime_artifact_safety import (
+    secure_append_runtime_text,
+)
+
 
 CONTROL_LOOP_RECEIPT_SCHEMA_VERSION = "reddog_resident_control_loop_receipt.v1"
 
@@ -59,13 +63,15 @@ def build_resident_control_loop_receipt(
     payload = {
         "schema_version": CONTROL_LOOP_RECEIPT_SCHEMA_VERSION,
         "accepted": bool(result.get("accepted")),
-        "status": str(result.get("status") or ""),
+        "status": _bounded_text(result.get("status"), 80),
         "rounds": _int(result.get("rounds")),
         "serial_progress": _int(result.get("serial_progress")),
         "claim_progress": _int(result.get("claim_progress")),
-        "receipt_ids": _string_tuple(result.get("receipt_ids")),
-        "rejection_reasons": _string_tuple(result.get("rejection_reasons")),
-        "created_at": str(created_at or ""),
+        "receipt_ids": _string_tuple(result.get("receipt_ids"), max_chars=256),
+        "rejection_reasons": _string_tuple(
+            result.get("rejection_reasons"), max_chars=512
+        ),
+        "created_at": _bounded_text(created_at, 80),
         "repo_root_digest": _digest(str(Path(repo_root).resolve())),
         "control_lock_acquired": result.get("control_lock_acquired") is True,
         "no_authority_issued": True,
@@ -93,18 +99,50 @@ def append_resident_control_loop_receipt(
         repo_root=repo_root,
         created_at=created_at,
     )
-    target = Path(path)
-    target.parent.mkdir(parents=True, exist_ok=True)
-    with target.open("a", encoding="utf-8", newline="\n") as handle:
-        handle.write(json.dumps(receipt.to_dict(), sort_keys=True, separators=(",", ":")))
-        handle.write("\n")
+    line = json.dumps(receipt.to_dict(), sort_keys=True, separators=(",", ":")) + "\n"
+    if len(line.encode("utf-8")) > 64 * 1024:
+        raise ValueError("resident_control_loop_receipt_too_large")
+    secure_append_runtime_text(
+        path,
+        line,
+        repo_root=repo_root,
+        validate_existing=_validate_existing_receipts,
+    )
     return receipt
 
 
-def _string_tuple(value: Any) -> tuple[str, ...]:
+def _string_tuple(value: Any, *, max_chars: int) -> tuple[str, ...]:
     if not isinstance(value, (list, tuple)):
         return ()
-    return tuple(str(item) for item in value if str(item or "").strip())
+    return tuple(
+        _bounded_text(item, max_chars)
+        for item in value[:128]
+        if str(item or "").strip()
+    )
+
+
+def _bounded_text(value: Any, max_chars: int) -> str:
+    return str(value or "").strip()[:max_chars]
+
+
+def _validate_existing_receipts(existing: str) -> None:
+    for line_number, raw in enumerate(existing.splitlines(), start=1):
+        if not raw.strip():
+            continue
+        if len(raw.encode("utf-8")) > 64 * 1024:
+            raise ValueError(f"resident_control_loop_receipt_line_too_large:{line_number}")
+        try:
+            payload = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ValueError(
+                f"resident_control_loop_receipt_chain_invalid_json:{line_number}"
+            ) from exc
+        if not isinstance(payload, dict):
+            raise ValueError(f"resident_control_loop_receipt_chain_invalid:{line_number}")
+        if payload.get("schema_version") != CONTROL_LOOP_RECEIPT_SCHEMA_VERSION:
+            raise ValueError(f"resident_control_loop_receipt_schema_invalid:{line_number}")
+        if not str(payload.get("receipt_id") or "").strip():
+            raise ValueError(f"resident_control_loop_receipt_id_missing:{line_number}")
 
 
 def _int(value: Any) -> int:

modules/communication/moltbot_bridge/src/reddog_resident_queue_binding_profile.py

@@ -18,6 +18,11 @@
 from pathlib import Path
 from typing import Mapping
 
+from modules.infrastructure.shared_utilities.runtime_artifact_safety import (
+    validate_runtime_artifact_path,
+    validate_runtime_root_path,
+)
+
 
 ENV_REDDOG_RESIDENT_QUEUE_BINDING_PROFILE = "REDDOG_RESIDENT_QUEUE_BINDING_PROFILE"
 PROFILE_SIGNED_0102_BOUNDED_CODE = "signed_0102_bounded_code"
@@ -187,29 +192,42 @@ def resident_queue_runtime_root_path(env: Mapping[str, str], repo_root: Path | s
         path = Path(raw)
         if not path.is_absolute():
             path = root.parent / path
-        return str(path.resolve())
+        return str(validate_runtime_root_path(path, repo_root=root))
     if resident_queue_binding_profile(env) not in RESIDENT_QUEUE_PROFILES:
         return ""
-    return str(root.parent / ".reddog" / "resident" / _repo_slug(root))
+    runtime_root = root.parent / ".reddog" / "resident" / _repo_slug(root)
+    return str(validate_runtime_root_path(runtime_root, repo_root=root))
 
 
 def resident_queue_runtime_file_path(
     env: Mapping[str, str],
     repo_root: Path | str,
     env_name: str,
 ) -> str:
-    """Return an explicit env path or a profile-derived runtime file path."""
+    """Return a confined explicit or profile-derived runtime file path."""
 
-    raw = str(env.get(env_name) or "").strip()
-    if raw:
-        return raw
     filename = PROFILE_RUNTIME_PATH_FILENAMES.get(env_name)
     if not filename:
         return ""
-    root = resident_queue_runtime_root_path(env, repo_root)
-    if not root:
+    raw = str(env.get(env_name) or "").strip()
+    runtime_root = resident_queue_runtime_root_path(env, repo_root)
+    if raw:
+        return str(
+            validate_runtime_artifact_path(
+                raw,
+                repo_root=repo_root,
+                allowed_root=runtime_root or None,
+            )
+        )
+    if not runtime_root:
         return ""
-    return str(Path(root) / filename)
+    return str(
+        validate_runtime_artifact_path(
+            Path(runtime_root) / filename,
+            repo_root=repo_root,
+            allowed_root=runtime_root,
+        )
+    )
 
 
 def resident_queue_artifact_generator_mode(env: Mapping[str, str]) -> str:

modules/communication/moltbot_bridge/tests/TestModLog.md

@@ -39,6 +39,11 @@
 **Truth boundary**: Tests use injected probes and do not perform live side
 effects. The production live canary remains unexecuted.
 
+## 2026-07-18: Runtime Artifact Confinement
+
+- Added source-path, runtime-root escape, malformed-chain, and concurrent
+  receipt append regression tests.
+
 ## 2026-07-18: REDDOG_HOLOINDEX_QUERY_OWNER_BOUNDARY_POC_PHASE1
 
 **Files**: test_reddog_holoindex_query_boundary.py (NEW),