fix(goals): own the lifecycle rules inward and give the projection a consumer

Three review findings, all reproduced at the previous head:

1. Forbidden dependency direction. The module imported
   `loopx.presentation.public_safety`, an outward dependency the control plane
   may not take. Redaction is now bounded and inward-safe inside the module and
   reuses `loopx.public_safe_text.find_private_text_match` for classification,
   so the shared private-text contract still has exactly one owner.

2. Material outcome semantics drifted. Evidence milestones kept a local
   outcome set and let `delivery_batch_scale == "multi_surface"` promote a run on
   its own. Driving the helper with `delivery_outcome=surface_only` plus
   `delivery_batch_scale=multi_surface` returned a reached milestone, but batch
   scale describes delivery width and the canonical typed rule
   (`MATERIAL_DELIVERY_OUTCOMES`) excludes `surface_only`. The module now
   consumes that canonical rule directly, and the smoke pins the full
   outcome x scale matrix in both directions.

3. No production consumer. The RFC's smallest slice requires one readout in
   status markdown. Status collection now attaches the projection and the
   presentation renderer prints phase, milestone and guard counts plus the
   blocking guards and the next transition. `loopx status` shows the readout
   without any extra IO: it derives only from payloads collection already
   gathered.

Verified: the smoke fails when the scale-promotion rule is restored and when the
presentation import is restored; the import-boundary suite is 15/15; status and
architecture suites are 29/29; docs governance passes; the four status smokes
pass; Ruff reports nothing new on the changed files.

Author: songoow

examples/control_plane/goal-artifact-lifecycle-projection-smoke.py

@@ -302,6 +302,123 @@ def assert_private_values_are_redacted_or_dropped() -> None:
     )
 
 
+def assert_batch_scale_never_promotes_an_outcome() -> None:
+    """The batch scale describes delivery width; it is not Goal evidence.
+
+    Regression for the counterexample the review raised: a `surface_only` run
+    also carrying `delivery_batch_scale=multi_surface` must not become a reached
+    milestone, because the canonical typed rule
+    (`MATERIAL_DELIVERY_OUTCOMES`) excludes `surface_only`.
+    """
+
+    for scale in ("single_surface", "multi_surface"):
+        for outcome in ("surface_only", "", "bogus", "multi_surface"):
+            projection = build_goal_artifact_lifecycle_projection(
+                goal_id=GOAL_ID,
+                goal={"id": GOAL_ID, "status": "active"},
+                run_history={
+                    "latest_runs": [
+                        {
+                            "delivery_outcome": outcome,
+                            "delivery_batch_scale": scale,
+                            "run_id": "run-1",
+                        }
+                    ]
+                },
+            )
+            assert projection["milestones"] == [], (outcome, scale, projection["milestones"])
+    # Every canonical material outcome still counts, at any scale.
+    for scale in ("single_surface", "multi_surface"):
+        for outcome in ("outcome_gap", "outcome_progress", "primary_goal_outcome"):
+            projection = build_goal_artifact_lifecycle_projection(
+                goal_id=GOAL_ID,
+                goal={"id": GOAL_ID, "status": "active"},
+                run_history={
+                    "latest_runs": [
+                        {
+                            "delivery_outcome": outcome,
+                            "delivery_batch_scale": scale,
+                            "run_id": "run-1",
+                        }
+                    ]
+                },
+            )
+            reached = [item for item in projection["milestones"] if item["reached"]]
+            assert [item["id"] for item in reached] == [outcome], (outcome, scale, reached)
+
+
+def assert_status_collection_attaches_a_readable_readout() -> None:
+    """The RFC's smallest slice includes one readout in status markdown.
+
+    Drives the same seam `loopx status` uses: collection attaches the
+    projection, the presentation renderer prints it. The projection is derived
+    from already-collected payloads, so this asserts no extra IO is required.
+    """
+
+    from loopx.control_plane.goals.artifact_lifecycle import (
+        attach_goal_artifact_lifecycle_projections,
+    )
+    from loopx.presentation.renderers.status_markdown import render_status_markdown
+
+    payload = {
+        "run_history": {
+            "goals": [
+                {
+                    "id": GOAL_ID,
+                    "status": "active",
+                    "acceptance": {"milestones": ["baseline_pass"]},
+                }
+            ]
+        },
+        "attention_queue": {
+            "items": [
+                {
+                    "goal_id": GOAL_ID,
+                    "user_todo_summary": {
+                        "gate_open_items": [
+                            {
+                                "todo_id": "todo_gate",
+                                "text": "approve the release",
+                                "status": "open",
+                                "action_kind": "publish",
+                            }
+                        ]
+                    },
+                    "agent_todo_summary": {"open_count": 0},
+                }
+            ]
+        },
+    }
+    attach_goal_artifact_lifecycle_projections(payload, history={"goals": []})
+    goal = payload["run_history"]["goals"][0]
+    projection = goal["artifact_lifecycle"]
+    assert projection["schema_version"] == GOAL_ARTIFACT_LIFECYCLE_PROJECTION_SCHEMA_VERSION
+    assert projection["lifecycle_phase"] == PHASE_WAITING_OWNER, projection
+    assert [guard["id"] for guard in projection["guards"]] == ["todo_gate"]
+    markdown = render_status_markdown(payload)
+    assert "artifact lifecycle: phase=waiting_owner" in markdown, markdown
+    assert "blocked by owner_decision (user): todo_gate" in markdown, markdown
+    assert_no_public_leak(projection)
+
+
+def assert_control_plane_imports_no_presentation_module() -> None:
+    """The projection may not depend outward on the presentation layer."""
+
+    import ast
+
+    source = (
+        REPO_ROOT / "loopx" / "control_plane" / "goals" / "artifact_lifecycle.py"
+    ).read_text(encoding="utf-8")
+    imported: list[str] = []
+    for node in ast.walk(ast.parse(source)):
+        if isinstance(node, ast.ImportFrom) and node.module:
+            imported.append(node.module)
+        elif isinstance(node, ast.Import):
+            imported.extend(alias.name for alias in node.names)
+    offenders = [name for name in imported if "presentation" in name]
+    assert offenders == [], offenders
+
+
 def main() -> int:
     assert_starting_phase_without_work()
     assert_declared_milestone_stays_unreached_while_gapped()
