diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index c47aadc0fe..4868cb4d66 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -87,6 +87,13 @@ P0 blocks correctness or continuity in the current user journey. P1 enables repe | **S12 Release, developer experience and community governance · P0 hygiene/P1** | Install/source validation, registration, DCO/PR, test layers, contributor routes and bilingual docs exist | Qualify first work and upgrade/rollback from clean machines/release artifacts; host/OS support follows the release contract. Reduce localization/test/review effort for useful changes; preserve exact-head evidence, fixtures, compatibility, maintainer routing and contributor credit; retire duplicate protocols/stale evidence | | **S13 Adoption, ecosystem and sustainability · P1 discovery/P2 pilots** | Public adoption loop, showcases, licensing/governance and observer-first product contract exist; paid PMF is unproven | Gather independent first/repeat usage and exit reasons; reproducible cases and pilots with fixed budgets/acceptance/rollback. Retain reusable adapters/delivery guides. Account for model/compute/storage/support and maintenance costs; only repeated demand justifies commercial hosting/support/distribution decisions, with no invented SLA or open-source-term change | +**S7 retry projection checkpoint.** Receipt waits introduced after stored resume +lanes were built also need the claim-scoped full selection source. Synthetic +composition and real File/SQLite CLI replay cover a pre-existing wait lane, +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. + ### Important Areas Outside RFCs and Their Owning Sources These directly determine whether a long-running team is usable. A directory or README provides an implementation/contract entry, not installed readiness; read availability from `loopx capability list/show` and the selected release. diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 8c0cafe1a2..653786f531 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -465,6 +465,10 @@ 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. +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. 类型化 deadline owner 会把新叠加的、更早到期的 Turn 重试纳入已有的完整来源 缓存，同时保留显示列表之外的更早监控时钟和未变化的缓存。同一个 typed batch @@ -474,6 +478,9 @@ current gates and the configured cadence floor still apply at and after due time 时钟、真实 gate 和独立可执行工作各自保持。这是自动通知解释的修正，不新增 配置或持久化状态。重试到期仅要求重新检查，不授予执行权限；当前 gate 和 已配置的调度周期下限仍然有效。 +stored resume lane 建立后新增的 receipt wait，沿同一个 claim-scoped 完整 +selection 来源读回。已有等待列表及其显示上限不能隐藏新重试；其它 Agent、 +被排除的执行者、身份不匹配或已被替换的等待不能由这条路径取得资格。 `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/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts index 05d877c740..97598183b2 100644 --- a/loopx/control_plane/todos/quota_selection.ts +++ b/loopx/control_plane/todos/quota_selection.ts @@ -325,6 +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 scopedReceiptRetry(row: JsonObject, due: bigint, agent: string | null): JsonObject | null { + const retry = object(row.blocked_retry); + return 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) + ? retry : null; +} function projectTodoFrontier(summary: JsonObject, now: bigint, agent: string | null): JsonObject { const projected = object(summary.frontier_deadline); const projectedTime = timestamp(projected?.next_due_at); @@ -346,14 +357,8 @@ function projectTodoFrontier(summary: JsonObject, now: bigint, agent: string | n 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); + const retry = scopedReceiptRetry(row, due, agent); + const receiptRetry = retry !== 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)) { @@ -418,6 +423,16 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject { const route = hasSourceContract ? routeContinuationLanes(request.route_items, agent, limit) : {}; const lanes = requireJsonObject(selected.lanes, "selection lanes"); const resumeBlocked = requireJsonObject(resume.resume_blocked_lanes, "resume blocked lanes"); + // An overlay can add a receipt wait after the stored resume lane was built. + // Read its scoped full-source row; an existing nonempty display lane must + // neither hide that retry nor impose its visibility limit on retry clocks. + const receiptWaits = now === null ? [] : rows(selection.items, + selection.available === undefined ? undefined : requireStringArray(selection.available, "available"), + observedAt).filter(row => { + const due = timestamp(row.payload.next_due_at); + return executableBy(row, agent) && due !== null && due > now && + scopedReceiptRetry(row.payload, due, agent) !== null; + }).map(row => row.payload); const frontier = now === null ? null : projectTodoFrontier({ ...requireJsonObject(selected.claim_visibility, "claim visibility"), ...requireJsonObject(resume.deferred_lanes, "deferred lanes"), @@ -427,6 +442,7 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject { resume_blocked_items: [ ...(Array.isArray(resumeBlocked.resume_blocked_items) ? resumeBlocked.resume_blocked_items : []), ...(resume.blocked_successor_items as JsonObject[]), + ...receiptWaits, ], frontier_deadline: request.frontier_deadline ?? null, monitor_open_items: lanes.monitor_items, diff --git a/tests/control_plane/test_blocked_retry_wait_context.py b/tests/control_plane/test_blocked_retry_wait_context.py index cd750a386f..2ba2bf27d4 100644 --- a/tests/control_plane/test_blocked_retry_wait_context.py +++ b/tests/control_plane/test_blocked_retry_wait_context.py @@ -15,7 +15,8 @@ DUE = "2026-10-01T00:05:00Z" -def source(*, user_class="user_action", newer=False, cached_monitor=False, monitor_due="2026-10-01T00:25:00Z"): +def source(*, user_class="user_action", newer=False, cached_monitor=False, + monitor_due="2026-10-01T00:25:00Z", existing_waits=0): 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}"} @@ -34,7 +35,14 @@ def source(*, user_class="user_action", newer=False, cached_monitor=False, monit "text": "Observe the independent source.", "watch_only": True, "cadence": "30m", "next_due_at": monitor_due, "target_key": "independent-source", - }] if cached_monitor else []), + }] if cached_monitor else []) + [{ + "todo_id": f"todo_existing_wait_{index}", "status": "open", "role": "agent", + "task_class": "advancement_task", "claimed_by": "agent-a", + "text": "Wait for an independent prerequisite.", + "resume_when": "resume_at:2035-01-01T00:00:00Z", "resume_ready": False, + "next_due_at": "2035-01-01T00:00:00Z", + "resume_condition": {"kind": "resume_at", "satisfied": False}, + } for index in range(existing_waits)], 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 {}), @@ -53,16 +61,35 @@ def source(*, user_class="user_action", newer=False, cached_monitor=False, monit 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"): + cached_monitor=False, monitor_due="2026-10-01T00:25:00Z", existing_waits=0, + retry_mismatch=None): 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, cached_monitor=cached_monitor, monitor_due=monitor_due) + raw = source(user_class=user_class, newer=newer, cached_monitor=cached_monitor, + monitor_due=monitor_due, existing_waits=existing_waits) before = deepcopy(raw) projected = overlay_active_turn_retries(raw, goal_id="goal-a", agent_id="agent-a", observed_at=now.isoformat()) + if retry_mismatch: + for item in projected["attention_queue"]["items"]: + for summary in (item["agent_todos"], item["project_asset"]["agent_todos"]): + for values in summary.values(): + for row in values if isinstance(values, list) else []: + if not isinstance(row, dict) or row.get("todo_id") != "todo_work": + continue + if retry_mismatch == "claim": + row["claimed_by"] = "agent-b" + elif retry_mismatch == "agent": + row["blocked_retry"] = {**row["blocked_retry"], "agent_id": "agent-b"} + elif retry_mismatch == "excluded": + row["excluded_agents"] = ["agent-a"] + elif retry_mismatch == "retry_todo": + row["blocked_retry"] = {**row["blocked_retry"], "todo_id": "todo_other"} + elif retry_mismatch == "canonical_wait": + row["resume_when"] = "todo_done:todo_dependency" result = build_quota_should_run(projected, goal_id="goal-a", agent_id="agent-a", scheduler_execution_context=scheduler_execution_context_for_runtime_profile( runtime or SchedulerRuntimeProfile.CODEX_APP_HEARTBEAT)) @@ -93,6 +120,25 @@ def test_pending_retry_is_not_a_nonblocking_notice_gate(monkeypatch): assert hint["pending_retry"]["execution_authority_granted"] is False +@pytest.mark.parametrize("existing_waits", [1, 12]) +def test_receipt_retry_survives_preexisting_wait_lanes(monkeypatch, existing_waits): + result = packet(monkeypatch, cached_monitor=True, existing_waits=existing_waits, floor=6) + 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"]["pending_retry"]["todo_id"] == "todo_work" + assert result["scheduler_hint"]["pending_retry"]["due_at"] == DUE + assert result["scheduler_hint"]["app_automation"]["recommended_interval_minutes"] == 6 + + +@pytest.mark.parametrize("mismatch", ["claim", "agent", "excluded", "retry_todo", "canonical_wait"]) +def test_full_source_retry_keeps_scope_and_receipt_fences(monkeypatch, mismatch): + result = packet(monkeypatch, cached_monitor=True, existing_waits=1, + retry_mismatch=mismatch) + assert "pending_retry" not in result["scheduler_hint"] + assert not result["agent_todo_summary"].get("blocked_retry_wait") + + @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) diff --git a/tests/control_plane/test_quota_retry_monitor_journey.py b/tests/control_plane/test_quota_retry_monitor_journey.py index 3019f576db..7a4c4e4422 100644 --- a/tests/control_plane/test_quota_retry_monitor_journey.py +++ b/tests/control_plane/test_quota_retry_monitor_journey.py @@ -20,7 +20,7 @@ @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("scenario", ["waiting", "notice_waiting", "notice_independent", "due_monitor", "independent", "no_retry", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"]) +@pytest.mark.parametrize("scenario", ["waiting", "notice_waiting", "notice_existing_wait", "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: @@ -46,7 +46,14 @@ def test_retry_monitor_guard_and_replay_keep_receipt_and_work_authority( if scenario in {"independent", "notice_independent", "earlier_independent"} else "" )) - if scenario in {"notice_waiting", "notice_independent", "earlier_notice", "equal_notice", "earlier_gate", "earlier_independent"}: + if scenario == "notice_existing_wait": + state.write_text(state.read_text() + + "\n- [ ] Wait for another prerequisite.\n" + f" \n") + + if scenario in {"notice_waiting", "notice_existing_wait", "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"