diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index 2ecc4e0676..45e08f5ad5 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -129,6 +129,27 @@ 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. 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. 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. +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. + **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 d250c7f3df..21d2b82af6 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,19 @@ 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 回读和 +长期／压缩历史回归保留原里程碑或真实缺口，避免行政收尾制造新的成果义务。源码 +覆盖已包括通知／真门禁叠加、原截止恢复，以及各自持有 Vision 和时钟时的原生取消 +认领／转派。canonical fixture 整合及原 Turn 记账检查后的完整受影响 CLI 套件通过； +保留此前失败及 base/head 归因。后续重构删除 Python 重试资格判定，由已有 TypeScript +结算 owner 批量核对行政／重试历史事实，供语义保留和旧状态读路使用。各维度仍独立， +保留原时钟及历史微秒到期边界；畸形 Turn 身份不再取得重试活动的正向证明。 +周期历史调用者也尊重同 Agent 最新工作的已接受重试直至原期限，不清零已累积的复盘 +计数。File/SQLite 真实 CLI 覆盖阈值、独立工作、真实用户门禁与自然到期恢复。 +完全相同的 Explore 指引改为同一 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 2773bd43a1..2fd96419ec 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 @@ -241,6 +255,40 @@ 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 保持原复核要求，这一读取 +分类不授予执行、完成、验收或结算权限。 + +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 @@ -479,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 @@ -494,11 +551,32 @@ 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 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 +589,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/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/blocked_retry.py b/loopx/control_plane/quota/blocked_retry.py index e1d517f015..092047e423 100644 --- a/loopx/control_plane/quota/blocked_retry.py +++ b/loopx/control_plane/quota/blocked_retry.py @@ -2,17 +2,16 @@ from __future__ import annotations -from datetime import datetime 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, + normalize_supported_todo_resume_when, +) 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( @@ -63,53 +62,13 @@ 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], *, 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. @@ -146,11 +105,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 @@ -158,8 +117,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"], @@ -170,6 +129,22 @@ 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": normalized} + for todo_id, retry in waits.items() + if (normalized := normalize_supported_todo_resume_when(retry["resume_when"])) + ] + 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 @@ -194,7 +169,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( @@ -207,9 +183,13 @@ 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, + # 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, }, } @@ -262,6 +242,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} @@ -283,6 +265,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/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/quota/settlement_phase.ts b/loopx/control_plane/quota/settlement_phase.ts index a1f09496f5..f2c867aaa6 100644 --- a/loopx/control_plane/quota/settlement_phase.ts +++ b/loopx/control_plane/quota/settlement_phase.ts @@ -1,8 +1,14 @@ -import type { SettlementIdentity } from "../effect_program.ts"; +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. @@ -54,6 +60,65 @@ 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; +} + +/** 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 9cea97190f..7cd43d8e5f 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, + qualifySettlementHistory, isCommittedMonitorPollEffect, isAcceptedProgressWriteback, isAcceptedReplanWriteback, @@ -1382,6 +1383,13 @@ export function readQuotaSettlementForAdmittedOwnerFromSnapshot( } export async function readQuotaSettlement(value: unknown): Promise { + 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/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index c8614c7cf8..373e26ee47 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -1176,6 +1176,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/runtime/run_context_retention.py b/loopx/control_plane/runtime/run_context_retention.py index 3cc249023d..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,6 +11,7 @@ from ..work_items.autonomous_replan_ack import ( AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, autonomous_replan_ack_recorded, + 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 @@ -142,7 +142,8 @@ def goal_semantic_history_from_runs( seen_retry_todos: set[tuple[str, str]] = set() observed_at = now_utc_iso() - for run in 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 ): @@ -157,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}) @@ -183,6 +181,7 @@ def goal_semantic_history_from_runs( if ( "latest_outcome_vision_checkpoint_run" not in context + and index not in qualification.accepted_blocked_replans and _run_has_outcome_vision_checkpoint(run) ): context["latest_outcome_vision_checkpoint_run"] = run @@ -200,6 +199,7 @@ def goal_semantic_history_from_runs( if ( "latest_material_milestone_run" not in context + 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..867be7b0fa 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; @@ -92,8 +97,10 @@ export function parseTodoTimestampMicros(value: string): bigint | null { 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/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/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/loopx/control_plane/work_items/autonomous_replan_ack.py b/loopx/control_plane/work_items/autonomous_replan_ack.py index 15d8280788..01d9d7cdbe 100644 --- a/loopx/control_plane/work_items/autonomous_replan_ack.py +++ b/loopx/control_plane/work_items/autonomous_replan_ack.py @@ -1,7 +1,8 @@ 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 from .progress_observation import FRESH_VISION_PATH_DISPOSITIONS @@ -14,6 +15,52 @@ AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK = AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW * 3 +class SettlementHistoryQualification(NamedTuple): + accepted_blocked_replans: frozenset[int] + active_turn_retries: frozenset[int] + + +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") + or (observed_at is not None and "blocked_retry" in run)] + if not candidates: + 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", "blocked_retry", + ) + result = effect_runtime_result("quota.settlement.read", { + "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) + 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: """Return whether an ACK binds exact typed trigger revisions. 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 d5ac47607e..fefdbd702d 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1375,6 +1375,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/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_final_outcome_diagnostics.py b/tests/control_plane/test_final_outcome_diagnostics.py index b542aeba9e..a710605d94 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,155 @@ 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 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, @@ -224,6 +379,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_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_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 50bbff178b..52fcd858c6 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1902,6 +1902,244 @@ 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, 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) + 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) + 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, + "--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" + return project, runtime, registry, before, closed + + +@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, 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( + 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) == prior_work_turns + 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 next(todo for todo in after["todos"] if todo["todo_id"] == TODO_ID) == before["todos"][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", + "--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) == prior_work_turns + + +@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, @@ -5031,6 +5269,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, @@ -5042,26 +5285,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 @@ -5075,6 +5322,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", @@ -6267,6 +6545,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/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/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index c6d2fa1b1f..ae400ee336 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_history_qualification_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"); } @@ -2066,3 +2075,59 @@ 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: {}}, + {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. + 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]); + // 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]], + ["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/); +}); 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" })); 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({