Sapientropic · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026
@@ -67,12 +67,12 @@ def chooser_tail_supported(tail: Sequence[str]) -> bool:
 
 def _action_candidates(payload: Mapping[str, Any]) -> list[dict[str, Any]]:
     candidates: list[dict[str, Any]] = []
-    for raw in [
-        payload.get("foreground_action"),
-        *(payload.get("safe_next_actions") or []),
-        *(payload.get("choices") if isinstance(payload.get("choices"), list) else []),
-        *(payload.get("write_actions") if isinstance(payload.get("write_actions"), list) else []),
-    ]:
+    raw_candidates: list[Any] = [payload.get("foreground_action")]
+    for key in ("safe_next_actions", "choices", "write_actions"):
+        value = payload.get(key)
+        if isinstance(value, list):
+            raw_candidates.extend(value)
+    for raw in raw_candidates:
         if not isinstance(raw, Mapping):
             continue
         action = normalize_foreground_action(raw)

@@ -186,7 +186,7 @@ def module_exception_payload(
         else []
     )
     action = search_actions[0] if search_actions else script_recovery_action(script_name, args)
-    payload = (
+    payload: dict[str, Any] = (
         {
             "ok": False,
             "error": cli_error_object(
@@ -231,7 +231,8 @@ def module_exception_payload(
 
 
 def render_module_exception_text(payload: dict[str, Any]) -> str:
-    error = payload.get("error") if isinstance(payload.get("error"), dict) else {}
+    raw_error = payload.get("error")
+    error: Mapping[str, Any] = raw_error if isinstance(raw_error, Mapping) else {}
     action = payload.get("foreground_action")
     action_map = action if isinstance(action, Mapping) else {}
     lines = [
@@ -252,7 +253,8 @@ def handle_module_exception(
     stderr_text: str = "",
 ) -> int:
     payload = module_exception_payload(script_name, args, exc, stderr_text=stderr_text)
-    error = payload.get("error") if isinstance(payload.get("error"), dict) else {}
+    raw_error = payload.get("error")
+    error: Mapping[str, Any] = raw_error if isinstance(raw_error, Mapping) else {}
     exit_code = cli_exit_code_for_error_code(str(error.get("code") or "runtime_error"))
     if args_request_json(args):
         emit_public_text(json.dumps(payload, ensure_ascii=False, indent=2), end="")

@@ -34,8 +34,10 @@
         "output_boundary",
         "policy_boundary",
         "privacy_boundary",
+        "risk_boundary",
         "source_boundary",
         "source_reopen_boundary",
+        "suppression_boundary",
         "write_boundary",
     }
 )

@@ -172,7 +172,9 @@ def build_perturbation_packet(
         reason_codes.append("inversion_requires_source_reopen_or_conflict_review")
     if policy.get("stale_conflict_checks_required"):
         reason_codes.append("stale_conflict_checks_required")
-    for reason in layer_diagnostics["reason_codes"]:
+    raw_layer_reasons = layer_diagnostics.get("reason_codes")
+    layer_reasons = raw_layer_reasons if isinstance(raw_layer_reasons, list) else []
+    for reason in layer_reasons:
         if reason not in reason_codes:
             reason_codes.append(str(reason))
 

@@ -194,13 +194,13 @@ def build_total_hexagram_encoding(
     if explicit_reviewed_state is None:
         observed: dict[int, list[dict[str, Any]]] = {line: [] for line in range(1, 7)}
         for signal in line_signals:
-            line = _line_index(signal.get("line"))
+            line_index = _line_index(signal.get("line"))
             value = _line_value(signal.get("value"))
-            if line is None or value is None:
+            if line_index is None or value is None:
                 continue
             if _private_scope(signal):
                 blocked = True
-                observed[line].append(
+                observed[line_index].append(
                     {
                         "value": value,
                         "authority": "blocked",
@@ -210,7 +210,7 @@ def build_total_hexagram_encoding(
                 )
                 continue
             refs = _safe_refs(signal.get("source_refs"))
-            observed[line].append(
+            observed[line_index].append(
                 {
                     "value": value,
                     "authority": "source_backed" if refs else "unknown",
@@ -221,8 +221,8 @@ def build_total_hexagram_encoding(
         for line, rows in observed.items():
             if not rows:
                 continue
-            values = {row["value"] for row in rows}
-            if len(values) > 1:
+            row_values = {row["value"] for row in rows}
+            if len(row_values) > 1:
                 ambiguous = True
                 line_slots[line] = {
                     "line": line,
@@ -292,15 +292,24 @@ def build_total_hexagram_encoding(
             "symbolic_advice_included": False,
         },
     }
-    if complete:
-        lines_tuple = tuple(int(value) for value in known_values)  # type: ignore[arg-type]
+    complete_values = [value for value in known_values if value in (0, 1)]
+    if complete and len(complete_values) == 6:
+        line_values = tuple(int(value) for value in complete_values)
+        lines_tuple = (
+            line_values[0],
+            line_values[1],
+            line_values[2],
+            line_values[3],
+            line_values[4],
+            line_values[5],
+        )
         hexagram = hexagram_from_lines(lines_tuple)
         hexagram_projection = hexagram.to_public_dict()
         hexagram_projection["bits_bottom_to_top"] = hexagram_projection["bitstring_bottom_to_top"]
         result["hexagram"] = hexagram_projection
         result["macro_state_hint"] = _state_hint(
             project=project,
-            lines=lines_tuple,  # type: ignore[arg-type]
+            lines=lines_tuple,
             changing_lines=changing_lines,
             active_layer=active_layer,
             momentum=momentum,

@@ -351,6 +351,58 @@ def _select_initial_foreground_action(
             foreground_action,
             context.recall_selector,
         )
+        if not current_source_route_policy.primary_deepen_followthrough_reopenable(
+            context.memory_packets,
+        ):
+            registry_fallback = recall_choices.registry_source_search_fallback_action(
+                context.recovery_cue
+            )
+            if registry_fallback:
+                registry_fallback["route_choice_posture"] = "route_note_requires_source_search"
+                registry_fallback["why"] = (
+                    "The top recall route is route-note navigation without a "
+                    "reopenable clean-source handle; search registered sources for "
+                    "the original cue anchors before deepening or claiming."
+                )
+                foreground_action = registry_fallback
+                safe_next_actions = [
+                    {
+                        "id": "refine_low_specificity_recall_cue",
+                        "label": "Refine low-specificity recall cue",
+                        "tool_name": "agent_recall",
+                        "arguments_template": {"query": "{tighter_cue}", "max": 3},
+                        "requires": ["tighter_cue"],
+                        "template_only": True,
+                        "command_template": 'aippocampus agent recall "{tighter_cue}" --json',
+                        "route_choice_posture": "route_note_requires_source_search",
+                        "mutation_risk": "read_only",
+                        "claim_boundary": "no_claim_before_reopen",
+                        "why": (
+                            "If source search does not find the right anchor, "
+                            "tighten the cue with a more specific phrase, object, "
+                            "person, or time clue."
+                        ),
+                    }
+                ]
+            else:
+                weak_route_recovery_card = _weak_route_recovery_card()
+                foreground_action = {
+                    "action_id": "recover_route_note_without_source_ref",
+                    "label": "Recover route-note recall",
+                    "tool_name": "search_memory",
+                    "why": (
+                        "The top route-note candidate has no reopenable source "
+                        "handle; provide a more specific cue before relying on it."
+                    ),
+                    "mutation_risk": "read_only",
+                    "claim_boundary": "no_claim_before_reopen",
+                } | context.search_fields
+            return ForegroundSelection(
+                foreground_action=foreground_action,
+                miss_recovery_card=miss_recovery_card,
+                weak_route_recovery_card=weak_route_recovery_card,
+                safe_next_actions=safe_next_actions,
+            )
         foreground_action = recall_choices.with_low_specificity_foreground_action(
             foreground_action,
             metrics=context.metrics,

@@ -23,6 +23,45 @@ def primary_current_source_reopenable(memory_packets: list[dict[str, Any]]) -> b
     )
 
 
+def primary_deepen_followthrough_reopenable(memory_packets: list[dict[str, Any]]) -> bool:
+    """Return whether the primary route is safe to expose as agent_deepen.
+
+    Low-specificity recall sometimes surfaces route-note navigation scents. A
+    route note is useful orientation, but it should not become the first
+    foreground `agent_deepen` action unless it carries joined clean-source
+    evidence that `agent_deepen` can reopen.
+    """
+
+    if not memory_packets:
+        return False
+    primary = memory_packets[0]
+    if primary.get("output_mode") != "reopenable_route":
+        return False
+    if not _route_note_like(primary):
+        return True
+    return _route_note_has_joined_source_ref(primary)
+
+
+def _route_note_like(packet: Mapping[str, Any]) -> bool:
+    markers = {
+        str(packet.get("route_kind") or ""),
+        str(packet.get("matched_cue_family") or ""),
+        str(packet.get("origin") or ""),
+        str(packet.get("route_origin") or ""),
+        str(packet.get("source") or ""),
+    }
+    return any("route_note" in marker.casefold() for marker in markers)
+
+
+def _route_note_has_joined_source_ref(packet: Mapping[str, Any]) -> bool:
+    for key in ("joined_evidence_refs", "source_refs"):
+        value = packet.get(key)
+        if isinstance(value, list) and any(isinstance(item, Mapping) for item in value):
+            return True
+    source_ref = packet.get("source_ref")
+    return isinstance(source_ref, Mapping) and bool(source_ref.get("message_id"))
+
+
 def apw_card_allows_primary(
     *,
     should_replace: bool,

@@ -10,6 +10,7 @@
     canonical_foreground_action_fields,
     foreground_shell_action,
 )
+from aippocampus_runtime.foreground_compact_language import compact_frontstage_projection
 from aippocampus_runtime.model.routing import DEFAULT_DEEPSEEK_API_KEY_ENV
 from aippocampus_runtime.ops.doctors.common import as_dict
 
@@ -146,11 +147,18 @@ def compact_provider_doctor_card(report: dict[str, Any]) -> dict[str, Any]:
             "frontstage_rule": "readiness and next check first; diagnostics stay in full detail",
         },
     }
-    return {
-        key: value
-        for key, value in card.items()
-        if value not in (None, "", [], {})
-    }
+    return compact_frontstage_projection(
+        {
+            key: value
+            for key, value in card.items()
+            if value not in (None, "", [], {})
+        },
+        extra_denied_keys={
+            "audit_json_available",
+            "boundary_summary",
+            "full_audit_command",
+        },
+    )
 
 
 def render_text(report: dict[str, Any]) -> str:

@@ -7,6 +7,7 @@
 
 from aippocampus_runtime import core
 from aippocampus_runtime.contracts import canonical_foreground_action_fields
+from aippocampus_runtime.foreground_compact_language import compact_frontstage_projection
 from aippocampus_runtime.ops.storage_governance_actions import (
     candidate_can_offer_compact_apply,
     storage_gc_summary_actions,
@@ -295,6 +296,8 @@ def bounded_cli_projection(
         action_fields = canonical_foreground_action_fields(
             summary_actions[0],
             safe_next_actions=summary_actions,
+            max_safe_next_actions=1,
+            safe_next_read_only_only=True,
         )
         next_steps = [
             "Continue normal work when cleanup was not the user goal.",
@@ -392,6 +395,19 @@ def bounded_cli_projection(
             canonical_foreground_action_fields(
                 primary,
                 safe_next_actions=[primary, *existing_actions],
+                max_safe_next_actions=1,
+                safe_next_read_only_only=True,
             )
         )
-    return projection
+    projection["details_available"] = True
+    return compact_frontstage_projection(
+        projection,
+        extra_denied_keys={
+            "comparable_metrics_command",
+            "full_audit_available",
+            "full_audit_flag",
+            "operator_audit_command",
+            "policy_model",
+            "report_sources",
+        },
+    )
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Mapping
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, cast
 
@@ -753,6 +754,7 @@ def explain(
 
 
 def build_macro_orientation_packet_fixture_report() -> dict[str, Any]:
+    fixture_now = datetime(2026, 6, 12, tzinfo=timezone.utc)
     moving = macro_state.build_macro_orientation_state(
         project="AIppocampus",
         hexagram="乾",
@@ -769,10 +771,15 @@ def build_macro_orientation_packet_fixture_report() -> dict[str, Any]:
         updated_at="2026-06-11T11:00:00Z",
         active_layer="人",
     )
-    current = macro_state.latest_project_macro_orientation([moving], project="AIppocampus")
+    current = macro_state.latest_project_macro_orientation(
+        [moving],
+        project="AIppocampus",
+        now=fixture_now,
+    )
     standing_projection = macro_state.latest_project_macro_orientation(
         [standing],
         project="AIppocampus",
+        now=fixture_now,
     )
     packet = _macro_memory_packet(current, project="AIppocampus")
     standing_packet = _macro_memory_packet(standing_projection, project="AIppocampus")

@@ -145,7 +145,7 @@ def _candidate_from_match(
         "candidate_id": f"registry-clean-source:{public_id}",
         "thread_key": _compact(thread_key, 120),
         "route_terms": route_terms,
-        "query_anchor_terms": unique_preserve(anchor_terms, limit=12),
+        "query_anchor_terms": unique_preserve(list(anchor_terms), limit=12),
         "actual_source_matched_terms": route_terms,
         "meaningful_matched_terms": route_terms,
         "anchor_quality": anchor_quality,

@@ -31,7 +31,8 @@ def _clean_source_messages_path(paths: Mapping[str, Any]) -> str:
 
 
 def _row_state(entry: Mapping[str, Any]) -> dict[str, bool]:
-    paths = entry.get("paths") if isinstance(entry.get("paths"), Mapping) else {}
+    raw_paths = entry.get("paths")
+    paths: Mapping[str, Any] = raw_paths if isinstance(raw_paths, Mapping) else {}
     provider_known = bool(
         str(entry.get("source_provider") or "").strip()
         or (isinstance(entry.get("session_meta"), Mapping) and entry.get("session_meta"))
@@ -57,8 +58,10 @@ def _count(rows: list[dict[str, bool]], key: str) -> int:
 
 
 def render_reachability_audit(report: Mapping[str, Any]) -> str:
-    counts = report.get("counts") if isinstance(report.get("counts"), Mapping) else {}
-    action = report.get("foreground_action") if isinstance(report.get("foreground_action"), Mapping) else {}
+    raw_counts = report.get("counts")
+    counts: Mapping[str, Any] = raw_counts if isinstance(raw_counts, Mapping) else {}
+    raw_action = report.get("foreground_action")
+    action: Mapping[str, Any] = raw_action if isinstance(raw_action, Mapping) else {}
     next_action = action.get("command") or action.get("command_template")
     lines = [
         f"registry source reachability: {report.get('status')}",
@@ -162,12 +165,7 @@ def registry_source_reachability_audit(
             }
             for entry, state in zip(entries, row_states, strict=True)
         ]
-    report.update(
-        canonical_foreground_action_fields(
-            actions[0] if actions else None,
-            safe_next_actions=actions,
-        )
-    )
+    report.update(canonical_foreground_action_fields(actions[0], safe_next_actions=actions))
     return report if include_paths else redact_private_paths(report)