fix(ci): reconcile exact-target recovery and settled readback

.github/workflows/python-tests.yml

@@ -197,7 +197,10 @@ jobs:
 
       - name: Qualify agent-facing CLI output
         env:
-          LOOPX_CLI_OUTPUT_BASE_REF: origin/${{ github.event.pull_request.base.ref || 'main' }}
+          # Compare the tested revision to its event base, even after main moves
+          # while this run is queued. Pin the selector's main ref as well.
+          LOOPX_CLI_OUTPUT_BASE_REF: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.sha }}
+          LOOPX_CLI_OUTPUT_MAIN_REF: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.sha }}
         run: python examples/control_plane/cli-output-budget-regression-smoke.py
 
   typescript-core:

docs/development/testing-and-quality.md

@@ -794,6 +794,9 @@ it does not grant execution quota, spending, or provider authority.
    sample window, workload and distribution; acknowledge noise. Keep the
    failing scenario and original result; do not shrink fixture populations,
    scan roots or sampling depth to obtain a pass.
+   CI comparisons must use the event's immutable base and candidate revisions,
+   including merge-queue bases. Do not resolve a moving branch after queueing:
+   unrelated changes on that branch are not regressions in the tested commit.
 2. **Inspect information value and redundancy.** Name the current consumer and
    decision each changed field supports. Remove derivable or unused copies when
    the consumer contract permits it. Similar rows in different lanes may serve
@@ -815,6 +818,8 @@ it does not grant execution quota, spending, or provider authority.
 1. **同口径测量。** 记录 base/head、负载、指标和测量边界。紧凑 JSON 字符、UTF-8
    字节、嵌套键数、真实 stdout 和 token 不可互换。延迟要保留样本窗口、负载和
    分布，并承认噪声。保留失败场景与原结果，不缩小 fixture、扫描范围或采样深度。
+   CI 对照必须固定事件的 base 和 candidate 提交，包括合并队列的 base；不能在排队
+   后重新解析移动分支，把其他提交的变化当作被测提交的回归。
 2. **分析信息价值与真实冗余。** 说明变化字段服务哪个消费者、哪个决策。合同允许时
    删除可推导或无人使用的副本；不同 lane 中相同的数据可能服务不同消费者，去重
    需要调用方迁移和语义等价验证。详情优先使用有界摘要和可达冷路径。不能删身份、

docs/reference/protocols/quota-monitor-observation-receipt-v0.md

@@ -76,9 +76,19 @@ advancement work remains active.
   Turn recomputes them. The automation stays active and quiet between Turns.
   Completing, superseding or archiving the Monitor cannot reopen its committed
   Turn. The settlement reader reports `replay_phase=settled`; CLI replay retains
-  the original identity in `heartbeat_receipt.settlement_identity` and has no
-  executable `selected_todo`. Consumers must use the receipt for historical
-  identity instead of requiring the Monitor to remain in the open frontier.
+  the original identity in `heartbeat_receipt.settlement_identity`, the durable
+  historical identity of the committed Turn, which stays readable after the
+  Monitor is completed, superseded or archived. While the bound Monitor still
+  projects as a current work item, the same identity also appears as
+  `selected_todo` and `agent_lane_next_action` with
+  `receipt_bound_monitor_phase=settled`. Those two work-lane projections are
+  conditional: once the bound Monitor is no longer reconstructible as current
+  work, both are absent and only the receipt retains the identity. Neither form
+  is executable selection: `should_run=false`, `must_attempt_work=false` and
+  `effective_action=heartbeat_settled_skip` remain authoritative. Consumers
+  must read historical identity from the receipt, must not require these fields
+  to be present, must not require the Monitor to remain in the open frontier,
+  and must not infer a new poll, delivery or spend from their presence.
   Uncommitted observation rows and auxiliary polls for another Todo do not
   qualify this closeout.
 
