diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index 9ac95f0b1c..116043a176 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -79,7 +79,7 @@ P0 blocks correctness or continuity in the current user journey. P1 enables repe | **S4 Runtime/host/daemon · P0/P1** | Attached/managed, Turn, broker, runtime connectors and Desktop repairs exist; registration does not establish executable capacity | Qualify multi-Turn supervision for one real supported combination; restart/cancel/drain/stop retain work and fence old executors. Then expand host parity, unique service-profile ownership, clean installation and upgrades; show unsupported adapter capabilities | | **S5 Frontend, Lark and human interaction · P0/P1** | Local chat, settings, proposals and partial Goal Channel verticals exist; Goal Overview, Tasks, the W1 work map, usage, acceptance readback and lifecycle stop ship as separate regions without one page owner; shared audience/session/work readback needs qualification | One journey spans settings, work graph, handoff, blockers, cost, corrections, artifacts and return. **Page:** the [Goal workspace](goal-workspace-surface-v0.md) composes that journey per Goal over shared typed projections: a persistent masthead with scoped Stop, vitals with one meaning per figure (named window and coverage, unknown is not zero), work grouped by owner, results and an all-work rail. M1–M2 come first, after first-screen preview approval. **Cases:** reconnect/repeated-click/stale/original-route. **Realtime IM** reuses Chat/Turn: isolate independent conversations on one listener, then qualify ordinary DM onboarding, explicit role selection, busy-session admission, provisional progress, media and exact permission callbacks through [the shared operational contract](capable-manager-semantic-handoff-v0.md#10-operational-contract). **Team:** [Live team workspace](live-team-workspace-v0.md) makes exchange, revision and original-coordinator continuation visible. Its [Work-scale map track](live-team-workspace-v0.md#11-delivery-order-and-relationship-to-aggressive-r2-progress) draws each Goal's typed Todo relations first (W1, the Goal workspace's Map lens), then live state and outputs on the same nodes. Then intelligent review, keyboard accessibility and a launcher-only command palette, bilingual terminology, actionable errors and offline degradation; interrupt only for actual decisions | | **S6 Materials, evidence, memory and learning · P1** | Authority registry, material lifecycle/frontier, decision context, reward memory and turn recall exist; explicit project material packets and source-verifier gates are implemented, while private source initialization/Core binding and ordinary-conversation adoption remain unqualified; direction baseline and parts of attribution remain proposed | Connect material revision→same-Agent read→decision reference→artifact/outcome. Expose expiry/revocation/source loss and forgetting policy. Handoff preserves decision-relevant summaries and authorized artifacts; qualify OpenViking/Obelisk as optional providers. Prove causal utility with controls, not relevance alone | -| **S7 Budget, scheduling and fleet scale · P0 observation/P1–P2 expansion** | Quota/scheduler and partial usage aggregates exist; full provider cost, distributed reservations and hundred-Agent concurrency need evidence | Separate configured budget, admission, consumption and estimates; unknown is not zero and replay cannot double-charge. R7 pagination/bounded summaries and [complete-history transport](typescript-control-plane-migration-v0.md), including refresh/replay/single-debit evidence beyond the RPC limit; provider/host limits, fairness, backpressure, event wake and isolation; report registration/activity/throughput and cost per accepted outcome separately. Receipt-backed semantic progress closes its original host Turn without completing an open/waiting Todo; File/SQLite guard replay preserves verified writeback when a later dependency wait changes the frontier and covers automatic quiet writeback during a receipt-owned retry, preserving independent work and monitor clocks. [Full quota and TurnEnvelope](../../reference/protocols/turn-envelope-v0.md) retain the original executable settlement command and single debit. Follow [the quota settlement contract](../../quota-allocation.md#receipt-backed-settlement-progress). External-wait authoring now derives successor readiness from the locked Todo snapshot through the existing resume owner, covering completed dependencies and monitor generation fences without trusting display flags. File/SQLite CLI coverage also rejects an original Turn’s step write into a newly selected Todo, preserves that Todo’s step/CAS and dependencies, and settles the original writeback once. Source-qualified continuation, installed adoption, live Host waiting-state cadence/readback and fleet scale remain separate acceptance; this bounded repair does not qualify fleet scale or host latency. | +| **S7 Budget, scheduling and fleet scale · P0 observation/P1–P2 expansion** | Quota/scheduler and partial usage aggregates exist; full provider cost, distributed reservations and hundred-Agent concurrency need evidence | Separate configured budget, admission, consumption and estimates; unknown is not zero and replay cannot double-charge. R7 pagination/bounded summaries and [complete-history transport](typescript-control-plane-migration-v0.md), including refresh/replay/single-debit evidence beyond the RPC limit; provider/host limits, fairness, backpressure, event wake and isolation; report registration/activity/throughput and cost per accepted outcome separately. Receipt-backed semantic progress closes its original host Turn without completing an open/waiting Todo; File/SQLite guard replay preserves verified writeback when a later dependency wait changes the frontier and covers automatic quiet writeback during a receipt-owned retry, preserving independent work and monitor clocks. The typed full-source batch keeps receipt retry identity independently of the earliest wakeup, reconciles earlier retries with retained caches/hidden clocks, and keeps ordinary notices non-gating across earlier/equal/later monitors; File/SQLite replay and canonical readback qualify this source boundary. [Full quota and TurnEnvelope](../../reference/protocols/turn-envelope-v0.md) retain the original executable settlement command and single debit. Follow [the quota settlement contract](../../quota-allocation.md#receipt-backed-settlement-progress). External-wait authoring now derives successor readiness from the locked Todo snapshot through the existing resume owner, covering completed dependencies and monitor generation fences without trusting display flags. File/SQLite CLI coverage also rejects an original Turn’s step write into a newly selected Todo, preserves that Todo’s step/CAS and dependencies, and settles the original writeback once. Source-qualified continuation, installed adoption, live Host waiting-state cadence/readback and fleet scale remain separate acceptance; this bounded repair does not qualify fleet scale or host latency. | | **S8 Capabilities, extensions and domain integration · P1/P2** | Capability catalog, extension lifecycle, hooks, engineering/research/content/office capabilities and computer-use contracts exist | First exercise the shared control plane with existing issue-fix/PR-review and material/research callers. Every provider has readiness/version/permissions/default-off/uninstall/rollback/isolation and real-entry evidence. New domain effects start with one simulated operation, not a marketplace or workflow DSL | | **S9 Identity, authority, privacy and trust · continuous P0/P1–P2 remote** | Public/private scope, capability gates, fencing and confirmation contracts belong to existing owners | R1/R3 cover sender/audience/artifact scope and stale authority; R6 authenticates tenant/Goal/actor/host, rotation/revocation and least privilege. Qualify credential custody, untrusted tool/document inputs, dependency supply chain, audit retention/deletion and vulnerability response through real paths; roles/messages/memory mint no write authority | | **S10 Reliability, diagnostics and operations · P0/P1** | Recovery/canary, read-only diagnostics prototype and DSH event adapter exist; C0/C1, overhead and full operations qualification are open | Failure classification→observable state→recovery drill→regression prevention; process/storage/network/delivery failures and data growth. Accepted Chat requests must settle even when runtime preparation fails before dispatch; qualify missing runtime assets, stop races and recovery without replay under [the shared conversation operational contract](capable-manager-semantic-handoff-v0.md#10-operational-contract). A context/provider read returning after the stop wait must honor the persisted Turn and exact Session claim: no late provider dispatch or handoff, even when a fresh request has completed. This bounded GQ08 repair does not certify upstream interrupt fidelity or effects already admitted elsewhere. Use [bounded repair lookup and targeted diagnostics](../../../skills/loopx-self-repair/references/targeted-diagnostics.md) to reduce redundant reads above the provider boundary; measure backend-specific cold/warm reads, writes and lock waits separately. Freeze SLO/RPO/RTO/capacity/retention boundaries and measure before qualification. Lark transport backpressure and drain retain the App consumer lease; unpersisted buffered lines are not accepted work or completion receipts. Runbooks include upgrade, restore, stop and human takeover; test counts do not prove recovery | diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 860b1fc4b7..c0db29e3f9 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -433,6 +433,26 @@ 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 receipt or changing the monitor's due time, hash, or material generation. +The typed deadline owner reconciles a newer, earlier receipt-owned retry with +an existing full-source deadline cache; hidden earlier monitor deadlines and +unchanged caches remain intact. The same typed batch derives `blocked_retry_wait` +from qualified full-source receipts independently of `frontier_deadline`, which +still names the earliest wakeup. A non-blocking `user_action` notice therefore +does not turn the retry into a human gate even when an earlier or equal monitor +owns that wakeup. The retry identity and original due time remain visible; the +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. + +类型化 deadline owner 会把新叠加的、更早到期的 Turn 重试纳入已有的完整来源 +缓存，同时保留显示列表之外的更早监控时钟和未变化的缓存。同一个 typed batch +从完整且合格的 receipt 来源独立投影 `blocked_retry_wait`；`frontier_deadline` +仍表示最早唤醒时间。即使更早或同刻的独立 monitor 拥有下一唤醒，普通 +`user_action` 提醒也不会把重试解释为人工门禁；原 retry 身份/到期时间、monitor +时钟、真实 gate 和独立可执行工作各自保持。这是自动通知解释的修正，不新增 +配置或持久化状态。重试到期仅要求重新检查，不授予执行权限；当前 gate 和 +已配置的调度周期下限仍然有效。 `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 diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 293fc529ec..421b42eb54 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -1291,8 +1291,9 @@ def _build_active_quota_payload( # gate explanation. Genuine gates and unknown scope keep their old route. notice_only = (user_summary.get("user_action_open_count", 0) > 0 and user_summary.get("all_open_count") == user_summary.get("user_action_open_count")) - retry_wait = (retry_deadline.get("blocked_retry") - if notice_only and retry_deadline.get("source") == "turn_settlement" else None) + retry_wait = ((prepared.agent_todo_summary or {}).get("blocked_retry_wait") or + (retry_deadline.get("blocked_retry") + if retry_deadline.get("source") == "turn_settlement" else None)) if notice_only else None execution_obligation = _execution_obligation( should_run=route.should_run, effective_action=route.effective_action, diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index ca5648b4a8..857b8fe114 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -629,9 +629,11 @@ def build_codex_app_scheduler_fallback_hint( def _pending_retry_projection(payload: dict[str, Any]) -> dict[str, Any]: - deadline = _dict_or_empty(_dict_or_empty(payload.get("agent_todo_summary")).get("frontier_deadline")) - retry = deadline.get("blocked_retry") - if deadline.get("source") != "turn_settlement" or not isinstance(retry, dict): + summary = _dict_or_empty(payload.get("agent_todo_summary")) + deadline = _dict_or_empty(summary.get("frontier_deadline")) + retry = summary.get("blocked_retry_wait") or (deadline.get("blocked_retry") + if deadline.get("source") == "turn_settlement" else None) + if not isinstance(retry, dict): return {} return {"pending_retry": {**retry, "execution_authority_granted": False, "recheck": "rerun quota at or after due_at; current gates and owner cadence still apply"}} @@ -649,6 +651,11 @@ def _receipt_retry_wait_profile( or due is None or due <= current_time): return None horizon = max(1, math.ceil((due - current_time).total_seconds() / 60)) + frontier = build_frontier_recheck_plan(payload, current_time=current_time) + if frontier: + # The retry explains the wait, but an independent earlier clock still + # owns the next wakeup. The owner cadence floor is applied afterward. + horizon = min(horizon, max(1, math.ceil(frontier["frontier_recheck_after_seconds"] / 60))) return {"interval_minutes": horizon, "retry": retry} if horizon < interval_minutes else None diff --git a/loopx/control_plane/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts index b2eb2c6484..05d877c740 100644 --- a/loopx/control_plane/todos/quota_selection.ts +++ b/loopx/control_plane/todos/quota_selection.ts @@ -325,13 +325,17 @@ function utcIso(micros: bigint): string { const date = new Date(Number(seconds * 1000n)).toISOString().slice(0, -5); return `${date}${fraction ? `.${String(fraction).padStart(6, "0")}` : ""}+00:00`; } -function projectTodoFrontierDeadline(summary: JsonObject, now: bigint, agent: string | null): JsonObject | null { +function projectTodoFrontier(summary: JsonObject, now: bigint, agent: string | null): JsonObject { const projected = object(summary.frontier_deadline); const projectedTime = timestamp(projected?.next_due_at); - if (projected && projectedTime !== null && projectedTime > now) return {...projected}; + const currentProjection = projected && projectedTime !== null && projectedTime > now ? projected : null; const seen = new Set(); - let selected: JsonObject | null = null, earliest: bigint | null = null; - for (const lane of frontierLanes) { + let selected: JsonObject | null = currentProjection; + let earliest: bigint | null = currentProjection ? projectedTime : null; + let addedRetryCount = 0; + let retryWait: JsonObject | null = null; + let retryDue: bigint | null = null; + for (const lane of currentProjection ? ["resume_blocked_items"] : frontierLanes) { const items = summary[lane]; if (!Array.isArray(items)) continue; for (const raw of items) { @@ -342,22 +346,38 @@ function projectTodoFrontierDeadline(summary: JsonObject, now: bigint, agent: st const id = identity(row), key = JSON.stringify([id, String(due)]); if (seen.has(key)) continue; seen.add(key); + const retry = object(row.blocked_retry); + const receiptRetry = row.status === "open" && row.task_class === "advancement_task" && + row.resume_ready === false && retry?.source === "turn_settlement" && + typeof retry.turn_instance_id === "string" && !!retry.turn_instance_id && + retry.agent_id === agent && agent !== null && typeof retry.goal_id === "string" && + (!row.claimed_by || row.claimed_by === retry.agent_id) && + row.resume_when === retry.resume_when && timestamp(retry.due_at) === due && + isBoundedBlockedRetry(retry, typeof row.todo_id === "string" ? row.todo_id : null); + // A receipt wait owns one unfinished Todo; the earliest wakeup may + // belong to a different monitor. Keep both facts from this full source. + if (receiptRetry && (retryDue === null || due < retryDue)) { + retryDue = due; + retryWait = retry; + } + if (currentProjection) { + // The full-source cache retains deadlines omitted from visible lanes. + // Only newly overlaid receipt retries can predate that current cache. + if (!receiptRetry || due >= projectedTime!) continue; + if (id !== currentProjection.identity) addedRetryCount += 1; + } if (earliest === null || due < earliest) { earliest = due; - const retry = object(row.blocked_retry); - const receiptRetry = row.status === "open" && row.task_class === "advancement_task" && - row.resume_ready === false && retry?.source === "turn_settlement" && - typeof retry.turn_instance_id === "string" && !!retry.turn_instance_id && - retry.agent_id === agent && agent !== null && typeof retry.goal_id === "string" && - (!row.claimed_by || row.claimed_by === retry.agent_id) && - row.resume_when === retry.resume_when && timestamp(retry.due_at) === due && - isBoundedBlockedRetry(retry, typeof row.todo_id === "string" ? row.todo_id : null); selected = {identity: id, source: receiptRetry ? "turn_settlement" : source(row, lane), next_due_at: utcIso(due), ...(receiptRetry ? {blocked_retry: retry} : {})}; } } } - return selected ? {schema_version: "todo_frontier_deadline_v0", ...selected, candidate_count: seen.size} : null; + const candidateCount = currentProjection + ? Math.max(1, Number(currentProjection.candidate_count) || 0) + addedRetryCount : seen.size; + const deadline = currentProjection && selected === currentProjection ? {...currentProjection} : + selected ? {schema_version: "todo_frontier_deadline_v0", ...selected, candidate_count: candidateCount} : null; + return {frontier_deadline: deadline, ...(retryWait ? {blocked_retry_wait: retryWait} : {})}; } /** One quota read boundary composes scope/claim selection, resume, hint @@ -397,10 +417,17 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject { const handoff = hasSourceContract ? handoffGateLanes(request.handoff_items, agent, limit) : {}; const route = hasSourceContract ? routeContinuationLanes(request.route_items, agent, limit) : {}; const lanes = requireJsonObject(selected.lanes, "selection lanes"); - const frontier = now === null ? null : projectTodoFrontierDeadline({ + const resumeBlocked = requireJsonObject(resume.resume_blocked_lanes, "resume blocked lanes"); + const frontier = now === null ? null : projectTodoFrontier({ ...requireJsonObject(selected.claim_visibility, "claim visibility"), ...requireJsonObject(resume.deferred_lanes, "deferred lanes"), - ...requireJsonObject(resume.resume_blocked_lanes, "resume blocked lanes"), ...handoff, + ...resumeBlocked, ...handoff, + // Retain the display lane and union the existing full-source successor + // waits before choosing a deadline; visibility limits are not clock limits. + resume_blocked_items: [ + ...(Array.isArray(resumeBlocked.resume_blocked_items) ? resumeBlocked.resume_blocked_items : []), + ...(resume.blocked_successor_items as JsonObject[]), + ], frontier_deadline: request.frontier_deadline ?? null, monitor_open_items: lanes.monitor_items, gate_open_items: lanes.gate_items, @@ -410,5 +437,5 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject { if (now !== null) lanes.gate_items = (lanes.gate_items as JsonObject[]).slice(0, 3); return {schema_version: "todo_quota_planning_v0", resume_planning: resume, ...selected, ...closure, ...(hasSourceContract ? {handoff_lanes: handoff, route_lanes: route} : {}), - ...(now === null ? {} : {frontier_deadline: frontier})}; + ...(now === null ? {} : frontier)}; } diff --git a/loopx/control_plane/todos/quota_summary.py b/loopx/control_plane/todos/quota_summary.py index fbea09267e..63f095dab3 100644 --- a/loopx/control_plane/todos/quota_summary.py +++ b/loopx/control_plane/todos/quota_summary.py @@ -299,6 +299,8 @@ def summarize_user_todos_for_quota( summary["closure_intent"] = planning["closure_intent"] if planning.get("frontier_deadline"): summary["frontier_deadline"] = planning["frontier_deadline"] + if planning.get("blocked_retry_wait"): + summary["blocked_retry_wait"] = planning["blocked_retry_wait"] monitor_writeback = todo_summary_monitor_writeback_contract(value) if monitor_writeback: summary["monitor_writeback"] = monitor_writeback @@ -677,6 +679,8 @@ def summarize_project_asset_todos_for_quota( } if planning.get("frontier_deadline"): summary["frontier_deadline"] = planning["frontier_deadline"] + if planning.get("blocked_retry_wait"): + summary["blocked_retry_wait"] = planning["blocked_retry_wait"] monitor_writeback = todo_summary_monitor_writeback_contract(value) if monitor_writeback: summary["monitor_writeback"] = monitor_writeback diff --git a/loopx/control_plane/work_items/execution_obligation.py b/loopx/control_plane/work_items/execution_obligation.py index 425b01bfb6..9b482fd8fc 100644 --- a/loopx/control_plane/work_items/execution_obligation.py +++ b/loopx/control_plane/work_items/execution_obligation.py @@ -219,7 +219,10 @@ def build_execution_obligation( "only controls whether to interrupt the user" ), } - if should_run and blocked_retry_wait: + # A quiet monitor route still owns a pending retry, not a user-action gate. + if blocked_retry_wait and ( + should_run or effective_action == EffectiveAction.MONITOR_QUIET_SKIP.value + ): return { "must_attempt_work": False, "kind": EffectiveAction.BLOCKED_WAIT.value, diff --git a/tests/control_plane/test_blocked_retry_wait_context.py b/tests/control_plane/test_blocked_retry_wait_context.py index 8557060b45..cd750a386f 100644 --- a/tests/control_plane/test_blocked_retry_wait_context.py +++ b/tests/control_plane/test_blocked_retry_wait_context.py @@ -15,34 +15,52 @@ DUE = "2026-10-01T00:05:00Z" -def source(*, user_class="user_action", newer=False): +def source(*, user_class="user_action", newer=False, cached_monitor=False, monitor_due="2026-10-01T00:25:00Z"): retry = {"schema_version": "quota_blocked_retry_v0", "source": "turn_settlement", "todo_id": "todo_work", "observed_at": "2026-10-01T00:00:00Z", "due_at": DUE, "resume_when": f"resume_at:{DUE}"} run = {"agent_id": "agent-a", "todo_id": "todo_work", "turn_instance_id": "turn-old", "delivery_outcome": "outcome_gap", "progress_observation": {"result_class": "blocked"}, "blocked_retry": retry} - return quota_status_payload(goal_id="goal-a", status="active", - recommended_action="Validate remaining authorized work.", claim_scope_agent_id="agent-a", + result = quota_status_payload(goal_id="goal-a", status="active", + recommended_action=("Wait for the recorded source." if cached_monitor else + "Validate remaining authorized work."), claim_scope_agent_id="agent-a", coordination={"registered_agents": ["agent-a", "agent-b"], "handoff_mode": "soft_claim"}, agent_todo_items=[{"todo_id": "todo_work", "status": "open", "role": "agent", "archive_state": "active", "task_class": "advancement_task", "claimed_by": "agent-a", - "text": "Validate remaining authorized work.", "action_kind": "validate"}], + "text": "Validate remaining authorized work.", "action_kind": "validate"}] + ([{ + "todo_id": "todo_watch", "status": "open", "role": "agent", + "task_class": "continuous_monitor", "claimed_by": "agent-a", + "text": "Observe the independent source.", "watch_only": True, + "cadence": "30m", "next_due_at": monitor_due, + "target_key": "independent-source", + }] if cached_monitor else []), user_todo_items=[{"todo_id": "todo_notice", "status": "open", "role": "user", "task_class": user_class, "bound_agent": "agent-a", **({"blocks_agent": "agent-a"} if user_class == "user_gate" else {}), "text": "Review the independent draft."}], latest_runs=([{"agent_id": "agent-a", "todo_id": "todo_work", "delivery_outcome": "outcome_progress"}] if newer else []) + [run]) + if cached_monitor: + for item in result["attention_queue"]["items"]: + for summary in (item["agent_todos"], item["project_asset"]["agent_todos"]): + summary["frontier_deadline"] = { + "schema_version": "todo_frontier_deadline_v0", + "identity": "todo_watch", "source": "continuous_monitor", + "next_due_at": monitor_due, "candidate_count": 1, + } + return result -def packet(monkeypatch, *, seconds=120, user_class="user_action", newer=False, runtime=None, floor=0): +def packet(monkeypatch, *, seconds=120, user_class="user_action", newer=False, runtime=None, floor=0, + cached_monitor=False, monitor_due="2026-10-01T00:25:00Z"): 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) - raw = source(user_class=user_class, newer=newer) + raw = source(user_class=user_class, newer=newer, cached_monitor=cached_monitor, monitor_due=monitor_due) before = deepcopy(raw) projected = overlay_active_turn_retries(raw, goal_id="goal-a", agent_id="agent-a", observed_at=now.isoformat()) result = build_quota_should_run(projected, goal_id="goal-a", agent_id="agent-a", @@ -75,6 +93,19 @@ def test_pending_retry_is_not_a_nonblocking_notice_gate(monkeypatch): assert hint["pending_retry"]["execution_authority_granted"] is False +@pytest.mark.parametrize("seconds,floor,interval", [(120, 0, 3), (295, 0, 1), (295, 6, 6)]) +def test_retry_overlay_precedes_a_later_cached_monitor_deadline(monkeypatch, seconds, floor, interval): + result = packet(monkeypatch, seconds=seconds, floor=floor, cached_monitor=True) + assert result["execution_obligation"]["kind"] == "blocked_wait" + assert result["interaction_contract"]["user_channel"]["non_gating_notice"] is True + assert result["scheduler_hint"]["cadence_class"] == "quiet_wait" + assert result["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] == interval + assert result["scheduler_hint"]["pending_retry"]["due_at"] == DUE + assert result["scheduler_hint"]["pending_retry"]["execution_authority_granted"] is False + assert result["agent_todo_summary"]["frontier_deadline"]["identity"] == "todo_work" + assert result["agent_todo_summary"]["frontier_deadline"]["candidate_count"] == 2 + + @pytest.mark.parametrize("seconds", [300, 301]) def test_at_or_after_due_retry_does_not_hold_work(monkeypatch, seconds): result = packet(monkeypatch, seconds=seconds) @@ -105,6 +136,14 @@ def test_real_user_gate_survives_retry_clock(monkeypatch): assert hint["pending_retry"]["due_at"] == DUE +def test_real_source_gate_is_not_reclassified_by_cached_retry(monkeypatch): + result = packet(monkeypatch, user_class="user_gate", cached_monitor=True) + assert result["interaction_contract"]["agent_channel"]["delivery_allowed"] is False + assert result["interaction_contract"]["agent_channel"]["must_attempt"] is False + assert not result["interaction_contract"]["user_channel"].get("non_gating_notice") + assert result["scheduler_hint"]["cadence_class"] == "human_gate" + + def test_owner_floor_clamps_retry_recheck(monkeypatch): result = packet(monkeypatch, floor=6) assert result["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] == 6 @@ -120,7 +159,7 @@ def test_goal_runtime_uses_same_retry_clock(monkeypatch): @pytest.mark.parametrize("mismatch", ["agent", "goal", "todo", "claim", "canonical_wait"]) def test_retry_projection_does_not_cross_scope_or_replace_a_canonical_wait(mismatch): - raw = source() + raw = source(cached_monitor=True) if mismatch == "todo": raw["run_history"]["goals"][0]["latest_runs"][0]["todo_id"] = "todo_other" for item in raw["attention_queue"]["items"]: @@ -140,3 +179,16 @@ def test_retry_projection_does_not_cross_scope_or_replace_a_canonical_wait(misma agent_id="agent-b" if mismatch == "agent" else "agent-a", observed_at="2026-10-01T00:02:00Z") assert result is raw assert result == before + + +@pytest.mark.parametrize("monitor_due", ["2026-10-01T00:03:00Z", DUE, "2026-10-01T00:25:00Z"]) +def test_retry_identity_survives_the_independent_next_wakeup(monkeypatch, monitor_due): + result = packet(monkeypatch, cached_monitor=True, monitor_due=monitor_due) + assert result["execution_obligation"]["kind"] == "blocked_wait" + assert result["interaction_contract"]["user_channel"]["non_gating_notice"] is True + assert result["scheduler_hint"]["cadence_class"] == "quiet_wait" + assert result["agent_todo_summary"]["blocked_retry_wait"]["due_at"] == DUE + assert result["scheduler_hint"]["pending_retry"]["due_at"] == DUE + frontier = result["agent_todo_summary"]["frontier_deadline"] + assert frontier["identity"] == ("todo_watch" if monitor_due <= DUE else "todo_work") + assert result["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] == (1 if monitor_due < DUE else 3) diff --git a/tests/control_plane/test_quota_retry_monitor_journey.py b/tests/control_plane/test_quota_retry_monitor_journey.py index d5d437ee9c..3019f576db 100644 --- a/tests/control_plane/test_quota_retry_monitor_journey.py +++ b/tests/control_plane/test_quota_retry_monitor_journey.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from datetime import UTC, datetime, timedelta from pathlib import Path from urllib.parse import quote @@ -19,7 +20,7 @@ @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("scenario", ["waiting", "due_monitor", "independent", "no_retry"]) +@pytest.mark.parametrize("scenario", ["waiting", "notice_waiting", "notice_independent", "due_monitor", "independent", "no_retry", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"]) def test_retry_monitor_guard_and_replay_keep_receipt_and_work_authority( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, scenario: str, ) -> None: @@ -42,9 +43,15 @@ def test_retry_monitor_guard_and_replay_keep_receipt_and_work_authority( "- [ ] Validate an independent result.\n" f" \n" - if scenario == "independent" else "" + if scenario in {"independent", "notice_independent", "earlier_independent"} else "" )) + if scenario in {"notice_waiting", "notice_independent", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"}: + state.write_text(state.read_text() + + "\n## User Todo\n\n- [ ] Review an independent draft.\n" + f" \n") + def run(*args: str): return cli._run_cli(registry, runtime, *args, cwd=project) @@ -76,6 +83,17 @@ def run(*args: str): assert code == 0, blocked assert blocked["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend" retry = blocked["blocked_retry"] + if scenario in {"earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"}: + due = retry["due_at"] if scenario == "equal_notice" else (datetime.now(UTC) + timedelta(seconds=120)).isoformat() + code, updated = run("todo", "update", "--goal-id", cli.GOAL_ID, + "--todo-id", "todo_retry_watch", "--agent-id", cli.AGENT_ID, + "--next-due-at", due, "--execute") + assert code == 0 and updated["ok"], updated + if scenario == "earlier_gate": + code, updated = run("todo", "update", "--goal-id", cli.GOAL_ID, + "--todo-id", "todo_retry_notice", "--agent-id", cli.AGENT_ID, + "--task-class", "user_gate", "--blocks-agent", cli.AGENT_ID, "--execute") + assert code == 0 and updated["ok"], updated guards = [] for _ in range(2): @@ -84,15 +102,23 @@ def run(*args: str): "--turn-instance-id", "turn-retry-next") assert code == 0 and guard["ok"], guard assert guard["heartbeat_receipt"]["status"] in {"committed", "replayed"} - if scenario == "waiting": + if scenario in {"waiting", "notice_waiting", "earlier_notice", "equal_notice"}: assert guard["effective_action"] == "monitor_quiet_skip" assert guard["execution_obligation"]["must_attempt_work"] is False + if scenario in {"notice_waiting", "earlier_notice", "equal_notice"}: + assert guard["execution_obligation"]["kind"] == "blocked_wait" + assert guard["interaction_contract"]["user_channel"]["non_gating_notice"] is True + assert guard["scheduler_hint"]["cadence_class"] == "quiet_wait" pending = guard["scheduler_hint"]["pending_retry"] assert pending["todo_id"] == cli.TODO_ID assert pending["due_at"] == retry["due_at"] + elif scenario == "earlier_gate": + assert guard["scheduler_hint"]["cadence_class"] == "human_gate" + assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is False + assert not guard["interaction_contract"]["user_channel"].get("non_gating_notice") else: assert guard["effective_action"] == "normal_run" - if scenario == "independent": + if scenario in {"independent", "notice_independent", "earlier_independent"}: assert guard["selected_todo"]["todo_id"] == "todo_retry_independent" guards.append(guard) assert guards[0]["heartbeat_receipt"]["event_id"] == guards[1]["heartbeat_receipt"]["event_id"] diff --git a/tests/control_plane_ts/quota_selection.test.ts b/tests/control_plane_ts/quota_selection.test.ts index d9f9aa3d29..897553b5d3 100644 --- a/tests/control_plane_ts/quota_selection.test.ts +++ b/tests/control_plane_ts/quota_selection.test.ts @@ -370,7 +370,7 @@ test("quota v2 validates closure in the existing batch while retaining v0/v1 wir }); -function planningDeadline(monitors: JsonObject[], gates: JsonObject[], fields: JsonObject = {}) { +function planningDeadline(monitors: JsonObject[], gates: JsonObject[], fields: JsonObject = {}, retries: JsonObject[] = []) { const items = [...monitors.map(payload => row(String(payload.todo_id), {payload, task_class: "continuous_monitor", required: [], targets: []})), ...gates.map(payload => row(String(payload.todo_id), {payload, task_class: "user_gate", gate: true, required: [], targets: []}))]; @@ -381,7 +381,13 @@ function planningDeadline(monitors: JsonObject[], gates: JsonObject[], fields: J has_deferred_count: false, has_visible_deferred_count: false, deferred_count: null, available_capabilities: null, sources: Object.fromEntries(["items", "backlog_items", "first_open_items", "deferred_items", "deferred_resume_candidates", "resume_blocked_items", "monitor_open_items", - "current_agent_claimed_monitor_items", "claimed_monitor_open_items"].map(key => [key, []]))}}); + "current_agent_claimed_monitor_items", "claimed_monitor_open_items"].map(key => [key, + key === "items" || key === "resume_blocked_items" ? retries.map(payload => ({ + payload: {text: "Wait for an authorized retry.", ...payload}, id: payload.todo_id, + status: payload.status, claim: payload.claimed_by ?? null, excluded: [], done: false, + resume: payload.resume_when, ready: false, ready_truthy: false, priority: 1, index: 1, + target_id: null, target_status: null, target_class: "advancement_task", + })) : []]))}}); } test("the existing batch keeps full gate deadlines before transport caps, scope and equal-time source", () => { const early = {todo_id: "early", next_due_at: "2026-10-01T08:01:00.000001+08:00"}; @@ -418,3 +424,42 @@ test("planning clock retains fresh projection extensions and rechecks stale/raw {current_time: "1969-12-31T23:59:58Z"}).frontier_deadline as JsonObject).next_due_at, "1969-12-31T23:59:59.000001+00:00"); }); + + +test("receipt retries precede later full-source caches without losing hidden earlier deadlines", () => { + const due = "2026-10-01T00:05:00Z"; + const retry = {schema_version: "quota_blocked_retry_v0", source: "turn_settlement", + todo_id: "retry", observed_at: "2026-10-01T00:00:00Z", due_at: due, + resume_when: `resume_at:${due}`, goal_id: "goal-a", agent_id: "agent-a", turn_instance_id: "turn-old"}; + const row = {todo_id: "retry", status: "open", task_class: "advancement_task", claimed_by: "agent-a", + resume_ready: false, resume_when: retry.resume_when, next_due_at: due, blocked_retry: retry, + resume_condition: {kind: "resume_at", satisfied: false}}; + const cache = {schema_version: "todo_frontier_deadline_v0", identity: "hidden-monitor", + source: "continuous_monitor", next_due_at: "2026-10-01T00:25:00Z", candidate_count: 25}; + const before = structuredClone({cache, row}); + assert.deepEqual(planningDeadline([], [], {frontier_deadline: cache}, [row, row]).frontier_deadline, + {schema_version: "todo_frontier_deadline_v0", identity: "retry", source: "turn_settlement", + next_due_at: "2026-10-01T00:05:00+00:00", blocked_retry: retry, candidate_count: 26}); + const later = Array.from({length: 8}, (_, index) => { + const id = `later-${index}`, due = "2026-10-01T00:20:00Z", resume = `resume_at:${due}`; + return {...row, todo_id: id, next_due_at: due, resume_when: resume, + blocked_retry: {...retry, todo_id: id, due_at: due, resume_when: resume}}; + }); + assert.equal((planningDeadline([], [], {frontier_deadline: cache}, [...later, row]) + .frontier_deadline as JsonObject).identity, "retry"); + const earlier = {...cache, next_due_at: "2026-10-01T00:01:00Z", extension: {retained: true}}; + assert.deepEqual(planningDeadline([], [], {frontier_deadline: earlier}, [row]).frontier_deadline, earlier); + for (const next_due_at of ["2026-10-01T00:01:00Z", due, "2026-10-01T00:25:00Z"]) { + const result = planningDeadline([], [], {frontier_deadline: {...cache, next_due_at}}, [...later, row]); + assert.deepEqual(result.blocked_retry_wait, retry, "the ninth retry retains identity independently of the earliest wakeup"); + } + const replaced = {...cache, identity: "retry", source: "turn_settlement", candidate_count: 1}; + assert.equal((planningDeadline([], [], {frontier_deadline: replaced}, [row]).frontier_deadline as JsonObject) + .candidate_count, 1); + assert.deepEqual({cache, row}, before); + for (const invalid of [{...row, claimed_by: "agent-b"}, {...row, blocked_retry: {...retry, agent_id: "agent-b"}}, + {...row, blocked_retry: {...retry, due_at: "2026-10-01T00:31:00Z"}}, {...row, resume_when: "todo_done:other"}]) { + assert.deepEqual(planningDeadline([], [], {frontier_deadline: cache}, [invalid]).frontier_deadline, cache); + assert.equal(planningDeadline([], [], {frontier_deadline: cache}, [invalid]).blocked_retry_wait, undefined); + } +});