fix(quota): fence settlement by exact GoalRef

docs/architecture/rfcs/goal-instance-identity-and-orphan-recovery-v0.md

@@ -840,6 +840,28 @@ promotion retain their own acceptance. No new paid cohort or soak is authorized.
   drain, unsupported/warm binary coverage, or any other M3 row.
   `execution_authority: false` and the overall activation hold remain.
 
+### 2026-09-30: M3 quota settlement owner candidate
+
+- **Baseline:** `3ec049e13`.
+- **Proposed:** Bind source-profile quota spend, replay, receipt repair, void,
+  settlement readback, and rolling-window accounting to the caller-captured
+  exact GoalRef. Python hands the ordered run-index and Goal-lifecycle lock
+  witnesses to the TypeScript accounting owner. TypeScript validates and claims
+  both witnesses, reuses `decideFirstPartyHostRuntime(require_current)`, and
+  keeps the owner fence through receipt and artifact commit.
+- **Evidence:** TypeScript and Python integration tests publish same-alias Goal
+  B after Goal A capture and prove that stale A writes nothing. They also cover
+  B-only spend and void, cross-instance replay and prepared-receipt repair
+  rejection, exact settlement readback, and per-instance rolling-window
+  accounting.
+- **Compatibility:** Non-source spend, replay, void, and readback requests omit
+  GoalRef and source admission. Their persisted records, receipts, response
+  payloads, and lock behavior retain the legacy shape.
+- **Remaining hold:** This qualifies only the `quota_settlement` inventory row.
+  Unsupported and warm binaries, downstream external-effect drain, and every
+  other unqualified M3 owner remain blocked. `execution_authority: false` and
+  the overall activation hold remain unchanged.
+
 ## Appendix B: Decision log
 
 | Date | Decision | Owner / approval | Alternatives | Normative sections changed |

docs/architecture/rfcs/goal-instance-identity-and-orphan-recovery-v0.zh-CN.md

@@ -761,6 +761,26 @@ service adoption、D1–D3 provider promotion 保留各自验收。不授权付
   binary 或其他 M3 行已完成。`execution_authority: false` 和总 activation hold
   保持不变。
 
+### 2026-09-30：M3 quota settlement owner 候选
+
+- **基线：** `3ec049e13`。
+- **候选实现：** Source profile 的 quota spend、replay、receipt repair、void、
+  settlement readback 与 rolling-window accounting 均绑定调用方预先捕获的精确
+  GoalRef。Python 按顺序把 run-index 与 Goal lifecycle lock witness 交接给
+  TypeScript accounting owner。TypeScript 校验并 claim 两个 witness，复用
+  `decideFirstPartyHostRuntime(require_current)`，并保持 owner fence，直到 receipt
+  与 artifact commit 完成。
+- **证据：** TypeScript 与 Python 集成测试在 Goal A 捕获后发布同名 Goal B，
+  证明迟到的 A 不产生任何写入。测试还覆盖 B 独立 spend/void、跨实例 replay
+  和 prepared receipt repair 拒绝、精确 settlement readback，以及按实例隔离的
+  rolling-window accounting。
+- **兼容性：** 非 source 的 spend、replay、void 与 readback 请求不携带 GoalRef
+  或 source admission；其持久化 record、receipt、响应 payload 和锁行为保持
+  legacy 形态。
+- **剩余 hold：** 本切片只资格化 `quota_settlement` inventory 行。不支持及常驻
+  binary、downstream external-effect drain 和其他未资格化 M3 owner 继续受阻。
+  `execution_authority: false` 和总 activation hold 保持不变。
+
 ## 附录 B：决策日志
 
 | 日期 | 决策 | Owner／批准 | 替代方案 | 变更的规范章节 |

loopx/capabilities/multi_subagent/cli.py

@@ -2,7 +2,13 @@
 
 from __future__ import annotations
 
+import argparse
+
 from ...agent_registry import load_goal_from_registry, registered_agent_ids_for_goal
+from ...control_plane.goals.first_party_host_admission import (
+    capture_first_party_host_goal_ref,
+)
+from ...control_plane.goals.source_session_registry_state import exact_goal_ref
 from ...orchestration import compact_orchestration_policy
 from .native_child_receipts import load_native_child_activity, record_native_child
 