@@ -313,6 +430,9 @@ def main() -> int:
     assert_unreached_milestone_blocks_closeout()
     assert_reached_milestones_still_allow_closeout()
     assert_private_values_are_redacted_or_dropped()
+    assert_batch_scale_never_promotes_an_outcome()
+    assert_status_collection_attaches_a_readable_readout()
+    assert_control_plane_imports_no_presentation_module()
     print("goal-artifact-lifecycle-projection-smoke ok")
     return 0
 

loopx/control_plane/goals/artifact_lifecycle.py

@@ -21,8 +21,11 @@
 import re
 from typing import Any
 
-from ...presentation.public_safety import redact_public_text
 from ...public_safe_text import find_private_text_match
+from ..work_items.delivery_outcome import (
+    MATERIAL_DELIVERY_OUTCOMES,
+    normalize_delivery_outcome,
+)
 
 GOAL_ARTIFACT_LIFECYCLE_PROJECTION_SCHEMA_VERSION = (
     "goal_artifact_lifecycle_projection_v0"
@@ -48,19 +51,38 @@
     r"\b(?:gh[pousr]_[A-Za-z0-9]{16,}|sk-[A-Za-z0-9_-]{16,}|AKIA[0-9A-Z]{16})\b"
 )
 
-# A material run outcome that a Goal's own acceptance can rest on.
-_MATERIAL_OUTCOMES = {"primary_goal_outcome", "outcome_progress", "multi_surface"}
+# Local path shapes this projection rewrites before publishing a label. The
+# control plane may not import the presentation layer's sanitizer, and the
+# lifecycle projection must not mint a second copy of the shared private-text
+# contract: it reuses `find_private_text_match` for classification and keeps
+# only this bounded, inward-safe rewrite.
+_PATH_SHAPES = (
+    re.compile(r"/(?:Users|home|private|tmp|var)/[^\s`|,)]+"),
+    re.compile(r"[A-Za-z]:\\\\Users\\\\[^\s`|,)]+"),
+)
+
+_TRUNCATION_MARKER = "..."
+
+
+def _bounded_redacted_text(value: Any, *, limit: int) -> str:
+    text = str(value or "").strip()
+    for pattern in _PATH_SHAPES:
+        text = pattern.sub("<local-path-redacted>", text)
+    text = re.sub(r"\s+", " ", text)
+    if len(text) > limit:
+        return text[: max(0, limit - 1)].rstrip() + _TRUNCATION_MARKER
+    return text
 
 
 def _compact_text(value: Any, *, limit: int = 240) -> str | None:
     """Bound and redact one label; never carry a raw body, path or credential."""
 
     if not isinstance(value, str):
         return None
