From c85c6ffc37ea5c5500a3ccb7c0e3af2a02c513f7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 05:48:04 +0800 Subject: [PATCH 1/2] fix(quota): bound settlement wait transport without losing facts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/effect_runtime.py | 1 + loopx/control_plane/effect_runtime_server.ts | 1 + loopx/control_plane/quota/blocked_retry.py | 25 ++- loopx/control_plane/quota/settlement.py | 1 + .../quota/unsettled_host_turn.py | 10 +- .../test_quota_settlement_cli.py | 210 ++++++++++++++++++ 6 files changed, 239 insertions(+), 9 deletions(-) diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 823c224e4d..55b41bd560 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -46,6 +46,7 @@ "todo.context.page", "work_item.context.project", "work_item.replan_semantics.project", + "quota.settlement.read", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", "goal.checkpoint_read_context.inspect_replay", diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 7f024eed59..a18b8941fe 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -32,6 +32,7 @@ const LOCAL_SNAPSHOT_METHODS = new Set([ "todo.context.page", "work_item.context.project", "work_item.replan_semantics.project", + "quota.settlement.read", "goal.checkpoint_read_context.resolve", "goal.checkpoint_read_context.commit", "goal.checkpoint_read_context.inspect_replay", diff --git a/loopx/control_plane/quota/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py index 45ce39108b..7d5e25e1f6 100644 --- a/loopx/control_plane/quota/blocked_retry.py +++ b/loopx/control_plane/quota/blocked_retry.py @@ -6,6 +6,7 @@ from typing import Any from ..todos.contract import normalize_todo_id +from ..todos.resume_condition import compact_todo_resume_items from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result BLOCKED_RETRY_SCHEMA_VERSION = "quota_blocked_retry_v0" @@ -14,6 +15,26 @@ TURN_SETTLEMENT_RETRY_SECONDS = 5 * 60 +def compact_settlement_wait_todos( + items: list[dict[str, Any]], *, todo_id: str, +) -> list[dict[str, Any]]: + """Transport all dependency facts and the bound wait's frozen observation. + + The TS blocked-wait owner needs the complete graph for cycle diagnosis, + but only the bound Todo's resume proof and Monitor clocks. Keep those + observations verbatim; prose and completion instructions are not inputs. + """ + facts = compact_todo_resume_items(items) + bound_fields = ( + "resume_condition", "last_checked_at", "next_due_at", "result_hash", + "material_change", "cadence", "expires_at", "watch_only", + ) + for source, row in zip((item for item in items if item.get("todo_id")), facts): + if source.get("todo_id") == todo_id: + row.update({field: source[field] for field in bound_fields if field in source}) + return facts + + def require_blocked_retry_wait( todo_fields: dict[str, Any] | None, *, @@ -30,11 +51,11 @@ def require_blocked_retry_wait( try: result = effect_runtime_result("quota.settlement.read", { "schema_version": "loopx_quota_blocked_wait_request_v0", - "todos": items, + "todos": compact_settlement_wait_todos(items, todo_id=todo_id), "todo_id": todo_id, "observed_at": observed_at, "allow_turn_settlement_retry": allow_turn_settlement_retry, - }) + }, large_local_snapshot=True) except EffectRuntimeRejected as exc: raise ValueError(str(exc)) from exc if not isinstance(result, dict): diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index 0539f8dccd..01d3b628c2 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/loopx/control_plane/quota/settlement.py @@ -378,6 +378,7 @@ def read(admission: Mapping[str, Any] | None) -> Any: else {} ), }, + large_local_snapshot=True, ) except EffectRuntimeRejected as exc: raise ValueError(str(exc)) from None diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py index 139c642ed7..0a8c2711f3 100644 --- a/loopx/control_plane/quota/unsettled_host_turn.py +++ b/loopx/control_plane/quota/unsettled_host_turn.py @@ -362,7 +362,7 @@ def apply_receipt_bound_wait_recovery( ) -> bool: """Read full provider facts only when a replay lost its executable binding.""" from ..todos.list_readback import list_goal_todos - from ..todos.resume_condition import compact_todo_resume_items + from .blocked_retry import compact_settlement_wait_todos from ..runtime.time import now_utc_iso source = list_goal_todos( @@ -370,18 +370,14 @@ def apply_receipt_bound_wait_recovery( ) # Keep transitive dependencies from this coherent read. A direct-target # slice loses longer cycles; compact facts keep prose out of the RPC. - bound: dict[str, Any] = next((row for row in source["todos"] if row.get("todo_id") == todo_id), {}) - items = compact_todo_resume_items(source["todos"]) - for row in items: - if row.get("todo_id") == todo_id: - row["resume_condition"] = bound.get("resume_condition") + items = compact_settlement_wait_todos(source["todos"], todo_id=todo_id) verdict = effect_runtime_result("quota.settlement.read", { "schema_version": "loopx_quota_receipt_bound_wait_request_v0", "todos": items, "todo_id": todo_id, "agent_id": agent_id, "turn_instance_id": turn_instance_id, "observed_at": now_utc_iso(), "monitor_phase": monitor_phase, "settlement_progress": settlement_progress, - }) + }, large_local_snapshot=True) if verdict.get("status") == "none": return False if verdict.get("status") != "recovery_required": diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 99171ee8b2..0dfc822d0b 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1801,10 +1801,12 @@ def test_scheduler_cache_misses_after_goal_configuration_changes_without_run( @pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("large_history", [False, True]) def test_typed_blocked_retry_with_peer_hard_lease( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, + large_history: bool, ) -> None: """A peer's overlapping execution lease must not strand exact closeout.""" from canonical_authority_fixture import ( @@ -1832,6 +1834,15 @@ def test_typed_blocked_retry_with_peer_hard_lease( todos = listed["todos"] for todo in todos: todo["required_write_scopes"] = ["src/**"] + if large_history: + # Complete authority remains large; display/prose bytes must not be + # retransmitted to the wait qualifier or removed from persisted Todos. + todos.extend({"todo_id": f"todo_history_{index:04d}", "role": "agent", + "schema_version": "todo_item_v0", "done": True, "source_section": "Agent Todo", + "status": "done", "task_class": "advancement_task", "archive_state": "active", + "text": "Previously validated work.", "note": "Historical details. " * 500} + for index in range(300)) + assert len(json.dumps(todos).encode()) > 2 * 1024 * 1024 projection = build_todo_runtime_shadow_projection( goal_id=GOAL_ID, todos=todos, handoff_mode="hard_lease", leases=[], ) @@ -1916,6 +1927,8 @@ def test_typed_blocked_retry_with_peer_hard_lease( assert not original.get("resume_when") assert original["completion_validation_required"] is True assert original["completion_validation_sha256"] + if large_history: + assert next(todo for todo in after["todos"] if todo["todo_id"] == "todo_history_0299")["note"] == todos[-1]["note"] next_rc, next_turn = _run_cli( registry_path, runtime, "quota", "should-run", "--codex-app", @@ -1938,6 +1951,203 @@ def test_typed_blocked_retry_with_peer_hard_lease( assert next_turn["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] <= 5 +@pytest.mark.parametrize("kind", ["todo_done", "monitor_changed", "resume_at", "implicit", "monitor"]) +def test_blocked_wait_transports_complete_lifecycle_facts_without_prose( + kind: str, monkeypatch: pytest.MonkeyPatch, +) -> None: + from loopx.control_plane import effect_runtime + from loopx.control_plane.quota.blocked_retry import require_blocked_retry_wait + + observed = "2026-01-01T00:00:00Z" + due = "2026-01-01T00:05:00Z" + waiting = {"todo_id": TODO_ID, "role": "agent", "status": "open", + "task_class": "advancement_task", "archive_state": "active", + "claimed_by": AGENT_ID, "note": "研究记录" * 250_000} + target = {"todo_id": ALTERNATIVE_TODO_ID, "role": "agent", "status": "open", + "task_class": "advancement_task", "archive_state": "active"} + if kind in {"todo_done", "monitor_changed", "resume_at"}: + resume = f"resume_at:{due}" if kind == "resume_at" else f"{kind}:{ALTERNATIVE_TODO_ID}" + waiting.update(resume_when=resume, resume_ready=False, + resume_condition={"schema_version": "todo_resume_condition_v0", "kind": kind, + "resume_when": resume, "satisfied": False, + "target_todo_id": ALTERNATIVE_TODO_ID, "target_status": "open", + "target_task_class": "advancement_task", "target_archive_state": "active"}) + if kind == "monitor_changed": + target.update(task_class="continuous_monitor", material_change_generation=7) + waiting["resume_monitor_generation"] = 7 + waiting["resume_condition"].update(target_task_class="continuous_monitor", + material_change_generation=7, baseline_generation=7) + elif kind == "monitor": + waiting.update(task_class="continuous_monitor", last_checked_at=observed, next_due_at=due, + result_hash="sha256:previous", material_change=False, material_change_generation=7, + cadence="5m", expires_at="2026-01-02T00:00:00Z", watch_only=True) + source = {"agent_todos": {"items": [waiting, target]}, "user_todos": {"items": []}} + before = json.dumps(source, sort_keys=True) + invoke = effect_runtime._request_with_info + requests = [] + + def observe_request(info, **kwargs): + if kwargs["method"] == "quota.settlement.read": + requests.append(kwargs["params"]) + return invoke(info, **kwargs) + + monkeypatch.setattr(effect_runtime, "_request_with_info", observe_request) + result = require_blocked_retry_wait(source, todo_id=TODO_ID, observed_at=observed, + allow_turn_settlement_retry=kind == "implicit") + assert len(requests) == 1 + assert len(json.dumps(requests[0]).encode()) < 4_096 + assert {row["todo_id"] for row in requests[0]["todos"]} == {TODO_ID, ALTERNATIVE_TODO_ID} + assert json.dumps(source, sort_keys=True) == before + assert result["todo_id"] == TODO_ID + if kind in {"todo_done", "monitor_changed"}: + assert result["schema_version"] == "quota_blocked_causal_wait_v0" + assert result["target_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + assert result["waiting_todo"]["resume_when"] == waiting["resume_when"] + elif kind == "monitor": + assert result["schema_version"] == "quota_monitor_unavailable_v0" + assert result["monitor_todo"] == {k: v for k, v in waiting.items() if k != "note"} + else: + assert result["schema_version"] == "quota_blocked_retry_v0" + assert result["due_at"] == due + + +def test_settlement_read_snapshot_is_explicit_and_preserves_complete_facts() -> None: + from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result + + request = {"schema_version": "loopx_quota_blocked_wait_request_v0", + "todos": [{"todo_id": TODO_ID, "role": "agent", "status": "open", + "task_class": "advancement_task", "note": "Complete source. " * 150_000}], + "todo_id": TODO_ID, "observed_at": "2026-01-01T00:00:00Z", + "allow_turn_settlement_retry": True} + assert len(json.dumps(request).encode()) > 2 * 1024 * 1024 + with pytest.raises(EffectRuntimeRejected, match="request is oversized"): + effect_runtime_result("quota.settlement.read", request) + result = effect_runtime_result("quota.settlement.read", request, large_local_snapshot=True) + assert result == {"schema_version": "quota_blocked_retry_v0", "source": "turn_settlement", + "todo_id": TODO_ID, "resume_when": "resume_at:2026-01-01T00:05:00Z", + "observed_at": "2026-01-01T00:00:00Z", "due_at": "2026-01-01T00:05:00Z"} + with pytest.raises(EffectRuntimeRejected, match="unavailable for this method"): + effect_runtime_result("runtime.ping", {}, large_local_snapshot=True) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_large_goal_preserves_bound_wait_recovery_and_once_only_settlement( + tmp_path: Path, provider: str, monkeypatch: pytest.MonkeyPatch, +) -> None: + from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _configure_selectable_alternative(project) + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + rc, listed = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) + assert rc == 0, listed + history = [{"schema_version": "todo_item_v0", "todo_id": f"todo_history_{i:04d}", + "role": "agent", "status": "done", "done": True, "task_class": "advancement_task", + "archive_state": "active", "source_section": "Agent Todo", + "text": "Previously validated work.", "note": "Historical details. " * 500} + for i in range(300)] + source = [*listed["todos"], *history] + assert len(json.dumps(source).encode()) > 2 * 1024 * 1024 + initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection( + goal_id=GOAL_ID, todos=source, handoff_mode="soft_claim", + ), state_path=state, provider=provider) + guard_args = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID) + binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, + "--turn-instance-id", "turn-large-causal-original") + rc, guard = _run_cli(registry, runtime, *guard_args, *binding) + assert rc == 0 and guard["heartbeat_receipt"]["closeout_required"], guard + rc, wait = _run_cli(registry, runtime, "todo", "update", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--todo-id", TODO_ID, + "--resume-when", f"todo_done:{ALTERNATIVE_TODO_ID}", + "--successor-todo-id", ALTERNATIVE_TODO_ID) + assert rc == 0, json.dumps(wait) + # The receipt-bound recovery adapter and the blocked writeback adapter must + # consume the same complete dependency graph, not a display-sized prefix. + rc, recovery = _run_cli(registry, runtime, *guard_args, *binding) + assert rc == 0, recovery + assert recovery["unsettled_host_turn_recovery"]["repair"] == "blocked_writeback" + assert recovery["unsettled_host_turn_recovery"]["binding_id"] == TODO_ID + rc, foreign = _run_cli(registry, runtime, *guard_args, *binding[:-3], + ALTERNATIVE_TODO_ID, *binding[-2:]) + assert rc == 1 and foreign["error_code"] == "heartbeat_receipt_identity_conflict", foreign + refresh = ("refresh-state", "--goal-id", GOAL_ID, *binding, + "--classification", "large_causal_wait", "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_gap", "--progress-result-class", "blocked", + "--progress-blocker-id", ALTERNATIVE_TODO_ID, "--progress-evidence-id", "evidence:pending-dependency", + "--no-global-sync", "--suppress-external-sinks") + for replay in (False, True): + rc, closed = _run_cli(registry, runtime, *refresh, cwd=project) + assert rc == 0, closed + assert closed["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend" + if replay: + assert closed["idempotent_replay"] is True + assert _classification_count(runtime, "large_causal_wait") == 1 + assert _spend_run_count(runtime) == 0 + successor = ("--agent-id", AGENT_ID, "--todo-id", ALTERNATIVE_TODO_ID, + "--turn-instance-id", "turn-large-causal-successor") + rc, next_turn = _run_cli(registry, runtime, *guard_args, *successor) + assert rc == 0, next_turn + assert next_turn["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + rc, advanced = _run_cli(registry, runtime, "refresh-state", "--goal-id", GOAL_ID, + *successor, "--classification", "large_goal_validated_progress", + "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress", + "--delivery-boundary", "in_flight_continuation", + "--no-global-sync", "--suppress-external-sinks", cwd=project) + assert rc == 0, advanced + spend = ("quota", "spend-slot", "--goal-id", GOAL_ID, *successor, + "--slots", "1", "--source", "heartbeat", "--execute") + for replay in (False, True): + rc, spent = _run_cli(registry, runtime, *spend) + assert rc == 0, spent + assert spent.get("idempotent_replay", False) is replay + assert _spend_run_count(runtime) == 1 + rc, historical = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID, + "--todo-id", history[-1]["todo_id"]) + assert rc == 0 and historical["todo"]["note"] == history[-1]["note"], historical + + +@pytest.mark.parametrize("invalid", ["cycle", "duplicate", "missing_target", "completed", "stale_generation"]) +def test_compact_blocked_wait_keeps_authority_rejections(invalid: str) -> None: + from loopx.control_plane.quota.blocked_retry import require_blocked_retry_wait + + waiting = {"todo_id": TODO_ID, "role": "agent", "status": "open", "archive_state": "active", + "task_class": "advancement_task", "resume_when": f"todo_done:{ALTERNATIVE_TODO_ID}", + "resume_ready": False, "note": "Historical details. " * 150_000, + "resume_condition": {"schema_version": "todo_resume_condition_v0", "kind": "todo_done", + "resume_when": f"todo_done:{ALTERNATIVE_TODO_ID}", "satisfied": False, + "target_todo_id": ALTERNATIVE_TODO_ID, "target_status": "open", + "target_task_class": "advancement_task", "target_archive_state": "active"}} + target = {"todo_id": ALTERNATIVE_TODO_ID, "role": "agent", "status": "open", + "archive_state": "active", "task_class": "advancement_task"} + rows = [waiting, target] + if invalid == "cycle": + target["resume_when"] = "todo_done:todo_transitive" + rows.append({**target, "todo_id": "todo_transitive", "resume_when": f"todo_done:{TODO_ID}"}) + error = "dependency cycle" + elif invalid == "duplicate": + rows.append(dict(waiting)) + error = "same unfinished Todo" + elif invalid == "missing_target": + rows.pop() + error = "registered pending causal target" + elif invalid == "completed": + waiting["status"] = "done" + error = "same unfinished Todo" + else: + waiting.update(resume_when=f"monitor_changed:{ALTERNATIVE_TODO_ID}", resume_monitor_generation=7) + waiting["resume_condition"].update(kind="monitor_changed", resume_when=waiting["resume_when"], + target_task_class="continuous_monitor", baseline_generation=7, + material_change_generation=6) + target.update(task_class="continuous_monitor", material_change_generation=7) + error = "stale causal target generation" + with pytest.raises(ValueError, match=error): + require_blocked_retry_wait({"agent_todos": {"items": rows}}, todo_id=TODO_ID, + observed_at="2026-01-01T00:00:00Z") + + def test_in_flight_progress_preserves_todo_across_heartbeat_settlements( tmp_path: Path, ) -> None: From 2b7b40e977a71c7069e60b181f596ee5e795b191 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 05:48:04 +0800 Subject: [PATCH 2/2] docs(quota): record complete settlement transport boundary Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/typescript-control-plane-migration-v0.md | 12 ++++++++++++ .../typescript-control-plane-migration-v0.zh-CN.md | 8 ++++++++ 2 files changed, 20 insertions(+) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index ea102e2072..73d851fc4d 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1393,6 +1393,18 @@ reduction remains a measured follow-up with complete-source parity, not a second Python policy. See the [history decision evidence](ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md). +Settlement wait adapters now share the existing resume-fact projection over the +complete dependency graph, retaining the bound Todo's resume proof and Monitor +clocks, hash and cadence. They do not retransmit unrelated Todo prose or truncate +canonical records. Blocked writeback, receipt-bound wait recovery and ordinary +`quota.settlement.read` readback explicitly reuse the same-UID, digest-bound +private snapshot channel when needed; the 2 MiB wire and 64 MiB snapshot limits +remain unchanged. The existing TS owner still qualifies dependencies, frozen +observations, original Turn identity, blocked closeout without spending and +once-only accounting. Isolated File/SQLite CLI validation covers a complete +multi-MiB Goal and replay; source validation does not establish installed-host +adoption or whole-Goal qualification. + **Checkpoint read-context transport.** A long-lived Goal can exceed the Effect runtime's 2 MiB socket limit even when the original Turn receipt is small: complete shared Goal prose and archived Todo facts are part of the read basis, diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 4960718ac3..2e02b62908 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -966,6 +966,14 @@ File/SQLite CLI 恢复覆盖超过 wire 预算的完整事实、后续独立工 游标/checkpoint 归约保留为以测量驱动、完整源语义一致 为前提的后续工作，不再造 Python 规则。见[历史决策证据](ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md)。 +结算等待适配器现在对完整依赖图共用既有 resume 事实投影，保留绑定 Todo 的 +resume 证明及 Monitor 时钟、hash、cadence；不重复传递无关正文，也不截断 +canonical 记录。受阻 writeback、回执绑定等待恢复和普通 `quota.settlement.read` +回读按需显式复用同 UID、摘要绑定的私有 snapshot 通道，2 MiB wire 和 64 MiB +snapshot 上限均不变。依赖资格、冻结观测、原 Turn 身份、受阻不扣额收尾及 +精确一次记账仍由既有 TS owner 判断。隔离 File/SQLite CLI 验证覆盖完整多 MiB +Goal 与重放；源码验证不代表已安装宿主采用或整 Goal 资格通过。 + **Checkpoint 恢复编写契约。** 现有 TS vision owner 只能投影校验器接受的恢复 选项：没有持久 vision 时，缺失的 checkpoint 必须提交 vision patch；有该基线时 才提供 unchanged reason。Authoring 契约复用校验器的 `todo_delta` 限制：保留前