@@ -286,9 +296,15 @@ using a complete read-only snapshot with disposable File/SQLite/PostgreSQL arms.
   未决 gate 和独立工作保留为诊断事实，由新 Turn 重新计算；自动化保持 active quiet。
   Monitor 完成、被替代或归档都不能重开已提交的 Turn。结算读取返回
   `replay_phase=settled`；CLI 重放通过 `heartbeat_receipt.settlement_identity`
-  保留原身份，不再投影可执行的 `selected_todo`。消费者应从回执读取历史身份，
-  不要求 Monitor 继续出现在未完成列表中。未提交的观察行，以及针对另一个 Todo
-  的辅助 poll，均不能构成该结算依据。
+  保留原身份，该回执是已提交 Turn 的持久历史身份，在 Monitor 完成、被替代或归档
+  后仍可读。当绑定的 Monitor 仍投影为当前工作项时，同一身份同时出现在
+  `selected_todo` 与 `agent_lane_next_action`（`receipt_bound_monitor_phase=settled`）。
+  这两个工作 lane 投影是有条件的：一旦绑定 Monitor 不再可重建为当前工作，二者均
+  不出现，只有回执保留身份。两种形式都不是可执行选择：`should_run=false`、
+  `must_attempt_work=false`、`effective_action=heartbeat_settled_skip` 仍然权威。
+  消费者应从回执读取历史身份，不得要求这些字段存在，也不得因字段存在推断新的
+  poll、delivery 或 spend，或要求 Monitor 继续出现在未完成列表中。未提交的观察行，
+  以及针对另一个 Todo 的辅助 poll，均不能构成该结算依据。
 
 ### 验收
 

loopx/capabilities/manager_context/__init__.py

@@ -11,6 +11,7 @@
     POLICY_SCHEMA as POLICY_SCHEMA,
     registered_context_recipients,
     source_context_authority,