-    collapsed = redact_public_text(value, limit=limit)
+    collapsed = _bounded_redacted_text(value, limit=limit)
     if not collapsed:
         return None
-    # The shared sanitizer covers local paths; this projection additionally
+    # The shared private-text contract classifies; this projection additionally
     # refuses a value that still matches a private-text or credential shape
     # rather than publishing a partly-redacted fragment.
     if find_private_text_match(collapsed) or _TOKEN_SHAPES.search(collapsed):
@@ -125,13 +147,14 @@ def _evidence_milestones(run_history: dict[str, Any]) -> list[dict[str, Any]]:
         record = _mapping(run)
         if not record:
             continue
-        outcome = str(record.get("delivery_outcome") or "").strip()
-        scale = str(record.get("delivery_batch_scale") or "").strip()
-        classification = str(record.get("classification") or "").strip()
-        if outcome not in _MATERIAL_OUTCOMES and scale != "multi_surface":
+        outcome = normalize_delivery_outcome(record.get("delivery_outcome"))
+        # Only a canonical material delivery outcome is Goal evidence. The batch
+        # scale describes how wide a delivery was, never whether it advanced the
+        # Goal, so it cannot promote `surface_only` into a reached milestone.
+        if outcome is None or outcome not in MATERIAL_DELIVERY_OUTCOMES:
             continue
-        milestone_id = outcome or scale or classification
-        if not milestone_id or milestone_id in seen:
+        milestone_id = outcome.value
+        if milestone_id in seen:
             continue
         seen.add(milestone_id)
         reference = _compact_text(record.get("evidence_ref") or record.get("run_id"), limit=120)
@@ -343,6 +366,49 @@ def build_goal_artifact_lifecycle_projection(
     }
 
 
