diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index b13a9a2843..3b73c9d358 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -385,7 +385,7 @@ export function createEffectRuntimeHandlers( lazyHandler(() => import("./quota/unsettled_host_turn_recovery.ts"), ({reduceUnsettledHostTurnRecovery}) => reduceUnsettledHostTurnRecovery), ], ["quota.turn_envelope.evaluate", lazyHandler(() => import("./quota/turn_envelope.ts"), ({evaluateTurnEnvelope}) => evaluateTurnEnvelope)], - ["quota.run_decision", lazyHandler(() => import("./quota/run_decision.ts"), ({resolveQuotaRunDecision}) => resolveQuotaRunDecision)], + ["quota.run_decision", lazyHandler(() => import("./quota/run_decision.ts"), ({admitQuotaRunDecision}) => admitQuotaRunDecision)], ["quota.scoped_override.project", lazyHandler(() => import("./quota/scoped_override.ts"), ({projectScopedOverride}) => projectScopedOverride)], ["task_lease.acquire.native", lazyHandler(() => import("./work_items/task_lease_acquire.ts"), ({executeTaskLeaseAcquire}) => executeTaskLeaseAcquire)], ["task_lease.inspect.native", lazyHandler(() => import("./work_items/task_lease_inspection.ts"), ({inspectTaskLease}) => inspectTaskLease)], diff --git a/loopx/control_plane/quota/decision_summary.py b/loopx/control_plane/quota/decision_summary.py index 3d9c5c2764..b97493042c 100644 --- a/loopx/control_plane/quota/decision_summary.py +++ b/loopx/control_plane/quota/decision_summary.py @@ -151,6 +151,7 @@ class QuotaRunDecision: quota: dict[str, Any] replan_decision_allowed: bool goal_route_hint_override: dict[str, Any] | None = None + agent_scope_frontier: dict[str, Any] | None = None def refine_quota_recommended_action( @@ -263,6 +264,7 @@ def resolve_quota_run_decision( registered_agent_ids: list[str], goal_frontier_projection: dict[str, Any] | None, task_orchestration_contract: dict[str, Any] | None, + vision_initialization_context: dict[str, Any] | None = None, ) -> QuotaRunDecision: """Apply the ordered final guards for one quota run decision.""" @@ -294,6 +296,7 @@ def resolve_quota_run_decision( "automation_prompt_upgrade_required": automation_prompt_upgrade_required, "replan_required": replan_required, "replan_applies": bool(scope.get("applies")), + "vision_initialization_context": vision_initialization_context, "goal_health_ok": goal_health_ok, "inbox_reply_due": inbox_reply_due, "inbox_material_review_due": inbox_material_review_due, diff --git a/loopx/control_plane/quota/run_decision.ts b/loopx/control_plane/quota/run_decision.ts index 751c96b286..950ebad9b5 100644 --- a/loopx/control_plane/quota/run_decision.ts +++ b/loopx/control_plane/quota/run_decision.ts @@ -2,6 +2,7 @@ import type {JsonObject} from "../effect_program.ts"; import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; import {requireBoolean, requireJsonObject} from "../runtime_decode.ts"; import {EffectiveAction} from "./effective_action.generated.ts"; +import {boundVisionBaselineAdmission} from "./vision_baseline_admission.ts"; interface RunFacts { normal_delivery_allowed: boolean; @@ -39,6 +40,7 @@ interface RunDecision extends JsonObject { quota_patch: JsonObject; replan_decision_allowed: boolean; goal_route_hint_override: JsonObject | null; + agent_scope_frontier: JsonObject | null; } function text(value: unknown, label: string): string { @@ -214,6 +216,7 @@ export function resolveQuotaRunDecision(params: JsonObject): JsonObject { should_run: false, effective_action: EffectiveAction.QUOTA_SKIP, reason: facts.reason, state: facts.state, quota_patch: {}, replan_decision_allowed: false, goal_route_hint_override: null, + agent_scope_frontier: null, }; applyDeliveryGuards(facts, decision); decision.effective_action = quotaEffectiveAction(facts, decision); @@ -224,3 +227,7 @@ export function resolveQuotaRunDecision(params: JsonObject): JsonObject { facts.task_orchestration_contract, facts.task_orchestration_lane_count); return decision; } + +export async function admitQuotaRunDecision(params: JsonObject): Promise { + return boundVisionBaselineAdmission(params, resolveQuotaRunDecision(params)); +} diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py index 5635697ada..8cfef490f6 100644 --- a/loopx/control_plane/quota/should_run.py +++ b/loopx/control_plane/quota/should_run.py @@ -670,6 +670,16 @@ def build_quota_should_run( receipt_bound_guard_turn_instance_id=receipt_bound_guard_turn_instance_id, receipt_bound_replan_review=receipt_bound_replan_review, ) + if runtime_root is not None and turn_instance_id and prepared.agent_frontier_id: + prepared.vision_initialization_context = { + "runtime_root": str(runtime_root), "goal_id": safe_goal_id, + "agent_id": prepared.agent_frontier_id, "turn_instance_id": turn_instance_id, + "receipt_bound": bool(receipt_bound_todo_id or receipt_bound_replan_obligation_id), + "todo_summary": prepared.agent_todo_summary, + "todo_items": prepared.agent_todo_planning_source_items, + "replan_obligation": prepared.replan_obligation, + "acceptance_gaps": prepared.goal_frontier_projection.get("acceptance_gaps"), + } route = _resolve_quota_should_run_route(prepared) route = _apply_selected_todo_guards( prepared, diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index a7071d940c..43fb27ef37 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -935,6 +935,7 @@ def _resolve_quota_should_run_route( registered_agent_ids=prepared.registered_agent_ids, goal_frontier_projection=prepared.goal_frontier_projection, task_orchestration_contract=prepared.task_orchestration_contract, + vision_initialization_context=prepared.vision_initialization_context, ) if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED: fields = settled_replay_fields() @@ -1179,7 +1180,8 @@ def _resolve_quota_should_run_route( agent_lane_next_action=agent_lane_next_action, ) agent_scope_frontier = ( - _selected_deferred_replan_frontier(prepared) + run_decision.agent_scope_frontier + or _selected_deferred_replan_frontier(prepared) or _agent_scope_no_candidate_frontier( agent_identity=prepared.agent_identity, agent_todo_summary=prepared.agent_todo_summary, diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index 0d856b0265..b6920fd2b6 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -185,6 +185,7 @@ class _QuotaDecisionPreparation: resolved_scheduler_context: SchedulerExecutionContextResolution delivery_continuity_anchor: dict[str, Any] | None delivery_continuity_todo: dict[str, Any] | None + vision_initialization_context: dict[str, Any] | None = None @property def inbox_priority_due(self) -> bool: diff --git a/loopx/control_plane/quota/vision_baseline_admission.ts b/loopx/control_plane/quota/vision_baseline_admission.ts new file mode 100644 index 0000000000..fe158658f4 --- /dev/null +++ b/loopx/control_plane/quota/vision_baseline_admission.ts @@ -0,0 +1,85 @@ +import {createHash} from "node:crypto"; +import {readFile} from "node:fs/promises"; +import {join} from "node:path"; +import type {JsonObject} from "../effect_program.ts"; +import {durableWriteJson, withFileMutationLock} from "../effect_runtime_io.ts"; +import {jsonObject, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; +import {AgentScopeFrontierAction} from "../agents/agent_scope_frontier.generated.ts"; +import {readAgent, readExcludedAgents} from "../todos/read_codec.ts"; + +const BASELINE_ADMISSION_SCHEMA = "vision_baseline_admission_v0"; + +interface VisionBaselineAdmission extends JsonObject { + schema_version: typeof BASELINE_ADMISSION_SCHEMA; + goal_id: string; + agent_id: string; + turn_instance_id: string; + replan_obligation_id: string | null; +} + +/** One initialization Turn per runtime/Goal/Agent; missing checkpoints cannot rearm it. */ +export async function boundVisionBaselineAdmission(params: JsonObject, decision: JsonObject): Promise { + const context = params.vision_initialization_context == null ? null + : requireJsonObject(params.vision_initialization_context, "vision_initialization_context"); + if (!context || decision.replan_decision_allowed !== true || context.receipt_bound === true || + decision.self_repair_allowed === true || decision.capability_repair_allowed === true || + decision.workspace_repair_allowed === true) return decision; + const agent = requireNonEmptyString(context.agent_id, "agent_id"); + const summary = requireJsonObject(context.todo_summary, "todo_summary"); + const counts = jsonObject(summary.work_counts); + // Unknown or incomplete projections cannot prove that all real work is blocked. + if (counts?.complete !== true || counts.agent_id !== agent || counts.advancement !== 0 || + summary.monitor_due_count !== 0 || Number(summary.monitor_schedule_gap_count ?? 0) > 0) return decision; + const rows = context.todo_items; + if (!Array.isArray(rows)) throw new TypeError("todo_items must be an array"); + const blocked = rows.map(row => requireJsonObject(row, "Todo")).filter(row => + row.status === "blocked" && row.done !== true && row.resume_ready !== true && + (row.task_class === "advancement_task" || row.task_class === "blocker") && + (!readAgent(row.claimed_by) || readAgent(row.claimed_by) === agent) && + !readExcludedAgents(row.excluded_agents).includes(agent)); + if (!blocked.length) return decision; + const obligation = requireJsonObject(context.replan_obligation, "replan_obligation"); + const triggers = obligation.triggers; + const gaps = context.acceptance_gaps; + if (!Array.isArray(triggers) || !triggers.length || !Array.isArray(gaps) || !gaps.length || + !triggers.every(raw => jsonObject(raw)?.kind === "vision_checkpoint_missing") || + !gaps.every(raw => { + const gap = jsonObject(raw); + return gap?.kind === "vision_checkpoint_missing" && gap.missing_baseline === true && gap.agent_id === agent; + })) return decision; + const root = requireNonEmptyString(context.runtime_root, "runtime_root"); + const scope = {goal_id: requireNonEmptyString(context.goal_id, "goal_id"), agent_id: agent}; + const turn = requireNonEmptyString(context.turn_instance_id, "turn_instance_id"); + const path = join(root, "vision-baseline-admissions", + `${createHash("sha256").update(JSON.stringify(scope)).digest("hex")}.json`); + return withFileMutationLock(path, async () => { + let previous: JsonObject | null = null; + try { previous = requireJsonObject(JSON.parse(await readFile(path, "utf8")), "baseline admission"); } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + if (previous && (previous.schema_version !== BASELINE_ADMISSION_SCHEMA || + previous.goal_id !== scope.goal_id || previous.agent_id !== agent)) { + throw new TypeError("Vision baseline admission scope/schema mismatch"); + } + if (!previous) { + const admission: VisionBaselineAdmission = {schema_version: BASELINE_ADMISSION_SCHEMA, ...scope, + turn_instance_id: turn, replan_obligation_id: requireNonEmptyString(obligation.obligation_id, "obligation_id")}; + await durableWriteJson(path, admission); + return decision; + } + // Reentry of the admitted Turn must retain its original work and settlement identity. + if (requireNonEmptyString(previous.turn_instance_id, "admitted Turn") === turn) return decision; + const reason = "Vision baseline initialization was already admitted; the scoped frontier remains blocked"; + const frontier = {schema_version: "agent_scope_frontier_v1", agent_id: agent, + action: AgentScopeFrontierAction.AGENT_SCOPE_WAIT, blocks_delivery: true, + quiet_noop_allowed: true, spend_policy: "no quota spend while waiting for runnable work or an owner change", + reason, recommended_action: "Wait for executable work or an explicit owner/replan change; retain the pending Vision baseline duty.", + candidate_counts: {current_agent_claimed_advancement_count: 0, unclaimed_advancement_count: 0, + blocked_wait_count: blocked.length}, + vision_baseline_admission: previous}; + return {...decision, normal_delivery_allowed: false, recovery_delivery_allowed: false, + should_run: false, replan_decision_allowed: false, + effective_action: AgentScopeFrontierAction.AGENT_SCOPE_WAIT, reason, agent_scope_frontier: frontier}; + }); +} diff --git a/tests/control_plane/test_missing_vision_baseline_wait.py b/tests/control_plane/test_missing_vision_baseline_wait.py new file mode 100644 index 0000000000..a20a660446 --- /dev/null +++ b/tests/control_plane/test_missing_vision_baseline_wait.py @@ -0,0 +1,120 @@ +"""A retained missing baseline cannot repeatedly wake a blocked-only Agent.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys + +import pytest + +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item +from loopx.control_plane.scheduler.execution_context import GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT +from loopx.quota import build_quota_should_run + +GOAL = "baseline-wait-fixture" +AGENT = "agent-baseline" + + +def status(*, goal=GOAL, agent=AGENT, extra=()): + checkpoint = {"schema_version": "vision_checkpoint_v0", "agent_id": agent, + "required": True, "satisfied": False, "decision": "missing_required", + "missing_baseline": True, + "triggers": [{"kind": "material_delivery_outcome", "delivery_outcome": "outcome_progress"}]} + return quota_status_payload( + goal_id=goal, status="active", recommended_action="Wait for external input.", + agent_todo_items=[quota_todo_item(todo_id="todo_blocked", title="External input pending.", + status="blocked", claimed_by=agent), *extra], + latest_runs=[{"classification": "state_refreshed", "generated_at": "2026-01-01T00:00:00Z", + "agent_id": agent, "progress_scope": "agent_lane", "vision_checkpoint": checkpoint}], + coordination={"agent_model": "peer_v1", "registered_agents": [agent]}, + quota_extra={"allowed_slots": 12, "spent_slots": 1}) + + +def guard(payload, runtime, turn, *, goal=GOAL, agent=AGENT, **kwargs): + return build_quota_should_run( + payload, goal_id=goal, agent_id=agent, runtime_root=runtime, turn_instance_id=turn, + scheduler_execution_context=GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT, + include_scheduler_detail=True, **kwargs) + + +def assert_wait(result): + assert result["effective_action"] == "agent_scope_wait" + assert result["should_run"] is False + assert result["normal_delivery_allowed"] is False + assert result["execution_obligation"]["must_attempt_work"] is False + channel = result["interaction_contract"]["agent_channel"] + assert channel["delivery_allowed"] is False + assert channel["quiet_noop_allowed"] is True + assert channel["must_attempt"] is False + assert result["scheduler_hint"]["action"] == "backoff_until_reassigned" + assert result["autonomous_replan_obligation"]["required"] is True + + +def test_first_initialization_then_fresh_turns_and_process_readback_wait(tmp_path): + payload = status() + first = guard(payload, tmp_path, "initialization") + assert first["effective_action"] == "autonomous_replan_required" + assert first["should_run"] is True + assert guard(payload, tmp_path, "initialization")["should_run"] is True + for turn in ("poll-1", "poll-2"): + assert_wait(guard(payload, tmp_path, turn)) + script = ( + "import json,sys; from pathlib import Path; " + "from tests.control_plane.test_missing_vision_baseline_wait import guard,assert_wait; " + "result=guard(json.load(sys.stdin),Path(sys.argv[1]),'process-readback'); " + "assert_wait(result); print(result['effective_action'])" + ) + completed = subprocess.run([sys.executable, "-c", script, str(tmp_path)], input=json.dumps(payload), + text=True, capture_output=True, check=True, + env={**os.environ, "LOOPX_USAGE_PING": "0"}) + assert completed.stdout.strip() == "agent_scope_wait" + assert len(list((tmp_path / "vision-baseline-admissions").glob("*.json"))) == 1 + + +@pytest.mark.parametrize("task_class", ["advancement_task", "continuous_monitor"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_real_work_and_due_monitor_resume_after_wait(tmp_path, task_class, provider): + from canonical_authority_fixture import initialize_canonical_authority + + guard(status(), tmp_path, "initialization") + assert_wait(guard(status(), tmp_path, "blocked")) + work = quota_todo_item(todo_id="todo_independent", title="Perform independent work.", + task_class=task_class, claimed_by=AGENT, archive_state="active", + **({"cadence": "30m", "next_due_at": "2000-01-01T00:00:00Z"} + if task_class == "continuous_monitor" else {})) + state = tmp_path / "state.md" + state.write_text("# Synthetic fixture\n\n## Agent Todo\n", encoding="utf-8") + initialize_canonical_authority( + tmp_path, GOAL, build_todo_runtime_shadow_projection( + goal_id=GOAL, todos=[work], handoff_mode="soft_claim"), + state_path=state, provider=provider) + result = guard(status(extra=[work]), tmp_path, "ready") + assert result["should_run"] is True + assert result["effective_action"] != "agent_scope_wait" + assert_wait(guard(status(), tmp_path, "blocked-again")) + + +def test_admission_is_scoped_to_goal_and_agent(tmp_path): + guard(status(), tmp_path, "initialization") + for goal, agent in (("another-goal", AGENT), (GOAL, "another-agent")): + result = guard(status(goal=goal, agent=agent), tmp_path, "initialization", goal=goal, agent=agent) + assert result["effective_action"] == "autonomous_replan_required" + assert len(list((tmp_path / "vision-baseline-admissions").glob("*.json"))) == 3 + + +def test_receipt_bound_replan_retains_original_settlement(tmp_path): + first = guard(status(), tmp_path, "initialization") + obligation = first["autonomous_replan_obligation"]["obligation_id"] + result = guard(status(), tmp_path, "original-owed-turn", + receipt_bound_replan_obligation_id=obligation) + assert result["should_run"] is True + assert result["autonomous_replan_obligation"]["obligation_id"] == obligation + + +def test_changed_checkpoint_timestamp_does_not_rearm_baseline(tmp_path): + guard(status(), tmp_path, "initialization") + updated = status() + updated["run_history"]["goals"][0]["latest_runs"][0]["generated_at"] = "2026-02-01T00:00:00Z" + assert_wait(guard(updated, tmp_path, "changed-checkpoint")) diff --git a/tests/control_plane_ts/fixtures/quota_run_decision.json b/tests/control_plane_ts/fixtures/quota_run_decision.json index 9e81c71c50..bbde525e54 100644 --- a/tests/control_plane_ts/fixtures/quota_run_decision.json +++ b/tests/control_plane_ts/fixtures/quota_run_decision.json @@ -33,7 +33,8 @@ "state": "eligible", "quota_patch": {}, "replan_decision_allowed": false, - "goal_route_hint_override": null + "goal_route_hint_override": null, + "agent_scope_frontier": null }, "cases": [ { diff --git a/tests/control_plane_ts/vision_baseline_admission.test.ts b/tests/control_plane_ts/vision_baseline_admission.test.ts new file mode 100644 index 0000000000..39c9b93f49 --- /dev/null +++ b/tests/control_plane_ts/vision_baseline_admission.test.ts @@ -0,0 +1,67 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import {mkdtemp, readFile, readdir, rm, writeFile} from "node:fs/promises"; +import {tmpdir} from "node:os"; +import {join} from "node:path"; +import {admitQuotaRunDecision} from "../../loopx/control_plane/quota/run_decision.ts"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; + +const fixture = JSON.parse(await readFile(new URL("./fixtures/quota_run_decision.json", import.meta.url), "utf8")); +function facts(root: string, turn = "initialization"): JsonObject { + return {...fixture.base_facts, replan_required: true, replan_applies: true, + vision_initialization_context: {runtime_root: root, goal_id: "baseline-fixture", agent_id: "agent-a", + turn_instance_id: turn, receipt_bound: false, + todo_summary: {work_counts: {complete: true, agent_id: "agent-a", advancement: 0}, monitor_due_count: 0}, + todo_items: [{todo_id: "todo_blocked", status: "blocked", task_class: "advancement_task", claimed_by: "agent-a"}], + replan_obligation: {obligation_id: "replan-baseline", triggers: [{kind: "vision_checkpoint_missing"}]}, + acceptance_gaps: [{kind: "vision_checkpoint_missing", missing_baseline: true, agent_id: "agent-a"}]}}; +} + +test("concurrent fresh Turns admit only one baseline initialization", async () => { + const root = await mkdtemp(join(tmpdir(), "vision-admission-")); + try { + const results = await Promise.all(["one", "two", "three"].map(turn => admitQuotaRunDecision(facts(root, turn)))); + assert.equal(results.filter(result => result.should_run === true).length, 1); + assert.equal(results.filter(result => result.effective_action === "agent_scope_wait").length, 2); + } finally { await rm(root, {recursive: true, force: true}); } +}); + +test("real obligations and unknown or runnable frontiers preserve their owner decisions", async () => { + const root = await mkdtemp(join(tmpdir(), "vision-admission-")); + try { + await admitQuotaRunDecision(facts(root)); + const patches: JsonObject[] = [ + {receipt_bound: true}, + {replan_obligation: {triggers: [{kind: "vision_checkpoint_missing"}, {kind: "periodic_review_due"}]}}, + {acceptance_gaps: [{kind: "vision_checkpoint_missing", missing_baseline: false, agent_id: "agent-a"}]}, + {todo_summary: {work_counts: {complete: false, agent_id: "agent-a", advancement: 0}, monitor_due_count: 0}}, + {todo_summary: {work_counts: {complete: true, agent_id: "agent-a", advancement: 1}, monitor_due_count: 0}}, + {todo_summary: {work_counts: {complete: true, agent_id: "agent-a", advancement: 0}, monitor_due_count: 1}}, + {todo_items: [{status: "blocked", task_class: "advancement_task", claimed_by: "agent-b"}]}, + {todo_items: [{status: "blocked", task_class: "advancement_task", excluded_agents: ["agent-a"]}]}, + {todo_items: [{status: "blocked", task_class: "advancement_task", resume_ready: true}]}, + ]; + for (const patch of patches) { + const request = facts(root, "fresh"); + Object.assign(request.vision_initialization_context as JsonObject, patch); + assert.equal((await admitQuotaRunDecision(request)).effective_action, "autonomous_replan_required"); + } + for (const patch of [{inbox_reply_due: true}, {workspace_guard: {reason: "Repair workspace"}}, + {automation_prompt_upgrade_required: true}, {replan_required: false}]) { + const result = await admitQuotaRunDecision({...facts(root, "fresh"), ...patch}); + assert.notEqual(result.effective_action, "agent_scope_wait"); + } + } finally { await rm(root, {recursive: true, force: true}); } +}); + +test("malformed durable state cannot silently rearm the allowance", async () => { + const root = await mkdtemp(join(tmpdir(), "vision-admission-")); + try { + await admitQuotaRunDecision(facts(root)); + const directory = join(root, "vision-baseline-admissions"); + const [file] = await readdir(directory); + await writeFile(join(directory, file), JSON.stringify({schema_version: "vision_baseline_admission_v0", + goal_id: "another-goal", agent_id: "agent-a", turn_instance_id: "first"})); + await assert.rejects(admitQuotaRunDecision(facts(root, "fresh")), /scope\/schema mismatch/); + } finally { await rm(root, {recursive: true, force: true}); } +});