From b8a837d19d650e307b56baad1f2a3778b648cabf Mon Sep 17 00:00:00 2001 From: song Date: Tue, 20 Oct 2026 14:21:43 +0800 Subject: [PATCH] fix(scheduler): isolate qualified dependency waits for main Rebuild PR #6883 from its own B2 changes on main 5853e151, removing the unrelated historical split ancestry. Preserve durable waits and scoped dispatch; keep the quota route owner below the existing statement limit. Verify configured acceptance across v1/v2 without weakening accepted, quiet, replay, or no-duplicate-dispatch assertions. This PR requires foundation #6784 before merge. The pure-main real feedback test intentionally remains a dependency gate; File and SQLite both pass in the foundation+B2+B4+B5 qualification tree. Original source: b74eb21b84eaab63509c4cbc54f04d44736b93d5 Original author: song Validated local source: ef657dd9f975a965582b45c3b824f5110767cec7 Validated tree: cc952d71ad9f2ba8101cdeda11e53b64beb95a34 Original history: codex/research-split-archive-20261020 Signed-off-by: song Signed-off-by: Codex --- .../goal-vision-replan-contract-v0.md | 23 ++ loopx/control_plane/goals/goal_vision_wait.py | 9 + .../control_plane/quota/should_run_packet.py | 46 ++- loopx/control_plane/scheduler/arbitration.py | 6 +- .../control_plane/scheduler/scheduler_hint.py | 25 +- loopx/control_plane/todos/resume_condition.py | 2 + loopx/control_plane/todos/resume_condition.ts | 15 +- loopx/control_plane/todos/resume_planning.ts | 7 +- .../work_items/action_portfolio.py | 12 + .../work_items/action_portfolio.ts | 12 +- .../work_items/action_selection_admission.ts | 8 +- .../work_items/action_selection_contract.py | 7 +- .../work_items/interaction_contract.ts | 61 ++- .../test_dependency_dispatch_cli.py | 302 +++++++++++++++ .../test_dependency_feedback_cli.py | 245 ++++++++++++ .../test_dependency_vision_wait_cli.py | 363 ++++++++++++++++++ .../test_goal_vision_blocked_successor.py | 9 +- .../test_scheduler_arbitration_transport.py | 15 + .../test_user_gate_lane_progress.py | 3 + .../control_plane_ts/action_portfolio.test.ts | 27 ++ .../action_selection_admission.test.ts | 25 ++ .../interaction_contract.test.ts | 59 +++ .../control_plane_ts/resume_planning.test.ts | 27 ++ .../scheduler_arbitration.test.ts | 48 +++ 24 files changed, 1309 insertions(+), 47 deletions(-) create mode 100644 tests/control_plane/test_dependency_dispatch_cli.py create mode 100644 tests/control_plane/test_dependency_feedback_cli.py create mode 100644 tests/control_plane/test_dependency_vision_wait_cli.py diff --git a/docs/reference/protocols/goal-vision-replan-contract-v0.md b/docs/reference/protocols/goal-vision-replan-contract-v0.md index 73df49e9a9..48a3d9a650 100644 --- a/docs/reference/protocols/goal-vision-replan-contract-v0.md +++ b/docs/reference/protocols/goal-vision-replan-contract-v0.md @@ -1122,6 +1122,29 @@ another uncovered binding. When there is no other selectable advancement, quota/ `vision_acceptance_gap` is deferred while that read model is active, so the agent can remain quiet instead of inventing duplicate successor work. +For a single exact successor without a bound heartbeat Turn, the scheduler can +keep inexpensive source rechecks alive beyond the unchanged-poll limit when +the resume owner supplies positive pending evidence. Alongside the existing +clock-backed `resume_at` case, `todo_done` requires an identified, unarchived +source of the supported task class that is still open or deferred. +`monitor_changed` requires the exact open Monitor, the successor's matching +generation baseline, and a current generation that has not advanced. The +Monitor source must also carry a valid next-due time, the runtime clock +observation, and explicit expiry evidence: either no expiry or an expiry later +than both the observation and the next due time. These source facts are +projected before compaction, including explicit null values that distinguish +known absence of expiry or archival from missing evidence. + +This quiet path grants no delivery or spend. Missing sources, incomplete proof, +an unscheduled or expired Monitor, and an existing Turn or settlement binding +retain their original supervision. Supplying a new Turn id to a quota command +is a bound operation, not an unbound scheduler recheck. Monitor-only waits keep +their existing schedule-based cadence caps; when a Monitor becomes due, fresh +quota admission must select its work before quiet waiting. A material poll then +advances the original generation fence so the successor can resume. Goal stop, +user gates and independently runnable advancement retain their existing +priority. + This is a read model, not a stored vision or todo lifecycle state. When the condition becomes ready, normal open-todo or deferred-successor routing resumes automatically and the active vision remains available for acceptance auditing. diff --git a/loopx/control_plane/goals/goal_vision_wait.py b/loopx/control_plane/goals/goal_vision_wait.py index 52188642a8..73844bc876 100644 --- a/loopx/control_plane/goals/goal_vision_wait.py +++ b/loopx/control_plane/goals/goal_vision_wait.py @@ -74,6 +74,7 @@ def _compact_resume_condition(value: Any) -> dict[str, Any]: "target_source_section", "target_task_class", "target_claimed_by", + "target_next_due_at", "capability", "provider", "scheduled_for", @@ -81,10 +82,17 @@ def _compact_resume_condition(value: Any) -> dict[str, Any]: "clock_provider", "material_change", "material_change_generation", + "baseline_generation", "invalid_state", ) if value.get(key) is not None } + if "target_archive_state" in value and value["target_archive_state"] is None: + # An explicitly unarchived source is positive evidence. An omitted + # source field must remain omitted for the typed resume owner. + compact["target_archive_state"] = None + if "target_expires_at" in value: + compact["target_expires_at"] = value["target_expires_at"] 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. @@ -281,6 +289,7 @@ def build_goal_vision_wait_state( "selected_todo_priority": selected.get("priority"), "selected_todo_claimed_by": selected.get("claimed_by"), "resume_when": selected.get("resume_when"), + "resume_monitor_generation": selected.get("resume_monitor_generation"), "resume_condition": _compact_resume_condition( selected.get("resume_condition") ), diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 4ebc286601..cec943ee86 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -470,10 +470,12 @@ def _resolve_agent_lane_delivery_route( delivery_preemptions: list[str], should_run: bool, normal_delivery_allowed: bool, -) -> dict[str, Any] | None: - """Project candidates once, then let TypeScript own their delivery route.""" +) -> tuple[dict[str, Any] | None, bool]: + """Return the selected route with its typed admission decision.""" if prepared.requested_action_todo_id is not None: + scoped_fallback = prepared.scoped_user_gate_fallback or {} + scoped_selected = scoped_fallback.get("selected_executable") or {} qualification = qualify_action_selection_from_inventory( requested_todo_id=prepared.requested_action_todo_id, candidate=prepared.requested_action_candidate, @@ -481,18 +483,27 @@ def _resolve_agent_lane_delivery_route( should_run=should_run, normal_delivery_allowed=normal_delivery_allowed, delivery_preemptions=delivery_preemptions, + scoped_fallback_todo_id=normalize_todo_id(scoped_selected.get("todo_id")), + quota_state=prepared.state, + goal_health_ok=prepared.goal_health_ok, ) prepared.action_selection_qualification = qualification + if (prepared.action_selection_qualification or {}).get("state") == "qualified": + # The typed selection owner may admit this exact scoped fallback while + # the broader user-gated delivery lane remains closed. Keep selected + # workspace/boundary guards active without reopening normal delivery. + should_run = True + if isinstance(prepared.guarded_agent_lane_next_action, dict): # Workspace/scope guards can change admission after initial selection. # Keep the selected identity frozen, but report the final qualification. - return prepared.guarded_agent_lane_next_action + return prepared.guarded_agent_lane_next_action, should_run if prepared.requested_action_todo_id is not None: qualification = prepared.action_selection_qualification or {} if qualification.get("state") != "qualified": - return None + return None, should_run if not isinstance(prepared.requested_action_candidate, dict): raise RuntimeError( "qualified action selection is missing its projected candidate" @@ -501,7 +512,7 @@ def _resolve_agent_lane_delivery_route( **prepared.requested_action_candidate, "selected_by": "agent_action_selection", "selection_binding": "pending_action_selection", - } + }, should_run item = prepared.item fallback = prepared.receipt_bound_agent_next_action @@ -548,7 +559,7 @@ def _resolve_agent_lane_delivery_route( (prepared.agent_identity or {}).get("agent_id") ) if not delivery_agent_id: - return fallback + return fallback, should_run continuity_todo = prepared.delivery_continuity_todo delivery_anchor = prepared.delivery_continuity_anchor @@ -567,7 +578,7 @@ def _resolve_agent_lane_delivery_route( and fallback is None and replan_candidate is None ): - return None + return None, should_run delivery_route = evaluate_delivery_route( agent_id=delivery_agent_id, previous_todo_id=( @@ -668,7 +679,7 @@ def _resolve_agent_lane_delivery_route( # The router may find an independent successor after the bound Todo # becomes unavailable. That successor cannot replace an already # committed settlement identity inside the same heartbeat Turn. - return None + return None, should_run boundary = delivery_route.get("boundary") if ( @@ -679,8 +690,8 @@ def _resolve_agent_lane_delivery_route( return { **selected_action, "delivery_boundary": boundary["delivery_boundary"], - } - return selected_action + }, should_run + return selected_action, should_run def _selected_deferred_replan_frontier( @@ -1106,7 +1117,7 @@ def _resolve_quota_should_run_route( capability_repair_allowed=capability_repair_allowed, workspace_repair_allowed=workspace_repair_allowed, ) - agent_lane_next_action = _resolve_agent_lane_delivery_route( + agent_lane_next_action, should_run = _resolve_agent_lane_delivery_route( prepared, due_monitor_attempt=due_monitor_attempt, receipt_bound_replan_decision=receipt_bound_replan_decision, @@ -1141,10 +1152,15 @@ def _resolve_quota_should_run_route( agent_todo_summary=prepared.agent_todo_summary, agent_lane_next_action=agent_lane_next_action, work_lane_contract=prepared.work_lane_contract, - candidate_should_run=bool( - (should_run and normal_delivery_allowed) - or ready_deferred_resume_candidates - ), + # A satisfied dependency is readiness, not admission. The + # existing scoped fallback above qualifies an independent + # successor under a user gate; an unqualified ready row must + # not resurrect delivery refused by the current quota owner. + candidate_should_run=bool(should_run and ( + normal_delivery_allowed + or (ready_deferred_resume_candidates + and prepared.goal_health_ok and state == "eligible") + )), ) ) if ( diff --git a/loopx/control_plane/scheduler/arbitration.py b/loopx/control_plane/scheduler/arbitration.py index 7d866f75b7..85d5101d9e 100644 --- a/loopx/control_plane/scheduler/arbitration.py +++ b/loopx/control_plane/scheduler/arbitration.py @@ -31,6 +31,7 @@ class SchedulerArbitration: mode: str errors: tuple[str, ...] = () scheduled_recheck_after_seconds: int | None = None + pending_dependency_wait: bool = False @property def ok(self) -> bool: @@ -60,8 +61,11 @@ def from_projection(cls, projection: Any) -> SchedulerArbitration: 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") + pending_dependency = projection.get("pending_dependency_wait", False) + if type(pending_dependency) is not bool: + raise TypeError("TypeScript scheduler pending dependency wait must be boolean") return cls(SchedulerDisposition(projection["disposition"]), projection["reason_code"], - projection["mode"], tuple(projection["errors"]), scheduled) + projection["mode"], tuple(projection["errors"]), scheduled, pending_dependency) def _scheduler_arbitration_facts( diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index af6c85c657..0f9044f6c4 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -1180,6 +1180,7 @@ def _monitor_bounded_wait_profile( cadence_class: str, default_interval_minutes: int, max_interval_minutes: int, + unchanged_poll_limit: int | None = 3, ) -> dict[str, Any]: """Cap a wait profile to the tightest continuous-monitor wakeup.""" @@ -1205,8 +1206,8 @@ def _monitor_bounded_wait_profile( "app_automation_initial_rrule": rrule_for_minutes(initial_interval), "app_automation_max_interval_minutes": max_interval_minutes, "unchanged_poll_backoff_multiplier": 2, - "local_scheduler_unchanged_poll_limit": 3, - "claude_code_loop_unchanged_poll_limit": 3, + "local_scheduler_unchanged_poll_limit": unchanged_poll_limit, + "claude_code_loop_unchanged_poll_limit": unchanged_poll_limit, **monitor_plan["reset_profile"], } if isinstance(monitor_plan, dict) @@ -1617,6 +1618,20 @@ def build_scheduler_hint( claude_limit=None, cadence_progression_override=progression, ) + if arbitration.pending_dependency_wait: + return builder.build( + action="backoff_until_state_change", + cadence_class="agent_scope_wait", + reason=( + "an exact successor has a registered pending dependency; keep quiet " + "source rechecks and follow fresh admission gates when it changes" + ), + codex_interval=10, + codex_max=60, + cli_limit=None, + claude_limit=None, + cadence_progression_override=[10, 20, 30, 60], + ) return builder.build( action="backoff_until_reassigned", cadence_class="agent_scope_wait", @@ -1634,11 +1649,13 @@ def build_scheduler_hint( ) if arbitration.disposition == SchedulerDisposition.MONITOR_WAIT: + unchanged_limit = None if arbitration.pending_dependency_wait else 3 monitor_profile = _monitor_bounded_wait_profile( payload, cadence_class="monitor_wait", default_interval_minutes=MONITOR_WAIT_HOST_FLOOR_MINUTES, max_interval_minutes=60, + unchanged_poll_limit=unchanged_limit, ) return builder.build( action="backoff_until_material_transition", @@ -1647,8 +1664,8 @@ def build_scheduler_hint( "monitor-only quiet polls should remain alive but use a slower " "cadence until material evidence, a blocker, or replan obligation appears" ), - cli_limit=3, - claude_limit=3, + cli_limit=unchanged_limit, + claude_limit=unchanged_limit, **monitor_profile, ) diff --git a/loopx/control_plane/todos/resume_condition.py b/loopx/control_plane/todos/resume_condition.py index 12ed1e69a5..d272d5904a 100644 --- a/loopx/control_plane/todos/resume_condition.py +++ b/loopx/control_plane/todos/resume_condition.py @@ -78,6 +78,8 @@ "resume_ready", "resume_monitor_generation", "material_change_generation", + "next_due_at", + "expires_at", ) diff --git a/loopx/control_plane/todos/resume_condition.ts b/loopx/control_plane/todos/resume_condition.ts index 06fbd79f19..d9d936c577 100644 --- a/loopx/control_plane/todos/resume_condition.ts +++ b/loopx/control_plane/todos/resume_condition.ts @@ -159,6 +159,8 @@ function todoItem(value: unknown, label: string): TodoItem { "source_section", "claimed_by", "task_repository", + "next_due_at", + "expires_at", ] as const) { const normalized = optionalString(raw[field], `${label}.${field}`); if (normalized !== undefined) item[field] = normalized; @@ -544,6 +546,17 @@ function conditionFor( condition.target_todo_id = spec.target; condition.target_status = monitor?.status ?? null; condition.target_task_class = monitor?.task_class ?? null; + if (monitor) { + // Source scheduling facts qualify quiet rechecks separately from the + // material-generation fence. They do not make the successor ready. + condition.target_archive_state = monitor.archive_state ?? null; + condition.target_next_due_at = monitor.next_due_at ?? null; + condition.target_expires_at = monitor.expires_at ?? null; + if (evaluatedAt) { + condition.evaluated_at = evaluatedAt.iso; + condition.clock_provider = "runtime_clock"; + } + } condition.baseline_generation = baseline ?? null; condition.material_change_generation = generation; condition.generation_fence = "strictly_greater_than_baseline"; @@ -674,7 +687,7 @@ export function evaluateTodoResumeConditions(value: unknown): JsonObject { const spec = parseResumeWhen(item.resume_when); return spec?.kind === "resume_at" && (!requestedKinds || requestedKinds.has(spec.kind)); }); - const evaluatedAt = needsRuntimeClock ? resumeTimestamp(request.evaluated_at) : null; + const evaluatedAt = resumeTimestamp(request.evaluated_at); if (needsRuntimeClock && !evaluatedAt) { throw new EffectRuntimeRequestError( "todo_resume_evaluation_request.evaluated_at must be a timezone-aware RFC3339 timestamp", diff --git a/loopx/control_plane/todos/resume_planning.ts b/loopx/control_plane/todos/resume_planning.ts index edbabe3f02..143942450b 100644 --- a/loopx/control_plane/todos/resume_planning.ts +++ b/loopx/control_plane/todos/resume_planning.ts @@ -8,6 +8,7 @@ import { import { diagnoseTodoResumeCondition, type ResumeConditionDiagnosis, evaluateTodoResumeConditions, TODO_RESUME_EVALUATION_REQUEST_SCHEMA_VERSION, + resumeConditionHasKnownPendingTarget, } from "./resume_condition.ts"; export const RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v0"; @@ -135,10 +136,12 @@ function successorWaits(resumeBlocked: readonly Item[], deferredItems: readonly return ordered(unique([...resumeBlocked, ...deferredItems.filter((entry) => entry.payload.resume_ready === false)]) .filter((entry) => { const lane = claimLane(entry, agent); + const condition = jsonObject(entry.payload.resume_condition); return entry.payload.task_class === "advancement_task" && (lane === "current_agent" || lane === "unclaimed") && entry.resume !== null && - jsonObject(entry.payload.resume_condition)?.satisfied === false && - entry.diagnosis?.state === "pending" && entry.diagnosis.kind !== "monitor_changed"; + condition?.satisfied === false && entry.diagnosis?.state === "pending" && + (entry.diagnosis.kind !== "monitor_changed" || + resumeConditionHasKnownPendingTarget(condition, entry.payload)); }).map((entry) => ({ ...entry, payload: { ...entry.payload, resume_when: entry.resume, resume_ready: false, } }))).sort((a, b) => Number(b.claim === agent) - Number(a.claim === agent)); diff --git a/loopx/control_plane/work_items/action_portfolio.py b/loopx/control_plane/work_items/action_portfolio.py index 5ad65201a6..0546956bbd 100644 --- a/loopx/control_plane/work_items/action_portfolio.py +++ b/loopx/control_plane/work_items/action_portfolio.py @@ -125,6 +125,9 @@ def qualify_action_selection( should_run: bool, normal_delivery_allowed: bool, delivery_preemptions: list[str], + scoped_fallback_todo_id: str | None = None, + quota_state: str | None = None, + goal_health_ok: bool = False, ) -> dict[str, Any]: """Adapt current Python projections into the TS-owned selection reducer.""" @@ -140,6 +143,9 @@ def qualify_action_selection( "should_run": should_run, "normal_delivery_allowed": normal_delivery_allowed, "delivery_preemptions": delivery_preemptions, + "scoped_fallback_todo_id": scoped_fallback_todo_id, + "quota_state": quota_state, + "goal_health_ok": goal_health_ok, }, ) except EffectRuntimeRejected as exc: @@ -159,6 +165,9 @@ def qualify_action_selection_from_inventory( should_run: bool, normal_delivery_allowed: bool, delivery_preemptions: list[str], + scoped_fallback_todo_id: str | None = None, + quota_state: str | None = None, + goal_health_ok: bool = False, ) -> dict[str, Any]: """Resolve the requested task class before invoking the typed reducer.""" @@ -186,6 +195,9 @@ def qualify_action_selection_from_inventory( should_run=should_run, normal_delivery_allowed=normal_delivery_allowed, delivery_preemptions=delivery_preemptions, + scoped_fallback_todo_id=scoped_fallback_todo_id, + quota_state=quota_state, + goal_health_ok=goal_health_ok, ) diff --git a/loopx/control_plane/work_items/action_portfolio.ts b/loopx/control_plane/work_items/action_portfolio.ts index 12f18cc34a..1d56a9ee57 100644 --- a/loopx/control_plane/work_items/action_portfolio.ts +++ b/loopx/control_plane/work_items/action_portfolio.ts @@ -385,6 +385,10 @@ export function qualifyActionSelection(value: unknown): ActionSelectionQualifica : []; const shouldRun = request.should_run === true; const normalDeliveryAllowed = request.normal_delivery_allowed === true; + const scopedFallbackTodoId = optionalNonEmptyString( + request.scoped_fallback_todo_id, + "action_selection_qualification_request.scoped_fallback_todo_id", + ); if (rawCandidate === null || rawCandidate === undefined) { const requestedTaskClass = request.requested_task_class === null || request.requested_task_class === undefined @@ -416,7 +420,13 @@ export function qualifyActionSelection(value: unknown): ActionSelectionQualifica "action_selection_qualification_request.candidate.todo_id must match requested_todo_id", ); } - if (!shouldRun || !normalDeliveryAllowed || preemptions.length > 0) { + // The existing gate-scope owner already proved this single fallback's + // independence. A pending selection may bind that same current Todo, while + // health, budget states, receipts and every other hard lane keep priority. + const scopedFallbackAllowed = scopedFallbackTodoId === requestedTodoId + && request.quota_state === "operator_gate" && request.goal_health_ok === true + && preemptions.length === 1 && preemptions[0] === "delivery_not_allowed"; + if (!scopedFallbackAllowed && (!shouldRun || !normalDeliveryAllowed || preemptions.length > 0)) { return { schema_version: ACTION_SELECTION_QUALIFICATION_SCHEMA_VERSION, state: "deferred", diff --git a/loopx/control_plane/work_items/action_selection_admission.ts b/loopx/control_plane/work_items/action_selection_admission.ts index c8f64a8409..22e4a0b34a 100644 --- a/loopx/control_plane/work_items/action_selection_admission.ts +++ b/loopx/control_plane/work_items/action_selection_admission.ts @@ -32,13 +32,19 @@ export function currentActionSelectionAdmission( const pending = selected.selection_binding === "pending_action_selection"; const delivery = pending && packet.normal_delivery_allowed === true && agent.must_attempt && agent.delivery_allowed === true; + const fallback = jsonObject(jsonObject(packet.scoped_user_gate_fallback)?.selected_executable); + const scopedDelivery = pending && qualification.state === "qualified" + && selectionTodoId(qualification.requested_todo_id) === requestedTodoId + && selectionTodoId(fallback?.todo_id) === selectedId + && packet.safe_bypass_allowed === true && packet.safe_bypass_kind === "scoped_user_gate_fallback" + && agent.must_attempt && agent.delivery_allowed === true; const workspaceRepair = pending && packet.workspace_repair_allowed === true && packet.effective_action === EffectiveAction.AGENT_WORKSPACE_REPAIR && agent.must_attempt && agent.delivery_allowed === false; const currentWork = !pending && agent.must_attempt; return {selected_todo_id: selectedId, admitted: selectedId !== null && selectedId === requestedTodoId - && packet.ok === true && packet.should_run === true && (delivery || workspaceRepair || currentWork)}; + && packet.ok === true && packet.should_run === true && (delivery || scopedDelivery || workspaceRepair || currentWork)}; } export function actionSelectionNeedsRecovery(packet: JsonObject, agent: AgentAdmissionFacts): boolean { diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py index 9f2041f049..c136e118a1 100644 --- a/loopx/control_plane/work_items/action_selection_contract.py +++ b/loopx/control_plane/work_items/action_selection_contract.py @@ -160,7 +160,7 @@ def delivery_spend_allowed( def action_selection_context(payload: Mapping[str, Any]) -> dict[str, Any]: """Transport bounded current facts; TypeScript owns every admission decision.""" fields = ("ok", "should_run", "effective_action", "normal_delivery_allowed", "recovery_delivery_allowed", - "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed", "safe_bypass_allowed", + "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed", "safe_bypass_allowed", "safe_bypass_kind", "actionable_by_codex", "spend_allowed_now", "spend_after_validation") result = {key: payload[key] for key in fields if key in payload} for field, keys in ( @@ -175,6 +175,11 @@ def action_selection_context(payload: Mapping[str, Any]) -> dict[str, Any]: result[field]["todo_id"] = normalize_todo_id(raw.get("todo_id")) if field == "action_selection_qualification" and isinstance(raw.get("selected_todo"), Mapping): result[field]["selected_todo"] = {"todo_id": normalize_todo_id(raw["selected_todo"].get("todo_id"))} + fallback = payload.get("scoped_user_gate_fallback") + if isinstance(fallback, Mapping) and isinstance(fallback.get("selected_executable"), Mapping): + result["scoped_user_gate_fallback"] = {"selected_executable": { + "todo_id": normalize_todo_id(fallback["selected_executable"].get("todo_id")), + }} return result diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index c002947625..1eb409efa1 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -16,10 +16,13 @@ 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"; +import { parseTodoTimestampMicros } from "../runtime_timestamp.ts"; +import { monitorIsActive } from "../todos/summary_lanes.ts"; -function exactBlockedSuccessorWait(request: JsonObject): JsonObject | null { +function exactBlockedSuccessorWait(request: JsonObject, monitorWait = false): JsonObject | null { const wait = jsonObject(request.vision_wait_state); - const exact = request.effective_action === AgentScopeFrontierAction.AGENT_SCOPE_WAIT && + const exact = (request.effective_action === AgentScopeFrontierAction.AGENT_SCOPE_WAIT || + monitorWait && request.effective_action === EffectiveAction.MONITOR_QUIET_SKIP) && 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 && @@ -27,8 +30,24 @@ function exactBlockedSuccessorWait(request: JsonObject): JsonObject | null { return exact && wait ? wait : null; } -/** Share the existing clock proof; a Turn binding always retains supervision. */ -function pendingDatedSuccessor(request: JsonObject, wait: JsonObject): JsonObject | null { +function hasScheduledMonitorSource(condition: JsonObject): boolean { + const time = (value: unknown): number | null => { + const micros = typeof value === "string" ? parseTodoTimestampMicros(value) : null; + return micros === null ? null : Number(micros) / 1_000_000; + }; + const observed = time(condition.evaluated_at), due = time(condition.target_next_due_at); + const expiry = time(condition.target_expires_at); + if (condition.clock_provider !== "runtime_clock" || observed === null || due === null || + !(condition.target_archive_state === null || condition.target_archive_state === "active") || + condition.target_expires_at !== null && expiry === null) return false; + // A Monitor must still be able to reach a poll. The ordinary quota owner + // retains due-work priority; an expired or unscheduled source cannot keep + // an unchanged external worker alive indefinitely. + return monitorIsActive({expiresAt: expiry}, observed) && monitorIsActive({expiresAt: expiry}, due); +} + +/** Share the resume owner's source proof; a Turn binding retains supervision. */ +function pendingAutomaticSuccessor(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)) && @@ -37,24 +56,27 @@ function pendingDatedSuccessor(request: JsonObject, wait: JsonObject): JsonObjec 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") { + (wait.selected_todo_claimed_by == null || wait.selected_todo_claimed_by === "" || + wait.selected_todo_claimed_by === request.agent_id) && + condition && ["resume_at", "todo_done", "monitor_changed"].includes(String(condition.kind))) { try { if (resumeConditionHasKnownPendingTarget(condition, { todo_id: wait.selected_todo_id, resume_when: wait.resume_when, - })) return condition; + resume_monitor_generation: wait.resume_monitor_generation, + }) && (condition.kind !== "monitor_changed" || hasScheduledMonitorSource(condition))) return condition; } catch { - // An incomplete or malformed clock proof cannot relax the contract. + // An incomplete or malformed source proof cannot relax the contract. } } return null; } -/** A clock-backed wait needs no observation before its scheduled instant. +/** A proven pending automatic successor needs only fresh source rechecks. * This read model never relaxes an existing Turn binding or settlement fence. - * Dependency waits retain their bounded observation/replan policy. */ + * Unknown dependencies retain their bounded observation/replan policy. */ export function projectBlockedSuccessorWaitObservation(request: JsonObject): JsonObject { const wait = exactBlockedSuccessorWait(request); - return {required: !!wait && pendingDatedSuccessor(request, wait) === null}; + return {required: !!wait && pendingAutomaticSuccessor(request, wait) === null}; } export const INTERACTION_CONTRACT_SCHEMA_VERSION = @@ -181,21 +203,24 @@ 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" && + const wait = exactBlockedSuccessorWait(request, disposition === "monitor_wait"); + const waiting = disposition === "agent_scope_wait" && + request.effective_action === AgentScopeFrontierAction.AGENT_SCOPE_WAIT || + disposition === "monitor_wait" && request.effective_action === EffectiveAction.MONITOR_QUIET_SKIP; + const condition = !errors.length && waiting && !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; + !facts.spend_allowed_now && !facts.spend_after_validation && wait + ? pendingAutomaticSuccessor(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( + const scheduled = disposition === "agent_scope_wait" && condition?.kind === "resume_at" ? {scheduled_recheck_after_seconds: Math.ceil( (Date.parse(String(condition.scheduled_for)) - Date.parse(String(condition.evaluated_at))) / 1000, )} : {}; + const dependency = condition && ["todo_done", "monitor_changed"].includes(String(condition.kind)) + ? {pending_dependency_wait: true} : {}; return {mode, errors: [...new Set(errors)], disposition: errors.length ? "consistency_repair" : disposition, - reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason, ...scheduled}; + reason_code: errors.length ? "scheduler_interaction_contract_inconsistent" : reason, ...scheduled, ...dependency}; } function decodeUserChannel(value: unknown): UserInteractionChannel { diff --git a/tests/control_plane/test_dependency_dispatch_cli.py b/tests/control_plane/test_dependency_dispatch_cli.py new file mode 100644 index 0000000000..068face639 --- /dev/null +++ b/tests/control_plane/test_dependency_dispatch_cli.py @@ -0,0 +1,302 @@ +"""Actual work after local gates and dated waits, using public local providers.""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import json +import shlex +import sys +from types import SimpleNamespace + +import pytest + +from loopx.control_plane.effect_runtime import restart_effect_runtime +import test_dependency_vision_wait_cli as dependency_fixtures +from test_dependency_vision_wait_cli import ( + AGENT, GOAL, REPO, add, canonical_todos, guard, update, wait_on, worker_args, +) + +public_goal = dependency_fixtures.public_goal +isolated_effect = dependency_fixtures.isolated_effect + +_RESULT = {"input": [1, 2, 3], "squares": [1, 4, 9], "sum": 14} + +_HOST = '''import json,subprocess,sys +from hashlib import sha256 +from pathlib import Path +from loopx.goal_mode_mcp import GoalModeMCPConfig,GoalModeMCPControlPlane +request=json.loads(Path(sys.argv[1]).read_text()) +def count(kind): + with Path(request["counts"]).open("a") as stream: + stream.write(json.dumps({"kind":kind})+"\\n") +def cli(*args): + process=subprocess.run([sys.executable,"-m","loopx.cli","--registry",request["registry"], + "--runtime-root",request["runtime"],"--format","json",*args],capture_output=True,text=True,check=True) + return json.loads(process.stdout) +count("dispatch") +control=GoalModeMCPControlPlane( + GoalModeMCPConfig(server_name="dependent-dispatch",runtime_profile="generic_cli",legacy_host_surface="claude_code"), + lambda:{"goal_id":request["goal_id"],"agent_id":request["agent_id"],"registry":request["registry"]}) +control.command_prefix=lambda:[sys.executable,"-m","loopx.cli","--runtime-root",request["runtime"]] +admission=json.loads(control.should_run()) +assert admission["should_run"] is True,admission +assert admission["interaction_contract"]["agent_channel"]["must_attempt"] is True,admission +assert admission["selected_todo"]["todo_id"]==request["todo_id"],admission +if admission["effective_action"]=="successor_replan_required": + current=cli("todo","list","--goal-id",request["goal_id"],"--todo-id",request["todo_id"])["todo"] + assert current["status"]=="deferred" and current["resume_condition"]["kind"]=="resume_at",current + assert current["resume_condition"]["satisfied"] is True,current + revision=cli("todo","list","--goal-id",request["goal_id"])["authority_read"]["provider_revision"] + cli("todo","update","--goal-id",request["goal_id"],"--agent-id",request["agent_id"], + "--todo-id",request["todo_id"],"--status","open","--clear-resume-when", + "--update-operation-id","activate-original-dated-work","--update-expected-provider-revision",revision) + admission=json.loads(control.should_run()) + assert admission["should_run"] is True and admission["selected_todo"]["todo_id"]==request["todo_id"],admission +count("model_boundary") +data=json.loads(Path(request["input"]).read_text()) +squares=[item*item for item in data] +result={"input":data,"squares":squares,"sum":sum(squares)} +artifact=(json.dumps(result,sort_keys=True)+"\\n").encode() +Path(request["artifact"]).write_bytes(artifact) +evidence="fixture:sum-of-squares-sha256:"+sha256(artifact).hexdigest() +vision={"schema_version":"goal_vision_replan_contract_v0","state":"no_followup", + "vision_patch":{"vision_summary":"Retain the requested work and each independent dependency.", + "acceptance_summary":"Compute the sum of squares for this task's declared input.", + "last_patch_summary":"This finite calculation passed its original validator."}, + "path_delta":{"schema_version":"goal_path_delta_v0","outcome":"no_change", + "prior_assumption":"The selected finite calculation still required execution.", + "observed_reality":"The declared inputs were processed and the actual result was validated.", + "evidence_refs":[evidence],"stopped":["This task is complete; other retained work keeps its own requirements."]}} +completion=json.loads(control.complete_task(request["todo_id"],request["agent_id"],evidence, + no_follow_up=True,agent_vision=vision)) +Path(request["receipt"]).write_text(json.dumps(completion)) +assert completion["ok"] is True,completion +print(json.dumps({"ok":True,"completed_todo_id":request["todo_id"]})) +''' + + +def _finite_work(goal, *, deferred=False): + project = goal.root / "project" + artifact = project / "calculation.json" + validator = [sys.executable, "-c", + "import json,pathlib,sys; assert json.loads(pathlib.Path(sys.argv[1]).read_text()) == " + repr(_RESULT), + str(artifact)] + options = ["--task-class", "advancement_task", "--action-kind", "validate", + "--validation-command-json", json.dumps(validator)] + if deferred: + options += ["--status", "deferred", "--resume-when", "resume_at:2026-01-01T00:10:00Z"] + return add(goal, "Compute the sum of squares for the declared input [1, 2, 3].", *options) + + +def _host(goal, todo_id, monkeypatch): + project = goal.root / "project" + monkeypatch.chdir(project) + monkeypatch.setenv("PYTHONPATH", str(REPO)) + input_path, artifact, counts, receipt, error = (project / name for name in + ("input.json", "calculation.json", "host-counts.jsonl", "host-receipt.json", "host-error.txt")) + input_path.write_text(json.dumps([1, 2, 3])) + host, request = project / "fake-calculation-host.py", project / "host-request.json" + host.write_text(_HOST) + request.write_text(json.dumps({"registry": str(goal.registry), "runtime": str(goal.runtime), + "goal_id": GOAL, "agent_id": AGENT, "todo_id": todo_id, "input": str(input_path), + "artifact": str(artifact), "counts": str(counts), "receipt": str(receipt)})) + launcher = project / "loopx-cli" + launcher.write_text("#!/bin/sh\nexec " + shlex.join([sys.executable, "-m", "loopx.cli"]) + ' "$@"\n') + launcher.chmod(0o700) + args = worker_args(goal) + args.cli_bin = str(launcher) + args.wake_cmd = shlex.join([sys.executable, str(host), str(request)]) + " 2>" + shlex.quote(str(error)) + args.wake_timeout_seconds = 180 + monkeypatch.syspath_prepend(str(REPO / "scripts")) + import external_scheduler_worker as worker + + def run(): + assert worker.run_worker(args) == 0, error.read_text() if error.exists() else "no Host error output" + + return SimpleNamespace(worker=worker, args=args, run=run, artifact=artifact, counts=counts, receipt=receipt) + + +def _gate(goal, target=None): + return goal.call("todo", "add", "--goal-id", GOAL, "--role", "user", "--task-class", "user_gate", + "--text", f"Approve access to {target or 'the unspecified dependent work'}.", + "--blocks-agent", AGENT, "--agent-id", AGENT, + *(("--unblocks-todo-id", target) if target else ()))["todo_id"] + + +def _single_execution(goal, host, todo_id): + assert json.loads(host.artifact.read_text()) == _RESULT + assert json.loads(host.receipt.read_text())["ok"] is True + assert [json.loads(line)["kind"] for line in host.counts.read_text().splitlines()] == ["dispatch", "model_boundary"] + rows = [json.loads(line) for line in (goal.runtime / f"goals/{GOAL}/runs/index.jsonl").read_text().splitlines()] + assert sum(row.get("classification") == "quota_slot_spent" and row.get("todo_id") == todo_id for row in rows) == 1 + current = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", todo_id)["todo"] + assert current["status"] == "done" + + +def test_local_wait_and_user_gates_preserve_only_verified_independent_work(public_goal, monkeypatch): + goal = public_goal + _, waiting = wait_on(goal, "todo_done") + independent = _finite_work(goal) + gated_work = add(goal, "Validate the separate artifact after owner approval.", + "--task-class", "advancement_task", "--action-kind", "validate") + gate = _gate(goal) + _gate(goal, gated_work) + monitor = add(goal, "Observe the result that needs separate owner approval.", + "--task-class", "continuous_monitor", "--action-kind", "observe", "--watch-only", + "--target-key", "owner-protected-result", "--cadence", "1h", "--next-due-at", "2026-01-01T00:00:00Z") + _gate(goal, monitor) + host = _host(goal, independent, monkeypatch) + + unknown = guard(goal) + assert unknown["requires_user_action"] is True and unknown["should_run"] is False + assert unknown.get("scoped_user_gate_fallback") is None + host.run() + assert not host.counts.exists() + + update(goal, gate, "--unblocks-todo-id", waiting) + fresh = guard(goal) + assert fresh["should_run"] is True and fresh["requires_user_action"] is True + assert fresh["selected_todo"]["todo_id"] == independent + assert fresh["safe_bypass_kind"] == "scoped_user_gate_fallback" + assert fresh["normal_delivery_allowed"] is False + + # The real Goal's compute=0 contract outranks its still-present exact gate + # and runnable independent work, including after a worker restart. + goal.call("configure-goal", "--goal-id", GOAL, "--quota-compute", "0", "--execute") + paused_source = goal.call("todo", "list", "--goal-id", GOAL) + for _ in range(2): + paused = guard(goal) + assert paused["quota"]["compute"] == 0 and paused["state"] == "paused" + assert paused["should_run"] is False and paused["safe_bypass_allowed"] is False + assert paused["interaction_contract"]["agent_channel"]["must_attempt"] is False + assert paused["scheduler_hint"]["action"] == "stop_until_explicit_resume" + host.run() + restart_effect_runtime() + after_pause = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after_pause) == canonical_todos(paused_source) + assert after_pause["authority_read"]["provider_revision"] == paused_source["authority_read"]["provider_revision"] + assert not host.counts.exists() and not host.artifact.exists() + rows = [json.loads(line) for line in (goal.runtime / f"goals/{GOAL}/runs/index.jsonl").read_text().splitlines()] + assert not any(row.get("classification") == "quota_slot_spent" for row in rows) + goal.call("configure-goal", "--goal-id", GOAL, "--quota-compute", "1", "--execute") + assert guard(goal)["selected_todo"]["todo_id"] == independent + + before = goal.call("todo", "list", "--goal-id", GOAL) + refused = goal.call("quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + "--runtime-profile", "generic_cli", "--todo-id", gated_work, + "--turn-instance-id", "refused-gated-selection", expected_exit=1) + assert refused["ok"] is False and refused["should_run"] is False + assert refused["safe_bypass_allowed"] is False + assert refused["action_selection_qualification"]["state"] == "deferred" + assert refused["interaction_contract"]["agent_channel"]["must_attempt"] is False + denied = goal.call("quota", "monitor-poll", "--execute", "--goal-id", GOAL, "--agent-id", AGENT, + "--runtime-profile", "generic_cli", "--todo-id", monitor, "--target-key", "owner-protected-result", + "--result-hash", "unauthorized-result", "--material-change", expected_exit=1) + assert denied["error_code"] == "monitor_poll_admission_rejected", denied + after = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after) == canonical_todos(before) + assert after["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + + host.run() + _single_execution(goal, host, independent) + retained = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", waiting)["todo"] + assert retained["status"] == "deferred" and retained["resume_condition"]["satisfied"] is False + blocked_monitor = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", monitor)["todo"] + assert blocked_monitor.get("result_hash") is None + restart_effect_runtime() + host.run() + _single_execution(goal, host, independent) + + next_work = add(goal, "Perform the next independent calculation only while the Goal is active.", + "--task-class", "advancement_task", "--action-kind", "validate") + assert guard(goal)["selected_todo"]["todo_id"] == next_work + goal.call("goal-lifecycle", "--goal-id", GOAL, "--operation", "stop", "--actor-kind", "owner", + "--reason", "Stop all work in this isolated fixture.", "--execute") + for _ in range(2): + restart_effect_runtime() + stopped = guard(goal) + assert stopped["should_run"] is False + assert stopped["scheduler_hint"]["action"] == "stop_until_explicit_resume" + host.run() + _single_execution(goal, host, independent) + + +@pytest.mark.parametrize("owner_gate", [False, True]) +def test_dated_wait_rechecks_current_readiness_and_dispatches_original_work_once(public_goal, monkeypatch, owner_gate): + from loopx.control_plane.todos import resume_condition + + goal = public_goal + waiting = _finite_work(goal, deferred=True) + clock = {"seconds": 0} + original = resume_condition.build_todo_resume_evaluation_request + + def evaluated(*args, **kwargs): + instant = datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(seconds=clock["seconds"]) + kwargs["evaluated_at"] = instant.isoformat().replace("+00:00", "Z") + return original(*args, **kwargs) + + monkeypatch.setattr(resume_condition, "build_todo_resume_evaluation_request", evaluated) + vision = goal.root / "dated-vision.json" + vision.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "vision_drift_detected", + "todo_delta": [f"activate:{waiting}"], "vision_patch": {"advancement_policy": "as_needed", + "acceptance_summary": "Compute the finite declared calculation when its registered time arrives.", + "replan_trigger_summary": "The dated calculation remains required."}})) + goal.call("refresh-state", "--goal-id", GOAL, "--agent-id", AGENT, "--classification", "dated_fixture", + "--agent-vision-json", str(vision), "--no-global-sync", "--suppress-external-sinks") + if owner_gate: + _gate(goal, waiting) + host = _host(goal, waiting, monkeypatch) + + def probe(command, **kwargs): + # Only the non-model clock is injected; each poll reloads canonical + # state through the actual public CLI and original resume evaluator. + return goal.call(*command[command.index("quota"):]) + + monkeypatch.setattr(host.worker, "run_quota_should_run", probe) + before = goal.call("todo", "list", "--goal-id", GOAL) + for seconds in (0, 300, 599): + clock["seconds"] = seconds + packet = guard(goal) + assert packet["should_run"] is False + host.run() + assert not host.counts.exists() + after_wait = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after_wait) == canonical_todos(before) + assert after_wait["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + + clock["seconds"] = 600 + due = guard(goal) + current = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", waiting)["todo"] + assert current["resume_when"] == "resume_at:2026-01-01T00:10:00Z" + assert current["resume_condition"]["satisfied"] is True + (goal.root / "due-readback.json").write_text(json.dumps({"quota": due, "todo": current}, indent=2)) + if owner_gate: + assert due["should_run"] is False and due["requires_user_action"] is True, json.dumps({key: due.get(key) + for key in ("effective_action", "requires_user_action", "safe_bypass_kind", "selected_todo", + "scoped_user_gate_fallback", "goal_frontier_projection", "interaction_contract")}) + else: + assert due["should_run"] is True and due["selected_todo"]["todo_id"] == waiting + host.run() + for seconds in (601, 900, 1200): + clock["seconds"] = seconds + restart_effect_runtime() + host.run() + if owner_gate: + assert not host.counts.exists() and not host.artifact.exists() + after = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after) == canonical_todos(before) + assert after["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + else: + _single_execution(goal, host, waiting) + assert guard(goal)["interaction_contract"]["mode"] == "terminal_no_followup" + + +def test_prose_time_cannot_create_a_dated_automatic_wait(public_goal): + goal = public_goal + before = goal.call("todo", "list", "--goal-id", GOAL) + rejected = goal.call("todo", "add", "--goal-id", GOAL, "--role", "agent", + "--claimed-by", AGENT, "--task-class", "advancement_task", "--status", "deferred", + "--text", "Resume this task ten minutes from now.", expected_exit=1) + assert "requires --resume-when" in json.dumps(rejected), rejected + after = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after) == canonical_todos(before) + assert after["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] diff --git a/tests/control_plane/test_dependency_feedback_cli.py b/tests/control_plane/test_dependency_feedback_cli.py new file mode 100644 index 0000000000..5ce75fa45e --- /dev/null +++ b/tests/control_plane/test_dependency_feedback_cli.py @@ -0,0 +1,245 @@ +"""Actual feedback sources, one analysis Host, and original acceptance receipts.""" +from __future__ import annotations + +from hashlib import sha256 +import json +import shlex +import sys + +from loopx.control_plane.effect_runtime import restart_effect_runtime +import test_dependency_vision_wait_cli as dependency_fixtures +from test_dependency_vision_wait_cli import ( + AGENT, GOAL, PEER, REPO, add, assert_quiet, canonical_todos, + guard, worker_args, +) + +public_goal = dependency_fixtures.public_goal +isolated_effect = dependency_fixtures.isolated_effect + +OPERATOR = "external-evaluator" + +_VERIFY = '''import json +from pathlib import Path +feedback=json.loads(Path("current-feedback.json").read_text()) +result=json.loads(Path("analysis.json").read_text()) +assert (feedback["candidate_id"],feedback["operation_id"],feedback["feedback_id"])==("C1","O1","F2") +assert [row["case"] for row in feedback["results"]]==["case-a","case-b","case-c"] +deltas=[row["candidate_score"]-row["baseline_score"] for row in feedback["results"]] +assert result=={"candidate_id":"C1","operation_id":"O1","feedback_id":"F2", + "covered_cases":[row["case"] for row in feedback["results"]],"score_deltas":deltas, + "max_delta":max(deltas),"conclusion":"no_improvement_on_declared_cases", + "global_optimality_proved":False,"stop_basis":"owner_authorized_finite_analysis_complete"} +assert max(deltas)<=0 +''' + +_HOST = '''import json,subprocess,sys +from hashlib import sha256 +from pathlib import Path +from loopx.goal_mode_mcp import GoalModeMCPConfig,GoalModeMCPControlPlane +request=json.loads(Path(sys.argv[1]).read_text()) +def count(kind): + with Path(request["counts"]).open("a") as stream: + stream.write(json.dumps({"kind":kind})+"\\n") +def cli(*args): + result=subprocess.run([sys.executable,"-m","loopx.cli","--registry",request["registry"], + "--runtime-root",request["runtime"],"--format","json",*args],capture_output=True,text=True,check=True) + return json.loads(result.stdout) +count("dispatch") +control=GoalModeMCPControlPlane( + GoalModeMCPConfig(server_name="feedback-fixture",runtime_profile="generic_cli",legacy_host_surface="claude_code"), + lambda:{"goal_id":request["goal_id"],"agent_id":request["agent_id"],"registry":request["registry"]}) +control.command_prefix=lambda:[sys.executable,"-m","loopx.cli","--runtime-root",request["runtime"]] +admission=json.loads(control.should_run()) +assert admission["should_run"] is True,admission +assert admission["interaction_contract"]["agent_channel"]["must_attempt"] is True,admission +assert admission["selected_todo"]["todo_id"]==request["todo_id"],admission +source=cli("todo","list","--goal-id",request["goal_id"],"--todo-id",request["source_id"])["todo"] +data=Path(request["feedback"]).read_bytes() +assert source["result_hash"]==sha256(data).hexdigest() +assert source["target_key"]=="candidate:C1:operation:O1" +count("model_boundary") +# This deterministic model substitute performs the requested finite analysis. +feedback=json.loads(data) +assert (feedback["candidate_id"],feedback["operation_id"],feedback["feedback_id"])==("C1","O1","F2") +deltas=[row["candidate_score"]-row["baseline_score"] for row in feedback["results"]] +assert max(deltas)<=0 +result={"candidate_id":"C1","operation_id":"O1","feedback_id":"F2", + "covered_cases":[row["case"] for row in feedback["results"]],"score_deltas":deltas, + "max_delta":max(deltas),"conclusion":"no_improvement_on_declared_cases", + "global_optimality_proved":False,"stop_basis":"owner_authorized_finite_analysis_complete"} +artifact=(json.dumps(result,sort_keys=True)+"\\n").encode() +Path(request["artifact"]).write_bytes(artifact) +evidence="fixture:finite-feedback-analysis-sha256:"+sha256(artifact).hexdigest() +vision={"schema_version":"goal_vision_replan_contract_v0","state":"no_followup", + "vision_patch":{"vision_summary":"Analyze the current C1/O1 feedback once.", + "acceptance_summary":"Report the declared finite cases and preserve unproved global claims.", + "last_patch_summary":"The owner-authorized finite analysis is complete."}, + "path_delta":{"schema_version":"goal_path_delta_v0","outcome":"no_change", + "prior_assumption":"The current evaluation still required analysis.", + "observed_reality":"The declared cases show no improvement; global optimality remains unproved.", + "evidence_refs":[evidence],"stopped":["The requested finite analysis is complete; no new experiment was authorized."]}} +# The original external wait retains its already-completed O1 successor. +# Goal-level no-follow-up must not overwrite that existing Todo relationship. +completion=json.loads(control.complete_task(request["todo_id"],request["agent_id"],evidence, + agent_vision=vision)) +Path(request["receipt"]).write_text(json.dumps({"completion":completion,"evidence":evidence,"vision":vision})) +assert completion["ok"] is True,completion +print(json.dumps({"ok":True,"completed_todo_id":request["todo_id"]})) +''' + + +def _change(goal, todo_id, agent, *options): + revision = goal.call("todo", "list", "--goal-id", GOAL)["authority_read"]["provider_revision"] + return goal.call("todo", "update", "--goal-id", GOAL, "--todo-id", todo_id, "--agent-id", agent, + "--update-operation-id", f"update-{revision}", "--update-expected-provider-revision", revision, *options) + + +def _monitor(goal, candidate, operation, due): + return add(goal, f"Observe candidate {candidate} evaluation {operation} feedback.", + "--task-class", "continuous_monitor", "--action-kind", "observe", "--watch-only", + "--target-key", f"candidate:{candidate}:operation:{operation}", "--cadence", "1h", + "--next-due-at", due, "--expires-at", "2100-01-01T00:00:00Z", owner=PEER) + + +def _observe(goal, monitor, candidate, operation, path, turn): + binding = ("--goal-id", GOAL, "--agent-id", PEER, "--turn-instance-id", turn) + admitted = goal.call("quota", "should-run", "--codex-app", *binding) + if admitted["heartbeat_receipt"]["status"] != "replayed": + assert admitted["selected_todo"]["todo_id"] == monitor, admitted + return goal.call("quota", "monitor-poll", "--execute", "--codex-app", *binding, + "--todo-id", monitor, "--target-key", f"candidate:{candidate}:operation:{operation}", + "--result-hash", sha256(path.read_bytes()).hexdigest(), "--material-change", + "--next-due-at", "2099-01-01T00:00:00Z") + + +def test_current_feedback_is_adopted_and_analyzed_once_with_old_feedback_retained(public_goal, monkeypatch): + goal = public_goal + project = goal.root / "project" + monkeypatch.chdir(project) + monkeypatch.setenv("PYTHONPATH", str(REPO)) + goal.call("configure-goal", "--goal-id", GOAL, "--registered-agent", AGENT, + "--registered-agent", PEER, "--registered-agent", OPERATOR, "--execute") + old_source = _monitor(goal, "C0", "O0", "2026-01-01T00:00:00Z") + source = _monitor(goal, "C1", "O1", "2099-01-01T00:00:00Z") + operation = add(goal, "Retain the result of evaluation O1 for candidate C1.", + "--task-class", "advancement_task", "--action-kind", "validate", owner=OPERATOR) + analysis = add(goal, "Analyze current C1/O1 feedback over exactly the three declared cases; retain unproved global claims.", + "--task-class", "advancement_task", "--action-kind", "validate") + _change(goal, analysis, AGENT, "--resume-when", f"monitor_changed:{source}", "--successor-todo-id", operation) + vision = project / "initial-vision.json" + vision.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "vision_drift_detected", + "todo_delta": [f"activate:{analysis}"], "vision_patch": { + "acceptance_summary": "Analyze current C1/O1 feedback over the three declared cases.", + "advancement_policy": "as_needed", "replan_trigger_summary": "Current feedback still requires its declared analysis."}})) + goal.call("refresh-state", "--goal-id", GOAL, "--agent-id", AGENT, "--classification", "feedback_fixture", + "--agent-vision-json", str(vision), "--no-global-sync", "--suppress-external-sinks") + verifier = project / "verify_analysis.py" + verifier.write_text(_VERIFY) + document = project / "acceptance.json" + document.write_text(json.dumps({"scope": {"kind": "selected_work", "todo_ids": [analysis]}, + "objective": "Analyze C1/O1 feedback once and stop after reporting the finite evidence and remaining unproved claims.", + "non_goals": ["No new experiment or global optimality proof is requested."], + "criteria": [{"id": "finite-feedback-analysis", "description": "Compute every declared case delta and preserve the bounded conclusion.", + "validation_argv": [sys.executable, "verify_analysis.py"], "validation_timeout_seconds": 5, + "validation_files": [{"path": "verify_analysis.py", "sha256": sha256(verifier.read_bytes()).hexdigest()}]}], + "bindings": [{"todo_id": analysis, "criterion_ids": ["finite-feedback-analysis"]}]})) + basis = goal.call("goal-acceptance", "inspect", "--goal-id", GOAL)["provider_revision"] + configured = goal.call("goal-acceptance", "configure", "--goal-id", GOAL, "--document", str(document), + "--expected-provider-revision", basis, "--operation-id", "confirm-current-feedback-analysis", "--execute") + acceptance = configured["goal_acceptance_contract"] + assert acceptance["status"] == "unverified" + assert acceptance["scope"] == {"kind": "selected_work", "todo_ids": [analysis]} + assert [criterion["id"] for criterion in acceptance["criteria"]] == ["finite-feedback-analysis"] + assert [ + (task["todo_id"], task["state"], task["criterion_ids"]) + for task in acceptance["tasks"] + ] == [(analysis, "ready", ["finite-feedback-analysis"])] + assert_quiet(guard(goal), analysis) + + old_feedback = project / "old-feedback.json" + old_feedback.write_text(json.dumps({"candidate_id": "C0", "operation_id": "O0", "feedback_id": "F0", "score": 100})) + old = _observe(goal, old_source, "C0", "O0", old_feedback, "feedback-C0-O0-F0") + assert old["replayed"] is False + assert _observe(goal, old_source, "C0", "O0", old_feedback, "feedback-C0-O0-F0")["replayed"] is True + current = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", source)["todo"] + assert current.get("material_change_generation", 0) == 0 + assert current.get("result_hash") is None + assert_quiet(guard(goal), analysis) + retained_old = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", old_source)["todo"] + assert retained_old["result_hash"] == sha256(old_feedback.read_bytes()).hexdigest() + + host, request = project / "fake-feedback-host.py", project / "host-request.json" + host.write_text(_HOST) + counts, artifact, receipt = (project / name for name in ("host-counts.jsonl", "analysis.json", "analysis-receipt.json")) + feedback = project / "current-feedback.json" + request.write_text(json.dumps({"goal_id": GOAL, "agent_id": AGENT, "registry": str(goal.registry), + "runtime": str(goal.runtime), "source_id": source, "todo_id": analysis, "feedback": str(feedback), + "artifact": str(artifact), "counts": str(counts), "receipt": str(receipt)})) + launcher = project / "loopx-cli" + launcher.write_text("#!/bin/sh\nexec " + shlex.join([sys.executable, "-m", "loopx.cli"]) + ' "$@"\n') + launcher.chmod(0o700) + args = worker_args(goal) + args.cli_bin = str(launcher) + host_error = project / "host-error.txt" + args.wake_cmd = shlex.join([sys.executable, str(host), str(request)]) + " 2>" + shlex.quote(str(host_error)) + args.wake_timeout_seconds = 180 + monkeypatch.syspath_prepend(str(REPO / "scripts")) + import external_scheduler_worker as worker + assert worker.run_worker(args) == 0, host_error.read_text() if host_error.exists() else "no Host error output" + restart_effect_runtime() + assert worker.run_worker(args) == 0, host_error.read_text() if host_error.exists() else "no Host error output" + assert not counts.exists() and not artifact.exists() + + feedback.write_text(json.dumps({"candidate_id": "C1", "operation_id": "O1", "feedback_id": "F2", "results": [ + {"case": "case-a", "baseline_score": 10, "candidate_score": 10}, + {"case": "case-b", "baseline_score": 20, "candidate_score": 19}, + {"case": "case-c", "baseline_score": 30, "candidate_score": 28}]})) + _change(goal, source, PEER, "--next-due-at", "2026-01-01T00:00:00Z") + adopted = _observe(goal, source, "C1", "O1", feedback, "feedback-C1-O1-F2") + assert adopted["replayed"] is False and adopted["material_change"] is True + assert _observe(goal, source, "C1", "O1", feedback, "feedback-C1-O1-F2")["replayed"] is True + goal.call("todo", "complete", "--goal-id", GOAL, "--todo-id", operation, "--agent-id", OPERATOR, + "--no-follow-up", "--evidence", "fixture:feedback-F2-sha256:" + sha256(feedback.read_bytes()).hexdigest()) + ready = guard(goal) + assert ready["should_run"] is True and ready["selected_todo"]["todo_id"] == analysis + assert ready["effective_action"] != "autonomous_replan_required" + assert worker.run_worker(args) == 0, host_error.read_text() if host_error.exists() else "no Host error output" + result = json.loads(artifact.read_text()) + assert result["score_deltas"] == [0, -1, -2] + assert result["global_optimality_proved"] is False + completed = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", analysis) + assert completed["todo"]["status"] == "done" + assert completed["todo"]["successor_todo_ids"] == [operation] + assert completed["todo"]["completion_continuation"] == "successor" + assert json.loads(receipt.read_text())["completion"]["ok"] is True + verified = goal.call("goal-acceptance", "verify", "--goal-id", GOAL, "--agent-id", AGENT, "--execute") + assert verified["checks_passed"] is True + assert verified["goal_acceptance_contract"]["status"] == "accepted" + + before_replay = goal.call("todo", "list", "--goal-id", GOAL) + settled = guard(goal) + (goal.root / "post-analysis-readback.json").write_text(json.dumps({"quota": settled, + "todos": before_replay, "acceptance": verified}, ensure_ascii=False, indent=2)) + assert settled["should_run"] is False, json.dumps({key: settled.get(key) for key in ( + "effective_action", "agent_lane_next_action", "goal_frontier_projection", "interaction_contract")}) + for _ in range(2): + assert _observe(goal, source, "C1", "O1", feedback, "feedback-C1-O1-F2")["replayed"] is True + restart_effect_runtime() + assert worker.run_worker(args) == 0 + after = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after) == canonical_todos(before_replay) + assert after["authority_read"]["provider_revision"] == before_replay["authority_read"]["provider_revision"] + final = guard(goal) + assert final["should_run"] is False + assert final["interaction_contract"]["agent_channel"]["must_attempt"] is False + # This accepted selected-work scope stops an empty analysis dispatch. The + # retained successor and other agents' monitors do not certify Goal closure. + assert final["goal_frontier_projection"].get("terminal_state") is None + acceptance = goal.call("goal-acceptance", "inspect", "--goal-id", GOAL)["goal_acceptance_contract"] + assert acceptance["status"] == "accepted" + assert acceptance["scope"] == {"kind": "selected_work", "todo_ids": [analysis]} + assert [json.loads(line)["kind"] for line in counts.read_text().splitlines()] == ["dispatch", "model_boundary"] + runs = [json.loads(line) for line in (goal.runtime / f"goals/{GOAL}/runs/index.jsonl").read_text().splitlines()] + assert sum(row.get("classification") == "quota_slot_spent" and row.get("todo_id") == analysis for row in runs) == 1 + assert sum(row.get("classification") == "quota_monitor_poll" and row.get("turn_instance_id") == "feedback-C1-O1-F2" + for row in runs) == 1 diff --git a/tests/control_plane/test_dependency_vision_wait_cli.py b/tests/control_plane/test_dependency_vision_wait_cli.py new file mode 100644 index 0000000000..b88899c7fe --- /dev/null +++ b/tests/control_plane/test_dependency_vision_wait_cli.py @@ -0,0 +1,363 @@ +"""Public creation and canonical source changes must reach the actual worker.""" +from __future__ import annotations + +import json +import io +import os +import sys +from argparse import Namespace +from contextlib import redirect_stdout +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from loopx.cli import main +from test_dated_vision_wait_cli import REPO, isolated_effect # noqa: F401 + +GOAL = "dependency-vision-fixture" +AGENT = "agent-waiter" +PEER = "agent-prerequisite" + + +@pytest.fixture(params=["file", "sqlite"]) +def public_goal(request, tmp_path, monkeypatch): + # Configuration, registry, Todo state and source facts all enter through the + # shipped CLI. No prebuilt provider projection or run history is installed. + request.getfixturevalue("isolated_effect") + monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite") + monkeypatch.setenv("LOOPX_USAGE_PING", "0") + project = tmp_path / "project" + project.mkdir() + registry, runtime = project / ".loopx/registry.json", tmp_path / "runtime" + + def call(*args, expected_exit=0): + output = io.StringIO() + with redirect_stdout(output): + result = main(["--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", *args]) + payload = json.loads(output.getvalue()) + assert result == expected_exit, payload + return payload + + storage = tmp_path / "storage.json" + storage.write_text(json.dumps({"schema_version": "loopx_goal_storage_defaults_v1", + "new_goal_provider": request.param, "canonical_creation": True, "new_goal_handoff_mode": "soft_claim"})) + preview = call("machine-config", "preview", "--namespace", "goal_storage", "--config-json", str(storage)) + call("machine-config", "apply", "--namespace", "goal_storage", "--config-json", str(storage), + "--expected-plan-revision", preview["plan_revision"], "--execute") + registered = call("project", "register", "--project-id", "dependency-fixture", "--project-kind", "work", + "--knowledge-root", str(project), "--goal-id", GOAL, + "--objective", "Validate the retained result after its prerequisite changes.", + "--acceptance", "The retained result is validated.", "--next-effect", "Wait for the prerequisite result.", + "--stop-condition", "Stop if the owner cancels.") + assert registered["storage_selection"]["authority_initialized"] is True + assert registered["storage_selection"]["provider"] == request.param + configured = call("configure-goal", "--goal-id", GOAL, "--registered-agent", AGENT, + "--registered-agent", PEER, "--execute") + assert configured["global_sync"]["selected_target"]["runtime_root"] == str(runtime) + return SimpleNamespace(call=call, registry=registry, runtime=runtime, root=tmp_path, + state=Path(registered["state_file"]), provider=request.param) + + +def add(goal, text, *options, owner=AGENT): + return goal.call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--text", text, + "--claimed-by", owner, *options)["todo_id"] + + +def update(goal, todo_id, *options): + listed = goal.call("todo", "list", "--goal-id", GOAL) + revision = listed["authority_read"]["provider_revision"] + return goal.call("todo", "update", "--goal-id", GOAL, "--todo-id", todo_id, "--agent-id", AGENT, + "--update-operation-id", f"update-{revision}", "--update-expected-provider-revision", revision, *options) + + +def wait_on(goal, kind, *, missing_source=False, monitor_due="2099-01-01T00:00:00Z", + monitor_expiry="2100-01-01T00:00:00Z", monitor_cadence="1h"): + if kind == "todo_done": + artifact = goal.root / "prerequisite.txt" + artifact.write_text("pending") + validator = [sys.executable, "-c", + "import pathlib,sys; assert pathlib.Path(sys.argv[1]).read_text() == 'ready'", str(artifact)] + source = add(goal, "Prepare the prerequisite artifact.", "--task-class", "advancement_task", + "--action-kind", "implement", "--validation-command-json", json.dumps(validator), owner=PEER) + waiting = add(goal, "Validate the retained result.", "--task-class", "advancement_task", + "--action-kind", "validate", "--status", "deferred", "--resume-when", + f"todo_done:{'todo_absent_source' if missing_source else source}") + else: + # The waiting Agent owns the Monitor too: its eventual due work must + # still run, or this wait would silence its only generation producer. + schedule = [] + if monitor_cadence is not None: + schedule.extend(["--cadence", monitor_cadence]) + if monitor_due is not None: + schedule.extend(["--next-due-at", monitor_due]) + if monitor_expiry is not None: + schedule.extend(["--expires-at", monitor_expiry]) + source = add(goal, "Observe the prerequisite artifact.", "--task-class", "continuous_monitor", + "--action-kind", "observe", "--target-key", "prerequisite-artifact", + *schedule) + independent = add(goal, "Validate an independent artifact.", "--task-class", "advancement_task", + "--action-kind", "validate", owner=PEER) + waiting = add(goal, "Validate the retained result.", "--task-class", "advancement_task", "--action-kind", "validate") + update(goal, waiting, "--resume-when", f"monitor_changed:{source}", "--successor-todo-id", independent) + vision = goal.root / "vision.json" + vision.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "vision_drift_detected", + "todo_delta": [f"activate:{waiting}"], "vision_patch": {"acceptance_summary": "Validate the retained result.", + "advancement_policy": "as_needed", "replan_trigger_summary": "The retained result still requires its prerequisite."}})) + goal.call("refresh-state", "--goal-id", GOAL, "--agent-id", AGENT, "--classification", "dependency_fixture", + "--agent-vision-json", str(vision), "--no-global-sync", "--suppress-external-sinks") + return source, waiting + + +def guard(goal, *options): + return goal.call("quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + "--runtime-profile", "generic_cli", "--include-detail", "scheduler", *options) + + +def assert_quiet(packet, waiting, *, monitor=False): + assert packet["should_run"] is False + assert packet["effective_action"] == ("monitor_quiet_skip" if monitor else "agent_scope_wait") + assert packet["vision_wait_state"]["selected_todo_id"] == waiting + channel = packet["interaction_contract"]["agent_channel"] + assert channel["must_attempt"] is False and channel["quiet_noop_allowed"] is True + assert channel["delivery_allowed"] is False + if not monitor: + assert not any("monitor-poll" in command for command in packet["interaction_contract"]["cli_channel"]["next_cli_actions"]) + hint = packet["scheduler_hint"] + assert hint["action"] == ("backoff_until_material_transition" if monitor else "backoff_until_state_change") + assert all(value is None for value 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 + + +def canonical_todos(listed): + # Resume conditions are fresh read models, including their clock witness. + return [{key: value for key, value in todo.items() if key not in ("resume_condition", "resume_ready")} + for todo in listed["todos"]] + + +def worker_args(goal): + return Namespace(cli_bin="loopx", registry=str(goal.registry), runtime_root=str(goal.runtime), + runtime_profile="generic_cli", goal_id=GOAL, agent_id=AGENT, state_file=str(goal.root / "scheduler.json"), + once=True, wake_cmd="unused", wake_stop_exit_code=75, + error_backoff_seconds=5, quota_timeout_seconds=90, wake_timeout_seconds=None) + + +def test_pending_dependency_stays_quiet_for_100_polls_restart_and_100_more(public_goal, monkeypatch): + goal = public_goal + source, waiting = wait_on(goal, "todo_done") + before = goal.call("todo", "list", "--goal-id", GOAL) + run_index = goal.runtime / f"goals/{GOAL}/runs/index.jsonl" + runs_before = [json.loads(line) for line in run_index.read_text().splitlines()] + monkeypatch.syspath_prepend(str(REPO / "scripts")) + import external_scheduler_worker as worker + + packets, sleeps = [], [] + counts = {"D": 0, "P": 0, "E": 0, "Q": 0} + + def probe(command, **kwargs): + # A new public command reloads the actual provider on every tick. + packet = goal.call(*command[command.index("quota"):]) + assert_quiet(packet, waiting) + assert packet["vision_wait_state"]["resume_condition"]["target_todo_id"] == source + packets.append(packet) + return packet + + def unexpected_wake(*args, **kwargs): + counts["D"] += 1 + raise AssertionError("an unchanged dependency dispatched new model work") + + class RestartWorker(Exception): + pass + + def advance(seconds): + sleeps.append(seconds) + if len(sleeps) in (100, 200): + raise RestartWorker + + monkeypatch.setattr(worker, "run_quota_should_run", probe) + monkeypatch.setattr(worker, "_run_wake", unexpected_wake) + args = worker_args(goal) + args.once = False + for expected in (100, 200): + # Interrupt only after the actual worker has persisted the tick, then + # reopen that same state route. No real hours or model calls are used. + with pytest.raises(RestartWorker): + worker.run_worker(args, sleep=advance) + persisted = json.loads(Path(args.state_file).read_text()) + assert persisted["unchanged_count"] == expected + assert len(packets) == expected + assert not persisted.get("replan_attempts") + after = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(after) == canonical_todos(before) + assert after["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + runs = [json.loads(line) for line in run_index.read_text().splitlines()] + counts["Q"] = sum(bool(run.get("quota_event")) for run in runs[len(runs_before):]) + events_path = goal.runtime / f"goals/{GOAL}/rollout-event-log.jsonl" + events = [json.loads(line) for line in events_path.read_text().splitlines()] + assert not any(event.get("event_kind") in ("quota_monitor_poll", "quota_spend") for event in events) + assert counts == {"D": 0, "P": 0, "E": 0, "Q": 0} + assert len(sleeps) == 200 and sleeps[-1] == 3600 + + +@pytest.mark.parametrize("kind", ["todo_done", "monitor_changed"]) +def test_real_worker_survives_restart_and_follows_its_dependency_source(public_goal, monkeypatch, kind): + goal = public_goal + source, waiting = wait_on(goal, kind) + before = goal.call("todo", "list", "--goal-id", GOAL) + first = guard(goal) + assert_quiet(first, waiting, monitor=kind == "monitor_changed") + if kind == "monitor_changed": + wait = first["vision_wait_state"] + assert wait["resume_monitor_generation"] == wait["resume_condition"]["baseline_generation"] == 0 + monkeypatch.syspath_prepend(str(REPO / "scripts")) + import external_scheduler_worker as worker + + packets, sleeps, wakes = [], [], [] + + def probe(command, **kwargs): + packet = goal.call(*command[command.index("quota"):]) + packets.append(packet) + return packet + + def advance(seconds): + sleeps.append(seconds) + assert len(sleeps) <= 2, "fresh source changes must leave dependency waiting" + if len(sleeps) == 2: + unchanged = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(unchanged) == canonical_todos(before) + assert unchanged["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + if kind == "todo_done": + (goal.root / "prerequisite.txt").write_text("ready") + goal.call("todo", "complete", "--goal-id", GOAL, "--todo-id", source, "--agent-id", PEER, + "--no-follow-up", "--evidence", str(goal.root / "prerequisite.txt")) + else: + update(goal, source, "--next-due-at", "2026-01-01T00:00:00Z") + + def wake(*args, **kwargs): + assert len(sleeps) == 2, {key: packets[-1].get(key) for key in ( + "effective_action", "autonomous_replan", "goal_frontier_projection", "selected_todo", "vision_wait_state")} + wakes.append(packets[-1]) + return SimpleNamespace(returncode=75, failure_kind=None) + + monkeypatch.setattr(worker, "run_quota_should_run", probe) + monkeypatch.setattr(worker, "_run_wake", wake) + args = worker_args(goal) + # Two separate invocations resume the persisted backoff; the live run then + # passes the old three-poll stop and observes the next real source event. + assert worker.run_worker(args, sleep=advance) == 0 + assert worker.run_worker(args, sleep=advance) == 0 + args.once = False + assert worker.run_worker(args, sleep=advance) == 0 + assert sleeps == ([3600, 3600] if kind == "monitor_changed" else [1800, 3600]) + assert len(packets) == 5 and len(wakes) == 1 + for packet in packets[:-1]: + assert_quiet(packet, waiting, monitor=kind == "monitor_changed") + assert packets[-1]["scheduler_hint"]["action"] == "run_now" + assert packets[-1]["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert packets[-1]["selected_todo"]["todo_id"] == (waiting if kind == "todo_done" else source) + if kind == "todo_done": + assert packets[-1]["effective_action"] == "successor_replan_required" + else: + binding = ("--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "actual-monitor-event") + admitted = goal.call("quota", "should-run", "--codex-app", *binding) + assert admitted["heartbeat_receipt"]["settlement_identity"]["todo_id"] == source + poll = goal.call("quota", "monitor-poll", "--execute", "--codex-app", *binding, + "--todo-id", source, "--result-hash", "new-prerequisite-result", "--material-change", + "--next-agent-todo", "Validate the new prerequisite result.", "--next-action-kind", "validate") + assert poll["material_change"] is True + resumed = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", waiting)["todo"] + assert resumed["resume_ready"] is True + assert resumed["resume_condition"]["material_change_generation"] > resumed["resume_monitor_generation"] + + +def test_unknown_dependency_keeps_bounded_supervision(public_goal): + goal = public_goal + _, waiting = wait_on(goal, "todo_done", missing_source=True) + packet = guard(goal) + assert packet["vision_wait_state"]["selected_todo_id"] == waiting + assert "target_status" not in packet["vision_wait_state"]["resume_condition"] + assert packet["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert set(packet["scheduler_hint"]["unchanged_poll"]["limits"].values()) == {3} + + +@pytest.mark.parametrize("expiry", ["2026-01-01T00:00:00Z", "2098-01-01T00:00:00Z"]) +def test_expired_or_unreachable_monitor_does_not_keep_waiting_forever(public_goal, expiry): + goal = public_goal + _, waiting = wait_on(goal, "monitor_changed", monitor_expiry=expiry) + packet = guard(goal) + assert packet["effective_action"] == "monitor_quiet_skip" + assert packet["should_run"] is False + assert packet["vision_wait_state"]["selected_todo_id"] == waiting + assert packet["vision_wait_state"]["resume_condition"]["target_expires_at"] == expiry + assert set(packet["scheduler_hint"]["unchanged_poll"]["limits"].values()) == {3} + + +def test_monitor_without_schedule_keeps_original_observation_admission(public_goal): + goal = public_goal + source, _ = wait_on(goal, "monitor_changed", monitor_due=None, monitor_cadence=None) + packet = guard(goal) + assert packet["should_run"] is True, packet + assert packet["effective_action"] == "external_evidence_observe" + assert packet["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert packet["scheduler_hint"]["action"] == "run_now" + monitor = goal.call("todo", "list", "--goal-id", GOAL, "--todo-id", source)["todo"] + assert not monitor.get("next_due_at") and not monitor.get("cadence") + + +def test_unavailable_provider_cannot_fall_back_or_wake_waiting_work(public_goal, monkeypatch): + from loopx.control_plane.effect_runtime import restart_effect_runtime + + goal = public_goal + _, waiting = wait_on(goal, "todo_done") + assert_quiet(guard(goal), waiting) + before = goal.call("todo", "list", "--goal-id", GOAL) + provider = goal.runtime / "authority" / f"{goal.provider}-v0" + retained = goal.root / "temporarily-unavailable-provider" + assert provider.is_dir() + restart_effect_runtime() + provider.rename(retained) + monkeypatch.syspath_prepend(str(REPO / "scripts")) + import external_scheduler_worker as worker + + def unavailable(command, **kwargs): + packet = goal.call(*command[command.index("quota"):], expected_exit=1) + assert packet.get("should_run") is not True, packet + raise RuntimeError("canonical dependency provider is unavailable") + + def unexpected_wake(*args, **kwargs): + raise AssertionError("an unavailable authority must never dispatch work") + + monkeypatch.setattr(worker, "run_quota_should_run", unavailable) + monkeypatch.setattr(worker, "_run_wake", unexpected_wake) + try: + assert worker.run_worker(worker_args(goal)) == 2 + assert not provider.exists() + finally: + retained.rename(provider) + restart_effect_runtime() + assert_quiet(guard(goal), waiting) + restored = goal.call("todo", "list", "--goal-id", GOAL) + assert canonical_todos(restored) == canonical_todos(before) + assert restored["authority_read"]["provider_revision"] == before["authority_read"]["provider_revision"] + + +def test_stop_and_independent_work_override_the_registered_wait(public_goal): + goal = public_goal + _, waiting = wait_on(goal, "todo_done") + assert_quiet(guard(goal), waiting) + goal.call("goal-lifecycle", "--goal-id", GOAL, "--operation", "stop", "--actor-kind", "owner", + "--reason", "Owner stopped this isolated test.", "--execute") + stopped = guard(goal) + assert stopped["scheduler_hint"]["action"] == "stop_until_explicit_resume" + assert stopped["should_run"] is False + goal.call("goal-lifecycle", "--goal-id", GOAL, "--operation", "resume", "--actor-kind", "owner", + "--reason", "Owner resumed this isolated test.", "--execute") + assert_quiet(guard(goal), waiting) + independent = add(goal, "Validate a newly available independent result.", "--task-class", "advancement_task", + "--action-kind", "validate") + fresh = guard(goal) + assert fresh["selected_todo"]["todo_id"] == independent + assert fresh["scheduler_hint"]["action"] == "run_now" + assert fresh["interaction_contract"]["agent_channel"]["must_attempt"] is True diff --git a/tests/control_plane/test_goal_vision_blocked_successor.py b/tests/control_plane/test_goal_vision_blocked_successor.py index 7b9f2ba3f5..63502e882f 100644 --- a/tests/control_plane/test_goal_vision_blocked_successor.py +++ b/tests/control_plane/test_goal_vision_blocked_successor.py @@ -678,6 +678,9 @@ def test_exact_blocked_successor_defers_only_open_vision_gap( assert wait["resume_when"] == f"todo_done:{BLOCKER_ID}" assert wait["resume_condition"]["target_todo_id"] == BLOCKER_ID assert wait["resume_condition"]["satisfied"] is False + assert wait["resume_condition"]["target_status"] == "open" + assert wait["resume_condition"]["target_task_class"] == "advancement_task" + assert wait["resume_condition"]["target_archive_state"] == "active" assert wait["deferred_acceptance_gap_count"] == 1 assert wait["automatic_resume"] is True assert guard["vision_wait_state"] == wait @@ -690,11 +693,11 @@ def test_exact_blocked_successor_defers_only_open_vision_gap( "blocked_successor_resume_pending" ) assert guard["interaction_contract"]["agent_channel"]["vision_wait_state"] == wait - assert guard["interaction_contract"]["agent_channel"]["must_attempt"] is True + assert guard["interaction_contract"]["agent_channel"]["must_attempt"] is False assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is False - assert guard["interaction_contract"]["agent_channel"]["quiet_noop_allowed"] is False + assert guard["interaction_contract"]["agent_channel"]["quiet_noop_allowed"] is True cli_actions = guard["interaction_contract"]["cli_channel"]["next_cli_actions"] - assert "quota monitor-poll" in cli_actions[0] + assert not any("quota monitor-poll" in action for action in cli_actions) assert "quota should-run" in cli_actions[1] assert "protocol_action_packet" not in guard cli_wait = guard["interaction_contract"]["cli_channel"]["vision_wait_state"] diff --git a/tests/control_plane/test_scheduler_arbitration_transport.py b/tests/control_plane/test_scheduler_arbitration_transport.py index 58506da3c5..84c5c935e5 100644 --- a/tests/control_plane/test_scheduler_arbitration_transport.py +++ b/tests/control_plane/test_scheduler_arbitration_transport.py @@ -33,6 +33,21 @@ def test_scheduled_recheck_codec_rejects_nonpositive_or_coerced_seconds(invalid) "scheduled_recheck_after_seconds": invalid}) +@pytest.mark.parametrize("invalid", [None, 0, 1, 1.5, "true", [], {}]) +def test_pending_dependency_codec_rejects_missing_or_coerced_proof(invalid): + with pytest.raises(TypeError, match="pending dependency wait"): + SchedulerArbitration.from_projection({"mode": "agent_scope_wait", "errors": [], + "disposition": "agent_scope_wait", "reason_code": "interaction_agent_scope_wait", + "pending_dependency_wait": invalid}) + + +def test_pending_dependency_codec_defaults_to_existing_bounded_supervision(): + projection = {"mode": "agent_scope_wait", "errors": [], "disposition": "agent_scope_wait", + "reason_code": "interaction_agent_scope_wait"} + assert not SchedulerArbitration.from_projection(projection).pending_dependency_wait + assert SchedulerArbitration.from_projection({**projection, "pending_dependency_wait": True}).pending_dependency_wait + + def encoded_size(value: dict) -> int: return len(json.dumps(value, separators=(",", ":")).encode()) diff --git a/tests/control_plane/test_user_gate_lane_progress.py b/tests/control_plane/test_user_gate_lane_progress.py index 319c9f55ad..a299b9b566 100644 --- a/tests/control_plane/test_user_gate_lane_progress.py +++ b/tests/control_plane/test_user_gate_lane_progress.py @@ -167,6 +167,7 @@ def test_unrelated_user_gate_allows_ready_deferred_successor_replan() -> None: assert selected["fallback_kind"] == "deferred_successor_replan" assert payload["decision"] == "successor_replan_required" assert payload["effective_action"] == "successor_replan_required" + assert payload["should_run"] is True assert payload["selected_todo"]["todo_id"] == "todo_ready_deferred" assert payload["interaction_contract"]["mode"] == "scoped_user_gate_fallback" assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is True @@ -296,6 +297,8 @@ def test_blocking_user_gate_backs_off_instead_of_polling_as_active_work() -> Non ) assert "scoped_user_gate_fallback" not in payload + assert payload["should_run"] is False + assert payload["effective_action"] == "operator_gate_notify" assert payload["interaction_contract"]["mode"] == "user_gate" assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is False expected_response_plan = { diff --git a/tests/control_plane_ts/action_portfolio.test.ts b/tests/control_plane_ts/action_portfolio.test.ts index 8d5e3de2c1..796b19b065 100644 --- a/tests/control_plane_ts/action_portfolio.test.ts +++ b/tests/control_plane_ts/action_portfolio.test.ts @@ -459,6 +459,33 @@ test("only runnable replan-only deferral permits one inline guard reentry", () = } }); +test("only the current exact scoped fallback may bind through an operator gate", () => { + const selected = candidate("todo_independent001", "Validate the independent result.", "P1"); + const request = { + schema_version: ACTION_SELECTION_QUALIFICATION_REQUEST_SCHEMA_VERSION, + requested_todo_id: selected.todo_id, candidate: selected, + should_run: false, normal_delivery_allowed: false, + delivery_preemptions: ["delivery_not_allowed"], + scoped_fallback_todo_id: selected.todo_id, quota_state: "operator_gate", goal_health_ok: true, + }; + const allowed = qualifyActionSelection(request); + assert.equal(allowed.state, "qualified"); + assert.equal((allowed.selected_todo as Record).todo_id, selected.todo_id); + for (const patch of [ + {scoped_fallback_todo_id: null}, {scoped_fallback_todo_id: "todo_gated001"}, + {goal_health_ok: false}, {quota_state: "paused"}, {quota_state: "throttled"}, + {quota_state: "unknown"}, {delivery_preemptions: []}, + ...["heartbeat_receipt", "autonomous_replan", "blocking_work_lane", "control_repair", + "ready_deferred_successor_priority_preemption"].map(reason => ({ + delivery_preemptions: [reason, "delivery_not_allowed"], + })), + ]) { + const denied = qualifyActionSelection({...request, ...patch}); + assert.equal(denied.state, "deferred", JSON.stringify(patch)); + assert.equal(denied.selected_todo, undefined); + } +}); + test("pending selection rejects a Todo absent from the current eligible set", () => { assert.deepEqual(qualifyActionSelection({ schema_version: ACTION_SELECTION_QUALIFICATION_REQUEST_SCHEMA_VERSION, diff --git a/tests/control_plane_ts/action_selection_admission.test.ts b/tests/control_plane_ts/action_selection_admission.test.ts index df86d32c27..ce7cb07c7d 100644 --- a/tests/control_plane_ts/action_selection_admission.test.ts +++ b/tests/control_plane_ts/action_selection_admission.test.ts @@ -37,6 +37,31 @@ test("workspace repair needs its repair route and explicit delivery refusal", () {...agent, delivery_allowed: false}).admitted, false); }); +test("qualified scoped delivery retains its exact identity and current channel permission", () => { + const packet: JsonObject = {...source("qualified"), normal_delivery_allowed: false, + safe_bypass_allowed: true, safe_bypass_kind: "scoped_user_gate_fallback", + selected_todo: {todo_id: "todo_target", selection_binding: "pending_action_selection"}, + scoped_user_gate_fallback: {selected_executable: {todo_id: "todo_target"}}}; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", agent).admitted, true); + assert.equal(packet.normal_delivery_allowed, false); + for (const patch of [ + {safe_bypass_allowed: false}, {safe_bypass_kind: "unrelated_bypass"}, + {scoped_user_gate_fallback: null}, + {scoped_user_gate_fallback: {selected_executable: {todo_id: "todo_other"}}}, + {action_selection_qualification: {state: "deferred", requested_todo_id: "todo_target"}}, + {action_selection_qualification: {state: "qualified", requested_todo_id: "todo_other"}}, + {ok: false}, {should_run: false}, + ]) { + assert.equal(currentActionSelectionAdmission({...packet, ...patch}, "todo_target", agent).admitted, + false, JSON.stringify(patch)); + } + for (const facts of [{must_attempt: false, delivery_allowed: true}, + {must_attempt: true, delivery_allowed: false}, {must_attempt: true, delivery_allowed: null}]) { + assert.equal(currentActionSelectionAdmission(packet, "todo_target", facts).admitted, false); + } + assert.equal(currentActionSelectionAdmission(packet, "todo_other", agent).admitted, false); +}); + test("pending delivery cannot borrow root permission from a denied or missing channel grant", () => { const packet = {...source(), normal_delivery_allowed: true, selected_todo: {todo_id: "todo_target", selection_binding: "pending_action_selection"}}; diff --git a/tests/control_plane_ts/interaction_contract.test.ts b/tests/control_plane_ts/interaction_contract.test.ts index ffbbd78bf1..adf03fee21 100644 --- a/tests/control_plane_ts/interaction_contract.test.ts +++ b/tests/control_plane_ts/interaction_contract.test.ts @@ -50,6 +50,65 @@ test("only a positively qualified unbound scheduled wait can omit observation", assert.deepEqual(projectBlockedSuccessorWaitObservation({...request, effective_action: "normal_run"}), {required: false}); }); +for (const kind of ["todo_done", "monitor_changed"]) { + test(`only a known pending ${kind} source can omit unbound successor observation`, () => { + const todo = {todo_id: "todo_waiting", status: "open", role: "agent", + task_class: "advancement_task", resume_when: `${kind}:todo_dependency`, resume_monitor_generation: 3}; + const source = {todo_id: "todo_dependency", status: "open", role: "agent", + task_class: kind === "todo_done" ? "advancement_task" : "continuous_monitor", + archive_state: "active", material_change_generation: 3, + next_due_at: "2099-01-01T00:00:00Z", expires_at: "2100-01-01T00:00:00Z"}; + const evaluate = (sources: JsonObject[]) => (evaluateTodoResumeConditions({ + schema_version: "todo_resume_evaluation_request_v0", items: [todo], source_items: sources, + evaluated_at: "2026-09-01T00:00:00Z", + }).conditions as JsonObject[])[0].condition as JsonObject; + const condition = evaluate([source]); + 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: "open", selected_todo_claimed_by: "agent-a", + resume_when: todo.resume_when, resume_monitor_generation: todo.resume_monitor_generation, + resume_condition: condition}; + const request = {effective_action: "agent_scope_wait", agent_id: "agent-a", vision_wait_state: wait}; + const before = structuredClone(request); + for (const heartbeat_receipt of [undefined, {status: "committed"}, {status: "replayed"}]) { + assert.deepEqual(projectBlockedSuccessorWaitObservation({...request, heartbeat_receipt}), {required: false}); + } + assert.deepEqual(request, before); + const proofFields = kind === "todo_done" + ? ["target_status", "target_task_class", "target_archive_state", "target_todo_id"] + : ["target_status", "target_task_class", "baseline_generation", "material_change_generation", "target_todo_id", + "target_archive_state", "target_next_due_at", "target_expires_at", "evaluated_at", "clock_provider"]; + const incomplete = proofFields.map(field => { + const copy = {...condition}; + delete copy[field]; + return {vision_wait_state: {...wait, resume_condition: copy}}; + }); + for (const patch of [ + ...incomplete, + {vision_wait_state: {...wait, resume_condition: evaluate([])}}, + {vision_wait_state: {...wait, resume_condition: evaluate([{...source, status: "done"}])}}, + {vision_wait_state: {...wait, resume_condition: {...condition, target_todo_id: "todo_waiting"}}}, + {vision_wait_state: {...wait, selected_todo_claimed_by: "other-agent"}}, + {vision_wait_state: {...wait, waiting_todo_count: 2}}, + {heartbeat_receipt: {status: "committed", settlement_identity: {todo_id: todo.todo_id}}}, + {heartbeat_receipt: {status: "replayed", pending_action_selection: {}}}, + {heartbeat_receipt: {status: "pending"}}, + {agent_id: "other-agent"}, + ...(kind === "monitor_changed" ? [ + {vision_wait_state: {...wait, resume_monitor_generation: undefined}}, + {vision_wait_state: {...wait, resume_monitor_generation: 2}}, + {vision_wait_state: {...wait, resume_condition: evaluate([{...source, material_change_generation: 4}])}}, + {vision_wait_state: {...wait, resume_condition: evaluate([{...source, next_due_at: undefined}])}}, + {vision_wait_state: {...wait, resume_condition: evaluate([{...source, expires_at: "2026-01-01T00:00:00Z"}])}}, + {vision_wait_state: {...wait, resume_condition: evaluate([{...source, expires_at: source.next_due_at}])}}, + {vision_wait_state: {...wait, resume_condition: evaluate([{...source, archive_state: "archived"}])}}, + ] : []), + ]) { + assert.deepEqual(projectBlockedSuccessorWaitObservation({...request, ...patch}), {required: true}, JSON.stringify(patch)); + } + }); +} + // The host-facing type cannot represent delivery without an agent attempt. // @ts-expect-error delivery requires must_attempt=true const invalidDeliveryChannel: AgentInteractionChannel = { diff --git a/tests/control_plane_ts/resume_planning.test.ts b/tests/control_plane_ts/resume_planning.test.ts index f0cbaff138..019b0bc4a8 100644 --- a/tests/control_plane_ts/resume_planning.test.ts +++ b/tests/control_plane_ts/resume_planning.test.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; import { projectTodoResumePlanning, RESUME_PLANNING_REQUEST } from "../../loopx/control_plane/todos/resume_planning.ts"; +import {evaluateTodoResumeConditions} from "../../loopx/control_plane/todos/resume_condition.ts"; function fact(id: string, overrides: JsonObject = {}): JsonObject { return { @@ -27,6 +28,32 @@ function request(overrides: JsonObject = {}): JsonObject { }; } +test("only the actual pending Monitor generation can cover an in-scope successor", () => { + const waiter = {todo_id: "todo_waiting", role: "agent", status: "open", task_class: "advancement_task", + text: "Validate after the watched artifact changes.", resume_when: "monitor_changed:todo_source", + resume_monitor_generation: 0}; + const source = {todo_id: "todo_source", role: "agent", status: "open", task_class: "continuous_monitor"}; + const evaluate = (items: JsonObject[]) => (evaluateTodoResumeConditions({ + schema_version: "todo_resume_evaluation_request_v0", items: [waiter], source_items: items, + }).conditions as JsonObject[])[0].condition as JsonObject; + const condition = evaluate([source]); + const waiting = fact(waiter.todo_id, {status: "open", claim: "agent-a", resume: waiter.resume_when, + target_id: source.todo_id, target_status: "open", target_class: "continuous_monitor", + payload: {...waiter, resume_ready: false, resume_condition: condition}}); + const project = (row: JsonObject) => { + const input = request(); + (input.sources as JsonObject).resume_blocked_items = [row]; + return projectTodoResumePlanning(input).blocked_successor_items as JsonObject[]; + }; + assert.equal(project(waiting)[0].todo_id, waiter.todo_id); + for (const changed of [evaluate([]), evaluate([{...source, material_change_generation: 1}]), + {...condition, baseline_generation: undefined}, {...condition, target_status: undefined}]) { + assert.equal(project({...waiting, payload: {...waiter, resume_ready: false, resume_condition: changed}}).length, 0); + } + assert.equal(project({...waiting, claim: "another-agent"}).length, 0); + assert.equal(project({...waiting, excluded: ["agent-a"]}).length, 0); +}); + test("capacity evaluation and claim lanes use one snapshot without mutating it", () => { const input = request({ available_capabilities: ["compiler"] }); (input.sources as JsonObject).items = [fact("todo_capacity", { claim: "agent-a" }), diff --git a/tests/control_plane_ts/scheduler_arbitration.test.ts b/tests/control_plane_ts/scheduler_arbitration.test.ts index 708a6a0efd..4b15a55cfa 100644 --- a/tests/control_plane_ts/scheduler_arbitration.test.ts +++ b/tests/control_plane_ts/scheduler_arbitration.test.ts @@ -205,3 +205,51 @@ test("unproven waits cannot relax finite scheduler supervision", () => { .scheduled_recheck_after_seconds, undefined); } }); + +test("only a proven dependency wait keeps source polling alive without a Turn obligation", () => { + for (const kind of ["todo_done", "monitor_changed"]) { + const input = datedWait(); + const wait = input.vision_wait_state as JsonObject; + const resume = `${kind}:todo_dependency`; + const condition = {schema_version: "todo_resume_condition_v0", kind, resume_when: resume, + target: "todo_dependency", target_todo_id: "todo_dependency", satisfied: false, + target_status: "open", target_archive_state: null, + target_task_class: kind === "todo_done" ? "advancement_task" : "continuous_monitor", + baseline_generation: 0, material_change_generation: 0, + target_next_due_at: "2099-01-01T00:00:00Z", target_expires_at: null, + evaluated_at: "2026-09-01T00:00:00Z", clock_provider: "runtime_clock"}; + input.vision_wait_state = {...wait, resume_when: resume, resume_monitor_generation: 0, resume_condition: condition}; + assert.deepEqual(projectSchedulerArbitration(input), {mode: "agent_scope_wait", errors: [], + disposition: "agent_scope_wait", reason_code: "interaction_agent_scope_wait", pending_dependency_wait: true}); + const monitor = {...input, ...facts("monitor_quiet_skip", false, false, false, true), + effective_action: "monitor_quiet_skip"}; + assert.deepEqual(projectSchedulerArbitration(monitor), {mode: "monitor_quiet_skip", errors: [], + disposition: "monitor_wait", reason_code: "interaction_monitor_quiet_wait", pending_dependency_wait: true}); + for (const patch of [{vision_wait_state: undefined}, {heartbeat_receipt: {status: "pending"}}, + {heartbeat_receipt: {status: "replayed", settlement_identity: {todo_id: "todo_scheduled"}}}, + {cli_channel: {spend_allowed_now: true}}, facts("normal_run"), + facts("user_gate", true, false, false, false)]) { + assert.equal(projectSchedulerArbitration({...monitor, ...patch}).pending_dependency_wait, undefined); + } + const pending = input.vision_wait_state as JsonObject; + for (const patch of [ + {vision_wait_state: {...pending, resume_condition: {...condition, target_status: undefined}}}, + {vision_wait_state: {...pending, selected_todo_claimed_by: "another-agent"}}, + {vision_wait_state: {...pending, waiting_todo_count: 2}}, + {heartbeat_receipt: {status: "committed", settlement_identity: {todo_id: wait.selected_todo_id}}}, + {heartbeat_receipt: {status: "replayed", pending_action_selection: {}}}, + {effective_action: "normal_run"}, + {cli_channel: {spend_allowed_now: true}}, + {cli_channel: {spend_after_validation: true}}, + facts("user_gate", true, false, false, false), + facts("monitor_quiet_skip", false, false, false, true), + facts("terminal_no_followup", false, false, false, true), + facts("successor_replan_required", false, true, false, false), + facts(), + ]) { + const result = projectSchedulerArbitration({...input, ...patch}); + assert.equal(result.pending_dependency_wait, undefined, JSON.stringify(patch)); + assert.equal(result.scheduled_recheck_after_seconds, undefined); + } + } +});