+def attach_goal_artifact_lifecycle_projections(
+    payload: dict[str, Any], *, history: dict[str, Any]
+) -> None:
+    """Attach one bounded lifecycle projection to every projected Goal.
+
+    Reads only payloads status collection already gathered, so the operator
+    readout costs no extra IO and grants no authority. A Goal the projection
+    cannot derive from is left without the key rather than given a placeholder.
+    """
+
+    sources = {str(goal.get("id")): goal for goal in _list(history.get("goals")) if _mapping(goal)}
+    run_history = _mapping(payload.get("run_history"))
+    items = _list(_mapping(payload.get("attention_queue")).get("items"))
+    for goal in _list(run_history.get("goals")):
+        record = _mapping(goal)
+        goal_id = str(record.get("id") or "").strip()
+        if not goal_id:
+            continue
+        source = {**sources.get(goal_id, {}), **record}
+        item = next(
+            (row for row in items if _mapping(row).get("goal_id") == goal_id), None
+        )
+        attention = _mapping(item)
+        projection = build_goal_artifact_lifecycle_projection(
+            goal_id=goal_id,
+            goal=source,
+            user_todo_summary=_mapping(
+                attention.get("user_todo_summary") or source.get("user_todo_summary")
+            ),
+            agent_todo_summary=_mapping(
+                attention.get("agent_todo_summary") or source.get("agent_todo_summary")
+            ),
+            run_history=run_history,
+            work_lane_contract=_mapping(
+                source.get("work_lane_contract") or attention.get("work_lane_contract")
+            ),
+            acceptance_gaps=_list(
+                source.get("acceptance_gaps") or attention.get("acceptance_gaps")
+            ),
+        )
+        record["artifact_lifecycle"] = projection
+
+
 __all__ = [
     "GOAL_ARTIFACT_LIFECYCLE_PROJECTION_SCHEMA_VERSION",
     "GUARD_KIND_EVIDENCE",
@@ -352,5 +418,6 @@ def build_goal_artifact_lifecycle_projection(
     "PHASE_QUALIFYING",
     "PHASE_STARTING",
     "PHASE_WAITING_OWNER",
+    "attach_goal_artifact_lifecycle_projections",
     "build_goal_artifact_lifecycle_projection",
 ]

loopx/control_plane/status/collection.py

@@ -9,6 +9,7 @@
 from typing import Any, Callable
 
 from ..goals.acceptance_observation import attach_goal_acceptance_observations
+from ..goals.artifact_lifecycle import attach_goal_artifact_lifecycle_projections
 from ..goals.contract_health import project_contract_health_for_goal
 from ..goals.activation import (
     GoalActivationState,
@@ -243,4 +244,5 @@ def collect_status(
             goal_channel_notification_projection
         )
     attach_goal_acceptance_observations(payload, history=history)
+    attach_goal_artifact_lifecycle_projections(payload, history=history)
     return payload

loopx/presentation/renderers/goal_artifact_lifecycle_markdown.py

@@ -0,0 +1,47 @@
+"""Render the derived Goal artifact lifecycle projection owned by status collection."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ...control_plane.goals.artifact_lifecycle import (
+    GOAL_ARTIFACT_LIFECYCLE_PROJECTION_SCHEMA_VERSION,
+)
+from ..markdown import as_dict, as_list, markdown_scalar
+
+
+def append_goal_artifact_lifecycle_markdown(
+    lines: list[str], goal: dict[str, Any]
+) -> None:
+    projection = as_dict(goal.get("artifact_lifecycle"))
+    if (
+        projection.get("schema_version")
+        != GOAL_ARTIFACT_LIFECYCLE_PROJECTION_SCHEMA_VERSION
+    ):
+        return
+    milestones = as_list(projection.get("milestones"))
+    reached = [
+        milestone
+        for milestone in milestones
+        if isinstance(milestone, dict) and milestone.get("reached") is True
+    ]
+    lines.append(
+        "  - artifact lifecycle: "
+        f"phase={markdown_scalar(projection.get('lifecycle_phase') or 'unknown')} "
+        f"milestones={len(reached)}/{len(milestones)} "
+        f"guards={len(as_list(projection.get('guards')))}"
+    )
+    for guard in as_list(projection.get("guards")):
+        if not isinstance(guard, dict) or guard.get("blocked") is not True:
+            continue
+        lines.append(
+            f"    - blocked by {markdown_scalar(guard.get('kind') or 'unknown')} "
+            f"({markdown_scalar(guard.get('owner') or 'unowned')}): "
+            f"{markdown_scalar(guard.get('id') or 'unknown')}"
+        )
+    for transition in as_list(projection.get("next_transitions")):
+        if isinstance(transition, dict):
+            lines.append(
+                f"    - next: {markdown_scalar(transition.get('target_phase') or 'unknown')} "
+                f"({markdown_scalar(transition.get('precondition') or 'unknown')})"
+            )

loopx/presentation/renderers/status_markdown.py

@@ -10,6 +10,7 @@
 from ...orchestration import orchestration_policy_summary
 from ..markdown import as_dict, as_list, markdown_scalar
 from .goal_acceptance_observation_markdown import append_goal_acceptance_observation_markdown
+from .goal_artifact_lifecycle_markdown import append_goal_artifact_lifecycle_markdown
 from .reward_memory_markdown import append_agent_reward_memory_markdown
 
 
@@ -246,6 +247,7 @@ def append_run_history_markdown(lines: list[str], run_history: dict[str, Any]) -
             f"unique_runs={goal.get('unique_runs')}"
         )
         append_goal_acceptance_observation_markdown(lines, goal)
+        append_goal_artifact_lifecycle_markdown(lines, goal)
         quota = goal.get("quota") if isinstance(goal.get("quota"), dict) else {}
         if quota:
             lines.append(