fix(status): use retained vision for acceptance observations

docs/architecture/rfcs/goal-artifact-lifecycle-projection-v0.md

@@ -111,6 +111,19 @@ state machine.
 4. Public-safe boundary stays identical to existing projections: no raw logs,
    evidence bodies, credentials, or local paths.
 
+### Acceptance observation source
+
+The existing `goal_acceptance_observation_projection_v0` discovers Agent lanes
+from recent runs and retained semantic history. It uses the existing
+`goal_frontier.semantic_history` module's vision-read interface. For a known
+Agent, present semantic history selects the current vision or its absence;
+retirement or a missing lane cannot revive an older recent row. Without
+semantic history, the legacy recent-list fallback remains. Unattributed legacy
+rows stay unattributed. Retained observations provide partial coverage, never
+acceptance certification. Public text validation and bounded output remain.
+The same observation module supplies the artifact lifecycle's evidence guards.
+This read-only correction adds no stored state, authority or execution gate.
+
 ## 4. Mid-Term Applications (2-4 releases)
 
 1. **Global goal board**: aggregate the per-goal projection across goals into

docs/architecture/rfcs/goal-artifact-lifecycle-projection-v0.zh-CN.md

@@ -64,6 +64,17 @@ Milestone 的达成：优先取 goal 声明的 acceptance 标记；没有声明
 3. 第一个消费者：status markdown 与 dashboard 的 goal 详情（与 `goal_channel_projection_v0` 并列展示 `goal_artifact_lifecycle_projection_v0`），让运维能回答例子里的三个问题。
 4. 公开边界与现有投影一致：不含 raw logs、evidence body、credentials、本地路径。
 
+### 验收观察的数据来源
+
+现有 `goal_acceptance_observation_projection_v0` 从近期记录和保留的语义历史中
+识别 Agent lane。它调用现有 `goal_frontier.semantic_history` module 的 vision
+读取 interface。对已知 Agent，存在的语义历史决定当前 vision 或其缺失状态。
+退休标记或缺失的 lane 不能让较早的近期记录重新生效。没有语义历史时，保留
+原有近期列表回退。未归属的历史记录仍不归属到其他 Agent。保留的观察只提供
+部分覆盖，不证明验收完成。公开文本校验和输出上限保持不变。同一个观察
+module 为 artifact lifecycle 提供证据 guard。这个只读修复不新增存储状态、
+权限或执行 gate。
+
 ## 4. 中期应用（2-4 个 release）
 
 1. **全局 goal board**：把 per-goal 投影跨 goal 聚合——哪些 goal 达成哪些 milestone、哪些 guard 阻塞了多少 goal、每个 goal 的 next transition 是什么。这是现有 `/loopx-global-*` 家族和 operator dashboard 的自然落点。

loopx/control_plane/goals/acceptance_observation.py

@@ -9,10 +9,8 @@
 from ..todos.user_gate import open_user_gate_todo_items
 from ..todos.contract import normalize_todo_decision_scope
 from ..runtime.public_safety import public_safe_compact_text, validate_public_safe_value
