From 696af80be2c5967d75d44f8f4082687e56a5118b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 10:31:51 +0800 Subject: [PATCH 1/9] fix: preserve receipt retry waits across vision replanning Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/quota-allocation.md | 16 ++++ loopx/control_plane/agents/agent_scope.py | 8 ++ .../goals/goal_frontier/__init__.py | 17 +++- loopx/control_plane/quota/blocked_retry.py | 20 ++++- .../control_plane/quota/should_run_packet.py | 1 + loopx/control_plane/todos/todo_semantics.py | 21 +++++ .../test_goal_vision_blocked_successor.py | 84 +++++++++++++++++++ .../test_quota_settlement_cli.py | 75 +++++++++++++++++ 8 files changed, 234 insertions(+), 8 deletions(-) diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 2773bd43a1..598fd775f0 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -499,6 +499,16 @@ claim-scoped full selection source. An existing wait lane or its display limit cannot hide a new retry; foreign, excluded, mismatched and replaced waits do not qualify through this path. +Ordinary open-Vision wait projection uses the same complete, Agent-scoped +runnable-work counts as selection. Commitment counts still include unfinished +work and cannot establish that it can execute during a receipt retry. The +existing resume owner supplies the retry's complete pending-clock proof to +Vision and host supervision. A causally linked pending retry can therefore +defer the open acceptance gap until its original due time; it neither closes +the Vision nor suppresses independent work, a missing checkpoint or an unrelated +acceptance obligation. Incomplete or differently scoped counts retain the +existing conservative frontier behavior. + 类型化 deadline owner 会把新叠加的、更早到期的 Turn 重试纳入已有的完整来源 缓存,同时保留显示列表之外的更早监控时钟和未变化的缓存。同一个 typed batch 从完整且合格的 receipt 来源独立投影 `blocked_retry_wait`;`frontier_deadline` @@ -511,6 +521,12 @@ stored resume lane 建立后新增的 receipt wait,沿同一个 claim-scoped selection 来源读回。已有等待列表及其显示上限不能隐藏新重试;其它 Agent、 被排除的执行者、身份不匹配或已被替换的等待不能由这条路径取得资格。 +普通开放 Vision 的等待投影复用 selection 的完整 Agent 范围可执行工作计数; +未完成承诺仍保留在 commitment 计数中,但不代表 receipt retry 期间可执行。 +既有 resume owner 向 Vision 和 Host 监督提供完整的未到期时钟证明;具有因果关联的 +retry 可暂缓普通开放验收缺口至原到期时间,不关闭 Vision,也不遮蔽独立工作、 +缺失 checkpoint 或无关验收义务;计数不完整或范围不匹配时仍保留原有保守判断。 + `quota should-run` also separates long-running observation from work that should advance the selected goal. When the selected goal's current projection is a dependency-only observation, the payload includes `work_lane_contract` with diff --git a/loopx/control_plane/agents/agent_scope.py b/loopx/control_plane/agents/agent_scope.py index 1dbf5fba29..fa6f6c2d18 100644 --- a/loopx/control_plane/agents/agent_scope.py +++ b/loopx/control_plane/agents/agent_scope.py @@ -34,6 +34,7 @@ todo_item_is_deferred, todo_item_task_class, todo_presentation_sort_key, + todo_scoped_runnable_advancement_count, ) from ..todos.summary_item import compact_todo_summary_item, compact_todo_summary_items from ..todos.user_gate import ( @@ -1350,6 +1351,7 @@ def _agent_scope_no_candidate_frontier( agent_lane_next_action: dict[str, Any] | None, work_lane_contract: dict[str, Any] | None, candidate_should_run: bool, + vision_wait_state: dict[str, Any] | None = None, ) -> dict[str, Any] | None: if not candidate_should_run or not _agent_identity_has_scoped_lane(agent_identity): return None @@ -1376,6 +1378,12 @@ def _agent_scope_no_candidate_frontier( agent_todo_summary, agent_id=agent_id, ) + if vision_wait_state and todo_scoped_runnable_advancement_count( + agent_todo_summary, agent_id=agent_id, + ) == 0: + # A committed retry keeps its unfinished commitment while selection + # proves there is no runnable work. Preserve the exact wait route. + current_count = unclaimed_count = 0 if current_count > 0 or unclaimed_count > 0: return None context = _AgentScopeNoCandidateContext( diff --git a/loopx/control_plane/goals/goal_frontier/__init__.py b/loopx/control_plane/goals/goal_frontier/__init__.py index cac41af39f..4122c52309 100644 --- a/loopx/control_plane/goals/goal_frontier/__init__.py +++ b/loopx/control_plane/goals/goal_frontier/__init__.py @@ -22,6 +22,7 @@ from ...todos.todo_semantics import ( todo_advancement_frontier_counts, todo_item_is_watch_only_monitor, + todo_scoped_runnable_advancement_count, todo_summary_open_task_counts, ) from ...todos.succession_warning import ( @@ -1544,15 +1545,23 @@ def build_goal_frontier_projection_context_from_status( agent_todo_summary=agent_todo_summary, agent_id=agent_id, ) + # Commitment floors include open work held by a receipt-owned retry. The + # typed selection projection, when complete for this lane, owns how much + # work is actually runnable; a commitment is not an execution candidate. + selectable_advancement_count = todo_scoped_runnable_advancement_count( + agent_todo_summary, agent_id=agent_id, + ) + if selectable_advancement_count is None: + selectable_advancement_count = ( + frontier_counts["current_agent_claimed_advancement_count"] + + frontier_counts["unclaimed_advancement_count"] + ) vision_wait_state = build_goal_vision_wait_state( agent_todo_summary=agent_todo_summary, source_items=agent_todo_source_items, agent_id=agent_id, acceptance_gaps=source_acceptance_gaps, - selectable_advancement_count=( - frontier_counts["current_agent_claimed_advancement_count"] - + frontier_counts["unclaimed_advancement_count"] - ), + selectable_advancement_count=selectable_advancement_count, ) acceptance_gaps = ( [gap for gap in source_acceptance_gaps if gap.get("kind") in GOAL_ACCEPTANCE_HOLD_TRIGGERS] diff --git a/loopx/control_plane/quota/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py index e1d517f015..14b2a4007c 100644 --- a/loopx/control_plane/quota/blocked_retry.py +++ b/loopx/control_plane/quota/blocked_retry.py @@ -6,7 +6,10 @@ from typing import Any from ..todos.contract import normalize_todo_id -from ..todos.resume_condition import compact_todo_resume_items +from ..todos.resume_condition import ( + compact_todo_resume_items, + evaluate_todo_resume_conditions, +) from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result BLOCKED_RETRY_SCHEMA_VERSION = "quota_blocked_retry_v0" @@ -170,6 +173,17 @@ def overlay_active_turn_retries( items = queue.get("items") if isinstance(queue, dict) else None if not isinstance(items, list): return status_payload + # Reuse the resume owner for complete clock proof. A bare satisfied=False + # suppresses selection but cannot qualify an automatic Vision/host wait. + retry_items = [ + {"todo_id": todo_id, "status": "open", "role": "agent", + "task_class": "advancement_task", "claimed_by": agent_id, + "resume_when": retry["resume_when"]} + for todo_id, retry in waits.items() + ] + resume_conditions = evaluate_todo_resume_conditions( + retry_items, source_items=retry_items, evaluated_at=observed_at, + ) projected_items = [] changed = False @@ -207,9 +221,7 @@ def overlay_summary(summary: Any) -> tuple[Any, bool]: "blocked_retry": retry, "next_due_at": retry["due_at"], "resume_condition": { - "kind": "resume_at", - "resume_when": resume, - "satisfied": False, + **resume_conditions[todo["todo_id"]], "blocked_retry": retry, }, } diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 3f2ef3399f..d56229b7d6 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -1175,6 +1175,7 @@ def _resolve_quota_should_run_route( agent_todo_summary=prepared.agent_todo_summary, agent_lane_next_action=agent_lane_next_action, work_lane_contract=prepared.work_lane_contract, + vision_wait_state=prepared.goal_frontier_projection.get("vision_wait_state"), # A satisfied dependency is readiness, not admission. The # existing scoped fallback above qualifies an independent # successor under a user gate; an unqualified ready row must diff --git a/loopx/control_plane/todos/todo_semantics.py b/loopx/control_plane/todos/todo_semantics.py index 9be2cf360e..e47fa48304 100644 --- a/loopx/control_plane/todos/todo_semantics.py +++ b/loopx/control_plane/todos/todo_semantics.py @@ -355,6 +355,27 @@ def agent_scoped_selectable_advancement_todo_ids( if (todo_id := normalize_todo_id(item.get("todo_id")))} +def todo_scoped_runnable_advancement_count( + summary: dict[str, Any] | None, *, agent_id: str | None, +) -> int | None: + """Decode the typed selection count; unknown sources keep legacy fences. + + Commitment floors intentionally include work held by receipt-owned waits. + They do not override a complete same-lane runnable-work projection. + """ + counts = (summary or {}).get("work_counts") + if ( + isinstance(counts, dict) + and counts.get("schema_version") == "todo_work_counts_v0" + and counts.get("complete") is True + and counts.get("agent_id") == agent_id + and type(counts.get("advancement")) is int + and counts["advancement"] >= 0 + ): + return counts["advancement"] + return None + + def todo_advancement_frontier_counts( summary: dict[str, Any] | None, *, agent_id: str | None, ) -> dict[str, int]: diff --git a/tests/control_plane/test_goal_vision_blocked_successor.py b/tests/control_plane/test_goal_vision_blocked_successor.py index 651fbe034a..ed8b6a8ff5 100644 --- a/tests/control_plane/test_goal_vision_blocked_successor.py +++ b/tests/control_plane/test_goal_vision_blocked_successor.py @@ -2,6 +2,7 @@ import json from copy import deepcopy +from datetime import timedelta import pytest @@ -245,6 +246,89 @@ def _quota(payload: dict) -> dict: ) +@pytest.mark.parametrize("policy", [None, "as_needed", "repeat_until_closed"]) +@pytest.mark.parametrize("seconds", [120, 300]) +@pytest.mark.parametrize("fault", [None, "unrelated_vision", "missing_checkpoint"]) +def test_receipt_retry_defers_linked_vision_until_its_original_due( + monkeypatch, policy, seconds, fault, +) -> None: + from loopx.control_plane.quota.blocked_retry import overlay_active_turn_retries + from tests.control_plane.test_blocked_retry_wait_context import START, source + + now = START + timedelta(seconds=seconds) + for path in ( + "loopx.control_plane.todos.quota_selection.now_utc", + "loopx.control_plane.todos.quota_summary.now_utc", + "loopx.control_plane.scheduler.monitor_todo.now_utc", + "loopx.control_plane.scheduler.scheduler_hint.now_utc", + ): + monkeypatch.setattr(path, lambda: now) + payload = source() + run = payload["run_history"]["goals"][0]["latest_runs"][0] + run["generated_at"] = START.isoformat() + run["progress_observation"] = { + "schema_version": "typed_progress_observation_v0", + "result_class": "blocked", + "work_item_id": "todo_work", + "blocker_id": "external-feedback", + "evidence_ids": ["evidence:pending-feedback"], + } + run["agent_vision"] = { + "schema_version": "goal_vision_replan_contract_v0", + "agent_id": "agent-a", + "state": "vision_patch_proposed", + "vision_patch": { + "acceptance_summary": "Qualify the original work after feedback arrives.", + **({"advancement_policy": policy} if policy else {}), + }, + "todo_delta": ["retain:todo_other" if fault == "unrelated_vision" else "retain:todo_work"], + } + if fault == "missing_checkpoint": + run["vision_checkpoint"] = { + "schema_version": "vision_checkpoint_v0", "agent_id": "agent-a", + "required": True, "satisfied": False, "decision": "missing_required", + "triggers": [{"kind": "material_delivery_outcome"}], + } + item = payload["attention_queue"]["items"][0] + # Build history projections after authoring the facts; cached absence of + # a checkpoint must not masquerade as a negative-case result. + payload = quota_status_payload( + goal_id="goal-a", status="active", recommended_action="Await qualified feedback.", + claim_scope_agent_id="agent-a", + coordination={"registered_agents": ["agent-a", "agent-b"], "agent_model": "peer_v1"}, + agent_todo_items=item["agent_todos"]["items"], + user_todo_items=item["user_todos"]["items"], latest_runs=[run], + ) + original = deepcopy(payload) + projected = overlay_active_turn_retries( + payload, goal_id="goal-a", agent_id="agent-a", observed_at=now.isoformat(), + ) + guard = build_quota_should_run( + projected, goal_id="goal-a", agent_id="agent-a", + scheduler_execution_context=GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT, + ) + assert payload == original + if fault == "missing_checkpoint": + assert "vision_checkpoint_missing" in { + gap["kind"] for gap in guard["goal_frontier_projection"]["acceptance_gaps"] + } + if fault == "missing_checkpoint" or fault == "unrelated_vision" and seconds < 300: + assert "vision_wait_state" not in guard["goal_frontier_projection"] + if seconds < 300: + assert guard["autonomous_replan_obligation"]["required"] is True + return + assert not guard.get("autonomous_replan_obligation") + if seconds < 300: + assert guard["execution_obligation"]["must_attempt_work"] is False + assert guard["agent_todo_summary"]["blocked_retry_wait"]["turn_instance_id"] == "turn-old" + assert guard["goal_frontier_projection"]["vision_wait_state"][ + "selected_todo_id" + ] == "todo_work" + else: + assert guard["selected_todo"]["todo_id"] == "todo_work" + assert "pending_retry" not in guard["scheduler_hint"] + + def _current_agent_blocker_status_payload( *, claimed_by: str = AGENT_ID, diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index a52e42f205..fe0556165f 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1917,6 +1917,81 @@ def test_typed_blocked_retry_with_peer_hard_lease( assert next_turn["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] <= 5 +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, +) -> None: + from canonical_authority_fixture import isolate_sqlite_runtime + from loopx.control_plane.quota.turn_envelope import build_turn_envelope + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _configure_completion_validation_todo(project) + _initialize_fixture_authority(project, runtime, provider=provider) + rc, before = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) + assert rc == 0, before + binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, + "--turn-instance-id", "turn-vision-feedback") + rc, admitted = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL_ID, *binding, cwd=project) + assert rc == 0, admitted + vision = tmp_path / "vision.json" + vision.write_text(json.dumps({ + "schema_version": "goal_vision_replan_contract_v0", + "state": "vision_patch_proposed", + "vision_patch": {"acceptance_summary": "Qualify the work after new feedback.", + "advancement_policy": "repeat_until_closed"}, + "todo_delta": [f"retain:{TODO_ID}"], + "path_delta": {"outcome": "no_change", "prior_assumption": "Feedback is required.", + "observed_reality": "No new feedback is available.", + "retained": ["The original work and acceptance."], + "evidence_refs": ["evidence:feedback-pending"]}, + }), encoding="utf-8") + rc, closed = _run_cli(registry, runtime, "refresh-state", "--goal-id", GOAL_ID, + *binding, "--classification", "feedback_pending", + "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_gap", "--progress-result-class", "blocked", + "--progress-blocker-id", "external-feedback", + "--progress-evidence-id", "evidence:feedback-pending", + "--agent-vision-json", str(vision), + "--no-global-sync", "--suppress-external-sinks", cwd=project) + assert rc == 0, closed + assert closed["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend" + rc, waiting = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", "turn-before-feedback-due", cwd=project) + assert rc == 0, waiting + assert waiting["execution_obligation"]["must_attempt_work"] is False + assert not waiting.get("autonomous_replan_obligation") + assert waiting["interaction_contract"]["agent_channel"]["delivery_allowed"] is False + assert waiting["agent_todo_summary"]["blocked_retry_wait"]["due_at"] == closed["blocked_retry"]["due_at"] + wait = waiting["goal_frontier_projection"]["vision_wait_state"] + assert wait["selected_todo_id"] == TODO_ID + assert wait["resume_condition"]["scheduled_for"] == closed["blocked_retry"]["due_at"] + envelope = build_turn_envelope(waiting) + assert envelope["contract_capsule"]["execution_obligation"]["must_attempt_work"] is False + rc, after = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) + assert rc == 0, after + assert after["todos"] == before["todos"] + assert _spend_run_count(runtime) == 0 + rc, independent = _run_cli( + registry, runtime, "todo", "add", "--goal-id", GOAL_ID, + "--role", "agent", "--task-class", "advancement_task", + "--action-kind", "validate", "--target-key", "independent-feedback", + "--claimed-by", AGENT_ID, "--text", "Validate independent evidence.", + ) + assert rc == 0, independent + rc, resumed = _run_cli( + registry, runtime, "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--turn-instance-id", "turn-independent-feedback", cwd=project, + ) + assert rc == 0, resumed + assert resumed["selected_todo"]["todo_id"] == independent["todo_id"] + assert resumed["execution_obligation"]["must_attempt_work"] is True + assert _spend_run_count(runtime) == 0 + + @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, From 4a1de4c751f00024f1bbcb8ade26a18f35320199 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 10:45:26 +0800 Subject: [PATCH 2/9] test: initialize canonical retry lease fixtures before public reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_quota_settlement_cli.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index fe0556165f..6fc3056e34 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1795,9 +1795,15 @@ def test_typed_blocked_retry_with_peer_hard_lease( "peer_independent_worktree_required": False, } registry_path.write_text(json.dumps(registry), encoding="utf-8") - rc, listed = _run_cli(registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID) - assert rc == 0, listed - todos = listed["todos"] + # Author the disposable source before promoting its canonical authority; + # the public reader correctly rejects an uninitialized provider. + summary = parse_active_state_todos( + state.read_text(encoding="utf-8"), state_path=state, item_limit=None, + ) + todos = [ + *(summary.get("user_todos", {}).get("items") or []), + *(summary.get("agent_todos", {}).get("items") or []), + ] for todo in todos: todo["required_write_scopes"] = ["src/**"] if large_history: From 9c94bc95e4310d7115ba14fe7ef1e606de0d3f85 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 11:49:10 +0800 Subject: [PATCH 3/9] fix: preserve outcome checkpoints across blocked replans Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/loopx-overall-roadmap-v0.md | 8 ++ .../rfcs/loopx-overall-roadmap-v0.zh-CN.md | 6 + docs/quota-allocation.md | 18 +++ .../goals/goal_frontier/outcome_continuity.py | 5 +- loopx/control_plane/quota/settlement_phase.ts | 34 ++++- .../quota/settlement_readback.ts | 6 + .../runtime/run_context_retention.py | 6 +- .../work_items/autonomous_replan_ack.py | 29 +++++ .../test_final_outcome_diagnostics.py | 120 ++++++++++++++++++ .../test_quota_settlement_cli.py | 79 ++++++++++++ .../quota_settlement_readback.test.ts | 9 ++ 11 files changed, 317 insertions(+), 3 deletions(-) diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index 6be776b0be..c90732c3ca 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -129,6 +129,14 @@ visibility limits, ordinary notices, current cadence floors and unchanged canonical waits/monitor clocks. Installed adoption and live host readback remain separate from this source repair. +**S3/S7 administrative closeout proposal.** [#7117](https://github.com/loopx-project/loopx/pull/7117) +also separates an accepted blocked replan from the preceding real outcome +checkpoint. File/SQLite CLI readback and long/compacted history regressions retain +the previous milestone or genuine gap without creating a new outcome obligation +from administrative closure. The complete local CLI suite still has failures; +source qualification, integration, installed adoption and measured benefit remain +distinct. Preserve those failures and the existing full-suite qualification work. + **S5/S11 long-horizon progress design.** The [research measurement refinement](long-horizon-harness-benchmark-research-program-v0.md#evidence-based-episode-evaluation-proposed-refinement) and [product reading journey](../../product/use-cases/auto-research/long-horizon-progress-design.md) connect native outcomes to research episodes, evidence and actual adoption. diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md index f5ca776204..32827bb7aa 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md @@ -104,6 +104,12 @@ managed 与 attached 的工作对话都应能持续在 LoopX 中进行:沿用 | **S12 发布、开发体验与社区治理 · P0 卫生/P1** | 安装、源码验证、扩展注册、DCO/PR、测试层级、contributor route 与双语文档已存在 | 从干净机器/发布包验一条首次工作和一次升级/回滚;host/OS 支持以 release contract 为准。缩短合理改动的定位、测试和 review 成本;公开精确 head、可重复 fixture、兼容窗口、维护者路由和贡献归属,退休重复协议及过时证据 | | **S13 采用、生态与商业可持续性 · P1 发现/P2 试点** | 公开 adoption loop、showcase、license/governance、observer-first 产品合同已存在;付费 PMF 未证明 | 先收集真实独立首次使用/重复使用/退出原因,做可复现案例和有固定预算/验收/回滚的试点;沉淀 reusable adapter 与交付手册。核算模型/计算/存储/支持成本及维护负担;满足重复需求后再决策商业托管边界、支持等级和分发,不承诺 SLA 或擅改开源条款 | +**S3/S7 行政收尾提案。** [#7117](https://github.com/loopx-project/loopx/pull/7117) +进一步区分已验收的阻塞重规划与前一个真实成果 checkpoint。File/SQLite CLI 回读和 +长期/压缩历史回归保留原里程碑或真实缺口,避免行政收尾制造新的成果义务。完整 +本地 CLI 套件仍有失败;源码资格、集成、安装采用与实测收益分别验证,保留失败 +记录和既有完整套件资格工作。 + **S5/S11 长程进展设计。** [研究计量细化](long-horizon-harness-benchmark-research-program-v0.zh-CN.md#基于证据的研究片段评估拟议细化) 与[产品阅读旅程](../../product/use-cases/auto-research/long-horizon-progress-design.zh-CN.md) 把原生结果关联到研究片段、证据与实际采用。研究侧牵头方法、基线/窗口/成本口径及独立 diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 598fd775f0..92bfcc87d7 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -241,6 +241,24 @@ selection of independent work remain unchanged. A previously committed debit is retained, and neither a bare blocker declaration nor an unqualified ACK closes the Turn. +History selection also distinguishes this accepted administrative blocked replan +from a delivered milestone. The existing TypeScript settlement owner qualifies +the persisted Run facts in one batch; a bare blocker, unaccepted ACK, mismatched +identity or unsupported checkpoint remains outcome-relevant. Both semantic +history and the legacy recent-Run reader retain the preceding real outcome +checkpoint, including any unresolved gap, instead of manufacturing a new +final-outcome obligation from the administrative closeout. The latest vision, +checkpoint and replan ACK remain visible. Missing checkpoints and genuine +delivery gaps still require their original review; this read classification +grants no execution, completion, acceptance or settlement authority. + +历史选取将已验收的行政阻塞重规划与成果里程碑区分。复用 TypeScript 结算 owner 批量 +核对持久 Run 的类型化事实;裸阻塞、未验收 ACK、身份错配和未知 checkpoint 均不能 +取得行政豁免。语义历史和旧 recent-Run 读路都保留前一个真实成果 checkpoint 及其 +未解决缺口,避免行政收尾反复制造新的成果复核义务。最新 vision、checkpoint 和 +重规划 ACK 仍然可见;缺失 checkpoint 与真实交付 gap 保持原复核要求,这一读取 +分类不授予执行、完成、验收或结算权限。 + An accepted `vision_checkpoint_v0` for `outcome_progress` also closes its exact Todo-bound Turn after the matching writeback and spend receipts commit. Both `in_flight_continuation` and `semantic_closeout` use their corresponding typed diff --git a/loopx/control_plane/goals/goal_frontier/outcome_continuity.py b/loopx/control_plane/goals/goal_frontier/outcome_continuity.py index 7fc43c248c..fb15fada91 100644 --- a/loopx/control_plane/goals/goal_frontier/outcome_continuity.py +++ b/loopx/control_plane/goals/goal_frontier/outcome_continuity.py @@ -5,6 +5,7 @@ from typing import Any from ...agents.agent_scope import agent_scope_item_claimed_by +from ...work_items.autonomous_replan_ack import accepted_blocked_replan_history_indexes from ..goal_vision_policy import ( COMPLETED_TODO_CHAIN_REPLAN_THRESHOLD, ) @@ -102,7 +103,9 @@ def _outcome_checkpoint_runs_for_goal( else None ) return [checkpoint_run] if isinstance(checkpoint_run, dict) else [] - return _latest_runs_for_goal(status_payload, goal_id=goal_id) + runs = _latest_runs_for_goal(status_payload, goal_id=goal_id) + administrative = accepted_blocked_replan_history_indexes(runs) + return [run for index, run in enumerate(runs) if index not in administrative] def latest_outcome_vision_checkpoint_from_status_payload( diff --git a/loopx/control_plane/quota/settlement_phase.ts b/loopx/control_plane/quota/settlement_phase.ts index a1f09496f5..2fa49f6c79 100644 --- a/loopx/control_plane/quota/settlement_phase.ts +++ b/loopx/control_plane/quota/settlement_phase.ts @@ -1,4 +1,4 @@ -import type { SettlementIdentity } from "../effect_program.ts"; +import { settlementIdentity, type SettlementIdentity } from "../effect_program.ts"; import { jsonObject } from "../runtime_decode.ts"; import { isBlockedLifecycleWait, isCausalBlockedWait, isMonitorUnavailableWait } from "./blocked_wait.ts"; import { isTurnScopedSettlementOutcome } from "../work_items/delivery_outcome.ts"; @@ -54,6 +54,38 @@ export function isAcceptedBlockedReplanWriteback( }); } +/** Classify persisted Run facts for history selection, without granting a + * settlement, debit or execution right. Reuse the settlement owner's exact + * semantic qualification; a caller-supplied "administrative" flag is no proof. + * One batch keeps history reads independent of per-Run bridge latency. */ +export function acceptedBlockedReplanHistoryIndexes(runs: unknown[]): number[] { + const accepted: number[] = []; + runs.forEach((value, index) => { + const run = jsonObject(value); + if (!run || run.todo_id != null || + ![run.goal_id, run.agent_id, run.turn_instance_id, run.replan_obligation_id] + .every(value => typeof value === "string" && value.trim().length > 0)) return; + const identity = settlementIdentity({ + goal_id: run.goal_id as string, agent_id: run.agent_id as string, + turn_instance_id: run.turn_instance_id as string, + replan_obligation_id: run.replan_obligation_id as string, + }); + const persisted = jsonObject(run.settlement_identity); + if (run.settlement_identity != null && (!persisted || + Object.entries(identity).some(([key, value]) => + (value === null ? persisted[key] ?? null : persisted[key]) !== value))) return; + // compact_vision_checkpoint omits the historical default boundary. Decode + // that existing representation here; live settlement still requires its + // explicit committed checkpoint and independently verified receipts. + const checkpoint = jsonObject(run.vision_checkpoint); + const decoded = checkpoint && checkpoint.delivery_boundary === undefined + ? {...run, vision_checkpoint: {...checkpoint, delivery_boundary: "semantic_closeout"}} + : run; + if (isAcceptedBlockedReplanWriteback(decoded, identity, run.replan_obligation_id)) accepted.push(index); + }); + return accepted; +} + /** A service-qualified capability duty can retire its exact admitted Turn. * The caller first verifies the durable writeback receipt. No Task/Goal * completion, progress or debit follows from this lifecycle evidence. */ diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 9cea97190f..b8ca2a93ae 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -34,6 +34,7 @@ import { import { isBoundedBlockedRetry, isAcceptedBlockedReplanWriteback, + acceptedBlockedReplanHistoryIndexes, isCommittedMonitorPollEffect, isAcceptedProgressWriteback, isAcceptedReplanWriteback, @@ -1382,6 +1383,11 @@ export function readQuotaSettlementForAdmittedOwnerFromSnapshot( } export async function readQuotaSettlement(value: unknown): Promise { + if (jsonObject(value)?.schema_version === "loopx_quota_blocked_replan_history_request_v0") { + const runs = jsonObject(value)?.runs; + if (!Array.isArray(runs)) throw new EffectRuntimeRequestError("blocked replan history requires Run facts"); + return {accepted_indexes: acceptedBlockedReplanHistoryIndexes(runs)}; + } if (jsonObject(value)?.schema_version === RECEIPT_BOUND_WAIT_REQUEST_SCHEMA) { return projectReceiptBoundWait(value); } diff --git a/loopx/control_plane/runtime/run_context_retention.py b/loopx/control_plane/runtime/run_context_retention.py index 3cc249023d..f457b8b682 100644 --- a/loopx/control_plane/runtime/run_context_retention.py +++ b/loopx/control_plane/runtime/run_context_retention.py @@ -12,6 +12,7 @@ from ..work_items.autonomous_replan_ack import ( AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, autonomous_replan_ack_recorded, + accepted_blocked_replan_history_indexes, ) from ..work_items.autonomous_replan_obligation import run_history_agent_id from ..work_items.replan_history_codec import REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS @@ -142,7 +143,8 @@ def goal_semantic_history_from_runs( seen_retry_todos: set[tuple[str, str]] = set() observed_at = now_utc_iso() - for run in runs: + administrative_indexes = accepted_blocked_replan_history_indexes(runs) + for index, run in enumerate(runs): if latest_owner_correction_run is None and isinstance( run.get("human_reward"), dict ): @@ -183,6 +185,7 @@ def goal_semantic_history_from_runs( if ( "latest_outcome_vision_checkpoint_run" not in context + and index not in administrative_indexes and _run_has_outcome_vision_checkpoint(run) ): context["latest_outcome_vision_checkpoint_run"] = run @@ -200,6 +203,7 @@ def goal_semantic_history_from_runs( if ( "latest_material_milestone_run" not in context + and index not in administrative_indexes and str(run.get("delivery_outcome") or "").strip() in MATERIAL_DELIVERY_OUTCOMES ): diff --git a/loopx/control_plane/work_items/autonomous_replan_ack.py b/loopx/control_plane/work_items/autonomous_replan_ack.py index 15d8280788..cf9a48ac49 100644 --- a/loopx/control_plane/work_items/autonomous_replan_ack.py +++ b/loopx/control_plane/work_items/autonomous_replan_ack.py @@ -2,6 +2,7 @@ from typing import Any +from ..effect_runtime import effect_runtime_result from ..runtime.time import parse_timestamp from .progress_observation import FRESH_VISION_PATH_DISPOSITIONS @@ -14,6 +15,34 @@ AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK = AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW * 3 +def accepted_blocked_replan_history_indexes(runs: list[dict[str, Any]]) -> set[int]: + """Transport persisted administrative candidates to the shared TS owner. + + This only selects history slots. It neither acknowledges a replan nor + replaces the receipt checks required for actual settlement. Missing legacy + evidence remains an ordinary outcome checkpoint, conservatively. + """ + candidates = [index for index, run in enumerate(runs) if run.get("replan_obligation_id")] + if not candidates: + return set() + fields = ( + "goal_id", "agent_id", "todo_id", "turn_instance_id", "replan_obligation_id", + "settlement_identity", "delivery_outcome", "progress_observation", + "autonomous_replan_ack", "vision_checkpoint", + ) + result = effect_runtime_result("quota.settlement.read", { + "schema_version": "loopx_quota_blocked_replan_history_request_v0", + "runs": [{key: runs[index][key] for key in fields if key in runs[index]} + for index in candidates], + }, large_local_snapshot=True) + indexes = result.get("accepted_indexes") if isinstance(result, dict) else None + if not isinstance(indexes, list) or any( + type(index) is not int or not 0 <= index < len(candidates) for index in indexes + ): + raise RuntimeError("TypeScript blocked replan history result must contain valid indexes") + return {candidates[index] for index in indexes} + + def ack_binds_trigger_checkpoints(ack: dict[str, Any] | None) -> bool: """Return whether an ACK binds exact typed trigger revisions. diff --git a/tests/control_plane/test_final_outcome_diagnostics.py b/tests/control_plane/test_final_outcome_diagnostics.py index b542aeba9e..55eb2fd46a 100644 --- a/tests/control_plane/test_final_outcome_diagnostics.py +++ b/tests/control_plane/test_final_outcome_diagnostics.py @@ -11,7 +11,13 @@ ) from loopx.control_plane.goals.goal_frontier.outcome_continuity import ( acceptance_gaps_from_outcome_checkpoint, + latest_outcome_vision_checkpoint_from_status_payload, ) +from loopx.control_plane.runtime.run_context_retention import ( + compact_goal_semantic_history, + goal_semantic_history_from_runs, +) +from loopx.status import compact_run from loopx.control_plane.quota.cli_projection import ( compact_quota_should_run_cli_payload, ) @@ -69,6 +75,110 @@ def material_run(): } +def blocked_replan_run(): + run = material_run() + obligation = "replan-0000000000000001" + run.update( + generated_at="2026-01-01T00:01:00Z", + turn_instance_id="turn-administrative-replan", + replan_obligation_id=obligation, + delivery_outcome="outcome_gap", + progress_observation={ + "schema_version": "typed_progress_observation_v0", + "result_class": "blocked", + "work_item_id": obligation, + "blocker_id": "source-not-ready", + "evidence_ids": ["evidence:source-readback"], + "fingerprint": "source-readback-fingerprint", + }, + autonomous_replan_ack={ + "schema_version": "autonomous_replan_ack_v0", + "source": "refresh_state", + "recorded": True, + "semantic_delta": { + "schema_version": "replan_semantic_delta_v0", + "accepted": True, + "obligation_id": obligation, + "observation_fingerprint": "source-readback-fingerprint", + "outcomes": ["new_concrete_blocker"], + "satisfying_outcomes": ["new_concrete_blocker"], + }, + }, + ) + run["vision_checkpoint"].update( + schema_version="vision_checkpoint_v0", delivery_boundary="semantic_closeout" + ) + run["vision_checkpoint"]["triggers"][0]["delivery_outcome"] = "outcome_gap" + return run + + +@pytest.mark.parametrize("semantic", [False, True]) +@pytest.mark.parametrize("prior_gap", [False, True]) +def test_administrative_replan_preserves_prior_real_outcome_checkpoint(semantic, prior_gap): + prior = material_run() + prior["agent_vision"]["vision_patch"]["acceptance_summary"] = "Verified milestone." + if prior_gap: + prior["vision_checkpoint"]["triggers"][0]["delivery_outcome"] = "outcome_gap" + admin = blocked_replan_run() + # The material checkpoint must survive well outside the recent display window. + runs = [admin, *[{"agent_id": AGENT, "classification": "quota_slot_spent"}] * 100, prior] + goal = {"id": GOAL, "latest_runs": [compact_run(run) for run in runs]} + if semantic: + history = goal_semantic_history_from_runs(runs) + context = history["agents"][0] + assert context["latest_outcome_vision_checkpoint_run"] is prior + assert context["latest_material_milestone_run"] is prior + assert context["latest_autonomous_replan_ack_run"] is admin + goal["semantic_history"] = compact_goal_semantic_history(history, compact_run=compact_run) + goal["latest_runs"] = [compact_run(admin)] + checkpoint = latest_outcome_vision_checkpoint_from_status_payload( + {"run_history": {"goals": [goal]}}, goal_id=GOAL, agent_id=AGENT + ) + assert checkpoint["generated_at"] == STAMP + gaps = acceptance_gaps_from_outcome_checkpoint(prior["agent_vision"], checkpoint) + assert bool(gaps) is prior_gap + + +@pytest.mark.parametrize("invalid", ["unaccepted", "identity", "fingerprint", "evidence", "unsatisfied", "ordinary", "claimed_only"]) +def test_unqualified_blocked_replan_cannot_hide_a_real_outcome_gap(invalid): + run = blocked_replan_run() + if invalid == "unaccepted": + run["autonomous_replan_ack"]["semantic_delta"]["accepted"] = False + elif invalid == "identity": + run["vision_checkpoint"]["agent_id"] = "other-agent" + elif invalid == "fingerprint": + run["autonomous_replan_ack"]["semantic_delta"]["observation_fingerprint"] = "other" + elif invalid == "evidence": + run["progress_observation"]["evidence_ids"] = [] + elif invalid == "unsatisfied": + run["vision_checkpoint"]["satisfied"] = False + elif invalid == "ordinary": + run["todo_id"] = "todo_delivery" + else: + run.pop("autonomous_replan_ack") + run["administrative_blocked_replan"] = True + history = goal_semantic_history_from_runs([run]) + assert history["agents"][0]["latest_outcome_vision_checkpoint_run"] is run + + +def test_administrative_history_qualification_is_batched_and_ordinary_reads_are_local(monkeypatch): + from loopx.control_plane.work_items import autonomous_replan_ack as codec + + calls = [] + real = codec.effect_runtime_result + + def observe(method, params, **kwargs): + calls.append((method, len(params["runs"]))) + return real(method, params, **kwargs) + + monkeypatch.setattr(codec, "effect_runtime_result", observe) + goal_semantic_history_from_runs([material_run()] * 100) + assert calls == [] + history = goal_semantic_history_from_runs([blocked_replan_run()] * 100 + [material_run()]) + assert calls == [("quota.settlement.read", 100)] + assert history["agents"][0]["latest_outcome_vision_checkpoint_run"]["generated_at"] == STAMP + + def decision(): status = quota_status_payload( goal_id=GOAL, @@ -224,6 +334,16 @@ def test_public_cli_diagnostic_reentry_preserves_guard_identity_without_spending runs = runtime / "goals" / GOAL / "runs" / "index.jsonl" runs.parent.mkdir(parents=True) runs.write_text(json.dumps(material_run()) + "\n") + # Consumer tests need an actual canonical provider; an unfenced Markdown + # fixture now correctly fails before final-outcome diagnosis is evaluated. + from canonical_authority_fixture import initialize_canonical_authority + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos + + todos = parse_active_state_todos(state.read_text(), state_path=state, item_limit=None) + initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection( + goal_id=GOAL, todos=todos["agent_todos"]["items"], handoff_mode="soft_claim", leases=[] + ), state_path=state) before = deepcopy((state.read_text(), runs.read_text())) results = [] for _ in range(2): diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 63655fdae9..bcf6446d45 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -6421,6 +6421,85 @@ def test_todoless_autonomous_replan_settles_quota_refresh_spend_chain( assert _spend_run_count(runtime) == 1 +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_blocked_replan_keeps_material_checkpoint_in_native_readback(tmp_path, monkeypatch, provider): + from canonical_authority_fixture import isolate_sqlite_runtime + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry_path = _write_fixture(tmp_path) + _configure_autonomous_replan_fixture(project, runtime, registry_path) + _initialize_fixture_authority(project, runtime, provider=provider) + stamp = "2026-01-01T00:00:00+00:00" + prior = { + "generated_at": stamp, "goal_id": GOAL_ID, "agent_id": AGENT_ID, + "classification": "verified_stage", "delivery_outcome": "outcome_progress", + "agent_vision": { + "agent_id": AGENT_ID, "state": "vision_active", "generated_at": stamp, + "vision_patch": {"vision_summary": "Verify the next outcome.", + "acceptance_summary": "Prior stage verified by evidence:stage."}, + "path_delta": {"outcome": "continue", "prior_assumption": "Verify the stage.", + "observed_reality": "The stage passed its independent check.", + "retained": ["Verify the next outcome."], "evidence_refs": ["evidence:stage"]}, + }, + "vision_checkpoint": { + "schema_version": "vision_checkpoint_v0", "agent_id": AGENT_ID, + "satisfied": True, "decision": "patched", "delivery_boundary": "semantic_closeout", + "triggers": [{"kind": "material_delivery_outcome", "delivery_outcome": "outcome_progress"}], + }, + } + runs = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + runs.write_text(json.dumps(prior) + "\n" + runs.read_text()) + turn = "blocked-replan-history-turn" + rc, guard = _run_cli(registry_path, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", turn) + assert rc == 0, guard + assert guard["decision"] == "autonomous_replan_required", guard + obligation = guard["replan_action_packet"]["obligation_id"] + packet = tmp_path / "vision.json" + packet.write_text(json.dumps({ + "schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed", + "vision_patch": {"vision_summary": "Continue after the source is available.", + "acceptance_summary": "The source is still unavailable; the outcome remains open."}, + "path_delta": {"outcome": "wait", "prior_assumption": "The source may be available.", + "observed_reality": "An independent read confirmed the source is unavailable.", + "retained": ["Keep the outcome open."], "evidence_refs": ["evidence:source-unavailable"]}, + })) + rc, closed = _run_cli( + registry_path, runtime, "refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", turn, "--replan-obligation-id", obligation, + "--classification", "source_wait_replanned", "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_gap", "--progress-result-class", "blocked", + "--progress-blocker-id", "source-not-ready", "--progress-evidence-id", "evidence:source-unavailable", + "--agent-vision-json", str(packet), "--no-global-sync", "--suppress-external-sinks", + ) + assert rc == 0, closed + assert closed["autonomous_replan_ack"]["semantic_delta"]["accepted"] is True + assert closed["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend" + assert closed["delivery_outcome"] == "outcome_gap" + assert _spend_run_count(runtime) == 0 + + # Read actual persisted facts through both status and subsequent admission. + # No receipt, checkpoint, history or clock is rewritten to obtain the result. + rc, status = _run_cli(registry_path, runtime, "status", "--goal-id", GOAL_ID) + assert rc == 0, status + goal = next(goal for goal in status["run_history"]["goals"] if goal["id"] == GOAL_ID) + context = next(row for row in goal["semantic_history"]["agents"] if row["agent_id"] == AGENT_ID) + assert context["latest_outcome_vision_checkpoint_run"]["generated_at"] == stamp + assert context["latest_material_milestone_run"]["generated_at"] == stamp + assert context["latest_autonomous_replan_ack_run"]["generated_at"] != stamp + for next_turn in (turn, "after-blocked-replan-history-turn"): + rc, readback = _run_cli(registry_path, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", next_turn) + assert rc == 0, readback + diagnostics = readback.get("vision_continuation_audit", {}).get("outcome_checkpoint_diagnostics", []) + assert diagnostics == [] + assert not any(trigger.get("kind") == "vision_outcome_checkpoint_required" + for trigger in readback.get("autonomous_replan_obligation", {}).get("triggers", [])) + assert _spend_run_count(runtime) == 0 + + def test_todoless_blocked_replan_settles_read_only_external_evidence_without_worktree( tmp_path: Path, ) -> None: diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index c6d2fa1b1f..811a8b9586 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -183,6 +183,14 @@ test("a service-qualified blocked replan closes only its exact Turn without debi assert.equal((accepted.progress as any).closeout_kind, "typed_blocked_writeback_no_spend"); assert.equal((accepted.spend as any).payload.ok, false); assert.equal((accepted.terminal_closeout as any).payload.ok, false); + const history = async (runs: unknown[]) => (await readQuotaSettlement({ + schema_version: "loopx_quota_blocked_replan_history_request_v0", runs, + })).accepted_indexes; + assert.deepEqual(await history([base, {...base, administrative_blocked_replan: true, + autonomous_replan_ack: null}]), [0]); + const compactCheckpoint = {...checkpoint} as Record; + delete compactCheckpoint.delivery_boundary; + assert.deepEqual(await history([{...base, vision_checkpoint: compactCheckpoint}]), [0]); await writeFile(index, JSON.stringify({...base, autonomous_replan_ack: {...ack, semantic_delta: {...semantic, satisfying_outcomes: ["fresh_vision_path_outcome"]}}}) + "\n"); assert.equal((await read()).replay_phase, "settled"); @@ -202,6 +210,7 @@ test("a service-qualified blocked replan closes only its exact Turn without debi {triggers: []}].map(patch => ({...base, vision_checkpoint: {...checkpoint, ...patch}})), ]; for (const changed of invalid) { + assert.deepEqual(await history([changed]), []); await writeFile(index, JSON.stringify(changed) + "\n"); assert.notEqual((await read()).replay_phase, "settled"); } From b9ca41b6904588ccdfdc84b528dac37fd4727bec Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 19:10:29 +0800 Subject: [PATCH 4/9] fix: preserve scoped receipt waits with nonblocking notices Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/loopx-overall-roadmap-v0.md | 9 +- .../rfcs/loopx-overall-roadmap-v0.zh-CN.md | 7 +- docs/quota-allocation.md | 11 + loopx/control_plane/quota/blocked_retry.py | 8 +- .../work_items/execution_obligation.py | 9 +- .../work_items/interaction_contract.py | 1 + .../test_quota_settlement_cli.py | 197 ++++++++++++++++-- 7 files changed, 217 insertions(+), 25 deletions(-) diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index c90732c3ca..163bf46d70 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -133,9 +133,12 @@ separate from this source repair. also separates an accepted blocked replan from the preceding real outcome checkpoint. File/SQLite CLI readback and long/compacted history regressions retain the previous milestone or genuine gap without creating a new outcome obligation -from administrative closure. The complete local CLI suite still has failures; -source qualification, integration, installed adoption and measured benefit remain -distinct. Preserve those failures and the existing full-suite qualification work. +from administrative closure. Source coverage now includes notice/gate overlap, +original-deadline recovery and fresh canonical unclaim/reassignment with each +Agent's own Vision and clock. The complete affected CLI suite passes after +canonical-fixture integration and original-Turn accounting checks; preserve the +earlier failures and their base/head attribution. Full-repository qualification, +integration, installed adoption and measured benefit remain distinct. **S5/S11 long-horizon progress design.** The [research measurement refinement](long-horizon-harness-benchmark-research-program-v0.md#evidence-based-episode-evaluation-proposed-refinement) and [product reading journey](../../product/use-cases/auto-research/long-horizon-progress-design.md) diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md index 32827bb7aa..a825f3f419 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md @@ -106,9 +106,10 @@ managed 与 attached 的工作对话都应能持续在 LoopX 中进行:沿用 **S3/S7 行政收尾提案。** [#7117](https://github.com/loopx-project/loopx/pull/7117) 进一步区分已验收的阻塞重规划与前一个真实成果 checkpoint。File/SQLite CLI 回读和 -长期/压缩历史回归保留原里程碑或真实缺口,避免行政收尾制造新的成果义务。完整 -本地 CLI 套件仍有失败;源码资格、集成、安装采用与实测收益分别验证,保留失败 -记录和既有完整套件资格工作。 +长期/压缩历史回归保留原里程碑或真实缺口,避免行政收尾制造新的成果义务。源码 +覆盖已包括通知/真门禁叠加、原截止恢复,以及各自持有 Vision 和时钟时的原生取消 +认领/转派。canonical fixture 整合及原 Turn 记账检查后的完整受影响 CLI 套件通过; +保留此前失败及 base/head 归因。整仓资格、集成、安装采用与实测收益仍分别验证。 **S5/S11 长程进展设计。** [研究计量细化](long-horizon-harness-benchmark-research-program-v0.zh-CN.md#基于证据的研究片段评估拟议细化) 与[产品阅读旅程](../../product/use-cases/auto-research/long-horizon-progress-design.zh-CN.md) diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 92bfcc87d7..63dbae1f1e 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -512,6 +512,17 @@ monitor clock, genuine gate and independent runnable work keep their own rules. This corrects the automatic notice explanation without adding configuration or persisted state. The pending retry is a recheck deadline, not execution authority: current gates and the configured cadence floor still apply at and after due time. + +A causal Vision wait retains the same non-gating notice and original retry +deadline; it must not emit a response plan asking the worker to wait for that +user action. Unscoped status overlays apply a receipt retry only to its current +claimant. An unclaimed Todo remains available to other Agents; the requesting +Agent's admission overlays only its own retry. History retains the original +Run and clock, and genuine gates still block their declared scope. + +关联 Vision 的等待保留同一非阻塞通知与原重试期限,不得再投影“等待用户”的响应 +计划。未分 Agent 的 status 只向当前认领者投影其重试;未认领 Todo 仍可由其他 Agent +选择,各 Agent 的准入只叠加自己的重试。原 Run/时钟和真正授权门禁保持原边界。 Receipt waits added after stored resume lanes were built are read from the same claim-scoped full selection source. An existing wait lane or its display limit cannot hide a new retry; foreign, excluded, mismatched and replaced waits do not diff --git a/loopx/control_plane/quota/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py index 14b2a4007c..7c04089b12 100644 --- a/loopx/control_plane/quota/blocked_retry.py +++ b/loopx/control_plane/quota/blocked_retry.py @@ -113,6 +113,7 @@ def overlay_active_turn_retries( goal_id: str, agent_id: str | None, observed_at: str, + include_unclaimed: bool = True, ) -> dict[str, Any]: """Project a receipt-owned wait into selection without editing hard-lease Todos. @@ -208,7 +209,8 @@ def overlay_summary(summary: Any) -> tuple[Any, bool]: and todo.get("status") == "open" and todo.get("task_class") == "advancement_task" and not todo.get("resume_when") - and todo.get("claimed_by") in (None, agent_id) + and (todo.get("claimed_by") == agent_id or + include_unclaimed and todo.get("claimed_by") is None) ): resume = retry["resume_when"] projected_rows.append( @@ -274,6 +276,8 @@ def overlay_status_turn_retries( History is the uncompressed control-plane snapshot, not the display tail. Qualification, supersession and clocks stay with the existing retry owner. + An unclaimed Todo is shared across lanes: only the requesting Agent's + admission may overlay its own retry there, never an unscoped status read. No canonical Todo, receipt, scheduler or quota mutation is performed here. """ payload = {"run_history": history, "attention_queue": queue} @@ -295,6 +299,6 @@ def overlay_status_turn_retries( for agent_id in agents: payload = overlay_active_turn_retries( payload, goal_id=goal["id"], agent_id=agent_id, - observed_at=observed_at, + observed_at=observed_at, include_unclaimed=False, ) return payload["attention_queue"] diff --git a/loopx/control_plane/work_items/execution_obligation.py b/loopx/control_plane/work_items/execution_obligation.py index 9b482fd8fc..1d8625c5d7 100644 --- a/loopx/control_plane/work_items/execution_obligation.py +++ b/loopx/control_plane/work_items/execution_obligation.py @@ -2,6 +2,7 @@ from typing import Any, cast +from ..agents.agent_scope_frontier import AgentScopeFrontierAction from ..quota.effective_action import EffectiveAction @@ -219,9 +220,13 @@ def build_execution_obligation( "only controls whether to interrupt the user" ), } - # A quiet monitor route still owns a pending retry, not a user-action gate. + # Quiet monitor and causal Vision routes retain the same qualified receipt + # wait. The typed interaction/scheduler owner still decides its disposition. if blocked_retry_wait and ( - should_run or effective_action == EffectiveAction.MONITOR_QUIET_SKIP.value + should_run or effective_action in { + EffectiveAction.MONITOR_QUIET_SKIP.value, + AgentScopeFrontierAction.AGENT_SCOPE_WAIT.value, + } ): return { "must_attempt_work": False, diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index ba009ab2f7..c7556da91d 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1357,6 +1357,7 @@ def _build_interaction_response_plan( if not ( user_channel.get("action_required") is True + and user_channel.get("non_gating_notice") is not True and user_channel.get("notify") == "NOTIFY" and agent_channel.get("must_attempt") is False and agent_channel.get("delivery_allowed") is False diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index f4402688cb..43f22decf1 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1902,16 +1902,17 @@ def test_typed_blocked_retry_with_peer_hard_lease( assert next_turn["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] <= 5 -@pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, -) -> None: +def _receipt_retry_with_vision_fixture(tmp_path, monkeypatch, provider, *, peer_agent=None): + """Create the wait through its actual admission and blocked-writeback owners.""" from canonical_authority_fixture import isolate_sqlite_runtime - from loopx.control_plane.quota.turn_envelope import build_turn_envelope - if provider == "sqlite": isolate_sqlite_runtime(tmp_path, monkeypatch) project, runtime, registry = _write_fixture(tmp_path) + if peer_agent is not None: + configuration = json.loads(registry.read_text()) + configuration["goals"][0]["coordination"]["registered_agents"].append(peer_agent) + configuration["goals"][0]["coordination"]["peer_independent_worktree_required"] = False + registry.write_text(json.dumps(configuration)) _configure_completion_validation_todo(project) _initialize_fixture_authority(project, runtime, provider=provider) rc, before = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) @@ -1943,22 +1944,55 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( "--no-global-sync", "--suppress-external-sinks", cwd=project) assert rc == 0, closed assert closed["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend" + return project, runtime, registry, before, closed + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("human_item", [None, "user_action", "user_gate"]) +def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, human_item: str | None, +) -> None: + from loopx.control_plane.quota.turn_envelope import build_turn_envelope + + project, runtime, registry, before, closed = _receipt_retry_with_vision_fixture( + tmp_path, monkeypatch, provider, + ) + if human_item is not None: + scope = "--bound-agent" if human_item == "user_action" else "--blocks-agent" + rc, human = _run_cli( + registry, runtime, "todo", "add", "--goal-id", GOAL_ID, "--role", "user", + "--agent-id", AGENT_ID, "--task-class", human_item, scope, AGENT_ID, + "--text", "Review independent work.", + ) + assert rc == 0, human rc, waiting = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", "turn-before-feedback-due", cwd=project) assert rc == 0, waiting assert waiting["execution_obligation"]["must_attempt_work"] is False + if human_item == "user_gate": + assert waiting["interaction_contract"]["user_channel"]["action_required"] is True + assert not waiting["interaction_contract"]["user_channel"].get("non_gating_notice") + assert waiting["scheduler_hint"]["action"] == "backoff_waiting_for_user" + assert _spend_run_count(runtime) == 0 + return assert not waiting.get("autonomous_replan_obligation") assert waiting["interaction_contract"]["agent_channel"]["delivery_allowed"] is False assert waiting["agent_todo_summary"]["blocked_retry_wait"]["due_at"] == closed["blocked_retry"]["due_at"] wait = waiting["goal_frontier_projection"]["vision_wait_state"] assert wait["selected_todo_id"] == TODO_ID assert wait["resume_condition"]["scheduled_for"] == closed["blocked_retry"]["due_at"] + if human_item == "user_action": + assert waiting["interaction_contract"]["user_channel"]["non_gating_notice"] is True + assert "response_plan" not in waiting["interaction_contract"] + assert waiting["scheduler_hint"]["reason_code"] == "blocked_wait" + assert waiting["scheduler_hint"]["action"] != "backoff_waiting_for_user" + assert waiting["scheduler_hint"]["pending_retry"]["due_at"] == closed["blocked_retry"]["due_at"] envelope = build_turn_envelope(waiting) assert envelope["contract_capsule"]["execution_obligation"]["must_attempt_work"] is False rc, after = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) assert rc == 0, after - assert after["todos"] == before["todos"] + assert next(todo for todo in after["todos"] if todo["todo_id"] == TODO_ID) == before["todos"][0] assert _spend_run_count(runtime) == 0 rc, independent = _run_cli( registry, runtime, "todo", "add", "--goal-id", GOAL_ID, @@ -1977,6 +2011,99 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( assert _spend_run_count(runtime) == 0 +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("reassign", [False, True]) +def test_fresh_peer_does_not_borrow_receipt_retry_after_canonical_handoff( + tmp_path, monkeypatch, provider, reassign, +): + peer = "codex-settlement-peer" + project, runtime, registry, before, closed = _receipt_retry_with_vision_fixture( + tmp_path, monkeypatch, provider, peer_agent=peer, + ) + # Give B its own causally related Vision through a real work Turn. Merely + # selecting an explicit Todo can bypass the stale unscoped projection and + # cannot establish that fresh automatic admission uses B's own frontier. + rc, added = _run_cli(registry, runtime, "todo", "add", "--goal-id", GOAL_ID, + "--role", "agent", "--claimed-by", peer, "--task-class", "advancement_task", + "--text", "Verify independent arithmetic.", "--action-kind", "validate", + "--target-key", "peer-arithmetic") + assert rc == 0, added + peer_binding = ("--goal-id", GOAL_ID, "--agent-id", peer, "--todo-id", added["todo_id"], + "--turn-instance-id", "turn-peer-arithmetic") + rc, admitted = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", + *peer_binding, cwd=project) + assert rc == 0, admitted + artifact = project / "verified-arithmetic.txt" + artifact.write_text("391\n") + assert int(artifact.read_text()) == 17 * 23 + peer_vision = tmp_path / "peer-vision.json" + peer_vision.write_text(json.dumps({ + "schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed", + "vision_patch": {"acceptance_summary": "Verify the retained original work after arithmetic.", + "advancement_policy": "repeat_until_closed"}, + "todo_delta": [f"retain:{TODO_ID}"], + "path_delta": {"outcome": "continue", "prior_assumption": "Verify arithmetic before the next work.", + "observed_reality": "The independent artifact equals 17 times 23.", + "retained": ["Verify the retained original work."], + "evidence_refs": ["evidence:verified-arithmetic"]}, + })) + rc, written = _run_cli(registry, runtime, "refresh-state", *peer_binding, + "--classification", "peer_arithmetic_verified", "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_progress", "--agent-vision-json", str(peer_vision), + "--no-global-sync", "--suppress-external-sinks", cwd=project) + assert rc == 0, written + rc, spent = _run_cli(registry, runtime, "quota", "spend-slot", *peer_binding, + "--slots", "1", "--source", "heartbeat", "--execute", cwd=project) + assert rc == 0, spent + rc, done = _run_cli(registry, runtime, "todo", "complete", *peer_binding, + "--evidence", str(artifact), "--successor-todo-id", TODO_ID, cwd=project) + assert rc == 0, done + # The current owner releases its assignment through the real canonical + # writer. No Run, clock, retry or receipt is patched for the observation. + rc, current = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) + assert rc == 0, current + change = ("--claimed-by", peer) if reassign else ("--clear-claim",) + rc, updated = _run_cli( + registry, runtime, "todo", "update", "--goal-id", GOAL_ID, "--todo-id", TODO_ID, + "--agent-id", AGENT_ID, "--update-operation-id", "transfer-after-feedback", + "--update-expected-provider-revision", current["authority_read"]["provider_revision"], *change, + ) + assert rc == 0, updated + rc, automatic = _run_cli( + registry, runtime, "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", peer, "--turn-instance-id", "turn-fresh-peer-after-feedback", cwd=project, + ) + assert rc == 0, automatic + assert automatic["agent_todo_summary"]["work_counts"]["advancement"] == 1 + assert not automatic.get("vision_wait_state") + assert not automatic["scheduler_hint"].get("pending_retry") + rc, fresh = _run_cli( + registry, runtime, "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", peer, "--todo-id", TODO_ID, + "--turn-instance-id", "turn-fresh-peer-after-feedback", cwd=project, + ) + assert rc == 0, fresh + assert fresh["selected_todo"]["todo_id"] == TODO_ID + assert fresh["execution_obligation"]["must_attempt_work"] is True + assert not fresh["agent_todo_summary"].get("blocked_retry_wait") + assert not fresh["scheduler_hint"].get("pending_retry") + assert _spend_run_count(runtime) == 1 + rc, readback = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) + assert rc == 0, readback + assert readback["todo"].get("claimed_by") == (peer if reassign else None) + rc, replay = _run_cli(registry, runtime, "refresh-state", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", "turn-vision-feedback", + "--classification", "feedback_pending", "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_gap", "--progress-result-class", "blocked", + "--progress-blocker-id", "external-feedback", "--progress-evidence-id", "evidence:feedback-pending", + "--agent-vision-json", str(tmp_path / "vision.json"), + "--no-global-sync", "--suppress-external-sinks", cwd=project) + assert rc == 0, replay + assert replay["idempotent_replay"] is True + assert replay["blocked_retry"] == closed["blocked_retry"] + assert _spend_run_count(runtime) == 1 + + @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, @@ -5106,6 +5233,11 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( first_rc, first = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID) assert first_rc == 0, first assert first["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID + # This result exists before the lifecycle wait; recovery must account for + # it without executing more work or rebinding the original Turn. + artifact = project / "verified-before-wait.txt" + artifact.write_text("391\n") + assert int(artifact.read_text()) == 17 * 23 update_rc, update = _run_cli( registry_path, runtime, "todo", "update", "--goal-id", GOAL_ID, @@ -5117,26 +5249,30 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( assert update_rc == 0, update replay_rc, replay = _run_cli(registry_path, runtime, *guard) assert replay_rc == 0, replay - assert replay["effective_action"] == "quota_skip" - assert replay["should_run"] is False + assert replay["effective_action"] == "unsettled_host_turn_recovery" + assert replay["normal_delivery_allowed"] is False + assert replay["unsettled_host_turn_recovery"]["binding_id"] == TODO_ID + assert replay["unsettled_host_turn_recovery"]["turn_instance_id"] == turn_id assert replay["heartbeat_receipt"]["status"] == "replayed" assert replay["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID - assert replay["selected_todo"]["todo_id"] == TODO_ID + assert replay.get("selected_todo") is None assert replay["work_lane_contract"]["obligation"] == ( - "wait_for_receipt_bound_deferred_todo" + "settle_bound_turn" ) - assert replay["work_lane_contract"]["must_attempt_work"] is False + assert replay["work_lane_contract"]["must_attempt_work"] is True assert replay["recommended_action"] == replay["work_lane_contract"]["action"] assert "independent alternative delivery" not in replay["recommended_action"] assert replay.get("agent_lane_next_action") is None interaction = replay["interaction_contract"] - assert interaction["agent_channel"]["must_attempt"] is False - assert interaction["cli_channel"]["spend_after_validation"] is False + assert interaction["agent_channel"]["must_attempt"] is True + assert interaction["agent_channel"]["delivery_allowed"] is False + assert interaction["cli_channel"]["spend_allowed_now"] is False + assert interaction["cli_channel"]["spend_after_validation"] is True assert ALTERNATIVE_TODO_ID not in json.dumps( interaction["cli_channel"].get("next_cli_actions", []) ) plan = interaction["cli_channel"].get("settlement_plan") - assert plan is None or plan["identity"]["todo_id"] == TODO_ID + assert plan["identity"] == first["heartbeat_receipt"]["settlement_identity"] assert _heartbeat_receipt_count(runtime, turn_id) == 1 assert _spend_run_count(runtime) == 0 @@ -5150,6 +5286,37 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( } assert _heartbeat_receipt_count(runtime, turn_id) == 1 + vision = tmp_path / "before-wait-vision.json" + vision.write_text(json.dumps({ + "schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed", + "vision_patch": {"acceptance_summary": "The arithmetic result is verified; the next work waits."}, + "path_delta": {"outcome": "continue", "prior_assumption": "Verify the result before waiting.", + "observed_reality": "The existing artifact equals 17 times 23; the next work is deferred.", + "retained": ["Keep the original future wait."], + "evidence_refs": ["evidence:verified-before-wait"]}, + })) + binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, + "--turn-instance-id", turn_id) + write_args = ("refresh-state", *binding, "--classification", "verified_before_wait", + "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress", + "--agent-vision-json", str(vision), "--no-global-sync", "--suppress-external-sinks") + rc, written = _run_cli(registry_path, runtime, *write_args, cwd=project) + assert rc == 0, written + rc, replayed_write = _run_cli(registry_path, runtime, *write_args, cwd=project) + assert rc == 0 and replayed_write["idempotent_replay"] is True, replayed_write + spend_args = ("quota", "spend-slot", *binding, "--slots", "1", "--source", "heartbeat", "--execute") + for _ in range(2): + rc, spent = _run_cli(registry_path, runtime, *spend_args, cwd=project) + assert rc == 0, spent + assert _classification_count(runtime, "verified_before_wait") == 1 + assert _spend_run_count(runtime) == 1 + rc, unchanged = _run_cli(registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID, + "--todo-id", TODO_ID) + assert rc == 0, unchanged + assert unchanged["todo"]["status"] == "deferred" + assert unchanged["todo"]["resume_when"] == "resume_at:2099-01-01T00:00:00Z" + assert artifact.read_text() == "391\n" + next_rc, next_turn = _run_cli( registry_path, runtime, "quota", "should-run", "--codex-app", From 89c0d7dc53945e562561ace8cb981d0b3836eabc Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:06:13 +0800 Subject: [PATCH 5/9] refactor(quota): centralize receipt retry history qualification Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/loopx-overall-roadmap-v0.md | 9 ++- .../rfcs/loopx-overall-roadmap-v0.zh-CN.md | 5 +- docs/quota-allocation.md | 16 +++++ loopx/control_plane/quota/blocked_retry.py | 58 ++----------------- loopx/control_plane/quota/settlement_phase.ts | 35 ++++++++++- .../quota/settlement_readback.ts | 12 ++-- .../runtime/run_context_retention.py | 14 ++--- loopx/control_plane/runtime_timestamp.ts | 17 ++++-- .../work_items/autonomous_replan_ack.py | 50 +++++++++++----- .../test_final_outcome_diagnostics.py | 45 ++++++++++++++ .../quota_settlement_readback.test.ts | 49 +++++++++++++++- 11 files changed, 217 insertions(+), 93 deletions(-) diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index 74d4d9b39e..b424ed72c9 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -137,8 +137,13 @@ from administrative closure. Source coverage now includes notice/gate overlap, original-deadline recovery and fresh canonical unclaim/reassignment with each Agent's own Vision and clock. The complete affected CLI suite passes after canonical-fixture integration and original-Turn accounting checks; preserve the -earlier failures and their base/head attribution. Full-repository qualification, -integration, installed adoption and measured benefit remain distinct. +earlier failures and their base/head attribution. The follow-on refactor removes +Python retry qualification and batches administrative/retry history facts through +the existing TypeScript settlement owner for semantic retention and legacy status. +It preserves independent outcomes, original clocks and historical microsecond +expiry; malformed Turn identities no longer establish a positive retry proof. +Full-repository qualification, comparable full caller cost, installed adoption +and measured long-horizon benefit remain distinct and incomplete. **S5/S11 long-horizon progress design.** The [research measurement refinement](long-horizon-harness-benchmark-research-program-v0.md#evidence-based-episode-evaluation-proposed-refinement) and [product reading journey](../../product/use-cases/auto-research/long-horizon-progress-design.md) diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md index 5b3a61638d..f56238f190 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md @@ -109,7 +109,10 @@ managed 与 attached 的工作对话都应能持续在 LoopX 中进行:沿用 长期/压缩历史回归保留原里程碑或真实缺口,避免行政收尾制造新的成果义务。源码 覆盖已包括通知/真门禁叠加、原截止恢复,以及各自持有 Vision 和时钟时的原生取消 认领/转派。canonical fixture 整合及原 Turn 记账检查后的完整受影响 CLI 套件通过; -保留此前失败及 base/head 归因。整仓资格、集成、安装采用与实测收益仍分别验证。 +保留此前失败及 base/head 归因。后续重构删除 Python 重试资格判定,由已有 TypeScript +结算 owner 批量核对行政/重试历史事实,供语义保留和旧状态读路使用。各维度仍独立, +保留原时钟及历史微秒到期边界;畸形 Turn 身份不再取得重试活动的正向证明。整仓 +资格、可比完整调用成本、安装采用与长程实测收益仍是分别验证的未完成项。 **S5/S11 长程进展设计。** [研究计量细化](long-horizon-harness-benchmark-research-program-v0.zh-CN.md#基于证据的研究片段评估拟议细化) 与[产品阅读旅程](../../product/use-cases/auto-research/long-horizon-progress-design.zh-CN.md) diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 63dbae1f1e..196c0886a5 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -259,6 +259,22 @@ grants no execution, completion, acceptance or settlement authority. 重规划 ACK 仍然可见;缺失 checkpoint 与真实交付 gap 保持原复核要求,这一读取 分类不授予执行、完成、验收或结算权限。 +The same settlement owner qualifies active receipt retries for semantic-history +selection and the legacy status overlay. Each reader sends one candidate batch; +ordinary history without candidates needs no runtime request. Administrative +acceptance and retry activity remain separate results. Python transports facts +and applies existing row selection instead of maintaining a second retry rule. +Historical reads retain aware timestamps, microsecond expiry and the original +deadline; live settlement validation is unchanged. A non-string or blank Turn +identity cannot establish an active retry. Full caller cost and installed +adoption require separate qualification. + +同一结算 owner 也为语义历史选取和旧状态叠加核对活动回执重试。每条读路只发送一批 +候选;没有候选的普通历史无需 runtime 请求。行政验收和重试活动分别返回,Python +传递事实并沿用已有行选取,不再维护第二套重试规则。历史读取保留带时区时间、微秒 +到期边界和原截止时间;实时结算校验不变。非字符串或空白 Turn 身份不能证明重试 +活动。完整调用成本与安装采用仍需分别验证。 + An accepted `vision_checkpoint_v0` for `outcome_progress` also closes its exact Todo-bound Turn after the matching writeback and spend receipts commit. Both `in_flight_continuation` and `semantic_closeout` use their corresponding typed diff --git a/loopx/control_plane/quota/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py index 7c04089b12..c252f562ad 100644 --- a/loopx/control_plane/quota/blocked_retry.py +++ b/loopx/control_plane/quota/blocked_retry.py @@ -2,7 +2,6 @@ from __future__ import annotations -from datetime import datetime from typing import Any from ..todos.contract import normalize_todo_id @@ -11,11 +10,7 @@ evaluate_todo_resume_conditions, ) from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result - -BLOCKED_RETRY_SCHEMA_VERSION = "quota_blocked_retry_v0" -MIN_RETRY_SECONDS = 60 -MAX_RETRY_SECONDS = 30 * 60 -TURN_SETTLEMENT_RETRY_SECONDS = 5 * 60 +from ..work_items.autonomous_replan_ack import qualify_settlement_history def compact_settlement_wait_todos( @@ -66,47 +61,6 @@ def require_blocked_retry_wait( return result -def active_turn_retry_for_run( - run: dict[str, Any], *, observed_at: str -) -> dict[str, Any] | None: - """Accept only the exact active, receipt-owned retry on a blocked work Run.""" - - todo_id = normalize_todo_id(run.get("todo_id")) - retry = run.get("blocked_retry") - progress = run.get("progress_observation") - if ( - not todo_id - or not isinstance(retry, dict) - or retry.get("schema_version") != BLOCKED_RETRY_SCHEMA_VERSION - or retry.get("source") != "turn_settlement" - or retry.get("todo_id") != todo_id - or run.get("delivery_outcome") != "outcome_gap" - or not isinstance(progress, dict) - or progress.get("result_class") != "blocked" - or not run.get("turn_instance_id") - ): - return None - try: - now = datetime.fromisoformat(observed_at.replace("Z", "+00:00")) - observed = datetime.fromisoformat( - str(retry["observed_at"]).replace("Z", "+00:00") - ) - due = datetime.fromisoformat(str(retry["due_at"]).replace("Z", "+00:00")) - delay = (due - observed).total_seconds() - except (KeyError, ValueError, TypeError): - return None - if ( - now.tzinfo is None - or observed.tzinfo is None - or due.tzinfo is None - or not MIN_RETRY_SECONDS <= delay <= MAX_RETRY_SECONDS - or due <= now - or retry.get("resume_when") != f"resume_at:{retry['due_at']}" - ): - return None - return retry - - def overlay_active_turn_retries( status_payload: dict[str, Any], *, @@ -150,11 +104,11 @@ def overlay_active_turn_retries( ) if not isinstance(runs, list): return status_payload + runs = [run for run in runs if isinstance(run, dict) and run.get("agent_id") == agent_id] + qualification = qualify_settlement_history(runs, observed_at=observed_at) waits: dict[str, dict[str, Any]] = {} seen: set[str] = set() - for run in runs: - if not isinstance(run, dict) or run.get("agent_id") != agent_id: - continue + for index, run in enumerate(runs): todo_id = normalize_todo_id(run.get("todo_id")) if not todo_id or todo_id in seen: continue @@ -162,8 +116,8 @@ def overlay_active_turn_retries( if classification.startswith(("quota_slot_", "quota_scheduler_")): continue seen.add(todo_id) - retry = active_turn_retry_for_run(run, observed_at=observed_at) - if retry is not None: + if index in qualification.active_turn_retries: + retry = run["blocked_retry"] waits[todo_id] = { **retry, "goal_id": goal_id, "agent_id": agent_id, "turn_instance_id": run["turn_instance_id"], diff --git a/loopx/control_plane/quota/settlement_phase.ts b/loopx/control_plane/quota/settlement_phase.ts index 2fa49f6c79..f2c867aaa6 100644 --- a/loopx/control_plane/quota/settlement_phase.ts +++ b/loopx/control_plane/quota/settlement_phase.ts @@ -1,8 +1,14 @@ import { settlementIdentity, type SettlementIdentity } from "../effect_program.ts"; import { jsonObject } from "../runtime_decode.ts"; +import { parseTodoTimestampMicros } from "../runtime_timestamp.ts"; +import { isCanonicalTodoId } from "../todos/identity.ts"; import { isBlockedLifecycleWait, isCausalBlockedWait, isMonitorUnavailableWait } from "./blocked_wait.ts"; import { isTurnScopedSettlementOutcome } from "../work_items/delivery_outcome.ts"; +function retryDelayIsBounded(seconds: number): boolean { + return Number.isFinite(seconds) && seconds >= 60 && seconds <= 30 * 60; +} + /** Historical entry point for blocked no-spend qualification: an advancement * retry/causal wait, blocked lifecycle or a Monitor unavailable attempt. * A lifecycle/Monitor closeout creates no retry clock or execution grant. */ @@ -19,7 +25,7 @@ export function isBoundedBlockedRetry(value: unknown, todoId: string | null): bo const observed = Date.parse(retry.observed_at); const due = Date.parse(retry.due_at); const delay = (due - observed) / 1000; - return Number.isFinite(delay) && delay >= 60 && delay <= 30 * 60; + return retryDelayIsBounded(delay); } /** The replan owner has already qualified the blocker/path and its evidence. @@ -86,6 +92,33 @@ export function acceptedBlockedReplanHistoryIndexes(runs: unknown[]): number[] { return accepted; } +/** Read-only history facts. Administrative closeout and a pending retry are + * independent dimensions; neither is a new effect, lease or completion grant. + * Use the same retry bound as settlement, with the aware historical codec so + * expiry keeps microseconds and supported ISO offsets across runtime changes. */ +export function qualifySettlementHistory(runs: unknown[], observedAt: string | null): { + accepted_indexes: number[]; active_retry_indexes: number[]; +} { + const active: number[] = []; + const now = observedAt === null ? null : parseTodoTimestampMicros(observedAt, true); + if (now !== null) runs.forEach((value, index) => { + const run = jsonObject(value), retry = jsonObject(run?.blocked_retry); + const progress = jsonObject(run?.progress_observation); + const todoId = typeof run?.todo_id === "string" ? run.todo_id.trim().toLowerCase() : ""; + if (!isCanonicalTodoId(todoId) || !retry || retry.schema_version !== "quota_blocked_retry_v0" + || retry.source !== "turn_settlement" || retry.todo_id !== todoId + || run?.delivery_outcome !== "outcome_gap" || progress?.result_class !== "blocked" + || typeof run?.turn_instance_id !== "string" || !run.turn_instance_id.trim() + || typeof retry.observed_at !== "string" || typeof retry.due_at !== "string" + || retry.resume_when !== `resume_at:${retry.due_at}`) return; + const observed = parseTodoTimestampMicros(retry.observed_at, true); + const due = parseTodoTimestampMicros(retry.due_at, true); + if (observed !== null && due !== null && due > now + && retryDelayIsBounded(Number(due - observed) / 1_000_000)) active.push(index); + }); + return {accepted_indexes: acceptedBlockedReplanHistoryIndexes(runs), active_retry_indexes: active}; +} + /** A service-qualified capability duty can retire its exact admitted Turn. * The caller first verifies the durable writeback receipt. No Task/Goal * completion, progress or debit follows from this lifecycle evidence. */ diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index b8ca2a93ae..7cd43d8e5f 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -34,7 +34,7 @@ import { import { isBoundedBlockedRetry, isAcceptedBlockedReplanWriteback, - acceptedBlockedReplanHistoryIndexes, + qualifySettlementHistory, isCommittedMonitorPollEffect, isAcceptedProgressWriteback, isAcceptedReplanWriteback, @@ -1383,10 +1383,12 @@ export function readQuotaSettlementForAdmittedOwnerFromSnapshot( } export async function readQuotaSettlement(value: unknown): Promise { - if (jsonObject(value)?.schema_version === "loopx_quota_blocked_replan_history_request_v0") { - const runs = jsonObject(value)?.runs; - if (!Array.isArray(runs)) throw new EffectRuntimeRequestError("blocked replan history requires Run facts"); - return {accepted_indexes: acceptedBlockedReplanHistoryIndexes(runs)}; + if (jsonObject(value)?.schema_version === "loopx_quota_history_qualification_request_v0") { + const input = jsonObject(value)!, runs = input.runs, observedAt = input.observed_at; + if (!Array.isArray(runs) || (observedAt != null && typeof observedAt !== "string")) { + throw new EffectRuntimeRequestError("history qualification requires Run facts and an optional observation clock"); + } + return qualifySettlementHistory(runs, typeof observedAt === "string" ? observedAt : null); } if (jsonObject(value)?.schema_version === RECEIPT_BOUND_WAIT_REQUEST_SCHEMA) { return projectReceiptBoundWait(value); diff --git a/loopx/control_plane/runtime/run_context_retention.py b/loopx/control_plane/runtime/run_context_retention.py index f457b8b682..16687ba82d 100644 --- a/loopx/control_plane/runtime/run_context_retention.py +++ b/loopx/control_plane/runtime/run_context_retention.py @@ -3,7 +3,6 @@ from collections.abc import Callable, Iterator from typing import Any -from ..quota.blocked_retry import active_turn_retry_for_run from ..work_items.delivery_outcome import ( MATERIAL_DELIVERY_OUTCOMES, PROGRESS_DELIVERY_OUTCOMES, @@ -12,7 +11,7 @@ from ..work_items.autonomous_replan_ack import ( AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, autonomous_replan_ack_recorded, - accepted_blocked_replan_history_indexes, + qualify_settlement_history, ) from ..work_items.autonomous_replan_obligation import run_history_agent_id from ..work_items.replan_history_codec import REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS @@ -143,7 +142,7 @@ def goal_semantic_history_from_runs( seen_retry_todos: set[tuple[str, str]] = set() observed_at = now_utc_iso() - administrative_indexes = accepted_blocked_replan_history_indexes(runs) + qualification = qualify_settlement_history(runs, observed_at=observed_at) for index, run in enumerate(runs): if latest_owner_correction_run is None and isinstance( run.get("human_reward"), dict @@ -159,10 +158,7 @@ def goal_semantic_history_from_runs( key = (agent_id, todo_id) if key not in seen_retry_todos: seen_retry_todos.add(key) - if ( - active_turn_retry_for_run(run, observed_at=observed_at) - is not None - ): + if index in qualification.active_turn_retries: active_blocked_retry_runs.append(run) context = contexts.setdefault(agent_id, {"agent_id": agent_id}) @@ -185,7 +181,7 @@ def goal_semantic_history_from_runs( if ( "latest_outcome_vision_checkpoint_run" not in context - and index not in administrative_indexes + and index not in qualification.accepted_blocked_replans and _run_has_outcome_vision_checkpoint(run) ): context["latest_outcome_vision_checkpoint_run"] = run @@ -203,7 +199,7 @@ def goal_semantic_history_from_runs( if ( "latest_material_milestone_run" not in context - and index not in administrative_indexes + and index not in qualification.accepted_blocked_replans and str(run.get("delivery_outcome") or "").strip() in MATERIAL_DELIVERY_OUTCOMES ): diff --git a/loopx/control_plane/runtime_timestamp.ts b/loopx/control_plane/runtime_timestamp.ts index 5f09c007b1..ea40bafdb3 100644 --- a/loopx/control_plane/runtime_timestamp.ts +++ b/loopx/control_plane/runtime_timestamp.ts @@ -40,13 +40,16 @@ export function parseIsoTimestamp(value: string): Date | null { /** Compatibility codec for datetime.fromisoformat inputs used by Todo metadata. * It keeps microseconds and offset seconds, which a JS Date cannot represent. - * Missing timezone means UTC, matching the existing Python runtime codec. */ -export function parseTodoTimestampMicros(value: string): bigint | null { - // The legacy wrapper replaces Z/z with +00:00 before fromisoformat, so - // these letters are timezone suffixes, never date/time separators. - const match = /^(\d{4}-\d{2}-\d{2}|\d{8}|\d{4}-W\d{2}(?:-[1-7])?|\d{4}W\d{2}[1-7]?)(?:[^Zz](.+))?$/u.exec(value); + * Missing timezone means UTC by default. The explicit aware-Python mode is + * for historical retry reads: datetime.fromisoformat(value.replace("Z", ...)) + * requires a zone, rejects lower-case z suffixes, but accepts z as a separator. + * Existing Todo callers keep their original codec without this mode. */ +export function parseTodoTimestampMicros(value: string, awarePythonDatetime = false): bigint | null { + // The Todo wrapper replaces Z/z; the aware retry wrapper replaces only Z. + const match = /^(\d{4}-\d{2}-\d{2}|\d{8}|\d{4}-W\d{2}(?:-[1-7])?|\d{4}W\d{2}[1-7]?)(?:([^Z])(.+))?$/u.exec(value); if (!match) return null; - const [, date, time] = match; + const [, date, separator, time] = match; + if (separator === "z" && !awarePythonDatetime) return null; let calendar: Date | null; if (date.includes("W")) { const week = /^(\d{4})-?W(\d{2})(?:-?([1-7]))?$/.exec(date)!; @@ -64,11 +67,13 @@ export function parseTodoTimestampMicros(value: string): bigint | null { if (!calendar || calendar.getUTCFullYear() < 1) return null; } if (time === undefined) { + if (awarePythonDatetime) return null; const result = BigInt(calendar.valueOf()) * 1000n; if (result < -62135596800000000n || result >= 253402300800000000n) return null; return result; } const parts = /^(.*?)(Z|z|[+-].*)?$/.exec(time)!; + if (awarePythonDatetime && (!parts[2] || parts[2] === "z")) return null; function clock(raw: string, offset: boolean): bigint | null { const parsed = /^(\d{2})(?:(:?)(\d{2})(?:\2(\d{2}))?)?(?:[.,](\d+))?$/.exec(raw); if (!parsed) return null; diff --git a/loopx/control_plane/work_items/autonomous_replan_ack.py b/loopx/control_plane/work_items/autonomous_replan_ack.py index cf9a48ac49..01d9d7cdbe 100644 --- a/loopx/control_plane/work_items/autonomous_replan_ack.py +++ b/loopx/control_plane/work_items/autonomous_replan_ack.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Any +from typing import Any, NamedTuple from ..effect_runtime import effect_runtime_result from ..runtime.time import parse_timestamp @@ -15,32 +15,50 @@ AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK = AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW * 3 -def accepted_blocked_replan_history_indexes(runs: list[dict[str, Any]]) -> set[int]: - """Transport persisted administrative candidates to the shared TS owner. +class SettlementHistoryQualification(NamedTuple): + accepted_blocked_replans: frozenset[int] + active_turn_retries: frozenset[int] - This only selects history slots. It neither acknowledges a replan nor - replaces the receipt checks required for actual settlement. Missing legacy - evidence remains an ordinary outcome checkpoint, conservatively. + +def qualify_settlement_history( + runs: list[dict[str, Any]], *, observed_at: str | None = None, +) -> SettlementHistoryQualification: + """Transport a history batch to its settlement owner, preserving originals. + + Candidate selection is an IO optimization, never positive qualification. + The TS result selects existing rows; it cannot acknowledge or settle work. + Missing legacy evidence remains an ordinary outcome checkpoint. """ - candidates = [index for index, run in enumerate(runs) if run.get("replan_obligation_id")] + candidates = [index for index, run in enumerate(runs) + if run.get("replan_obligation_id") + or (observed_at is not None and "blocked_retry" in run)] if not candidates: - return set() + return SettlementHistoryQualification(frozenset(), frozenset()) fields = ( "goal_id", "agent_id", "todo_id", "turn_instance_id", "replan_obligation_id", "settlement_identity", "delivery_outcome", "progress_observation", - "autonomous_replan_ack", "vision_checkpoint", + "autonomous_replan_ack", "vision_checkpoint", "blocked_retry", ) result = effect_runtime_result("quota.settlement.read", { - "schema_version": "loopx_quota_blocked_replan_history_request_v0", + "schema_version": "loopx_quota_history_qualification_request_v0", + "observed_at": observed_at, "runs": [{key: runs[index][key] for key in fields if key in runs[index]} for index in candidates], }, large_local_snapshot=True) - indexes = result.get("accepted_indexes") if isinstance(result, dict) else None - if not isinstance(indexes, list) or any( - type(index) is not int or not 0 <= index < len(candidates) for index in indexes - ): - raise RuntimeError("TypeScript blocked replan history result must contain valid indexes") - return {candidates[index] for index in indexes} + def indexes(field: str) -> frozenset[int]: + values = result.get(field) if isinstance(result, dict) else None + if not isinstance(values, list) or any( + type(index) is not int or not 0 <= index < len(candidates) for index in values + ): + raise RuntimeError("TypeScript settlement history result must contain valid indexes") + return frozenset(candidates[index] for index in values) + + return SettlementHistoryQualification(indexes("accepted_indexes"), indexes("active_retry_indexes")) + + +def accepted_blocked_replan_history_indexes(runs: list[dict[str, Any]]) -> frozenset[int]: + """Legacy outcome reader needs administrative qualification, not a clock.""" + return qualify_settlement_history(runs).accepted_blocked_replans def ack_binds_trigger_checkpoints(ack: dict[str, Any] | None) -> bool: diff --git a/tests/control_plane/test_final_outcome_diagnostics.py b/tests/control_plane/test_final_outcome_diagnostics.py index 55eb2fd46a..a710605d94 100644 --- a/tests/control_plane/test_final_outcome_diagnostics.py +++ b/tests/control_plane/test_final_outcome_diagnostics.py @@ -179,6 +179,51 @@ def observe(method, params, **kwargs): assert history["agents"][0]["latest_outcome_vision_checkpoint_run"]["generated_at"] == STAMP +def test_retry_history_uses_one_batch_and_never_retimes_the_source(monkeypatch): + from loopx.control_plane.work_items import autonomous_replan_ack as codec + + waiting = { + "agent_id": AGENT, "todo_id": "todo_pending001", "turn_instance_id": "pending-turn", + "delivery_outcome": "outcome_gap", "progress_observation": {"result_class": "blocked"}, + "blocked_retry": { + "schema_version": "quota_blocked_retry_v0", "source": "turn_settlement", + "todo_id": "todo_pending001", "observed_at": STAMP, + "due_at": "2026-01-01T00:05:00.000001Z", + "resume_when": "resume_at:2026-01-01T00:05:00.000001Z", + }, + } + original = deepcopy(waiting) + calls = [] + real = codec.effect_runtime_result + + def observe(method, params, **kwargs): + calls.append((method, len(params["runs"]))) + return real(method, params, **kwargs) + + monkeypatch.setattr(codec, "effect_runtime_result", observe) + monkeypatch.setattr("loopx.control_plane.runtime.run_context_retention.now_utc_iso", + lambda: "2026-01-01T00:05:00Z") + history = goal_semantic_history_from_runs([waiting] * 100 + [material_run()]) + assert calls == [("quota.settlement.read", 100)] + assert history["active_blocked_retry_runs"] == [waiting] + assert waiting == original + # One microsecond later the original clock expires; it is not renewed. + monkeypatch.setattr("loopx.control_plane.runtime.run_context_retention.now_utc_iso", + lambda: "2026-01-01T00:05:00.000001Z") + assert goal_semantic_history_from_runs([waiting])["active_blocked_retry_runs"] == [] + + +@pytest.mark.parametrize("indexes", [[True], [-1], [1], "0", None]) +def test_history_transport_rejects_malformed_qualification(indexes, monkeypatch): + from loopx.control_plane.work_items import autonomous_replan_ack as codec + + monkeypatch.setattr(codec, "effect_runtime_result", lambda *args, **kwargs: { + "accepted_indexes": [], "active_retry_indexes": indexes, + }) + with pytest.raises(RuntimeError, match="valid indexes"): + codec.qualify_settlement_history([{"blocked_retry": {}}], observed_at=STAMP) + + def decision(): status = quota_status_payload( goal_id=GOAL, diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index 811a8b9586..5fab14eefc 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -184,7 +184,7 @@ test("a service-qualified blocked replan closes only its exact Turn without debi assert.equal((accepted.spend as any).payload.ok, false); assert.equal((accepted.terminal_closeout as any).payload.ok, false); const history = async (runs: unknown[]) => (await readQuotaSettlement({ - schema_version: "loopx_quota_blocked_replan_history_request_v0", runs, + schema_version: "loopx_quota_history_qualification_request_v0", runs, })).accepted_indexes; assert.deepEqual(await history([base, {...base, administrative_blocked_replan: true, autonomous_replan_ack: null}]), [0]); @@ -2075,3 +2075,50 @@ test( ); }, ); + +test("history retry qualification preserves its original aware clock and exact expiry", async () => { + const run = {todo_id: "todo_pending001", turn_instance_id: "turn-pending", + delivery_outcome: "outcome_gap", progress_observation: {result_class: "blocked"}, + blocked_retry: {schema_version: "quota_blocked_retry_v0", source: "turn_settlement", + todo_id: "todo_pending001", observed_at: "2026-01-01T00:00:00.000001Z", + due_at: "2026-01-01T00:05:00.000001Z", resume_when: "resume_at:2026-01-01T00:05:00.000001Z"}}; + const read = (runs: unknown[], observed_at: unknown) => readQuotaSettlement({ + schema_version: "loopx_quota_history_qualification_request_v0", runs, observed_at, + }); + assert.deepEqual((await read([run], "2026-01-01T00:05:00Z")).active_retry_indexes, [0]); + assert.deepEqual((await read([run], "2026-01-01T00:05:00.000001Z")).active_retry_indexes, []); + assert.deepEqual((await read([run], "2026-01-01T08:05:00+08:00")).active_retry_indexes, [0]); + for (const clock of [undefined, null, "2026-01-01T00:01:00", "2026-01-01", "2026-02-30T00:01:00Z", + "2026-01-01T00:01:00z"]) { + assert.deepEqual((await read([run], clock)).active_retry_indexes, []); + } + for (const patch of [{source: "todo"}, {todo_id: "todo_other001"}, {resume_when: null}, + {observed_at: "2026-01-01T00:00:00"}, {due_at: "2026-02-30T00:05:00Z"}]) { + const bad = {...run, blocked_retry: {...run.blocked_retry, ...patch}}; + assert.deepEqual((await read([bad], "2026-01-01T00:01:00Z")).active_retry_indexes, []); + } + for (const patch of [{delivery_outcome: "outcome_progress"}, {progress_observation: null}, + {todo_id: null}, {turn_instance_id: null}, {turn_instance_id: {}}]) { + assert.deepEqual((await read([{...run, ...patch}], "2026-01-01T00:01:00Z")).active_retry_indexes, []); + } + // Historical Python datetime inputs retain ISO-week dates and offset seconds. + const legacy = {...run, blocked_retry: {...run.blocked_retry, + observed_at: "2026-W01-4T00:00:00+00:00:01", due_at: "2026-W01-4T00:05:00+00:00:01", + resume_when: "resume_at:2026-W01-4T00:05:00+00:00:01"}}; + assert.deepEqual((await read([legacy], "2026-01-01T00:01:00Z")).active_retry_indexes, [0]); + const legacySeparator = {...run, blocked_retry: {...run.blocked_retry, + observed_at: "2026-01-01z00:00:00Z", due_at: "2026-01-01z00:05:00Z", + resume_when: "resume_at:2026-01-01z00:05:00Z"}}; + assert.deepEqual((await read([legacySeparator], "2026-01-01T00:01:00Z")).active_retry_indexes, [0]); + for (const [observed_at, due_at, expected] of [ + ["2026-01-01T00:00:00Z", "2026-01-01T00:01:00Z", [0]], + ["2026-01-01T00:00:00Z", "2026-01-01T00:30:00Z", [0]], + ["2026-01-01T00:00:00.000001Z", "2026-01-01T00:01:00Z", []], + ["2026-01-01T00:00:00Z", "2026-01-01T00:30:00.000001Z", []], + ] as const) { + const bounded = {...run, blocked_retry: {...run.blocked_retry, observed_at, due_at, + resume_when: `resume_at:${due_at}`}}; + assert.deepEqual((await read([bounded], "2026-01-01T00:00:30Z")).active_retry_indexes, expected); + } + await assert.rejects(read([run], 123), /observation clock/); +}); From 598a4de1d8b7717481f9a19521eafe3c129e15cd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:24:45 +0800 Subject: [PATCH 6/9] fix(quota): preserve aware history clock range comparisons Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/runtime_timestamp.ts | 6 ++++-- .../quota_settlement_readback.test.ts | 11 ++++++++++- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/runtime_timestamp.ts b/loopx/control_plane/runtime_timestamp.ts index ea40bafdb3..867be7b0fa 100644 --- a/loopx/control_plane/runtime_timestamp.ts +++ b/loopx/control_plane/runtime_timestamp.ts @@ -97,8 +97,10 @@ export function parseTodoTimestampMicros(value: string, awarePythonDatetime = fa offset = parts[2][0] === "-" ? -parsed : parsed; } const result = BigInt(calendar.valueOf()) * 1000n + local - offset; - // Python datetime requires UTC results within 0001-01-01T00:00:00Z to 9999-12-31T23:59:59.999999Z. + // Todo normalization requires a representable UTC datetime. Historical + // aware-datetime comparisons also accept valid local dates whose offset + // places their instant outside that UTC range; no UTC datetime is built. // -62135596800000000n is 0001-01-01T00:00:00Z, 253402300800000000n is 10000-01-01T00:00:00Z. - if (result < -62135596800000000n || result >= 253402300800000000n) return null; + if (!awarePythonDatetime && (result < -62135596800000000n || result >= 253402300800000000n)) return null; return result; } diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index 5fab14eefc..ae400ee336 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -2098,7 +2098,8 @@ test("history retry qualification preserves its original aware clock and exact e assert.deepEqual((await read([bad], "2026-01-01T00:01:00Z")).active_retry_indexes, []); } for (const patch of [{delivery_outcome: "outcome_progress"}, {progress_observation: null}, - {todo_id: null}, {turn_instance_id: null}, {turn_instance_id: {}}]) { + {todo_id: null}, {turn_instance_id: null}, {turn_instance_id: {}}, + {turn_instance_id: true}, {turn_instance_id: 1}, {turn_instance_id: " "}]) { assert.deepEqual((await read([{...run, ...patch}], "2026-01-01T00:01:00Z")).active_retry_indexes, []); } // Historical Python datetime inputs retain ISO-week dates and offset seconds. @@ -2110,6 +2111,14 @@ test("history retry qualification preserves its original aware clock and exact e observed_at: "2026-01-01z00:00:00Z", due_at: "2026-01-01z00:05:00Z", resume_when: "resume_at:2026-01-01z00:05:00Z"}}; assert.deepEqual((await read([legacySeparator], "2026-01-01T00:01:00Z")).active_retry_indexes, [0]); + // Aware Python comparisons need not normalize the instant into a UTC datetime. + for (const start of ["0001-01-01T00:00:00+08:00", "9999-12-31T23:00:00-08:00"]) { + const due = start.replace(":00:00", ":05:00"), now = start.replace(":00:00", ":01:00"); + const edge = {...run, blocked_retry: {...run.blocked_retry, + observed_at: start, due_at: due, resume_when: `resume_at:${due}`}}; + assert.deepEqual((await read([edge], now)).active_retry_indexes, [0]); + assert.deepEqual((await read([edge], due)).active_retry_indexes, []); + } for (const [observed_at, due_at, expected] of [ ["2026-01-01T00:00:00Z", "2026-01-01T00:01:00Z", [0]], ["2026-01-01T00:00:00Z", "2026-01-01T00:30:00Z", [0]], From 87206c28756b9aa526b3db3c8dae5ebf3b19b69b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:28:26 +0800 Subject: [PATCH 7/9] fix(quota): preserve receipt waits across periodic review Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/loopx-overall-roadmap-v0.md | 5 ++ .../rfcs/loopx-overall-roadmap-v0.zh-CN.md | 5 +- docs/quota-allocation.md | 14 +++++ loopx/control_plane/quota/cli_projection.py | 38 ++++++++++++ .../work_items/replan_history.ts | 17 ++++++ .../work_items/replan_history_codec.py | 7 +++ .../test_quota_cli_projection.py | 61 +++++++++++++++++++ .../test_quota_settlement_cli.py | 36 +++++++++-- tests/control_plane_ts/replan_history.test.ts | 25 ++++++++ 9 files changed, 202 insertions(+), 6 deletions(-) diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index b424ed72c9..45e08f5ad5 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -142,6 +142,11 @@ Python retry qualification and batches administrative/retry history facts throug the existing TypeScript settlement owner for semantic retention and legacy status. It preserves independent outcomes, original clocks and historical microsecond expiry; malformed Turn identities no longer establish a positive retry proof. +The periodic-history caller also respects the latest same-Agent accepted retry +until its original deadline without resetting earned review counts. File/SQLite +real CLI tests cover the threshold, independent work, real user gates and natural +deadline recovery. Identical Explore guidance is referenced within the same CLI +packet, preserving all clauses and standalone reads. Full-repository qualification, comparable full caller cost, installed adoption and measured long-horizon benefit remain distinct and incomplete. diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md index f56238f190..21d2b82af6 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md @@ -111,7 +111,10 @@ managed 与 attached 的工作对话都应能持续在 LoopX 中进行:沿用 认领/转派。canonical fixture 整合及原 Turn 记账检查后的完整受影响 CLI 套件通过; 保留此前失败及 base/head 归因。后续重构删除 Python 重试资格判定,由已有 TypeScript 结算 owner 批量核对行政/重试历史事实,供语义保留和旧状态读路使用。各维度仍独立, -保留原时钟及历史微秒到期边界;畸形 Turn 身份不再取得重试活动的正向证明。整仓 +保留原时钟及历史微秒到期边界;畸形 Turn 身份不再取得重试活动的正向证明。 +周期历史调用者也尊重同 Agent 最新工作的已接受重试直至原期限,不清零已累积的复盘 +计数。File/SQLite 真实 CLI 覆盖阈值、独立工作、真实用户门禁与自然到期恢复。 +完全相同的 Explore 指引改为同一 CLI 包内引用,保留全部条款及独立读取。整仓 资格、可比完整调用成本、安装采用与长程实测收益仍是分别验证的未完成项。 **S5/S11 长程进展设计。** [研究计量细化](long-horizon-harness-benchmark-research-program-v0.zh-CN.md#基于证据的研究片段评估拟议细化) diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 196c0886a5..4829a3d87d 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -112,6 +112,20 @@ The cadence does not interrupt a host, schedule a Turn, spend quota or grant authority. The executor must settle work and re-enter quota before starting the next work Turn for the review obligation to take effect. +A reached periodic count alone does not preempt the latest same-Agent work's +accepted, unexpired receipt retry. The shared settlement owner qualifies that +retry and its original deadline. The counter is not acknowledged or reset: +review becomes due again at expiry. New work or evidence-driven triggers and +independent runnable Todos retain their normal paths; another Agent's wait +cannot suppress this Agent's review. + +The compact CLI packet may replace an identical Explore writeback-guidance +mirror with `result_writeback_guidance_ref`, pointing to the complete settlement +attachment in the same packet. Every clause remains available without another +guard call. Differing text, standalone Explore reads and `--include-detail vision` +retain the original guidance. This changes presentation, not work or settlement +obligations. + ```bash # Preview, apply, and read back an explicit Goal threshold (supported: 1–6). loopx configure-goal --goal-id example --execution-replan-after-turns 5 diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py index e37dd763ae..5a4dab688d 100644 --- a/loopx/control_plane/quota/cli_projection.py +++ b/loopx/control_plane/quota/cli_projection.py @@ -769,6 +769,42 @@ def _reference_inline_goal_acceptance(payload: dict[str, Any]) -> dict[str, Any] return result +def _reference_explore_writeback_guidance(payload: dict[str, Any]) -> dict[str, Any]: + """Keep identical optional instructions once in this same CLI packet. + + This is transport-only reference projection, after all decisions. Preserve + standalone Explore reads, differing text and the complete settlement owner. + """ + contract = payload.get("interaction_contract", {}) + agent = contract.get("agent_channel", {}) + context = agent.get("work_context", {}) + sources = context.get("sources", []) + steps = contract.get("cli_channel", {}).get("settlement_plan", {}).get("ordered_steps", []) + for step_index, step in enumerate(steps): + if step.get("kind") != "durable_writeback": + continue + for attachment_index, attachment in enumerate(step.get("optional_attachments", [])): + guidance = attachment.get("guidance") + if attachment.get("capability_id") != "explore" or not isinstance(guidance, str) or not guidance: + continue + ref = (f"$.interaction_contract.cli_channel.settlement_plan.ordered_steps[{step_index}]" + f".optional_attachments[{attachment_index}].guidance") + updated = list(sources) + for index, source in enumerate(sources): + content = source.get("content", {}) + graph = content.get("graph", {}) + if source.get("kind") != "explore_turn_context" or graph.get("result_writeback_guidance") != guidance: + continue + graph = dict(graph) + graph.pop("result_writeback_guidance") + graph["result_writeback_guidance_ref"] = ref + updated[index] = {**source, "content": {**content, "graph": graph}} + if updated != sources: + return {**payload, "interaction_contract": {**contract, "agent_channel": { + **agent, "work_context": {**context, "sources": updated}}}} + return payload + + def compact_quota_should_run_cli_payload( payload: dict[str, Any], *, @@ -864,6 +900,8 @@ def compact_quota_should_run_cli_payload( agent_lane_next_action.pop("content_revision", None) compact["agent_lane_next_action"] = agent_lane_next_action compact = _reference_inline_goal_acceptance(compact) + if not include_vision_detail: + compact = _reference_explore_writeback_guidance(compact) return _promote_interaction_contract( _promote_runtime_capability_reentry(compact) ) diff --git a/loopx/control_plane/work_items/replan_history.ts b/loopx/control_plane/work_items/replan_history.ts index 317fb149f4..905d1ebcce 100644 --- a/loopx/control_plane/work_items/replan_history.ts +++ b/loopx/control_plane/work_items/replan_history.ts @@ -8,6 +8,7 @@ import { import { projectTodoResumePlanning } from "../todos/resume_planning.ts"; import { monitorIsActive } from "../todos/summary_lanes.ts"; import { authorityUnicodeCompare } from "../coordination/authority_store_codec.ts"; +import { qualifySettlementHistory } from "../quota/settlement_phase.ts"; export const PROGRESS_RESULT_CLASSES = [ "advanced", "unchanged", "blocked", "exploration_exhausted", "no_followup", @@ -27,6 +28,7 @@ interface Monitor { key: string | null; } interface Run { + retryFacts: JsonObject; agent: string | null; publicAgent: string | null; monitorAgent: string | null; @@ -53,6 +55,7 @@ interface Todos { resume: JsonObject | null; } interface Request { + retryObservedAt: string | null; operation: "all" | "progress" | "periodic" | "monitor_streak" | "frontier_monitor_streak"; observedAt: number | null; agent: string | null; @@ -128,6 +131,7 @@ function decode(value: unknown): Request { } return { operation, observedAt, + retryObservedAt: optionalNonEmptyString(raw.retry_observed_at, "retry_observed_at"), agent: optionalNonEmptyString(raw.agent_id, "agent_id"), monitorAgent: optionalNonEmptyString(raw.monitor_agent_id, "monitor_agent_id"), neutral: new Set(requireStringArray(raw.neutral_classifications, "neutral_classifications")), @@ -139,6 +143,9 @@ function decode(value: unknown): Request { runs: rows(raw.runs, "runs", row => { const monitor = requireJsonObject(row.monitor, "monitor"); return { + retryFacts: {todo_id: row.todo_id, blocked_retry: row.blocked_retry, + delivery_outcome: row.delivery_outcome, progress_observation: row.progress, + turn_instance_id: row.turn_id}, agent: optionalNonEmptyString(row.agent_id, "run.agent_id"), publicAgent: optionalNonEmptyString(row.public_agent_id, "public_agent_id"), monitorAgent: optionalNonEmptyString(row.monitor_agent_id, "run.monitor_agent_id"), @@ -247,6 +254,16 @@ function periodicTrigger(runs: readonly Run[], request: Request, qualified?: Rea if (durable.length >= request.periodic) break; } if (durable.length < request.periodic) return null; + // A periodic counter supplies no new evidence about the latest accepted + // blocked work. Respect that lane's original bounded retry until it expires; + // do not ACK/reset the counter. Fresh work, typed stalls and external feedback + // retain their own triggers, and independent Todos remain selectable. + const latestWork = runs.find(run => run.classification && + (run.classification !== "quota_monitor_poll" || run.monitor.material === true)); + const lane = request.agent ?? sole(runs.map(run => run.agent)); + if (lane && latestWork?.agent === lane && qualifySettlementHistory( + [latestWork.retryFacts], request.retryObservedAt, + ).active_retry_indexes.length) return null; return { kind: "periodic_review_due", section: "run_history", text: `latest ${durable.length} ${qualified ? "settled work Turns" : "durable public run records"} since last autonomous replan reached periodic review threshold ${request.periodic}`, diff --git a/loopx/control_plane/work_items/replan_history_codec.py b/loopx/control_plane/work_items/replan_history_codec.py index b26b789444..8985a4002d 100644 --- a/loopx/control_plane/work_items/replan_history_codec.py +++ b/loopx/control_plane/work_items/replan_history_codec.py @@ -82,6 +82,9 @@ def _run_fact(run: Mapping[str, Any], ack_recorded: Callable[..., bool]) -> dict "turn_id": _progress_turn_instance_id(run), "accepted_ack": ack_recorded(run), "progress": progress_observation_from_run(run), + **({"todo_id": run.get("todo_id"), "blocked_retry": run["blocked_retry"], + "delivery_outcome": run.get("delivery_outcome")} + if isinstance(run.get("blocked_retry"), dict) else {}), "monitor": { "target_id": str(target.get("target_id") or "").strip() or None, "mode": str(target.get("monitor_mode") or "").strip() or None, @@ -186,6 +189,10 @@ def project_replan_history( # Carry one observation for the complete current source. Expiry is # interpreted by the shared TypeScript Monitor owner, not this codec. params["observed_at"] = now_utc().timestamp() + if operation in {"all", "periodic"}: + # Preserve the native retry's microsecond deadline; TS reuses the + # settlement owner instead of a second expiry policy or bridge call. + params["retry_observed_at"] = now_utc().isoformat() if effective_turn_cadence is not None: params["settlement_source"] = effective_turn_cadence["settlement_source"] params["periodic_threshold"] = effective_turn_cadence["threshold"] diff --git a/tests/control_plane/test_quota_cli_projection.py b/tests/control_plane/test_quota_cli_projection.py index 7ae0e43846..d496ab15a9 100644 --- a/tests/control_plane/test_quota_cli_projection.py +++ b/tests/control_plane/test_quota_cli_projection.py @@ -89,6 +89,67 @@ def test_progress_only_replan_does_not_invent_a_vision_input() -> None: assert compact_quota_should_run_cli_payload(payload) == payload +@pytest.mark.parametrize("variant", ["identical", "different", "missing-owner", "wrong-capability", "wrong-step"]) +def test_explore_guidance_reference_preserves_every_clause_in_same_packet(variant: str) -> None: + guidance = "Keep the lease through evidence delivery. Preserve negative results. " * 80 + attachment = {"capability_id": "other" if variant == "wrong-capability" else "explore", + "guidance": guidance, "required": False} + graph = {"result_writeback_guidance": guidance + (" Preserve a new obligation." if variant == "different" else ""), + "writeback_results": [{"status": "refuted", "observation": "Measured negative result."}]} + source = {"kind": "explore_turn_context", "content": {"graph": graph}} + steps = [{"kind": "validation"}, {"kind": "validation" if variant == "wrong-step" else "durable_writeback", + "optional_attachments": [] if variant == "missing-owner" else [attachment]}] + payload = {"interaction_contract": {"agent_channel": {"must_attempt": True, "work_context": {"sources": [source]}}, + "cli_channel": {"settlement_plan": {"ordered_steps": steps}}}} + before = deepcopy(payload) + compact = compact_quota_should_run_cli_payload(payload) + projected = compact["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["graph"] + if variant == "identical": + assert projected["result_writeback_guidance_ref"] == ( + "$.interaction_contract.cli_channel.settlement_plan.ordered_steps[1].optional_attachments[0].guidance") + assert "result_writeback_guidance" not in projected + restored = deepcopy(compact) + replacement = restored["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["graph"] + replacement.pop("result_writeback_guidance_ref") + replacement["result_writeback_guidance"] = restored["interaction_contract"]["cli_channel"][ + "settlement_plan"]["ordered_steps"][1]["optional_attachments"][0]["guidance"] + assert restored == before + assert len(json.dumps(compact)) < len(json.dumps(before)) - 4000 + assert compact_quota_should_run_cli_payload(compact) == compact + else: + assert projected == graph + assert compact_quota_should_run_cli_payload(payload, include_vision_detail=True) == before + assert payload == before + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_real_cli_explore_reference_resolves_to_standalone_guidance(tmp_path, monkeypatch, provider): + from canonical_authority_fixture import isolate_sqlite_runtime + from tests.control_plane import test_quota_settlement_cli as fixture + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = fixture._write_fixture(tmp_path) + config = json.loads(registry.read_text()) + config["goals"][0]["explore_graph"] = {"enabled": True} + registry.write_text(json.dumps(config)) + fixture._initialize_fixture_authority(project, runtime, provider=provider) + binding = ("--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID) + rc, guard = fixture._run_cli(registry, runtime, "quota", "should-run", "--codex-app", + *binding, "--todo-id", fixture.TODO_ID, "--turn-instance-id", "guidance-read", cwd=project) + assert rc == 0, guard + sources = guard["interaction_contract"]["agent_channel"]["work_context"]["sources"] + graph = next(source["content"]["graph"] for source in sources if source.get("kind") == "explore_turn_context") + assert "result_writeback_guidance_ref" in graph + steps = guard["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"] + owner = next(attachment["guidance"] for step in steps if step["kind"] == "durable_writeback" + for attachment in step["optional_attachments"] if attachment["capability_id"] == "explore") + rc, standalone = fixture._run_cli(registry, runtime, "explore", "turn-context", *binding, cwd=project) + assert rc == 0, standalone + assert owner == standalone["graph"]["result_writeback_guidance"] + assert fixture._spend_run_count(runtime) == 0 + + def _items(count: int, *, prefix: str) -> list[dict[str, object]]: return [ { diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 43f22decf1..37640967d8 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1902,12 +1902,18 @@ def test_typed_blocked_retry_with_peer_hard_lease( assert next_turn["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] <= 5 -def _receipt_retry_with_vision_fixture(tmp_path, monkeypatch, provider, *, peer_agent=None): +def _receipt_retry_with_vision_fixture( + tmp_path, monkeypatch, provider, *, peer_agent=None, prior_work_turns=0, +): """Create the wait through its actual admission and blocked-writeback owners.""" from canonical_authority_fixture import isolate_sqlite_runtime if provider == "sqlite": isolate_sqlite_runtime(tmp_path, monkeypatch) project, runtime, registry = _write_fixture(tmp_path) + if prior_work_turns: + configuration = json.loads(registry.read_text()) + configuration["goals"][0]["quota"]["allowed_slots"] = 30 + registry.write_text(json.dumps(configuration)) if peer_agent is not None: configuration = json.loads(registry.read_text()) configuration["goals"][0]["coordination"]["registered_agents"].append(peer_agent) @@ -1915,6 +1921,24 @@ def _receipt_retry_with_vision_fixture(tmp_path, monkeypatch, provider, *, peer_ registry.write_text(json.dumps(configuration)) _configure_completion_validation_todo(project) _initialize_fixture_authority(project, runtime, provider=provider) + for index in range(prior_work_turns): + prior_binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--todo-id", TODO_ID, "--turn-instance-id", f"verified-work-{index}") + rc, admitted = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", + *prior_binding, cwd=project) + assert rc == 0, admitted + assert admitted["execution_obligation"]["must_attempt_work"] is True + artifact = project / "verified-arithmetic.txt" + artifact.write_text(str(17 * (23 + index))) + assert int(artifact.read_text()) == 17 * (23 + index) + rc, written = _run_cli(registry, runtime, "refresh-state", *prior_binding, + "--classification", "arithmetic_verified", "--delivery-batch-scale", "single_surface", + "--delivery-outcome", "outcome_progress", "--delivery-boundary", "in_flight_continuation", + "--no-global-sync", "--suppress-external-sinks", cwd=project) + assert rc == 0, written + rc, spent = _run_cli(registry, runtime, "quota", "spend-slot", *prior_binding, + "--slots", "1", "--source", "heartbeat", "--execute", cwd=project) + assert rc == 0 and spent["appended"], spent rc, before = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) assert rc == 0, before binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, @@ -1949,13 +1973,15 @@ def _receipt_retry_with_vision_fixture(tmp_path, monkeypatch, provider, *, peer_ @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("human_item", [None, "user_action", "user_gate"]) +@pytest.mark.parametrize("prior_work_turns", [0, 5]) def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, human_item: str | None, + prior_work_turns: int, ) -> None: from loopx.control_plane.quota.turn_envelope import build_turn_envelope project, runtime, registry, before, closed = _receipt_retry_with_vision_fixture( - tmp_path, monkeypatch, provider, + tmp_path, monkeypatch, provider, prior_work_turns=prior_work_turns, ) if human_item is not None: scope = "--bound-agent" if human_item == "user_action" else "--blocks-agent" @@ -1974,7 +2000,7 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( assert waiting["interaction_contract"]["user_channel"]["action_required"] is True assert not waiting["interaction_contract"]["user_channel"].get("non_gating_notice") assert waiting["scheduler_hint"]["action"] == "backoff_waiting_for_user" - assert _spend_run_count(runtime) == 0 + assert _spend_run_count(runtime) == prior_work_turns return assert not waiting.get("autonomous_replan_obligation") assert waiting["interaction_contract"]["agent_channel"]["delivery_allowed"] is False @@ -1993,7 +2019,7 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( rc, after = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID) assert rc == 0, after assert next(todo for todo in after["todos"] if todo["todo_id"] == TODO_ID) == before["todos"][0] - assert _spend_run_count(runtime) == 0 + assert _spend_run_count(runtime) == prior_work_turns rc, independent = _run_cli( registry, runtime, "todo", "add", "--goal-id", GOAL_ID, "--role", "agent", "--task-class", "advancement_task", @@ -2008,7 +2034,7 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( assert rc == 0, resumed assert resumed["selected_todo"]["todo_id"] == independent["todo_id"] assert resumed["execution_obligation"]["must_attempt_work"] is True - assert _spend_run_count(runtime) == 0 + assert _spend_run_count(runtime) == prior_work_turns @pytest.mark.parametrize("provider", ["file", "sqlite"]) diff --git a/tests/control_plane_ts/replan_history.test.ts b/tests/control_plane_ts/replan_history.test.ts index 9a8eed1e0c..26aa7e0e85 100644 --- a/tests/control_plane_ts/replan_history.test.ts +++ b/tests/control_plane_ts/replan_history.test.ts @@ -60,6 +60,31 @@ test("periodic work counts distinct logical turns and keeps the existing public assert.equal(trigger(many(19)), null); }); +test("periodic count respects the latest lane retry without erasing the earned review", () => { + const due = "2026-01-01T00:05:00.123456Z"; + const retry = run(21, {todo_id: "todo_feedback", delivery_outcome: "outcome_gap", + progress: {...observation, result_class: "blocked"}, blocked_retry: { + schema_version: "quota_blocked_retry_v0", source: "turn_settlement", + todo_id: "todo_feedback", observed_at: "2026-01-01T00:00:00.123456Z", + due_at: due, resume_when: `resume_at:${due}`, + }}); + const rows = [retry, ...many(20)]; + const before = {retry_observed_at: "2026-01-01T00:05:00.123455Z"}; + const original = structuredClone(rows); + assert.equal(trigger(rows, before), null); + assert.equal(trigger(rows, {...before, operation: "periodic"}), null); + assert.equal(trigger(rows, {retry_observed_at: due})?.kind, "periodic_review_due"); + assert.equal(trigger(rows)?.kind, "periodic_review_due", "legacy requests have no current clock proof"); + assert.equal(trigger([{...retry, agent_id: "peer"}, ...many(20)], before)?.kind, "periodic_review_due"); + assert.equal(trigger([run(22), ...rows], before)?.kind, "periodic_review_due", "new work supersedes the wait"); + assert.equal(trigger([{...retry, delivery_outcome: "outcome_progress"}, ...many(20)], before)?.kind, + "periodic_review_due", "a bare retry field is not accepted blocked work"); + assert.equal(trigger([retry, {...retry, turn_id: "previous"}, ...many(20)], before)?.kind, + "typed_progress_repeat", "a concrete repeated stall is independent evidence"); + assert.equal(trigger([poll(22), ...rows], before), null, "quiet polls do not manufacture new evidence"); + assert.deepEqual(rows, original); +}); + test("periodic review counts material work, not quiet monitor receipts", () => { const quiet = many(20, n => poll(n, { mode: n > 18 ? "due_monitor_observed_without_material_transition" : "monitor_quiet_until_material_transition" })); From 8262862cdf99f612095f8259ac8e2e34175e3573 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 22:11:50 +0800 Subject: [PATCH 8/9] test(quota): exercise an earned review during receipt wait Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_quota_settlement_cli.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 37640967d8..52fcd858c6 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1983,6 +1983,16 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli( project, runtime, registry, before, closed = _receipt_retry_with_vision_fixture( tmp_path, monkeypatch, provider, prior_work_turns=prior_work_turns, ) + if prior_work_turns: + # The default cadence is six. Reach a real review threshold only after + # the blocked Turn was admitted, using the configuration owner rather + # than fabricated history or settlement receipts. + rc, configured = _run_cli( + registry, runtime, "configure-goal", "--goal-id", GOAL_ID, + "--execution-replan-after-turns", str(prior_work_turns), "--execute", cwd=project, + ) + assert rc == 0, configured + assert configured["after"]["execution_profile"]["replan_after_effective_turns"] == prior_work_turns if human_item is not None: scope = "--bound-agent" if human_item == "user_action" else "--blocks-agent" rc, human = _run_cli( From bdf7131d42ae22cff8b402f1c8d0437baa1e0be3 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 24 Oct 2026 00:39:47 +0800 Subject: [PATCH 9/9] fix: preserve receipt clock precision in Todo wait projections Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/quota-allocation.md | 9 +++++ loopx/control_plane/quota/blocked_retry.py | 20 ++++++++--- loopx/control_plane/todos/resume_condition.py | 15 ++++---- loopx/control_plane/todos/resume_condition.ts | 23 +++++++----- .../test_status_receipt_retry_projection.py | 23 +++++++++++- .../test_todo_resume_condition_adapter.py | 11 ++++++ .../todo_resume_condition.test.ts | 35 +++++++++++++++++++ 7 files changed, 117 insertions(+), 19 deletions(-) diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 4829a3d87d..2fd96419ec 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -527,6 +527,15 @@ Turn without spending quota. The retry suppresses only the blocked Todo for the same Agent until due or superseded by newer work, so independent eligible Todos can still be selected. Historical blocked writebacks without a bounded retry retain their prior spend readback; no debit is retroactively erased. +Receipt clocks can contain microseconds. Todo `resume_at` authoring and readback +accept up to six fractional digits and compare the exact instant; existing +second/millisecond normalization and receipt identities are unchanged. A valid +historical receipt with a Python datetime spelling outside Todo authoring syntax +retains its original selection fence, without fabricating an automatic clock +proof or rewriting the receipt. +回执时钟可能包含微秒。Todo `resume_at` 的写入与读取接受最多六位小数并精确比较, +原有秒/毫秒的规范化和回执身份保持不变。合法历史回执若使用 Todo 写入语法之外的 +Python datetime 格式,仍保留原选择限制,不伪造自动时钟证明,也不改写历史回执。 Automatic no-spend stall writeback uses the same receipt-owned retry projection as live selection, including its post-writeback readback. A future unchanged monitor therefore stays quiet during that wait without failing the heartbeat diff --git a/loopx/control_plane/quota/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py index c252f562ad..092047e423 100644 --- a/loopx/control_plane/quota/blocked_retry.py +++ b/loopx/control_plane/quota/blocked_retry.py @@ -8,6 +8,7 @@ from ..todos.resume_condition import ( compact_todo_resume_items, evaluate_todo_resume_conditions, + normalize_supported_todo_resume_when, ) from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from ..work_items.autonomous_replan_ack import qualify_settlement_history @@ -133,11 +134,16 @@ def overlay_active_turn_retries( retry_items = [ {"todo_id": todo_id, "status": "open", "role": "agent", "task_class": "advancement_task", "claimed_by": agent_id, - "resume_when": retry["resume_when"]} + "resume_when": normalized} for todo_id, retry in waits.items() + if (normalized := normalize_supported_todo_resume_when(retry["resume_when"])) ] - resume_conditions = evaluate_todo_resume_conditions( - retry_items, source_items=retry_items, evaluated_at=observed_at, + normalized_clock = normalize_supported_todo_resume_when(f"resume_at:{observed_at}") + resume_conditions = ( + evaluate_todo_resume_conditions( + retry_items, source_items=retry_items, + evaluated_at=normalized_clock.removeprefix("resume_at:"), + ) if retry_items and normalized_clock else {} ) projected_items = [] changed = False @@ -177,7 +183,13 @@ def overlay_summary(summary: Any) -> tuple[Any, bool]: "blocked_retry": retry, "next_due_at": retry["due_at"], "resume_condition": { - **resume_conditions[todo["todo_id"]], + # Historical Python datetime spellings may not + # be valid authoring syntax. Keep their original + # selection fence without inventing clock proof. + **resume_conditions.get(normalize_todo_id(todo["todo_id"]), { + "kind": "resume_at", "resume_when": resume, + "satisfied": False, + }), "blocked_retry": retry, }, } diff --git a/loopx/control_plane/todos/resume_condition.py b/loopx/control_plane/todos/resume_condition.py index d272d5904a..61e4145b6d 100644 --- a/loopx/control_plane/todos/resume_condition.py +++ b/loopx/control_plane/todos/resume_condition.py @@ -43,7 +43,7 @@ ) _RESUME_AT_PATTERN = re.compile( r"^resume_at:(?P[1-9]\d{3}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}" - r"(?:\.\d{1,3})?(?:Z|[+-]\d{2}:\d{2}))$", + r"(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2}))$", re.IGNORECASE, ) _PR_REF_NUMBER_PATTERN = re.compile( @@ -123,11 +123,14 @@ def normalize_todo_resume_when(value: Any) -> str | None: return None if not 1000 <= utc.year <= 9999: return None - canonical = ( - utc.isoformat(timespec="milliseconds") - if utc.microsecond - else utc.replace(microsecond=0).isoformat() - ).replace("+00:00", "Z") + if utc.microsecond % 1000: + canonical = utc.strftime("%Y-%m-%dT%H:%M:%S") + "." + f"{utc.microsecond:06d}".rstrip("0") + "Z" + else: + canonical = ( + utc.isoformat(timespec="milliseconds") + if utc.microsecond + else utc.replace(microsecond=0).isoformat() + ).replace("+00:00", "Z") return f"{TODO_RESUME_KIND_RESUME_AT}:{canonical}" candidate = candidate.lower() if candidate.startswith(f"{TODO_RESUME_KIND_RESUME_AT}:"): diff --git a/loopx/control_plane/todos/resume_condition.ts b/loopx/control_plane/todos/resume_condition.ts index d9d936c577..df0db77312 100644 --- a/loopx/control_plane/todos/resume_condition.ts +++ b/loopx/control_plane/todos/resume_condition.ts @@ -54,7 +54,7 @@ const RESUME_PATTERN = /^[a-z][a-z\d_-]{0,31}(?::[a-z\d_.:@#/-]{1,181})?$/; const PR_RESUME_PATTERN = /^pr_merged:(?:[a-z\d_.-]{1,80}\/[a-z\d_.-]{1,100})?#[1-9]\d{0,8}$/; const RESUME_AT_PATTERN = - /^resume_at:(?[1-9]\d{3}-(?\d{2})-(?\d{2})[tT](?\d{2}):(?\d{2}):(?\d{2})(?:\.(?\d{1,3}))?(?[zZ]|(?[+-])(?\d{2}):(?\d{2})))$/; + /^resume_at:(?[1-9]\d{3}-(?\d{2})-(?\d{2})[tT](?\d{2}):(?\d{2}):(?\d{2})(?:\.(?\d{1,6}))?(?[zZ]|(?[+-])(?\d{2}):(?\d{2})))$/; const GITHUB_PULL_URL_PATTERN = /^https:\/\/github\.com\/([^/]+\/[^/]+)\/pull\/(\d+)(?:\b|\/|#|\?)/i; const PR_REF_PATTERN = @@ -74,11 +74,13 @@ interface ResumeSpec { interface ResumeTimestamp { iso: string; - milliseconds: number; + microseconds: bigint; } -function normalizedTimestamp(milliseconds: number): string { - return new Date(milliseconds).toISOString().replace(".000Z", "Z"); +function normalizedTimestamp(milliseconds: number, remainderMicros: number): string { + const iso = new Date(milliseconds).toISOString(); + if (remainderMicros === 0) return iso.replace(".000Z", "Z"); + return iso.replace(/Z$/, `${String(remainderMicros).padStart(3, "0").replace(/0+$/, "")}Z`); } /** Strict RFC3339 date-time parsing for user-authored wakeups. A timezone is @@ -92,7 +94,9 @@ function resumeTimestamp(value: unknown): ResumeTimestamp | null { .slice(0, 19) .split(/[-tT:]/) .map((part) => Number.parseInt(part, 10)); - const fraction = Number.parseInt((match.groups.fraction ?? "").padEnd(3, "0") || "0", 10); + const fractionText = (match.groups.fraction ?? "").padEnd(6, "0"); + const fraction = Number.parseInt(fractionText.slice(0, 3), 10); + const remainderMicros = Number.parseInt(fractionText.slice(3), 10); const local = new Date(0); local.setUTCFullYear(year, month - 1, day); local.setUTCHours(hour, minute, second, fraction); @@ -106,7 +110,10 @@ function resumeTimestamp(value: unknown): ResumeTimestamp | null { if (Number.isNaN(milliseconds)) return null; const utcYear = new Date(milliseconds).getUTCFullYear(); if (utcYear < 1000 || utcYear > 9999) return null; - return { iso: normalizedTimestamp(milliseconds), milliseconds }; + // Date supplies the checked calendar/offset, but must not round a receipt's + // microsecond deadline down and release work before its original clock. + return { iso: normalizedTimestamp(milliseconds, remainderMicros), + microseconds: BigInt(milliseconds) * 1000n + BigInt(remainderMicros) }; } function parsedResumeAt(value: string): ResumeTimestamp | null { @@ -514,7 +521,7 @@ function conditionFor( if (!scheduled || !evaluatedAt) { return { ...condition, invalid_state: "runtime_clock_unavailable" }; } - const satisfied = evaluatedAt.milliseconds >= scheduled.milliseconds; + const satisfied = evaluatedAt.microseconds >= scheduled.microseconds; const receiptId = `resume_at_${createHash("sha256") .update(`${item.todo_id}\n${spec.normalized}`, "utf8") .digest("hex") @@ -650,7 +657,7 @@ export function resumeConditionHasKnownPendingTarget(condition: JsonObject, wait && condition.material_change === false && condition.material_change_generation === 0 && condition.resume_receipt === null - && evaluated.milliseconds < scheduled.milliseconds; + && evaluated.microseconds < scheduled.microseconds; } default: return false; } diff --git a/tests/control_plane/test_status_receipt_retry_projection.py b/tests/control_plane/test_status_receipt_retry_projection.py index 576669a6b0..1c686067bf 100644 --- a/tests/control_plane/test_status_receipt_retry_projection.py +++ b/tests/control_plane/test_status_receipt_retry_projection.py @@ -89,11 +89,32 @@ def test_status_uses_retained_semantic_runs_before_display_compaction(): assert projected["items"][0]["agent_todos"]["items"][0]["blocked_retry"]["turn_instance_id"] == "original-attempt" +def test_status_preserves_microsecond_boundary_and_historical_clock_fence(): + observed, due = "2030-01-01T00:00:00.123456+00:00", "2030-01-01T00:05:00.123456+00:00" + _, run, history, queue = fixture(observed, due) + projected = overlay_status_turn_retries(history, queue, observed_at="2030-01-01T00:05:00.123455Z") + row = projected["items"][0]["agent_todos"]["items"][0] + assert row["blocked_retry"]["due_at"] == due + assert row["resume_condition"]["scheduled_for"] == "2030-01-01T00:05:00.123456Z" + assert row["resume_condition"]["satisfied"] is False + assert overlay_status_turn_retries(history, queue, observed_at=due) is queue + # Python's historical receipt reader accepts a basic date-time spelling; + # reading it does not expand the stricter Todo authoring grammar. + run["blocked_retry"].update(observed_at="20300101T000000+0000", due_at="20300101T000500+0000", + resume_when="resume_at:20300101T000500+0000") + historical = overlay_status_turn_retries(history, queue, observed_at="2030-01-01T00:02:00Z") + row = historical["items"][0]["agent_todos"]["items"][0] + assert row["resume_condition"]["satisfied"] is False + assert row["blocked_retry"]["due_at"] == "20300101T000500+0000" + assert "clock_provider" not in row["resume_condition"] + assert overlay_status_turn_retries(history, queue, observed_at="2030-01-01T00:05:00Z") is queue + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("channel", ["cli", "http"]) def test_real_provider_status_projects_wait_without_writes(tmp_path: Path, monkeypatch, provider, channel): isolate_sqlite_runtime(tmp_path, monkeypatch) - now = datetime.now(UTC) + now = datetime.now(UTC).replace(microsecond=123456) observed, due = now.isoformat(), (now + timedelta(minutes=5)).isoformat() todo, run, _, _ = fixture(observed, due) state = tmp_path / "ACTIVE_GOAL_STATE.md" diff --git a/tests/control_plane/test_todo_resume_condition_adapter.py b/tests/control_plane/test_todo_resume_condition_adapter.py index df39c7cc54..af69e70579 100644 --- a/tests/control_plane/test_todo_resume_condition_adapter.py +++ b/tests/control_plane/test_todo_resume_condition_adapter.py @@ -2,6 +2,8 @@ from typing import Any +import pytest + from loopx.control_plane.effect_program import interpret_quota_should_run_packet from loopx.control_plane.quota.should_run import build_quota_should_run from loopx.control_plane.testing.quota_fixtures import quota_status_payload @@ -65,6 +67,15 @@ def fake_runtime(method: str, request: dict[str, Any]) -> dict[str, Any]: ] +@pytest.mark.parametrize("fraction", ["123", "123000", "123450", "000001", "123456"]) +def test_resume_at_python_readback_matches_typed_authoring_precision(fraction: str) -> None: + value = f"resume_at:2030-01-01T08:00:00.{fraction}+08:00" + expected = fraction[:3] if fraction[3:] == "000" else fraction.rstrip("0") + canonical = f"resume_at:2030-01-01T00:00:00.{expected}Z" + assert resume_condition.normalize_todo_resume_when(value) == canonical + assert resume_condition.require_supported_todo_resume_when(value) == canonical + + def test_timezone_aware_resume_at_is_stable_across_ticks_and_summary_restarts() -> None: authored = "resume_at:2026-09-14T09:30:00+08:00" normalized = resume_condition.require_supported_todo_resume_when(authored) diff --git a/tests/control_plane_ts/todo_resume_condition.test.ts b/tests/control_plane_ts/todo_resume_condition.test.ts index e9c2f1a170..fe48647af7 100644 --- a/tests/control_plane_ts/todo_resume_condition.test.ts +++ b/tests/control_plane_ts/todo_resume_condition.test.ts @@ -11,6 +11,7 @@ import { diagnoseTodoResumeCondition, normalizeTodoResumeWhen, planTodoExternalWaitTransition, + resumeConditionHasKnownPendingTarget, } from "../../loopx/control_plane/todos/resume_condition.ts"; function todo( @@ -92,6 +93,40 @@ test("resume_at emits one stable due receipt across repeated ticks and restart", }), /evaluated_at must be a timezone-aware RFC3339 timestamp/); }); +test("receipt clock precision preserves the last microsecond before due", () => { + for (const date of ["1969-12-31", "2030-01-01"]) { + const resume_when = `resume_at:${date}T08:00:00.123456+08:00`; + const item = todo("todo_scheduled", "open", "advancement_task", {resume_when}); + const evaluate = (fraction: string) => { + const result = evaluateTodoResumeConditions({ + schema_version: TODO_RESUME_EVALUATION_REQUEST_SCHEMA_VERSION, + items: [item], source_items: [], evaluated_at: `${date}T00:00:00.${fraction}Z`, + }); + return (result.conditions as Array<{condition: Record}>)[0].condition; + }; + assert.equal(normalizeTodoResumeWhen({ + schema_version: TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION, resume_when, + }), `resume_at:${date}T00:00:00.123456Z`); + const before = evaluate("123455"); + assert.equal(before.satisfied, false); + assert.equal(before.scheduled_for, `${date}T00:00:00.123456Z`); + assert.equal(before.evaluated_at, `${date}T00:00:00.123455Z`); + assert.equal(before.resume_receipt, null); + assert.equal(resumeConditionHasKnownPendingTarget(before, item), true); + const due = evaluate("123456"); + assert.equal(due.satisfied, true); + assert.equal(resumeConditionHasKnownPendingTarget(due, item), false); + assert.deepEqual(due.resume_receipt, evaluate("123457").resume_receipt); + } + for (const fraction of ["123", "123000", "123450", "000001"]) { + const canonical = fraction === "123000" ? "123" : fraction.replace(/0+$/, ""); + assert.equal(normalizeTodoResumeWhen({ + schema_version: TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION, + resume_when: `resume_at:2030-01-01T00:00:00.${fraction}Z`, + }), `resume_at:2030-01-01T00:00:00.${canonical}Z`); + } +}); + test("live monitor completion is invalid, but historical completion remains satisfied", () => { for (const status of ["open", "blocked", "deferred", "done"]) { const result = evaluateTodoResumeConditions({