@@ -16,6 +22,7 @@ def register_native_child_commands(subparsers, add_format):
     parser.add_argument("--goal-id", required=True)
     parser.add_argument("--agent-id", required=True)
     parser.add_argument("--turn-instance-id", required=True)
+    parser.add_argument("--goal-instance-id", help=argparse.SUPPRESS)
     parser.add_argument("--operation-id", help="Stable identity for one host-native child operation.")
     parser.add_argument("--stage", choices=("decision", "result", "review"))
     parser.add_argument("--operation", choices=("spawn", "followup", "skip"))
@@ -31,6 +38,17 @@ def handle_native_child_command(args, registry_path, runtime_root, print_payload
     if args.command != "native-child":
         return None
     try:
+        goal_instance_id = str(
+            getattr(args, "goal_instance_id", None) or ""
+        ).strip()
+        goal_ref = (
+            exact_goal_ref(args.goal_id, goal_instance_id)
+            if goal_instance_id
+            else capture_first_party_host_goal_ref(
+                registry_path=registry_path,
+                goal_id=args.goal_id,
+            )
+        )
         goal = load_goal_from_registry(registry_path, args.goal_id)
         if goal is None or args.agent_id not in registered_agent_ids_for_goal(goal):
             raise ValueError("coordinator is not registered for this Goal")
@@ -50,6 +68,7 @@ def handle_native_child_command(args, registry_path, runtime_root, print_payload
             payload = {"ok": True, "native_child_activity": load_native_child_activity(
                 runtime_root, goal_id=args.goal_id, agent_id=args.agent_id,
                 turn_instance_id=args.turn_instance_id, configured_limit=configured_limit,
+                registry_path=registry_path, goal_ref=goal_ref,
             )}
         else:
             if not args.operation_id or not args.stage or not args.outcome:
@@ -62,6 +81,7 @@ def handle_native_child_command(args, registry_path, runtime_root, print_payload
                 entrypoint_id=args.entrypoint_id, reason_code=args.reason_code,
                 evidence_ref=args.evidence_ref, validation_ref=args.validation_ref,
                 execute=args.execute,
+                registry_path=registry_path, goal_ref=goal_ref,
             )
     except (OSError, ValueError, KeyError, RuntimeError) as exc:
         payload = {"ok": False, "error": str(exc)}

loopx/capabilities/multi_subagent/native_child_receipts.py

@@ -13,6 +13,7 @@
 from pathlib import Path
 from typing import Any
 
+from ...control_plane.quota.accounting_admission import quota_accounting_admission
 from ...control_plane.quota.settlement import read_heartbeat_settlement
 from ...control_plane.runtime.public_safety import validate_public_safe_value
 from ...rollout_event_log import (
@@ -64,20 +65,33 @@ def _details(event: Mapping[str, Any]) -> dict[str, Any]:
 
 def _events_for_turn(
     events: Sequence[Mapping[str, Any]], *, goal_id: str, agent_id: str, turn_instance_id: str,
+    goal_ref: Mapping[str, Any] | None = None,
 ) -> list[dict[str, Any]]:
     return [dict(event) for event in events
             if event.get("event_kind") in EVENT_KINDS.values()
             and event.get("goal_id") == goal_id
             and event.get("agent_id") == agent_id
-            and event.get("run_id") == turn_instance_id]
+            and event.get("run_id") == turn_instance_id
+            and (
+                event.get("goal_ref") == dict(goal_ref)
+                if goal_ref is not None
+                else "goal_ref" not in event
+            )]
 
 
 def native_child_activity(
     events: Sequence[Mapping[str, Any]], *, goal_id: str, agent_id: str,
     turn_instance_id: str, configured_limit: int,
+    goal_ref: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """One read model for CLI, agent-context and product projections."""
-    rows = _events_for_turn(events, goal_id=goal_id, agent_id=agent_id, turn_instance_id=turn_instance_id)
+    rows = _events_for_turn(
+        events,
+        goal_id=goal_id,
+        agent_id=agent_id,
+        turn_instance_id=turn_instance_id,
+        goal_ref=goal_ref,
+    )
     operations: dict[str, dict[str, Any]] = {}
     for event in rows:
         details = _details(event)
@@ -134,32 +148,80 @@ def native_child_activity(
 def load_native_child_activity(
     runtime_root: Path, *, goal_id: str, agent_id: str,
     turn_instance_id: str, configured_limit: int,
+    goal_ref: Mapping[str, Any] | None = None,
+    registry_path: Path | None = None,
 ) -> dict[str, Any]:
-    source = iter_rollout_events(rollout_event_log_path(runtime_root, goal_id))
-    events = [event for event in source
-              if event.get("event_kind") in EVENT_KINDS.values()
-              and event.get("agent_id") == agent_id
-              and event.get("run_id") == turn_instance_id]
-    return native_child_activity(
-        events, goal_id=goal_id, agent_id=agent_id,
-        turn_instance_id=turn_instance_id, configured_limit=configured_limit,
-    )
+    with quota_accounting_admission(
+        runtime_root=runtime_root,
+        registry_path=registry_path,
+        goal_id=goal_id,
+        goal_ref=goal_ref,
+        operation="native-child-read",
+        lock_legacy_index=False,
+    ) as source_admission:
+        if source_admission is not None:
+            readback = read_heartbeat_settlement(
+                runtime_root,
+                goal_id=goal_id,
+                agent_id=agent_id,
+                todo_id=None,
+                turn_instance_id=turn_instance_id,
+                resolve_original_binding=True,
+                registry_path=registry_path,
+                goal_ref=goal_ref,
+                source_admission=source_admission,
+                borrow_source_admission=True,
+            )
+            if readback is None or readback.identity.value is None:
+                raise ValueError(
+                    "native child read requires an admitted, settlement-bound Turn guard"
+                )
+        source = iter_rollout_events(
+            rollout_event_log_path(runtime_root, goal_id)
+        )
+        events = [
+            event
+            for event in source
+            if event.get("event_kind") in EVENT_KINDS.values()
+            and event.get("agent_id") == agent_id
+            and event.get("run_id") == turn_instance_id
+            and (
+                event.get("goal_ref") == dict(goal_ref)
+                if goal_ref is not None
+                else "goal_ref" not in event
+            )
+        ]
+        return native_child_activity(
+            events,
+            goal_id=goal_id,
+            agent_id=agent_id,
+            turn_instance_id=turn_instance_id,
+            configured_limit=configured_limit,
+            goal_ref=goal_ref,
+        )
 
 
 def latest_native_child_activity(
     events: Sequence[Mapping[str, Any]], *, goal_id: str, configured_limit: int,
+    goal_ref: Mapping[str, Any] | None = None,
 ) -> dict[str, Any] | None:
     """Expose only the latest reported Turn in existing Goal status surfaces."""
     observations = [event for event in events
                     if event.get("goal_id") == goal_id
                     and event.get("event_kind") in EVENT_KINDS.values()
-                    and event.get("agent_id") and event.get("run_id")]
+                    and event.get("agent_id") and event.get("run_id")
+                    and (
+                        event.get("goal_ref") == dict(goal_ref)
+                        if goal_ref is not None
+                        else "goal_ref" not in event
+                    )]
     if not observations:
         return None
     latest = max(observations, key=lambda event: str(event.get("recorded_at") or ""))
     return native_child_activity(
         events, goal_id=goal_id, agent_id=str(latest["agent_id"]),
         turn_instance_id=str(latest["run_id"]), configured_limit=configured_limit,
+        goal_ref=goal_ref,
     )
 
 
@@ -212,13 +274,16 @@ def _normalized_fields(
     raise ValueError("stage must be decision, result or review")
 
 
-def record_native_child(
+def _record_native_child(
     *, runtime_root: Path, goal_id: str, agent_id: str,
     turn_instance_id: str, operation_id: str, configured_limit: int,
     stage: str, outcome: str, operation: str | None = None,
     entrypoint_id: str | None = None, reason_code: str | None = None,
     evidence_ref: str | None = None, validation_ref: str | None = None,
     execute: bool = False,
+    registry_path: Path | None = None,
+    goal_ref: Mapping[str, Any] | None = None,
+    source_admission: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Preview or append a typed report; never launch a child or spend quota."""
     goal_id = _id(goal_id, field="goal_id")
@@ -233,7 +298,13 @@ def record_native_child(
     )
     log_path = rollout_event_log_path(runtime_root, goal_id)
     events = load_rollout_events(log_path)
-    prior = _events_for_turn(events, goal_id=goal_id, agent_id=agent_id, turn_instance_id=turn_instance_id)
+    prior = _events_for_turn(
+        events,
+        goal_id=goal_id,
+        agent_id=agent_id,
+        turn_instance_id=turn_instance_id,
+        goal_ref=goal_ref,
+    )
     existing = next((event for event in prior
                      if event.get("case_id") == operation_id
                      and event.get("event_kind") == EVENT_KINDS[stage]), None)
@@ -244,6 +315,9 @@ def report_admission() -> Mapping[str, Any]:
         readback = read_heartbeat_settlement(
             runtime_root, goal_id=goal_id, agent_id=agent_id, todo_id=None,
             turn_instance_id=turn_instance_id, resolve_original_binding=True,
+            registry_path=registry_path, goal_ref=goal_ref,
+            source_admission=source_admission,
+            borrow_source_admission=source_admission is not None,
         )
         if readback is None or readback.identity.value is None or readback.identity.failure is not None:
             reason = (readback.identity.failure.reason
@@ -261,7 +335,8 @@ def report_admission() -> Mapping[str, Any]:
     def validate_transition(observed: Sequence[Mapping[str, Any]]) -> None:
         admission = report_admission()
         current = _events_for_turn(observed, goal_id=goal_id, agent_id=agent_id,
-                                   turn_instance_id=turn_instance_id)
+                                   turn_instance_id=turn_instance_id,
+                                   goal_ref=goal_ref)
         decisions = {str(item.get("case_id")): item for item in current
                      if item.get("event_kind") == EVENT_KINDS["decision"]}
         if stage == "decision":
@@ -293,12 +368,20 @@ def validate_transition(observed: Sequence[Mapping[str, Any]]) -> None:
         goal_id=goal_id, event_kind=EVENT_KINDS[stage], agent_id=agent_id,
         run_id=turn_instance_id, case_id=operation_id, status=fields["outcome"],
         details=fields, recorded_at=(existing or {}).get("recorded_at"),
+        goal_ref=goal_ref,
     )
     appended = False
     if execute:
         stored, appended = append_rollout_event_once(
             log_path, event,
-            identity_fields=("goal_id", "event_kind", "agent_id", "run_id", "case_id"),
+            identity_fields=(
+                "goal_id",
+                "event_kind",
+                "agent_id",
+                "run_id",
+                "case_id",
+                *(("goal_ref",) if goal_ref is not None else ()),
+            ),
             precondition=lambda: validate_transition(load_rollout_events(log_path)),
         )
         if _details(stored) != fields:
@@ -318,5 +401,46 @@ def validate_transition(observed: Sequence[Mapping[str, Any]]) -> None:
             events if execute or existing else [*events, event], goal_id=goal_id,
             agent_id=agent_id, turn_instance_id=turn_instance_id,
             configured_limit=configured_limit,
+            goal_ref=goal_ref,
         ),
     }
+
+
+def record_native_child(
+    *, runtime_root: Path, goal_id: str, agent_id: str,
+    turn_instance_id: str, operation_id: str, configured_limit: int,
+    stage: str, outcome: str, operation: str | None = None,
+    entrypoint_id: str | None = None, reason_code: str | None = None,
+    evidence_ref: str | None = None, validation_ref: str | None = None,
+    execute: bool = False, registry_path: Path | None = None,
+    goal_ref: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Preview or append one report under its exact quota owner."""
+
+    with quota_accounting_admission(
+        runtime_root=runtime_root,
+        registry_path=registry_path,
+        goal_id=goal_id,
+        goal_ref=goal_ref,
+        operation="native-child-report",
+        lock_legacy_index=False,
+    ) as source_admission:
+        return _record_native_child(
+            runtime_root=runtime_root,
+            goal_id=goal_id,
+            agent_id=agent_id,
+            turn_instance_id=turn_instance_id,
+            operation_id=operation_id,
+            configured_limit=configured_limit,
+            stage=stage,
+            outcome=outcome,
+            operation=operation,
+            entrypoint_id=entrypoint_id,
+            reason_code=reason_code,
+            evidence_ref=evidence_ref,
+            validation_ref=validation_ref,
+            execute=execute,
+            registry_path=registry_path,
+            goal_ref=goal_ref,
+            source_admission=source_admission,
+        )