-from .goal_vision_read_model import (
-    acceptance_gaps_from_agent_vision,
-    latest_agent_vision_from_runs,
-)
+from .goal_vision_read_model import acceptance_gaps_from_agent_vision
+from .goal_frontier.semantic_history import latest_agent_vision_from_status_payload
 from .goal_frontier.outcome_continuity import (
     acceptance_gaps_from_outcome_checkpoint,
     latest_outcome_vision_checkpoint_from_status_payload,
@@ -72,13 +70,15 @@ def build_goal_acceptance_observation(
     """Reuse acceptance rules over already-read runs, keeping per-agent lineage.
 
     The collected history and Todo projections are bounded; empty observations
-    cannot establish complete Goal acceptance. No file, quota, or mutation IO.
+    cannot establish complete Goal acceptance. Retained lane history, when
+    present, owns current vision identity. No file, quota, or mutation IO.
     """
     item = _dict(item)
     asset = _dict(item.get("project_asset"))
     runs = _rows(goal.get("latest_runs"))
     goal_id = _text(goal.get("id")) or "unknown"
     runs = [run for run in runs if run.get("goal_id") in (None, goal.get("id"))]
+    semantic_contexts = _rows(_dict(goal.get("semantic_history")).get("agents"))
     historical_progress: list[dict[str, Any]] = []
     seen_flags: set[str] = set()
     for source in runs:
@@ -110,13 +110,20 @@ def build_goal_acceptance_observation(
 
     agents = list(
         dict.fromkeys(
-            str(
-                _dict(run.get("agent_vision")).get("agent_id")
-                or run.get("agent_id")
-                or ""
-            )
-            for run in runs
-            if isinstance(run.get("agent_vision"), dict)
+            [
+                str(
+                    _dict(run.get("agent_vision")).get("agent_id")
+                    or run.get("agent_id")
+                    or ""
+                )
+                for run in runs
+                if isinstance(run.get("agent_vision"), dict)
+            ]
+            + [
+                str(context.get("agent_id") or "").strip()
+                for context in semantic_contexts
+                if str(context.get("agent_id") or "").strip()
+            ]
         )
     )
     gaps: list[dict[str, Any]] = []
@@ -132,8 +139,9 @@ def build_goal_acceptance_observation(
             )
             == agent
         ]
-        vision = latest_agent_vision_from_runs(
-            agent_runs, goal_id=goal_id, agent_id=agent or None
+        vision = latest_agent_vision_from_status_payload(
+            {"run_history": {"goals": [{**goal, "latest_runs": agent_runs}]}},
+            goal_id=goal_id, agent_id=agent or None,
         )
         patch = _dict(_dict(vision).get("vision_patch"))
         checkpoint = latest_outcome_vision_checkpoint_from_status_payload(
@@ -216,7 +224,7 @@ def build_goal_acceptance_observation(
         "goal_id": goal_id,
         "read_only": True,
         "acceptance_assessed": False,
-        "coverage": "partial" if item or runs else "unavailable",
+        "coverage": "partial" if item or runs or semantic_contexts else "unavailable",
         "missing_sources": sources_missing,
         "truncated": len(gaps) > OBSERVATION_LIMIT or len(guards) > OBSERVATION_LIMIT,
         "historical_progress": historical_progress[:OBSERVATION_LIMIT],

tests/control_plane/test_goal_acceptance_observation.py

@@ -2,8 +2,11 @@
 
 import copy
 import json
+from datetime import datetime, timezone
 from pathlib import Path
 
+import pytest
+
 from loopx.control_plane.goals.acceptance_observation import (
     build_goal_acceptance_observation,
 )
@@ -87,6 +90,105 @@ def test_latest_vision_is_per_agent_and_closed_lane_does_not_hide_other_lane():
     assert [gap["owner"] for gap in result["acceptance_gaps"]] == ["agent-b"]
 
 
+@pytest.mark.parametrize("mode,expected", [
+    ("retained", [("agent-a", "Current acceptance evidence")]),
+    ("replaced", [("agent-a", "Current acceptance evidence")]),
+    ("retired", []),
+    ("absent", []),
+    ("misattributed", []),
+    ("private", [("agent-a", None)]),
+])
+def test_acceptance_uses_the_current_semantic_lane(mode, expected):
+    stale = vision_run(acceptance="Obsolete acceptance evidence")
+    current = vision_run(acceptance="Current acceptance evidence")
+    current.update(run_id="current-vision", generated_at="2026-09-02T00:00:00Z")
+    foreign = {**vision_run("foreign-agent"), "goal_id": "another-goal"}
+    context = {"agent_id": "agent-a", "latest_agent_vision_run": current}
+    if mode == "retired":
+        context = {"agent_id": "agent-a", "vision_retired_at": "2026-09-02T00:00:00Z"}
+    elif mode == "misattributed":
+        context["latest_agent_vision_run"] = vision_run("agent-b")
+    elif mode == "private":
+        current["agent_vision"]["vision_patch"]["acceptance_summary"] = "x" * 500 + " /Users/private/report.json"
+    goal = {
+        "id": "acceptance-demo",
+        "latest_runs": [foreign] + ([] if mode in {"retained", "private"} else [stale]),
+        "semantic_history": {"agents": [] if mode == "absent" else [context]},
+    }
+    before = copy.deepcopy(goal)
+    result = build_goal_acceptance_observation(goal, None)
+    assert [(gap["owner"], gap["evidence_required"]) for gap in result["acceptance_gaps"]] == expected
+    assert result["acceptance_assessed"] is False
+    assert result["coverage"] == "partial"
+    validate_public_safe_value(result)
+    assert goal == before
+
+
+def test_unattributed_legacy_vision_is_not_assigned_to_a_retained_peer():
+    goal = {
+        "id": "acceptance-demo", "latest_runs": [vision_run("")],
+        "semantic_history": {"agents": [{
+            "agent_id": "agent-b", "latest_agent_vision_run": vision_run("agent-b"),
+        }]},
+    }
+    result = build_goal_acceptance_observation(goal, {})
+    assert [gap["owner"] for gap in result["acceptance_gaps"]] == [None, "agent-b"]
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("retired", [False, True])
+def test_status_keeps_current_acceptance_when_recent_history_changes(tmp_path, provider, retired):
+    from canonical_authority_fixture import promoted_create_fixture
+    from loopx.state_refresh import refresh_state_run
+
+    registry, runtime, state = promoted_create_fixture(
+        tmp_path, provider=provider, goal_id="acceptance-demo", registered_agents=["agent-a", "agent-b"],
+    )
+    configuration = json.loads(registry.read_text(encoding="utf-8"))
+    configuration["goals"][0].update(domain="software", adapter={"kind": "read_only_project_map_v0"})
+    registry.write_text(json.dumps(configuration), encoding="utf-8")
+    produced = refresh_state_run(
+        registry_path=registry, runtime_root_override=str(runtime), goal_id="acceptance-demo",
+        project=tmp_path / "project", state_file=state, classification="state_refreshed",
+        recommended_action="Verify the accepted outcome.", agent_id="agent-a",
+        agent_vision_packet={"vision_patch": {
+            "vision_summary": "Deliver the independently verified outcome.",
+            **vision_run()["agent_vision"]["vision_patch"],
+        }}, dry_run=False, sync_global=False,
+    )
+    assert produced["ok"] is True, produced
+    runs = runtime / "goals" / "acceptance-demo" / "runs"
+    index = runs / "index.jsonl"
+    source_paths = [registry, state, index, Path(produced["json_path"])]
+    with index.open("a", encoding="utf-8") as stream:
+        for number in range(1 if retired else 30):
+            run_path = runs / f"observation-{number}.json"
+            record = {
+                "goal_id": "acceptance-demo", "agent_id": "agent-a" if retired else "agent-b",
+                "generated_at": datetime.now(timezone.utc).isoformat(),
+                "classification": "state_refreshed" if retired else "monitor_poll",
+                "json_path": str(run_path),
+            }
+            if retired:
+                record["vision_checkpoint"] = {"satisfied": True, "decision": "retired_or_superseded"}
+            run_path.write_text(json.dumps(record), encoding="utf-8")
+            stream.write(json.dumps(record) + "\n")
+            source_paths.append(run_path)
+    before = {path: path.read_bytes() for path in source_paths}
+    for display_limit in (0, 5):
+        payload = collect_status(
+            registry_path=registry, runtime_root_override=str(runtime), limit=display_limit,
+        )
+        assert payload["ok"] is True, payload.get("contract")
+        goal = payload["run_history"]["goals"][0]
+        assert len(goal["latest_runs"]) <= display_limit
+        observation = goal["acceptance_observation"]
+        assert [gap["owner"] for gap in observation["acceptance_gaps"]] == ([] if retired else ["agent-a"])
+        assert observation["acceptance_assessed"] is False
+        validate_public_safe_value(observation)
+    assert {path: path.read_bytes() for path in source_paths} == before
+
+
 def test_missing_history_and_empty_observations_are_never_complete():
     result = build_goal_acceptance_observation(
         {"id": "acceptance-demo", "lifecycle_flags": [{}, "connected"]}, None