diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index e85181c8e6..6a0da627ca 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -1503,8 +1503,19 @@ That hint is not a delivery permission. It is the cross-runtime wait policy: `run_now` keeps the active cadence for required work; `backoff_waiting_for_user` slows Codex App and stops CLI/Claude loops after repeated unchanged polls; `backoff_until_reassigned` handles peer reassignment waits without dropping -agent-to-agent handoff cadence too quickly; -`backoff_until_material_transition` handles monitor-only quiet polls; and +agent-to-agent handoff cadence too quickly. + +`backoff_until_state_change` also keeps a verified, automatically resumable +`resume_at` successor alive through quiet rechecks. The scheduler bounds its +backoff by the remaining scheduled wait, subject to host cadence floors, and +does not apply the ordinary three-unchanged-poll stop to this clock-backed wait. +This requires the current agent's exact pending successor and an unbound Turn; +missing or invalid clock evidence, another agent's claim, and manual/dependency +waits retain their existing supervision. A fresh due-time guard still owns +admission; keeping the host alive grants no model wake, quota spend, new time +budget or permission. Explicit stop, pause and user gates still take precedence. + +`backoff_until_material_transition` handles monitor-only quiet polls, and `backoff_until_fresh_evidence` handles mapped or post-handoff no-op waits. `preserve_current_schedule` handles `heartbeat_settled_skip`: settlement closes one Turn, not the Goal or its remaining Todos. Replay keeps the installed host diff --git a/loopx/control_plane/scheduler/arbitration.py b/loopx/control_plane/scheduler/arbitration.py index d52e875f92..7d866f75b7 100644 --- a/loopx/control_plane/scheduler/arbitration.py +++ b/loopx/control_plane/scheduler/arbitration.py @@ -5,6 +5,8 @@ from enum import Enum from typing import Any +from ..goals.goal_vision_wait import exact_blocked_successor_wait_state + SCHEDULER_ARBITRATION_SCHEMA_VERSION = "scheduler_arbitration_v0" @@ -28,6 +30,7 @@ class SchedulerArbitration: reason_code: str mode: str errors: tuple[str, ...] = () + scheduled_recheck_after_seconds: int | None = None @property def ok(self) -> bool: @@ -54,12 +57,16 @@ def from_projection(cls, projection: Any) -> SchedulerArbitration: or not isinstance(projection.get("errors"), list) or any(not isinstance(error, str) for error in projection["errors"])): raise TypeError("TypeScript scheduler arbitration shape mismatch") + scheduled = projection.get("scheduled_recheck_after_seconds") + if scheduled is not None and (type(scheduled) is not int or scheduled <= 0): + raise TypeError("TypeScript scheduler scheduled recheck must be positive integer seconds") return cls(SchedulerDisposition(projection["disposition"]), projection["reason_code"], - projection["mode"], tuple(projection["errors"])) + projection["mode"], tuple(projection["errors"]), scheduled) def _scheduler_arbitration_facts( contract: Any, agent_scope_frontier_actions: Collection[str] = (), + *, payload: dict[str, Any] | None = None, ) -> dict[str, Any]: """Keep legacy Python scalar/Mapping codecs; transport only consumed facts.""" contract = contract if isinstance(contract, Mapping) else {} @@ -79,6 +86,14 @@ def _scheduler_arbitration_facts( wait = agent.get("wait") if isinstance(agent, Mapping) else None source = wait.get("source") if isinstance(wait, Mapping) else None channels["agent_channel"]["wait_source"] = source if isinstance(source, str) and len(source) <= 64 else None - return {"schema_version": contract.get("schema_version"), + facts = {"schema_version": contract.get("schema_version"), "mode": str(contract.get("mode") or "").strip(), **channels, "agent_scope_frontier_actions": [str(action) for action in agent_scope_frontier_actions]} + wait = exact_blocked_successor_wait_state(payload) + if wait: + identity = payload.get("agent_identity") + facts.update({"effective_action": payload.get("effective_action"), + "agent_id": identity.get("agent_id") if isinstance(identity, Mapping) else None, + "vision_wait_state": wait, + "heartbeat_receipt": payload.get("heartbeat_receipt")}) + return facts diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index 1f4a0d12c8..af6c85c657 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -1601,6 +1601,22 @@ def build_scheduler_hint( ) if arbitration.disposition == SchedulerDisposition.AGENT_SCOPE_WAIT: + if arbitration.scheduled_recheck_after_seconds is not None: + horizon = max(1, math.ceil(arbitration.scheduled_recheck_after_seconds / 60)) + progression = sorted({min(bucket, horizon) for bucket in (10, 20, 30, 60)}) + return builder.build( + action="backoff_until_state_change", + cadence_class="agent_scope_wait", + reason=( + "an automatic clock-backed successor is pending; keep quiet rechecks " + "until its scheduled instant, then follow fresh admission gates" + ), + codex_interval=progression[0], + codex_max=progression[-1], + cli_limit=None, + claude_limit=None, + cadence_progression_override=progression, + ) return builder.build( action="backoff_until_reassigned", cadence_class="agent_scope_wait", diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 7f94137b8f..938f9c2c7e 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1102,7 +1102,7 @@ def _interaction_reads_and_arbitration( selected = payload.get("selected_todo") or {} orchestration = payload.get("task_orchestration_contract") or {} projection_params = { - "arbitration_facts": _scheduler_arbitration_facts(contract, agent_scope_frontier_actions), + "arbitration_facts": _scheduler_arbitration_facts(contract, agent_scope_frontier_actions, payload=payload), "required_reads": [dict(item) for item in reads if isinstance(item, dict) and item.get("command")] if isinstance(reads, list) else [], "goal_id": payload.get("goal_id"), diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index ab21b6ee6d..c002947625 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -17,34 +17,44 @@ 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 { +function exactBlockedSuccessorWait(request: JsonObject): JsonObject | null { 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}; + return exact && wait ? wait : null; +} + +/** Share the existing clock proof; a Turn binding always retains supervision. */ +function pendingDatedSuccessor(request: JsonObject, wait: JsonObject): JsonObject | null { 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 && + if (unbound && typeof request.agent_id === "string" && !!request.agent_id && + 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}; + })) return condition; } catch { // An incomplete or malformed clock proof cannot relax the contract. } } - return {required: true}; + return null; +} + +/** 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 = exactBlockedSuccessorWait(request); + return {required: !!wait && pendingDatedSuccessor(request, wait) === null}; } export const INTERACTION_CONTRACT_SCHEMA_VERSION = @@ -171,9 +181,21 @@ export function projectSchedulerArbitration(value: unknown): JsonObject { errors.push(...interactionConsistencyIssues(facts).map(issue => issue.code)); const [disposition, reason] = schedulerDisposition(facts, modes, user.non_gating_notice === true && agent.wait_source === "turn_settlement"); + const wait = exactBlockedSuccessorWait(request); + const condition = !errors.length && disposition === "agent_scope_wait" && + !facts.must_attempt && !facts.user_required && facts.quiet_noop_allowed && + !facts.spend_allowed_now && !facts.spend_after_validation && wait && + (wait.selected_todo_claimed_by == null || wait.selected_todo_claimed_by === "" || + wait.selected_todo_claimed_by === request.agent_id) + ? pendingDatedSuccessor(request, wait) : null; + // The resume-condition owner has verified both timestamps and a future target. + // This private duration bounds host rechecks; it grants no execution/spend. + const scheduled = condition ? {scheduled_recheck_after_seconds: Math.ceil( + (Date.parse(String(condition.scheduled_for)) - Date.parse(String(condition.evaluated_at))) / 1000, + )} : {}; return {mode, errors: [...new Set(errors)], disposition: errors.length ? "consistency_repair" : disposition, - reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason}; + reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason, ...scheduled}; } function decodeUserChannel(value: unknown): UserInteractionChannel { diff --git a/tests/control_plane/scheduler_arbitration_fixture.py b/tests/control_plane/scheduler_arbitration_fixture.py index cb9f8427b4..5d05882294 100644 --- a/tests/control_plane/scheduler_arbitration_fixture.py +++ b/tests/control_plane/scheduler_arbitration_fixture.py @@ -19,7 +19,7 @@ def project_native_arbitration( "delivery_allowed": False, "selection_required": False, "has_replan": False, "settlement_only": False, "arbitration_facts": _scheduler_arbitration_facts( - payload.get("interaction_contract"), agent_scope_frontier_actions, + payload.get("interaction_contract"), agent_scope_frontier_actions, payload=payload, ), }) return SchedulerArbitration.from_projection(result["arbitration"]) diff --git a/tests/control_plane/test_dated_vision_wait_cli.py b/tests/control_plane/test_dated_vision_wait_cli.py index a409817c03..7e79954e71 100644 --- a/tests/control_plane/test_dated_vision_wait_cli.py +++ b/tests/control_plane/test_dated_vision_wait_cli.py @@ -151,9 +151,74 @@ def test_due_vision_wait_restores_explicit_resume(tmp_path, isolated_effect): 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 guard["scheduler_hint"]["action"] == "run_now" + assert all(limit is None for limit in guard["scheduler_hint"]["unchanged_poll"]["limits"].values()) assert state.read_bytes() == original +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_real_cli_worker_keeps_automatic_wait_alive_and_honors_explicit_stop( + tmp_path, isolated_effect, monkeypatch, provider, +): + from argparse import Namespace + from canonical_authority_fixture import initialize_canonical_authority + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + + monkeypatch.syspath_prepend(str(REPO / "scripts")) + import external_scheduler_worker as worker + + registry, runtime, state = fixture(tmp_path) + 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 + initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection( + goal_id=GOAL, handoff_mode="soft_claim", todos=listed["todos"], + ), state_path=state, provider=provider) + original_state = state.read_bytes() + packets, sleeps = [], [] + + def real_probe(command, **kwargs): + # Adapt process transport only: every reply is produced by the real CLI + # against the isolated authority, with a fresh Turn identity. + rc, packet = cli(registry, runtime, *command[command.index("quota"):], + "--turn-instance-id", f"scheduled-worker-{len(packets)}") + assert rc == 0, packet + packets.append(packet) + return packet + + def advance_host(seconds): + sleeps.append(seconds) + assert len(sleeps) <= 4, "explicit stop must end the actual worker" + if len(sleeps) == 4: + stopped = json.loads(registry.read_text()) + stopped["goals"][0]["activation_state"] = "stopped" + registry.write_text(json.dumps(stopped)) + + monkeypatch.setattr(worker, "run_quota_should_run", real_probe) + monkeypatch.setattr(worker, "_run_wake", lambda *a, **kw: pytest.fail("a pending wait cannot wake the model")) + args = Namespace(cli_bin="loopx", registry=str(registry), runtime_root=str(runtime), + runtime_profile="generic_cli", goal_id=GOAL, agent_id=AGENT, + state_file=str(tmp_path / "scheduler-state.json"), once=False, wake_cmd="unused", + error_backoff_seconds=5, quota_timeout_seconds=90, wake_timeout_seconds=None) + assert worker.run_worker(args, sleep=advance_host) == 0 + assert sleeps == [600, 1200, 1800, 3600] + assert len(packets) == 5 + for packet in packets[:-1]: + assert packet["should_run"] is False + assert packet["vision_wait_state"]["automatic_resume"] is True + hint = packet["scheduler_hint"] + assert hint["action"] == "backoff_until_state_change" + assert all(limit is None for limit in hint["unchanged_poll"]["limits"].values()) + assert set(hint["unchanged_poll"]["after_limits"].values()) == {"continue"} + assert hint["unchanged_poll"]["final_quota_replan_check_enabled"] is False + assert packets[-1]["scheduler_hint"]["action"] == "stop_until_explicit_resume" + assert packets[-1]["scheduler_hint"]["cadence_class"] == "goal_stopped" + assert state.read_bytes() == original_state + 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_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: diff --git a/tests/control_plane/test_scheduler_arbitration_transport.py b/tests/control_plane/test_scheduler_arbitration_transport.py index b678cbba26..58506da3c5 100644 --- a/tests/control_plane/test_scheduler_arbitration_transport.py +++ b/tests/control_plane/test_scheduler_arbitration_transport.py @@ -25,6 +25,14 @@ def contract() -> dict: "cli_channel": {"spend_allowed_now": False, "spend_after_validation": False}} +@pytest.mark.parametrize("invalid", [True, False, 0, -1, 1.5, "1800", []]) +def test_scheduled_recheck_codec_rejects_nonpositive_or_coerced_seconds(invalid): + with pytest.raises(TypeError, match="scheduled recheck"): + SchedulerArbitration.from_projection({"mode": "agent_scope_wait", "errors": [], + "disposition": "agent_scope_wait", "reason_code": "interaction_agent_scope_wait", + "scheduled_recheck_after_seconds": invalid}) + + def encoded_size(value: dict) -> int: return len(json.dumps(value, separators=(",", ":")).encode()) diff --git a/tests/control_plane/test_scheduler_interaction_arbitration.py b/tests/control_plane/test_scheduler_interaction_arbitration.py index 0366419f48..7b4ee2a9c3 100644 --- a/tests/control_plane/test_scheduler_interaction_arbitration.py +++ b/tests/control_plane/test_scheduler_interaction_arbitration.py @@ -564,3 +564,37 @@ def test_python_source_codec_does_not_coerce_non_boolean_authority(flag) -> None assert arbitration.disposition is SchedulerDisposition.CONSISTENCY_REPAIR assert arbitration.errors == ("interaction_contract.user_channel.action_required_must_be_boolean",) assert payload == original + + +@pytest.mark.parametrize("minutes,progression", [ + (1, [1]), (11, [10, 11]), (25, [10, 20, 25]), (61, [10, 20, 30, 60]), +]) +def test_automatic_dated_wait_bounds_rechecks_without_finite_stop(minutes, progression): + payload = _payload(mode="agent_scope_wait", should_run=False, user_required=False, + must_attempt=False, delivery_allowed=False, quiet_noop_allowed=True) + evaluated = datetime(2099, 1, 1, tzinfo=timezone.utc) + due = (evaluated + timedelta(minutes=minutes)).isoformat().replace("+00:00", "Z") + resume = f"resume_at:{due}" + payload["vision_wait_state"] = { + "schema_version": "goal_vision_wait_state_v0", "state": "waiting", + "reason_code": "exact_blocked_successor", "automatic_resume": True, + "agent_id": payload["agent_identity"]["agent_id"], "waiting_todo_count": 1, + "selected_todo_id": "todo_scheduled", "selected_todo_status": "deferred", + "resume_when": resume, "resume_condition": { + "schema_version": "todo_resume_condition_v0", "kind": "resume_at", + "resume_when": resume, "target": due, "satisfied": False, + "scheduled_for": due, "evaluated_at": "2099-01-01T00:00:00Z", + "clock_provider": "runtime_clock", "material_change": False, + "material_change_generation": 0, "resume_receipt": None, + }, + } + original = deepcopy(payload) + hint = build_scheduler_hint(payload, agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS, + scheduler_execution_context=scheduler_execution_context_for_runtime_profile("generic_cli"), + include_detail=True) + local = hint["cold_path_detail"]["local_scheduler"] + assert hint["action"] == "backoff_until_state_change" + assert local["example_progression_minutes"] == progression + assert all(limit is None for limit in hint["unchanged_poll"]["limits"].values()) + assert set(hint["unchanged_poll"]["after_limits"].values()) == {"continue"} + assert payload == original diff --git a/tests/control_plane_ts/scheduler_arbitration.test.ts b/tests/control_plane_ts/scheduler_arbitration.test.ts index 81b36f7e12..708a6a0efd 100644 --- a/tests/control_plane_ts/scheduler_arbitration.test.ts +++ b/tests/control_plane_ts/scheduler_arbitration.test.ts @@ -141,3 +141,67 @@ test("the required-read reply retains its reads and complete scheduler dispositi assert.deepEqual(result, {required_reads: [], arbitration: {mode: "bounded_delivery", disposition: "active_work", reason_code: "interaction_agent_attempt_required", errors: []}}); }); + +function datedWait(): JsonObject { + const resume = "resume_at:2099-01-01T01:00:00Z"; + return {...facts("agent_scope_wait", false, false, false, true), + effective_action: "agent_scope_wait", agent_id: "scheduled-agent", + heartbeat_receipt: {status: "committed"}, + vision_wait_state: {schema_version: "goal_vision_wait_state_v0", state: "waiting", + reason_code: "exact_blocked_successor", agent_id: "scheduled-agent", + automatic_resume: true, waiting_todo_count: 1, selected_todo_id: "todo_scheduled", + selected_todo_status: "deferred", selected_todo_claimed_by: "scheduled-agent", + resume_when: resume, resume_condition: {schema_version: "todo_resume_condition_v0", + kind: "resume_at", resume_when: resume, target: "2099-01-01T01:00:00Z", + satisfied: false, scheduled_for: "2099-01-01T01:00:00Z", evaluated_at: "2099-01-01T00:30:00Z", + clock_provider: "runtime_clock", material_change: false, material_change_generation: 0, + resume_receipt: null}}}; +} + +test("only an unbound pending automatic dated successor carries a scheduled recheck", () => { + const input = datedWait(); + const before = structuredClone(input); + assert.deepEqual(projectSchedulerArbitration(input), {mode: "agent_scope_wait", errors: [], + disposition: "agent_scope_wait", reason_code: "interaction_agent_scope_wait", + scheduled_recheck_after_seconds: 1800}); + assert.deepEqual(input, before); + for (const claimed of [null, ""]) { + assert.equal(projectSchedulerArbitration({...input, vision_wait_state: { + ...input.vision_wait_state as JsonObject, selected_todo_claimed_by: claimed, + }}).scheduled_recheck_after_seconds, 1800); + } +}); + +test("unproven waits cannot relax finite scheduler supervision", () => { + const input = datedWait(); + const wait = input.vision_wait_state as JsonObject; + const condition = wait.resume_condition as JsonObject; + const patches: JsonObject[] = [ + {effective_action: "agent_scope_exhausted"}, {agent_id: "other-agent"}, + {heartbeat_receipt: {status: "pending"}}, + {heartbeat_receipt: {status: "committed", settlement_identity: {todo_id: "todo_scheduled"}}}, + {heartbeat_receipt: {status: "replayed", pending_action_selection: {}}}, + ...[{automatic_resume: false}, {schema_version: "future"}, {state: "ready"}, + {waiting_todo_count: 2}, {selected_todo_status: "done"}, {selected_todo_claimed_by: "other-agent"}, + {agent_id: "other-agent"}, {resume_when: "resume_at:2098-01-01T00:00:00Z"}] + .map(patch => ({vision_wait_state: {...wait, ...patch}})), + ...[{satisfied: true}, {evaluated_at: condition.scheduled_for}, {evaluated_at: "invalid"}, + {clock_provider: "caller_clock"}, {material_change: true}, {resume_receipt: {}}, + {scheduled_for: "2098-01-01T00:00:00Z"}, {kind: "todo_done"}] + .map(patch => ({vision_wait_state: {...wait, resume_condition: {...condition, ...patch}}})), + ]; + for (const patch of patches) { + assert.equal(projectSchedulerArbitration({...input, ...patch}).scheduled_recheck_after_seconds, + undefined, JSON.stringify(patch)); + } + for (const higher of [facts("user_gate", true, false, false, false), + facts("terminal_no_followup", false, false, false, true), facts()]) { + const result = projectSchedulerArbitration({...input, ...higher}); + assert.equal(result.scheduled_recheck_after_seconds, undefined); + assert.notEqual(result.disposition, "agent_scope_wait"); + } + for (const field of ["spend_allowed_now", "spend_after_validation"]) { + assert.equal(projectSchedulerArbitration({...input, cli_channel: {[field]: true}}) + .scheduled_recheck_after_seconds, undefined); + } +});