+    source_context_target_authority,
 )
 from ...control_plane.collaboration.goal_instance_scope import (
     collaboration_goal_scope,
@@ -86,6 +87,15 @@ def authority(
     grant = source_context_authority(runtime_root, registry_path, session, turn)
     return {**grant, "instruction": INSTRUCTION} if grant["mode"] == "context_only" else grant
 
+
+def target_authority(
+    runtime_root: Path, *, session: dict, turn: dict, target: dict
+) -> dict:
+    """Authorize one target already validated by an exact Goal scope."""
+    grant = source_context_target_authority(runtime_root, session, turn, target)
+    return {**grant, "instruction": INSTRUCTION} if grant["mode"] == "context_only" else grant
+
+
 def deliver(
     runtime_root: Path, registry_path: Path, *, session: dict, turn: dict, request: dict
 ) -> dict:
@@ -101,7 +111,16 @@ def deliver(
             goal_scope,
             operation="request_create",
         )
-        grant = authority(runtime_root, registry_path, session, turn)
+        grant = (
+            target_authority(
+                runtime_root,
+                session=session,
+                turn=turn,
+                target=target,
+            )
+            if goal_scope.exact
+            else authority(runtime_root, registry_path, session, turn)
+        )
         if target not in grant["targets"]:
             raise ValueError("context recipient is not authorized or registered")
         content = str(turn.get("message") or "")

loopx/capabilities/manager_context/roundtrip.py

@@ -12,7 +12,7 @@
 from datetime import datetime, timezone, timedelta
 from uuid import uuid4
 
-from . import _root, _read, _write, _hash, authority
+from . import _root, _read, _write, _hash, authority, target_authority
 from .tracking import _entry, _now
 from ...file_lock import (
     LockAcquisitionPolicy,
@@ -380,7 +380,7 @@ def _exact_return_scope(registry, reply):
     return collaboration_goal_scope(
         registry,
         goal_id=reply["goal_id"],
-        agents=(),
+        agents=(reply["agent_id"],),
         caller_goal_ref=reply["goal_ref"],
     )
 
@@ -418,8 +418,13 @@ def _exact_return_context(root, registry, store, path, state_path, now):
             or not turn
         ):
             raise ValueError("original_conversation_unavailable")
-        grant = authority(root, registry, session, turn)
         target = {key: row[key] for key in ("goal_id", "agent_id")}
+        grant = target_authority(
+            root,
+            session=session,
+            turn=turn,
+            target=target,
+        )
         if (
             target not in grant["targets"]
             or grant.get("source_id") != row["source_id"]

loopx/control_plane/collaboration/source_grant_observation.py

@@ -33,27 +33,26 @@ def registered_context_recipients(registry: dict) -> dict:
     return {"active_goal_ids": active_goals, "available": available}
 
 
-def source_context_authority(
-    runtime_root: Path, registry_path: Path, session: dict, turn: dict
+def _source_context_grant(
+    runtime_root: Path,
+    session: dict,
+    turn: dict,
+    available_rows: list[dict],
 ) -> dict:
-    """Return only a write-only recipient catalog; no cross-audience Goal evidence."""
-    if registry_path is None:
-        return {"mode": "unavailable", "targets": []}
-    try:
-        registry = load_project_registry(registry_path)
-        if not isinstance(registry, dict):
-            raise ValueError("invalid registry")
-        require_runtime_compatible_project_registry(
-            registry, operation="context source recipient observation"
-        )
-    except (OSError, ValueError, TypeError):
-        return {"mode": "unavailable", "targets": []}
-    observed = registered_context_recipients(registry)
-    available = {(row["goal_id"], row["agent_id"]) for row in observed["available"]}
+    available = {
+        (row["goal_id"], row["agent_id"])
+        for row in available_rows
+        if isinstance(row, dict)
+        and isinstance(row.get("goal_id"), str)
+        and isinstance(row.get("agent_id"), str)
+    }
     scope = conversation_scope(session, origin=turn.get("origin", "unknown"))
     if scope["private_conversation"] and turn.get("origin") == "web":
-        allowed = {target for target in available
-                   if scope["goal_ids"] is None or target[0] in scope["goal_ids"]}
+        allowed = {
+            target
+            for target in available
+            if scope["goal_ids"] is None or target[0] in scope["goal_ids"]
+        }
         source_id = "web:" + _hash([session["session_id"], turn["client_turn_id"]])
     else:
         if scope["kind"] != "external_audience":
@@ -74,19 +73,68 @@ def source_context_authority(
             if policy.get("schema_version") != POLICY_SCHEMA:
                 raise ValueError("invalid policy")
             grants = policy.get("sources", {}).get(ingress["channel"], {})
-            selected = effect_runtime_result("collaboration.source.recipients", {
-                "source": grants, "sender_id": ingress["sender_id"],
-                "available": observed["available"],
-            })
-            allowed = {(v["goal_id"], v["agent_id"]) for v in selected["targets"]}
+            selected = effect_runtime_result(
+                "collaboration.source.recipients",
+                {
+                    "source": grants,
+                    "sender_id": ingress["sender_id"],
+                    "available": available_rows,
+                },
+            )
+            allowed = {
+                (value["goal_id"], value["agent_id"])
+                for value in selected["targets"]
+            }
             source_id = ingress["source_id"]
-        except (OSError, ValueError, KeyError, TypeError, AttributeError, EffectRuntimeRejected):
+        except (
+            OSError,
+            ValueError,
+            KeyError,
+            TypeError,
+            AttributeError,
+            EffectRuntimeRejected,
+        ):
             return {"mode": "unavailable", "targets": []}
     targets = [
-        {"goal_id": g, "agent_id": a} for g, a in sorted(allowed & available)
+        {"goal_id": goal_id, "agent_id": agent_id}
+        for goal_id, agent_id in sorted(allowed & available)
     ]
     return {
         "mode": "context_only",
         "targets": targets,
         "source_id": source_id,
     }
+
+
+def source_context_target_authority(
+    runtime_root: Path,
+    session: dict,
+    turn: dict,
+    target: dict,
+) -> dict:
+    """Authorize one target whose exact Goal scope was already validated."""
+    return _source_context_grant(runtime_root, session, turn, [target])
+
+
+def source_context_authority(
+    runtime_root: Path, registry_path: Path, session: dict, turn: dict
+) -> dict:
+    """Return only a write-only recipient catalog; no cross-audience Goal evidence."""
+    if registry_path is None:
+        return {"mode": "unavailable", "targets": []}
+    try:
+        registry = load_project_registry(registry_path)
+        if not isinstance(registry, dict):
+            raise ValueError("invalid registry")
+        require_runtime_compatible_project_registry(
+            registry, operation="context source recipient observation"
+        )
+    except (OSError, ValueError, TypeError):
+        return {"mode": "unavailable", "targets": []}
+    observed = registered_context_recipients(registry)
+    return _source_context_grant(
+        runtime_root,
+        session,
+        turn,
+        observed["available"],
+    )

loopx/control_plane/quota/settlement_precedence.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 from .effective_action import EffectiveAction
+from .selected_todo_projection import selected_todo_projection
 from ..work_items.work_lane import work_lane_contract_is_receipt_bound_monitor_settled
 
 from typing import Any
@@ -129,10 +130,24 @@ def apply_settled_monitor_precedence(payload: dict[str, Any]) -> None:
     recorded_action = payload.get("agent_lane_next_action")
     clear_quota_action_projections(payload)
     payload.update(settled_replay_fields())
-    if (
-        isinstance(recorded_action, dict)
-        and recorded_action.get("selection_binding") == "heartbeat_receipt"
-        and isinstance(lane, dict)
-        and recorded_action.get("todo_id") == lane.get("selected_todo_id")
-    ):
-        payload["agent_lane_next_action"] = recorded_action
+    bound_action = next(
+        (
+            candidate
+            for candidate in (
+                recorded_action,
+                lane.get("receipt_bound_monitor_item"),
+            )
+            if isinstance(candidate, dict)
+            and candidate.get("selection_binding") == "heartbeat_receipt"
+            and candidate.get("todo_id") == lane.get("selected_todo_id")
+        ),
+        None,
+    )
+    if bound_action is not None:
+        payload["agent_lane_next_action"] = bound_action
+    selected_todo = selected_todo_projection(
+        agent_lane_next_action=bound_action,
+        work_lane_contract=lane,
+    )
+    if selected_todo is not None:
+        payload["selected_todo"] = selected_todo

loopx/semantics/project_registry_io_manifest_v1.json

@@ -215,15 +215,15 @@
     },
     {
       "site": "loopx/capabilities/manager_context/__init__.py::<module>.configure_delivery_target.update::codec_read:load_project_registry#1",
-      "line": 358,
+      "line": 377,
       "column": 51,
       "kind": "codec_read",
       "api": "load_project_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/capabilities/manager_context/__init__.py::<module>.configure_evidence_scope::codec_read:load_project_registry#1",
-      "line": 313,
+      "line": 332,
       "column": 16,
       "kind": "codec_read",
       "api": "load_project_registry",
@@ -247,7 +247,7 @@
     },
     {
       "site": "loopx/capabilities/manager_context/roundtrip.py::<module>.drain::codec_read:load_project_registry#1",
-      "line": 879,
+      "line": 884,
       "column": 22,
       "kind": "codec_read",
       "api": "load_project_registry",
@@ -1015,7 +1015,7 @@
     },
     {
       "site": "loopx/control_plane/collaboration/source_grant_observation.py::<module>.source_context_authority::codec_read:load_project_registry#1",
-      "line": 43,
+      "line": 126,
       "column": 20,
       "kind": "codec_read",
       "api": "load_project_registry",