From 44363a4e9a493adbee04b9a578d9e0b87e28df1e Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 14 Oct 2026 09:38:48 +0800 Subject: [PATCH] fix(quota): keep unbound scheduled vision waits quiet Signed-off-by: huangruiteng --- docs/quota-allocation.md | 14 ++ .../control_plane/effect_runtime_handlers.ts | 1 + loopx/control_plane/goals/goal_vision_wait.py | 10 + .../work_items/interaction_contract.py | 17 +- .../work_items/interaction_contract.ts | 32 +++ .../test_dated_vision_wait_cli.py | 201 ++++++++++++++++++ .../interaction_contract.test.ts | 39 ++++ 7 files changed, 309 insertions(+), 5 deletions(-) create mode 100644 tests/control_plane/test_dated_vision_wait_cli.py diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 36b8f72c04..411513d291 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -883,6 +883,20 @@ deferred-successor selection runs again. Missing vision checkpoints, closed stage succession, unsupported conditions, and continuous-monitor-as-gate repairs remain fail-closed and are not hidden by this wait path. +For a single exact scheduled successor, the interaction contract allows a +quiet no-op while the canonical `resume_at` condition has positive runtime-clock +proof that it is still pending. It does not require a `monitor-poll` from an +unbound heartbeat receipt or derive new work merely from repeated pre-due +wakes. The scheduled instant and Todo lifecycle remain unchanged; once due, +the normal explicit resume route applies. Existing bound Turns still owe +their original settlement. Dependency waits, mixed waiting frontiers, and +missing or invalid clock proofs retain the bounded observation policy. + +单个精确的定时 successor 若有原生运行时时钟证明尚未到期，交互契约允许安静等待， +不要求无 Todo 绑定的 heartbeat 回执执行 `monitor-poll`，也不因重复提前唤醒而派生 +新工作。原恢复时间与 Todo 生命周期保留；到期后走正常显式恢复流程。已有绑定的 +Turn 仍需结算原工作；依赖等待、混合等待前沿和缺失或无效的时钟证据仍沿用有界观察策略。 + For completed handoff gates, record that rationale structurally as `no_followup=true`; status projects the gate as `cleared_no_followup` instead of waking the blocked agent with `successor_replan_required`. diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index d07e0c1b2f..9765fcbdc5 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -751,6 +751,7 @@ export function createEffectRuntimeHandlers( ["turn.host_todo_completion.evaluate", lazyHandler(() => import("./turn_driver/host_todo_completion.ts"), ({evaluateHostTodoCompletion}) => evaluateHostTodoCompletion)], ["work_item.interaction_reads.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionRequiredReads}) => projectInteractionRequiredReads)], ["work_item.interaction_consistency.validate", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionConsistencyValidation}) => projectInteractionConsistencyValidation)], + ["work_item.blocked_wait_observation.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectBlockedSuccessorWaitObservation}) => projectBlockedSuccessorWaitObservation)], ["work_item.context.plan", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({planInteractionWorkContext}) => planInteractionWorkContext)], ["work_item.context.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionWorkContext}) => projectInteractionWorkContext)], ["work_item.replan_settlement.project", lazyHandler(() => import("./work_items/replan_settlement.ts"), ({projectReplanSettlementContract}) => projectReplanSettlementContract)], diff --git a/loopx/control_plane/goals/goal_vision_wait.py b/loopx/control_plane/goals/goal_vision_wait.py index f453c5a6ec..52188642a8 100644 --- a/loopx/control_plane/goals/goal_vision_wait.py +++ b/loopx/control_plane/goals/goal_vision_wait.py @@ -76,9 +76,19 @@ def _compact_resume_condition(value: Any) -> dict[str, Any]: "target_claimed_by", "capability", "provider", + "scheduled_for", + "evaluated_at", + "clock_provider", + "material_change", + "material_change_generation", + "invalid_state", ) if value.get(key) is not None } + if value.get("kind") == "resume_at": + # Keep the complete positive clock proof for the typed interaction + # owner, including the explicit absence of an already-triggered receipt. + compact["resume_receipt"] = value.get("resume_receipt") compact["satisfied"] = False return compact diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index acf9215976..15b63ac929 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -81,11 +81,18 @@ class InteractionContractPacket(_InteractionContractRequired, total=False): def _blocked_successor_wait_observation_required(payload: Mapping[str, object]) -> bool: - return bool( - payload.get("effective_action") - == AgentScopeFrontierAction.AGENT_SCOPE_WAIT.value - and exact_blocked_successor_wait_state(payload) - ) + wait = exact_blocked_successor_wait_state(payload) + if payload.get("effective_action") != AgentScopeFrontierAction.AGENT_SCOPE_WAIT.value or not wait: + return False + identity = payload.get("agent_identity") + agent_id = identity.get("agent_id") if isinstance(identity, Mapping) else None + result = effect_runtime_result("work_item.blocked_wait_observation.project", { + "effective_action": payload.get("effective_action"), + "agent_id": agent_id, + "vision_wait_state": wait, + "heartbeat_receipt": payload.get("heartbeat_receipt"), + }) + return result["required"] is True def _user_todo_item_is_explicitly_non_gating(item: dict[str, Any]) -> bool: diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index ad4f1abb97..f1c79ab7e2 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -14,6 +14,38 @@ import { EffectiveAction } from "../quota/effective_action.generated.ts"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; import type { JsonObject } from "../effect_program.ts"; +import { AgentScopeFrontierAction } from "../agents/agent_scope_frontier.generated.ts"; +import { resumeConditionHasKnownPendingTarget } from "../todos/resume_condition.ts"; + +/** A clock-backed wait needs no observation before its scheduled instant. + * This read model never relaxes an existing Turn binding or settlement fence. + * Dependency waits retain their bounded observation/replan policy. */ +export function projectBlockedSuccessorWaitObservation(request: JsonObject): JsonObject { + const wait = jsonObject(request.vision_wait_state); + const exact = request.effective_action === AgentScopeFrontierAction.AGENT_SCOPE_WAIT && + wait?.schema_version === "goal_vision_wait_state_v0" && wait.state === "waiting" && + wait.reason_code === "exact_blocked_successor" && wait.automatic_resume === true && + typeof wait.selected_todo_id === "string" && !!wait.selected_todo_id && + typeof wait.resume_when === "string" && !!wait.resume_when; + if (!exact || !wait) return {required: false}; + const receipt = jsonObject(request.heartbeat_receipt); + const unbound = request.heartbeat_receipt == null || (!!receipt && + ["committed", "replayed"].includes(String(receipt.status)) && + !("settlement_identity" in receipt) && !("pending_action_selection" in receipt)); + const condition = jsonObject(wait.resume_condition); + if (unbound && wait.agent_id === request.agent_id && wait.waiting_todo_count === 1 && + ["open", "deferred"].includes(String(wait.selected_todo_status)) && + condition?.kind === "resume_at") { + try { + if (resumeConditionHasKnownPendingTarget(condition, { + todo_id: wait.selected_todo_id, resume_when: wait.resume_when, + })) return {required: false}; + } catch { + // An incomplete or malformed clock proof cannot relax the contract. + } + } + return {required: true}; +} export const INTERACTION_CONTRACT_SCHEMA_VERSION = "loopx_interaction_contract_v0"; diff --git a/tests/control_plane/test_dated_vision_wait_cli.py b/tests/control_plane/test_dated_vision_wait_cli.py new file mode 100644 index 0000000000..58992f95ab --- /dev/null +++ b/tests/control_plane/test_dated_vision_wait_cli.py @@ -0,0 +1,201 @@ +"""Real heartbeat coverage for an unbound, automatically scheduled Vision wait.""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile + +import pytest + +from loopx.control_plane.effect_runtime import restart_effect_runtime + +REPO = Path(__file__).resolve().parents[2] +GOAL = "scheduled-vision-fixture" +AGENT = "agent-scheduled" +TODO = "todo_scheduled_delivery" +FUTURE = "resume_at:2099-01-01T00:00:00Z" + + +@pytest.fixture +def isolated_effect(tmp_path: Path, monkeypatch): + # Runtime-root alone does not isolate a managed Effect process. + for variable in ("TMPDIR", "TEMP", "TMP"): + monkeypatch.setenv(variable, str(tmp_path)) + monkeypatch.setattr(tempfile, "tempdir", str(tmp_path)) + yield + restart_effect_runtime() + + +def fixture(root: Path, resume: str = FUTURE) -> tuple[Path, Path, Path]: + project, runtime = root / "project", root / "runtime" + state = project / f".codex/goals/{GOAL}/ACTIVE_GOAL_STATE.md" + state.parent.mkdir(parents=True) + state.write_text( + "---\nstatus: active\nowner_mode: goal\n" + 'objective: "Deliver the scheduled public fixture."\n' + "updated_at: 2026-01-01T00:00:00Z\n---\n\n# Scheduled fixture\n\n" + "## Objective\n\nDeliver the scheduled public fixture.\n\n" + "## Next Action\n\n- Resume at the scheduled instant.\n\n" + "## Agent Todo\n\n- [ ] [P1] Deliver the scheduled fixture.\n" + f" \n", + encoding="utf-8", + ) + registry = project / ".loopx/registry.json" + registry.parent.mkdir() + registry.write_text(json.dumps({ + "schema_version": "0.1", "common_runtime_root": str(runtime), + "goals": [{"id": GOAL, "repo": str(project), "status": "active", + "domain": "scheduled-vision-fixture", + "adapter": {"kind": "read_only_project_map_v0", "status": "connected-read-only"}, + "authority_sources": [], + "state_file": str(state.relative_to(project)), + "coordination": {"registered_agents": [AGENT], "agent_model": "peer_v1"}, + "quota": {"compute": 1.0, "window_hours": 24, "allowed_slots": 20}}], + }), encoding="utf-8") + runs = runtime / f"goals/{GOAL}/runs" + runs.mkdir(parents=True) + run = {"goal_id": GOAL, "agent_id": AGENT, + "classification": "scheduled_vision_fixture", "generated_at": "2026-01-01T00:00:00Z", + "progress_scope": "agent_lane", "agent_vision": { + "schema_version": "goal_vision_replan_contract_v0", "agent_id": AGENT, + "state": "vision_drift_detected", "todo_delta": [f"activate:{TODO}"], + "vision_patch": {"acceptance_summary": "Deliver this same scheduled fixture.", + "advancement_policy": "as_needed", + "replan_trigger_summary": "The scheduled acceptance remains open."}}} + (runs / "vision.json").write_text(json.dumps(run), encoding="utf-8") + (runs / "index.jsonl").write_text(json.dumps({**run, "json_path": str(runs / "vision.json")}) + "\n", encoding="utf-8") + return registry, runtime, state + + +def cli(registry: Path, runtime: Path, *args: str) -> tuple[int, dict]: + result = subprocess.run( + [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", *args], + cwd=REPO, env={**os.environ, "PYTHONPATH": str(REPO), "LOOPX_USAGE_PING": "0"}, + capture_output=True, text=True, check=False, timeout=90, + ) + return result.returncode, json.loads(result.stdout) + + +@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +def test_future_vision_wait_is_quiet_without_binding_or_poll(tmp_path, isolated_effect, monkeypatch, provider): + from canonical_authority_fixture import initialize_canonical_authority + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + + registry, runtime, state = fixture(tmp_path) + if provider != "legacy": + if provider == "sqlite": + monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite") + rc, listed = cli(registry, runtime, "todo", "list", "--goal-id", GOAL) + assert rc == 0, listed + projection = build_todo_runtime_shadow_projection(goal_id=GOAL, handoff_mode="soft_claim", todos=listed["todos"]) + initialize_canonical_authority(runtime, GOAL, projection, state_path=state, provider=provider) + original_state = state.read_bytes() + args = ("quota", "should-run", "--codex-app", "--goal-id", GOAL, + "--agent-id", AGENT, "--turn-instance-id", "scheduled-before-due") + rc, first = cli(registry, runtime, *args) + assert rc == 0, json.dumps(first) + assert first["decision"] == "agent_scope_wait" + assert first["heartbeat_receipt"]["status"] == "committed" + assert "settlement_identity" not in first["heartbeat_receipt"] + # An explicit unbound poll still fails closed; the repair changes guidance, + # not the command's identity qualification. + poll_rc, poll = cli(registry, runtime, "quota", "monitor-poll", "--execute", + "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT, + "--turn-instance-id", "scheduled-before-due") + assert poll_rc == 1, poll + assert poll["error_code"] == "heartbeat_receipt_identity_conflict" + contract = first["interaction_contract"] + assert contract["agent_channel"]["must_attempt"] is False + assert contract["agent_channel"]["quiet_noop_allowed"] is True + assert contract["agent_channel"]["delivery_allowed"] is False + assert not any("monitor-poll" in action for action in contract["cli_channel"]["next_cli_actions"]) + assert first["vision_wait_state"]["resume_when"] == FUTURE + assert first["goal_frontier_projection"]["replan_required"] is False + assert first["heartbeat_receipt"].get("closeout_required") is not True + rc, replay = cli(registry, runtime, *args) + assert rc == 0, replay + assert replay["heartbeat_receipt"]["status"] == "replayed" + assert replay["interaction_contract"]["agent_channel"]["must_attempt"] is False + for turn in ("scheduled-next-wake", "scheduled-third-wake"): + rc, later = cli(registry, runtime, *args[:-1], turn) + assert rc == 0, later + assert later["effective_action"] == "agent_scope_wait" + assert later["interaction_contract"]["agent_channel"]["must_attempt"] is False + assert later["vision_wait_state"]["resume_when"] == FUTURE + assert later["goal_frontier_projection"]["replan_required"] is False + assert state.read_bytes() == original_state + rc, listed = cli(registry, runtime, "todo", "list", "--goal-id", GOAL) + assert rc == 0, listed + scheduled = next(todo for todo in listed["todos"] if todo["todo_id"] == TODO) + assert scheduled["status"] == "deferred" + assert scheduled["resume_when"] == FUTURE + runs = [json.loads(line) for line in (runtime / f"goals/{GOAL}/runs/index.jsonl").read_text().splitlines()] + assert not any(run.get("classification") in ("quota_monitor_poll", "quota_slot_spent") for run in runs) + + +def test_due_vision_wait_restores_explicit_resume(tmp_path, isolated_effect): + registry, runtime, state = fixture(tmp_path, "resume_at:2026-01-01T00:00:00Z") + original = state.read_bytes() + rc, guard = cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "scheduled-due") + assert rc == 0, guard + assert guard["effective_action"] == "successor_replan_required" + assert guard["selected_todo"]["todo_id"] == TODO + assert guard["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is False + assert any("todo update" in action and "--clear-resume-when" in action + for action in guard["interaction_contract"]["cli_channel"]["next_cli_actions"]) + assert "vision_wait_state" not in guard + assert state.read_bytes() == original + + +def test_independent_work_is_not_silenced_by_scheduled_wait(tmp_path, isolated_effect): + registry, runtime, state = fixture(tmp_path) + with state.open("a", encoding="utf-8") as output: + output.write("\n- [ ] [P2] Validate independent public evidence.\n" + f" \n") + rc, guard = cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "independent-work") + assert rc == 0, guard + assert guard["selected_todo"]["todo_id"] == "todo_independent" + assert guard["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True + + +def test_bound_monitor_keeps_exact_turn_and_agent_fences(tmp_path, isolated_effect): + registry, runtime, state = fixture(tmp_path) + with state.open("a", encoding="utf-8") as output: + output.write("\n- [ ] [P1-monitor] Observe a public result.\n" + f" \n") + rc, guard = cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "bound-monitor") + assert rc == 0, guard + assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == "todo_result_monitor" + assert guard["interaction_contract"]["agent_channel"]["must_attempt"] is True + original = state.read_bytes() + for turn, agent in [("stale-monitor-turn", AGENT), ("bound-monitor", "agent-foreign")]: + rc, refused = cli(registry, runtime, "quota", "monitor-poll", "--execute", "--codex-app", + "--goal-id", GOAL, "--agent-id", agent, "--turn-instance-id", turn, + "--todo-id", "todo_result_monitor", "--result-hash", "public-result-1") + assert rc == 1, refused + assert refused["error_code"] == "heartbeat_receipt_identity_conflict" + assert state.read_bytes() == original + rc, poll = cli(registry, runtime, "quota", "monitor-poll", "--execute", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "bound-monitor", + "--todo-id", "todo_result_monitor", "--result-hash", "public-result-1") + assert rc == 0, poll + assert poll["appended"] is True + assert poll["material_change"] is False + rc, after = cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "after-monitor") + assert rc == 0, after + assert after["effective_action"] != "unsettled_host_turn_recovery" + assert after["interaction_contract"]["agent_channel"]["must_attempt"] is False diff --git a/tests/control_plane_ts/interaction_contract.test.ts b/tests/control_plane_ts/interaction_contract.test.ts index 9c4b21140f..4bc5d2963e 100644 --- a/tests/control_plane_ts/interaction_contract.test.ts +++ b/tests/control_plane_ts/interaction_contract.test.ts @@ -9,8 +9,47 @@ import { projectInteractionConsistencyValidation, projectInteractionRequiredReads, projectInteractionWorkContext, + projectBlockedSuccessorWaitObservation, type AgentInteractionChannel, } from "../../loopx/control_plane/work_items/interaction_contract.ts"; +import { evaluateTodoResumeConditions } from "../../loopx/control_plane/todos/resume_condition.ts"; + +test("only a positively qualified unbound scheduled wait can omit observation", () => { + const todo = {todo_id: "todo_scheduled", status: "deferred", role: "agent", + task_class: "advancement_task", resume_when: "resume_at:2099-01-01T00:00:00Z"}; + const evaluated = evaluateTodoResumeConditions({schema_version: "todo_resume_evaluation_request_v0", + items: [todo], source_items: [], evaluated_at: "2026-09-01T00:00:00Z"}); + const condition = (evaluated.conditions as JsonObject[])[0].condition as JsonObject; + const wait = {schema_version: "goal_vision_wait_state_v0", state: "waiting", + reason_code: "exact_blocked_successor", automatic_resume: true, agent_id: "agent-a", waiting_todo_count: 1, + selected_todo_id: todo.todo_id, selected_todo_status: "deferred", + resume_when: todo.resume_when, resume_condition: condition}; + const request = {effective_action: "agent_scope_wait", agent_id: "agent-a", vision_wait_state: wait}; + assert.deepEqual(projectBlockedSuccessorWaitObservation(request), {required: false}); + for (const status of ["committed", "replayed"]) { + assert.deepEqual(projectBlockedSuccessorWaitObservation({...request, + heartbeat_receipt: {status, settlement_binding_owed: true}}), {required: false}); + } + for (const patch of [ + {heartbeat_receipt: {status: "committed", settlement_identity: {todo_id: todo.todo_id}}}, + {heartbeat_receipt: {status: "committed", settlement_identity: {todo_id: "todo_other"}}}, + {heartbeat_receipt: {status: "committed", pending_action_selection: {todo_id: "todo_other"}}}, + {heartbeat_receipt: {status: "unknown"}}, + {heartbeat_receipt: "invalid"}, + {agent_id: "agent-foreign"}, + {vision_wait_state: {...wait, waiting_todo_count: 2}}, + {vision_wait_state: {...wait, selected_todo_status: "done"}}, + {vision_wait_state: {...wait, resume_condition: {...condition, satisfied: true}}}, + {vision_wait_state: {...wait, resume_condition: {...condition, evaluated_at: "2099-01-01T00:00:00Z"}}}, + {vision_wait_state: {...wait, resume_condition: {...condition, clock_provider: "author"}}}, + {vision_wait_state: {...wait, resume_condition: {...condition, scheduled_for: "2098-01-01T00:00:00Z"}}}, + {vision_wait_state: {...wait, resume_when: "resume_at:2099-01-01T00:00:00"}}, + {vision_wait_state: {...wait, resume_condition: {kind: "resume_at", satisfied: false}}}, + {vision_wait_state: {...wait, resume_when: "todo_done:todo_dependency", + resume_condition: {kind: "todo_done", satisfied: false}}}, + ]) assert.deepEqual(projectBlockedSuccessorWaitObservation({...request, ...patch}), {required: true}); + assert.deepEqual(projectBlockedSuccessorWaitObservation({...request, effective_action: "normal_run"}), {required: false}); +}); // The host-facing type cannot represent delivery without an agent attempt. // @ts-expect-error delivery requires must_attempt=true