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,14 @@ 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 same
+  receipt-bound identity is also projected as `selected_todo` and, when its
+  recorded monitor item is available, `agent_lane_next_action` with
+  `receipt_bound_monitor_phase=settled`. These are historical readback, not
+  executable selection: `should_run=false`, `must_attempt_work=false` and
+  `effective_action=heartbeat_settled_skip` remain authoritative. Consumers
+  must not require the Monitor to remain in the open frontier or infer a new
+  poll, delivery or spend from the presence of these identity fields.
   Uncommitted observation rows and auxiliary polls for another Todo do not
   qualify this closeout.
 

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",

tests/control_plane/test_monitor_followthrough_contract.py

@@ -634,7 +634,8 @@ def test_same_turn_material_monitor_poll_is_no_spend_closeout_before_successor(
         runtime_root=runtime,
     )
     assert replay["heartbeat_receipt"]["settlement_identity"]["todo_id"] == admitted["todo_id"]
-    assert replay.get("selected_todo") is None
+    assert replay["selected_todo"]["todo_id"] == admitted["todo_id"]
+    assert replay["agent_lane_next_action"]["receipt_bound_monitor_phase"] == "settled"
     assert replay["should_run"] is False
     assert replay["effective_action"] == "heartbeat_settled_skip"
     assert replay["execution_obligation"]["must_attempt_work"] is False
@@ -800,7 +801,8 @@ def test_same_turn_unchanged_monitor_poll_is_already_settled(tmp_path: Path) ->
         runtime_root=runtime,
     )
     assert replay["heartbeat_receipt"]["settlement_identity"]["todo_id"] == monitor["todo_id"]
-    assert replay.get("selected_todo") is None
+    assert replay["selected_todo"]["todo_id"] == monitor["todo_id"]
+    assert replay["agent_lane_next_action"]["receipt_bound_monitor_phase"] == "settled"
     assert replay["effective_action"] == "heartbeat_settled_skip"
     assert replay["execution_obligation"]["must_attempt_work"] is False
     assert replay["heartbeat_recommendation"]["agent_must_attempt"] is False

tests/control_plane/test_runtime_source_read_batching.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import os
 from pathlib import Path
 
 import pytest
@@ -9,8 +10,11 @@
 
 
 def serial_fingerprint(root: Path) -> str:
-    """Independent reference: names and raw bytes, not decoded source text."""
+    """Independent reference: release identity, names, and raw source bytes."""
     digest = hashlib.sha256()
+    digest.update(b"loopx_effect_runtime_source_instance_v1\0")
+    digest.update(os.fsencode(os.path.normcase(os.fspath(root.resolve()))))
+    digest.update(b"\0")
     for path in sorted(p for p in root.rglob("*") if p.suffix in {".ts", ".json"}):
         digest.update(path.relative_to(root).as_posix().encode("utf-8"))
         digest.update(path.read_bytes())

tests/control_plane/test_settled_monitor_user_gate.py

@@ -83,7 +83,8 @@ def call(*args):
         claimed_by=AGENT_ID, agent_id=AGENT_ID)
     replay = call(*guard_args)
     assert replay["heartbeat_receipt"]["settlement_identity"] == admitted["heartbeat_receipt"]["settlement_identity"]
-    assert replay.get("selected_todo") is None
+    assert replay["selected_todo"]["todo_id"] == monitor["todo_id"]
+    assert replay["agent_lane_next_action"]["receipt_bound_monitor_phase"] == "settled"
     assert replay["should_run"] is False
     assert replay["safe_bypass_allowed"] is False
     assert replay["execution_obligation"]["must_attempt_work"] is False