diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index feca1885fe..d062ef9704 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -226,6 +226,10 @@ Python retains registry, document/material IO, independent Vision-wait consumers and compatibility exports. Its replaced summary, filter and result coordination is retired. Persisted completion spellings use shared token data and compatibility codecs without scalar RPCs; +Vision-wait state projection now composes typed resume planning, current-agent +blocker eligibility, causal lineage coverage and result assembly in one TypeScript +effect. Python retains source-snapshot encoding, result validation, historical +wait-state decoding and independent consumers. `todo.summary.project`, `todo.completion_state.normalize` and `todo.completion_state.require_metadata` are retired after their production callers exit. Existing semantic owners execute in process inside the transaction. diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 701cf99955..992e92a335 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -185,7 +185,10 @@ Host fence 和真实 provider IO。有界扫描只提供读取提示，写入锁 依赖与闭合判断仍使用完整源。 Python 保留 registry、文档／材料 IO、独立的 Vision-wait 消费者及兼容导出。 被替代的摘要、筛选与结果协调已退役。持久化 completion 拼写使用共享 token 数据与兼容 codec， -不再逐标量调用 RPC；实际生产调用者退出后，退役 `todo.summary.project`、 +不再逐标量调用 RPC； +Vision-wait 状态投影现在由一次 TypeScript effect 组合类型化 resume planning、当前 Agent blocker +资格、因果血缘覆盖和结果组装。Python 保留源快照编码、结果校验、历史等待态解码及独立消费者。 +实际生产调用者退出后，退役 `todo.summary.project`、 `todo.completion_state.normalize` 与 `todo.completion_state.require_metadata`。 事务在进程内执行既有语义 owner。 diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 2c1a1a8e6b..44044e12cb 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -364,6 +364,7 @@ export function createEffectRuntimeHandlers( ["goal.checkpoint_read_context.commit", lazyHandler(() => import("./goals/checkpoint_commit.ts"), ({commitCheckpoint}) => commitCheckpoint)], ["goal.checkpoint_read_context.inspect_replay", lazyHandler(() => import("./goals/checkpoint_commit.ts"), ({inspectCheckpointReplay}) => inspectCheckpointReplay)], ["goal.vision_wait.coverage", lazyHandler(() => import("./goals/vision_wait_coverage.ts"), ({projectVisionWaitCoverage}) => projectVisionWaitCoverage)], + ["goal.vision_wait.project", lazyHandler(() => import("./goals/vision_wait_coverage.ts"), ({projectGoalVisionWaitState}) => projectGoalVisionWaitState)], ["goal.shared_goal_alignment.project", lazyHandler(() => import("./goals/shared_goal_alignment.ts"), ({projectSharedGoalAlignment}) => projectSharedGoalAlignment)], ["goal.operator_actions.project", lazyHandler(() => import("./goals/operator_actions.ts"), ({projectGoalOperatorActions}) => projectGoalOperatorActions)], ["goal.amendment.execute", lazyHandler(() => import("./goals/goal_amendment_authority.ts"), ({goalAmendmentLocal}) => goalAmendmentLocal)], diff --git a/loopx/control_plane/goals/goal_vision_wait.py b/loopx/control_plane/goals/goal_vision_wait.py index 73844bc876..afb680418b 100644 --- a/loopx/control_plane/goals/goal_vision_wait.py +++ b/loopx/control_plane/goals/goal_vision_wait.py @@ -5,17 +5,22 @@ from typing import Any from ..effect_runtime import effect_runtime_result -from ..todos.todo_semantics import todo_item_excludes_agent -from ..todos.contract import ( - TODO_TASK_CLASS_BLOCKER, - normalize_todo_claimed_by, - normalize_todo_id, - normalize_todo_status, -) -from ..todos.resume_planning import project_todo_resume_planning +from ..todos.contract import normalize_todo_id +from ..todos.resume_planning import build_todo_resume_planning_request GOAL_VISION_WAIT_STATE_SCHEMA_VERSION = "goal_vision_wait_state_v0" -VISION_ACCEPTANCE_GAP_KIND = "vision_acceptance_gap" + + +def _vision_wait_blocker_rows(items: list[dict[str, Any]]) -> list[dict[str, Any]]: + fields = ("todo_id", "task_class", "status", "priority", "claimed_by", "excluded_agents", "reason") + return [{key: item[key] for key in fields if key in item} for item in items if isinstance(item, dict)] + + +def _vision_wait_lineage_rows(items: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None: + if items is None: + return None + fields = ("todo_id", "successor_todo_ids", "resume_condition") + return [{key: item[key] for key in fields if key in item} for item in items if isinstance(item, dict)] def exact_blocked_successor_wait_state(value: Any) -> dict[str, Any]: @@ -57,174 +62,6 @@ def exact_blocked_successor_frontier_identity(value: Any) -> str | None: ).hexdigest()[:16] -def _compact_resume_condition(value: Any) -> dict[str, Any]: - if not isinstance(value, dict): - return {} - compact = { - key: value.get(key) - for key in ( - "schema_version", - "resume_when", - "satisfied", - "kind", - "target", - "target_todo_id", - "target_status", - "target_archive_state", - "target_source_section", - "target_task_class", - "target_claimed_by", - "target_next_due_at", - "capability", - "provider", - "scheduled_for", - "evaluated_at", - "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. - compact["resume_receipt"] = value.get("resume_receipt") - compact["satisfied"] = False - return compact - - -def _acceptance_gap_causal_todo_ids( - gaps: list[dict[str, Any]], -) -> set[str]: - todo_ids: set[str] = set() - for gap in gaps: - for key in ( - "todo_id", - "current_todo_id", - "blocked_todo_id", - "completed_todo_id", - "successor_todo_id", - ): - todo_id = normalize_todo_id(gap.get(key)) - if todo_id: - todo_ids.add(todo_id) - for key in ( - "vision_todo_ids", - "lineage_todo_ids", - "successor_todo_ids", - "completed_todo_ids", - ): - raw_values = gap.get(key) - values = raw_values if isinstance(raw_values, list) else [] - todo_ids.update( - todo_id for value in values if (todo_id := normalize_todo_id(value)) - ) - return todo_ids - - -def _wait_lineage_edges(items: list[dict[str, Any]]) -> list[list[str]]: - edges: set[tuple[str, str]] = set() - for item in items: - todo_id = normalize_todo_id(item.get("todo_id")) - if not todo_id: - continue - for value in item.get("successor_todo_ids") or []: - successor = normalize_todo_id(value) - if successor: - edges.add((todo_id, successor)) - condition = item.get("resume_condition") - if isinstance(condition, dict) and condition.get("satisfied") is False: - target = normalize_todo_id(condition.get("target_todo_id")) - if target: - # A prerequisite can explain its waiting successor, never the - # reverse: shared prerequisites do not cover unrelated siblings. - edges.add((target, todo_id)) - return [list(edge) for edge in sorted(edges)] - - -def _covered_wait_items( - *, - agent_todo_summary: dict[str, Any] | None, - agent_id: str | None, - causal_todo_ids: set[str], - source_items: list[dict[str, Any]] | None, - lineage_source_items: list[dict[str, Any]] | None, -) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: - if source_items is not None: - # Source rows have already passed the canonical resume evaluator. Never - # rebuild them from the quota's bounded deferred/backlog display lanes. - agent_todo_summary = { - "items": source_items, - "deferred_items": [ - i for i in source_items if i.get("status") == "deferred" - ], - "resume_blocked_items": [ - i - for i in source_items - if i.get("resume_ready") is False and i.get("status") == "open" - ], - "current_agent_blocker_items": source_items, - } - - raw_blockers = (agent_todo_summary or {}).get("current_agent_blocker_items") - blocker_items = raw_blockers if isinstance(raw_blockers, list) else [] - safe_agent_id = normalize_todo_claimed_by(agent_id) - blocker_items = [ - item - for item in blocker_items - if isinstance(item, dict) - and safe_agent_id is not None - and item.get("task_class") == TODO_TASK_CLASS_BLOCKER - and normalize_todo_status(item.get("status")) == "blocked" - and str(item.get("reason") or "").strip() - and normalize_todo_claimed_by(item.get("claimed_by")) == safe_agent_id - and not todo_item_excludes_agent(item, agent_id=safe_agent_id) - ] - candidates = project_todo_resume_planning( - agent_todo_summary or {}, agent_id=agent_id - )["blocked_successor_items"] - coverage = effect_runtime_result( - "goal.vision_wait.coverage", - { - "causal_todo_ids": sorted(causal_todo_ids), - "waiting_todo_ids": [i["todo_id"] for i in candidates if i.get("todo_id")], - "blocker_todo_ids": [ - i["todo_id"] for i in blocker_items if i.get("todo_id") - ], - "edges": _wait_lineage_edges( - lineage_source_items - if lineage_source_items is not None - else source_items - if source_items is not None - else candidates - ), - }, - ) - if ( - not isinstance(coverage, dict) - or coverage.get("schema_version") != "vision_wait_coverage_v0" - ): - raise RuntimeError("TypeScript vision wait coverage shape mismatch") - if coverage.get("covered") is not True: - return [], [] - witnesses = set(coverage["witness_todo_ids"]) - blocker_items = [i for i in blocker_items if i.get("todo_id") in witnesses] - candidates = ( - [] - if blocker_items - else [i for i in candidates if i.get("todo_id") in witnesses] - ) - return blocker_items, candidates - - def build_goal_vision_wait_state( *, agent_todo_summary: dict[str, Any] | None, @@ -241,98 +78,34 @@ def build_goal_vision_wait_state( checkpoints and closed-stage successor requirements remain strict. """ - gaps = [gap for gap in (acceptance_gaps or []) if isinstance(gap, dict)] - if not gaps or any(gap.get("kind") != VISION_ACCEPTANCE_GAP_KIND for gap in gaps): - return None - if selectable_advancement_count > 0: - return None - - # A gap with no causal link cannot borrow another gap's wait witness. - if any(not _acceptance_gap_causal_todo_ids([gap]) for gap in gaps): - return None - causal_todo_ids = _acceptance_gap_causal_todo_ids(gaps) - if agent_todo_summary: - for proof in (agent_todo_summary or {}).get("vision_wait_states") or []: - if ( - isinstance(proof, dict) - and proof.get("schema_version") == GOAL_VISION_WAIT_STATE_SCHEMA_VERSION - and proof.get("state") == "waiting" - and proof.get("agent_id") == agent_id - and proof.get("causal_todo_ids") == sorted(causal_todo_ids) - ): - return dict(proof) - blocker_items, candidates = _covered_wait_items( - agent_todo_summary=agent_todo_summary, - agent_id=agent_id, - causal_todo_ids=causal_todo_ids, - source_items=source_items, - lineage_source_items=lineage_source_items, - ) - if candidates: - selected = candidates[0] - waiting_todo_ids = [ - todo_id - for todo_id in ( - normalize_todo_id(item.get("todo_id")) for item in candidates[:5] - ) - if todo_id - ] - payload: dict[str, Any] = { - "schema_version": GOAL_VISION_WAIT_STATE_SCHEMA_VERSION, - "state": "waiting", - "reason_code": "exact_blocked_successor", - "agent_id": agent_id, - "waiting_todo_count": len(candidates), - "waiting_todo_ids": waiting_todo_ids, - "selected_todo_id": normalize_todo_id(selected.get("todo_id")), - "selected_todo_status": selected.get("status"), - "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") - ), - "causal_todo_ids": sorted(causal_todo_ids), - "deferred_acceptance_gap_count": len(gaps), - "deferred_acceptance_gap_kinds": [VISION_ACCEPTANCE_GAP_KIND], - "automatic_resume": True, - "resume_behavior": ( - "when resume_ready becomes true, restore ordinary open-todo or " - "deferred-successor routing without closing the active vision" - ), + summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {} + planning_input = summary + if source_items is not None: + planning_input = { + "items": source_items, + "deferred_items": [item for item in source_items if item.get("status") == "deferred"], + "resume_blocked_items": [ + item for item in source_items + if item.get("resume_ready") is False and item.get("status") == "open" + ], + "current_agent_blocker_items": source_items, } - return {key: value for key, value in payload.items() if value is not None} - - if not blocker_items: - return None - selected = blocker_items[0] - waiting_todo_ids = [ - todo_id - for todo_id in ( - normalize_todo_id(item.get("todo_id")) for item in blocker_items[:5] - ) - if todo_id - ] - payload = { - "schema_version": GOAL_VISION_WAIT_STATE_SCHEMA_VERSION, - "state": "waiting", - "reason_code": "current_agent_blocker", + blocker_items = source_items if source_items is not None else summary.get("current_agent_blocker_items") + wait_summary = {"vision_wait_states": summary.get("vision_wait_states"), + "current_agent_blocker_items": _vision_wait_blocker_rows(blocker_items or [])} + lineage_items = lineage_source_items if lineage_source_items is not None else source_items + result = effect_runtime_result("goal.vision_wait.project", { + "agent_todo_summary": wait_summary, "agent_id": agent_id, - "waiting_todo_count": len(blocker_items), - "waiting_todo_ids": waiting_todo_ids, - "selected_todo_id": normalize_todo_id(selected.get("todo_id")), - "selected_todo_status": selected.get("status"), - "selected_todo_priority": selected.get("priority"), - "selected_todo_claimed_by": selected.get("claimed_by"), - "blocker_reason": str(selected.get("reason") or "").strip(), - "causal_todo_ids": sorted(causal_todo_ids), - "deferred_acceptance_gap_count": len(gaps), - "deferred_acceptance_gap_kinds": [VISION_ACCEPTANCE_GAP_KIND], - "automatic_resume": False, - "resume_behavior": ( - "re-evaluate the active vision when the blocker is resolved or " - "superseded, or when runnable scoped advancement appears" - ), - } - return {key: value for key, value in payload.items() if value is not None} + "acceptance_gaps": acceptance_gaps or [], + "selectable_advancement_count": selectable_advancement_count, + "lineage_source_items": _vision_wait_lineage_rows(lineage_items), + "resume_planning_request": build_todo_resume_planning_request(planning_input, agent_id=agent_id), + }) + if result is not None and ( + not isinstance(result, dict) + or result.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION + or result.get("state") != "waiting" + ): + raise RuntimeError("TypeScript vision wait state shape mismatch") + return result diff --git a/loopx/control_plane/goals/vision_wait_coverage.ts b/loopx/control_plane/goals/vision_wait_coverage.ts index c34a39ae34..94c3b29e49 100644 --- a/loopx/control_plane/goals/vision_wait_coverage.ts +++ b/loopx/control_plane/goals/vision_wait_coverage.ts @@ -1,12 +1,16 @@ import type { JsonObject } from "../effect_program.ts"; -import { requireJsonObject, requireStringArray } from "../runtime_decode.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { optionalNonEmptyString, requireInteger, requireJsonObject, requireStringArray } from "../runtime_decode.ts"; +import { projectTodoResumePlanning } from "../todos/resume_planning.ts"; -/** Every causal obligation needs its own wait witness; shared prerequisites - * do not make sibling work interchangeable. Inputs are evaluated Todo facts, - * never model-maintained alternate-route declarations. */ -export function projectVisionWaitCoverage(value: unknown): JsonObject { - const request = requireJsonObject(value, "vision wait coverage"); +const WAIT_STATE_SCHEMA = "goal_vision_wait_state_v0"; +const VISION_GAP = "vision_acceptance_gap"; +const CAUSAL_ID_KEYS = ["todo_id", "current_todo_id", "blocked_todo_id", "completed_todo_id", "successor_todo_id", + "vision_todo_ids", "lineage_todo_ids", "successor_todo_ids", "completed_todo_ids"] as const; + +/** Every causal obligation needs its own witness; shared prerequisites do not cover sibling work. */ +export function projectVisionWaitCoverage(requestPayload: unknown): JsonObject { + const request = requireJsonObject(requestPayload, "vision wait coverage"); const roots = requireStringArray(request.causal_todo_ids, "causal_todo_ids"); const waits = new Set(requireStringArray(request.waiting_todo_ids, "waiting_todo_ids")); const blockers = new Set(requireStringArray(request.blocker_todo_ids, "blocker_todo_ids")); @@ -26,21 +30,178 @@ export function projectVisionWaitCoverage(value: unknown): JsonObject { const pending = [root]; let covered = false; while (pending.length) { - const id = pending.pop()!; - if (seen.has(id)) continue; - seen.add(id); - if (waits.has(id) || blockers.has(id)) { - witnesses.add(id); - covered = true; - } - for (const next of graph.get(id) ?? []) pending.push(next); + const todoId = pending.pop()!; + if (seen.has(todoId)) continue; + seen.add(todoId); + if (waits.has(todoId) || blockers.has(todoId)) { witnesses.add(todoId); covered = true; } + for (const next of graph.get(todoId) ?? []) pending.push(next); } if (!covered) uncovered.add(root); } - return { - schema_version: "vision_wait_coverage_v0", - covered: roots.length > 0 && uncovered.size === 0, - witness_todo_ids: [...witnesses].sort(), - uncovered_todo_ids: [...uncovered].sort(), + return {schema_version: "vision_wait_coverage_v0", covered: roots.length > 0 && uncovered.size === 0, + witness_todo_ids: [...witnesses].sort(), uncovered_todo_ids: [...uncovered].sort()}; +} + +function normalizedTodoId(candidate: unknown): string | null { + if (typeof candidate !== "string") return null; + const id = candidate.trim().toLowerCase(); + return /^todo_[a-z0-9_-]{3,64}$/.test(id) ? id : null; +} + +function visionGapRows(value: unknown): JsonObject[] { + return Array.isArray(value) ? value.filter((gap): gap is JsonObject => + typeof gap === "object" && gap !== null && !Array.isArray(gap)) : []; +} + +function causalTodoIds(gaps: readonly JsonObject[]): string[] | null { + const causal = new Set(); + for (const gap of gaps) { + let linked = false; + for (const key of CAUSAL_ID_KEYS) { + const ids = Array.isArray(gap[key]) ? gap[key] as unknown[] : [gap[key]]; + for (const candidate of ids) { + const id = normalizedTodoId(candidate); + if (id) { causal.add(id); linked = true; } + } + } + if (!linked) return null; + } + return [...causal].sort(); +} + +function cachedWaitProof(summary: JsonObject, agentId: string | null, causalIds: string[]): JsonObject | null { + for (const proof of Array.isArray(summary.vision_wait_states) ? summary.vision_wait_states : []) { + if (typeof proof !== "object" || proof === null || Array.isArray(proof)) continue; + const cachedWait = proof as JsonObject; + if (cachedWait.schema_version === WAIT_STATE_SCHEMA && cachedWait.state === "waiting" && + cachedWait.agent_id === (agentId ?? null) && JSON.stringify(cachedWait.causal_todo_ids) === JSON.stringify(causalIds)) { + return cachedWait; + } + } + return null; +} + +function normalizedAgentId(candidate: unknown): string | null { + if (typeof candidate !== "string") return null; + const id = candidate.trim().toLowerCase().replaceAll(" ", "-"); + return /^[a-z][a-z0-9_.:@-]{0,79}$/.test(id) ? id : null; +} + +function currentAgentBlockers(rows: unknown, agentId: string | null): JsonObject[] { + if (!Array.isArray(rows) || !agentId) return []; + return rows.filter((rawRow): rawRow is JsonObject => { + if (typeof rawRow !== "object" || rawRow === null || Array.isArray(rawRow)) return false; + const blocker = rawRow as JsonObject; + const excludedRows = Array.isArray(blocker.excluded_agents) ? blocker.excluded_agents + : typeof blocker.excluded_agents === "string" ? blocker.excluded_agents.split(",") : []; + return blocker.task_class === "blocker" && blocker.status === "blocked" && + normalizedAgentId(blocker.claimed_by) === agentId && !excludedRows.some(id => normalizedAgentId(id) === agentId); + }); +} + +function waitLineageEdges(rows: unknown): string[][] { + const edges = new Set(); + for (const rawRow of Array.isArray(rows) ? rows : []) { + if (typeof rawRow !== "object" || rawRow === null || Array.isArray(rawRow)) continue; + const lineageRow = rawRow as JsonObject; + const id = normalizedTodoId(lineageRow.todo_id); + if (!id) continue; + for (const next of Array.isArray(lineageRow.successor_todo_ids) ? lineageRow.successor_todo_ids : []) { + const successor = normalizedTodoId(next); + if (successor) edges.add(JSON.stringify([id, successor])); + } + const condition = typeof lineageRow.resume_condition === "object" && lineageRow.resume_condition !== null + ? lineageRow.resume_condition as JsonObject : {}; + const target = condition.satisfied === false ? normalizedTodoId(condition.target_todo_id) : null; + if (target) edges.add(JSON.stringify([target, id])); + } + return [...edges].map(edge => JSON.parse(edge) as string[]); +} + +function projectWaitCoverage(causalIds: string[], candidates: JsonObject[], blockers: JsonObject[], lineageRows: unknown): JsonObject { + return projectVisionWaitCoverage({causal_todo_ids: causalIds, + waiting_todo_ids: candidates.map(candidate => candidate.todo_id).filter((id): id is string => typeof id === "string"), + blocker_todo_ids: blockers.map(blocker => blocker.todo_id).filter((id): id is string => typeof id === "string"), + edges: waitLineageEdges(lineageRows)}); +} + +function selectedWaitItems(blockers: JsonObject[], candidates: JsonObject[], witnesses: Set) { + const selectedBlockers = blockers.filter(blocker => witnesses.has(String(blocker.todo_id))); + const selectedCandidates = selectedBlockers.length ? [] : candidates.filter(candidate => witnesses.has(String(candidate.todo_id))); + return {waitItems: selectedCandidates.length ? selectedCandidates : selectedBlockers, automatic: selectedCandidates.length > 0}; +} + +function compactResumeCondition(resumeCondition: unknown): JsonObject { + const condition = typeof resumeCondition === "object" && resumeCondition !== null && !Array.isArray(resumeCondition) + ? resumeCondition as JsonObject : {}; + const keys = ["schema_version", "resume_when", "satisfied", "kind", "target", "target_todo_id", "target_status", + "target_archive_state", "target_source_section", "target_task_class", "target_claimed_by", "target_next_due_at", + "capability", "provider", "scheduled_for", "evaluated_at", "clock_provider", "material_change", + "material_change_generation", "baseline_generation", "invalid_state", "target_expires_at"]; + const compact: JsonObject = Object.fromEntries(keys.filter(key => key in condition).map(key => [key, condition[key]])); + if (condition.kind === "resume_at") compact.resume_receipt = condition.resume_receipt ?? null; + compact.satisfied = false; + return compact; +} + +function automaticResumeFields(selected: JsonObject): JsonObject { + const fields: JsonObject = {}; + for (const key of ["resume_when", "resume_monitor_generation"] as const) { + if (selected[key] !== undefined && selected[key] !== null) fields[key] = selected[key]; + } + fields.resume_condition = compactResumeCondition(selected.resume_condition); + return fields; +} + +function blockerReasonField(selectedTodo: JsonObject): JsonObject { + return typeof selectedTodo.reason === "string" && selectedTodo.reason.trim() + ? {blocker_reason: selectedTodo.reason.trim()} : {}; +} + +function waitStatePayload(input: {causalIds: string[]; gapCount: number; selectedTodo: JsonObject; waitItems: JsonObject[]; + agentId: string | null; automatic: boolean}): JsonObject { + const {causalIds, gapCount, selectedTodo, waitItems, agentId, automatic} = input; + const payload: JsonObject = { + schema_version: WAIT_STATE_SCHEMA, state: "waiting", + reason_code: automatic ? "exact_blocked_successor" : "current_agent_blocker", + agent_id: agentId, waiting_todo_count: waitItems.length, + waiting_todo_ids: waitItems.slice(0, 5).map(candidate => normalizedTodoId(candidate.todo_id)).filter((id): id is string => id !== null), + selected_todo_id: normalizedTodoId(selectedTodo.todo_id), selected_todo_status: selectedTodo.status ?? null, + selected_todo_priority: selectedTodo.priority ?? null, selected_todo_claimed_by: selectedTodo.claimed_by ?? null, + ...(automatic ? automaticResumeFields(selectedTodo) : blockerReasonField(selectedTodo)), + causal_todo_ids: causalIds, deferred_acceptance_gap_count: gapCount, + deferred_acceptance_gap_kinds: [VISION_GAP], automatic_resume: automatic, + resume_behavior: automatic + ? "when resume_ready becomes true, restore ordinary open-todo or deferred-successor routing without closing the active vision" + : "re-evaluate the active vision when the blocker is resolved or superseded, or when runnable scoped advancement appears", }; + return Object.fromEntries(Object.entries(payload).filter(([, field]) => field !== null)); +} + +function projectVisionWaitFromFacts(input: {request: JsonObject; gaps: JsonObject[]; causalIds: string[]}): JsonObject | null { + const {request, gaps, causalIds} = input; + const agentId = optionalNonEmptyString(request.agent_id, "agent_id"); + const summary = request.agent_todo_summary === null ? {} : requireJsonObject(request.agent_todo_summary, "agent_todo_summary"); + const cachedWait = cachedWaitProof(summary, agentId, causalIds); + if (cachedWait) return cachedWait; + const resumePlan = projectTodoResumePlanning(requireJsonObject(request.resume_planning_request, "resume_planning_request")); + const blockers = currentAgentBlockers(summary.current_agent_blocker_items, agentId); + const candidates = Array.isArray(resumePlan.blocked_successor_items) ? resumePlan.blocked_successor_items as JsonObject[] : []; + const lineageRows = Array.isArray(request.lineage_source_items) ? request.lineage_source_items : candidates; + const coverage = projectWaitCoverage(causalIds, candidates, blockers, lineageRows); + if (coverage.covered !== true) return null; + const waitSelection = selectedWaitItems(blockers, candidates, new Set(coverage.witness_todo_ids as string[])); + if (!waitSelection.waitItems.length) return null; + return waitStatePayload({causalIds, gapCount: gaps.length, selectedTodo: waitSelection.waitItems[0], + waitItems: waitSelection.waitItems, agentId, automatic: waitSelection.automatic}); +} + +/** Composes resume planning and causal witness evaluation in one runtime call. */ +export function projectGoalVisionWaitState(requestValue: unknown): JsonObject | null { + const request = requireJsonObject(requestValue, "vision wait state"); + const gaps = visionGapRows(request.acceptance_gaps); + const selectableCount = requireInteger(request.selectable_advancement_count, "selectable_advancement_count"); + if (!gaps.length || gaps.some(gap => gap.kind !== VISION_GAP) || selectableCount !== 0) return null; + const causalIds = causalTodoIds(gaps); + return causalIds ? projectVisionWaitFromFacts({request, gaps, causalIds}) : null; } diff --git a/tests/control_plane/test_goal_vision_blocked_successor.py b/tests/control_plane/test_goal_vision_blocked_successor.py index ed8b6a8ff5..79e16ae064 100644 --- a/tests/control_plane/test_goal_vision_blocked_successor.py +++ b/tests/control_plane/test_goal_vision_blocked_successor.py @@ -596,22 +596,11 @@ def test_current_agent_blocker_outranks_exact_blocked_successor() -> None: ] == WAITING_ID -@pytest.mark.parametrize( - ("claimed_by", "reason"), - [ - (PRIMARY_AGENT, "Peer-owned blocker."), - (AGENT_ID, None), - ], - ids=["other-agent", "missing-reason"], -) -def test_unscoped_or_malformed_blocker_does_not_defer_open_vision_gap( - claimed_by: str, - reason: str | None, -) -> None: +def test_peer_owned_blocker_does_not_defer_open_vision_gap() -> None: guard = _quota( _current_agent_blocker_status_payload( - claimed_by=claimed_by, - reason=reason, + claimed_by=PRIMARY_AGENT, + reason="Peer-owned blocker.", ) ) @@ -620,6 +609,16 @@ def test_unscoped_or_malformed_blocker_does_not_defer_open_vision_gap( assert guard["goal_frontier_projection"]["replan_required"] is True +@pytest.mark.parametrize("reason", [None, "", "Waiting for assignment."]) +def test_blocker_reason_text_does_not_change_wait_eligibility(reason: str | None) -> None: + guard = _quota(_current_agent_blocker_status_payload(reason=reason)) + + wait = guard["goal_frontier_projection"].get("vision_wait_state") + assert wait is not None + assert wait["reason_code"] == "current_agent_blocker" + assert wait["selected_todo_id"] == PROJECTED_BLOCKER_ID + + def test_unbound_legacy_blocker_ack_cannot_clear_vision_replan_obligation() -> None: obligation = { "schema_version": "autonomous_replan_obligation_v0", diff --git a/tests/control_plane/test_vision_wait_coverage.py b/tests/control_plane/test_vision_wait_coverage.py index 0be6961dfb..a506540d55 100644 --- a/tests/control_plane/test_vision_wait_coverage.py +++ b/tests/control_plane/test_vision_wait_coverage.py @@ -3,12 +3,14 @@ from __future__ import annotations import json +import shutil from pathlib import Path import pytest from loopx.cli import main as cli_main from loopx.control_plane.goals.goal_vision import normalize_goal_vision_packet +from loopx.control_plane.goals import goal_vision_wait from loopx.control_plane.scheduler.execution_context import ( scheduler_execution_context_for_runtime_profile, ) @@ -19,6 +21,7 @@ enforce_open_replan_writeback, ) from loopx.quota import build_quota_should_run +from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture GOAL = "fallback-wait-fixture" AGENT = "worker" @@ -98,6 +101,20 @@ def fallback_state(metadata: str) -> str: ) +def test_vision_wait_composition_uses_one_runtime_crossing(monkeypatch) -> None: + methods = [] + original = goal_vision_wait.effect_runtime_result + + def counted(method, request): + methods.append(method) + return original(method, request) + + monkeypatch.setattr(goal_vision_wait, "effect_runtime_result", counted) + decision() + + assert methods == ["goal.vision_wait.project"] + + @pytest.mark.parametrize( "extra_count,reverse", [(0, False), (8, False), (40, False), (40, True)] ) @@ -146,6 +163,7 @@ def state(metadata: str | None) -> str: runtime = tmp_path / "runtime" runs = runtime / "goals" / GOAL / "runs" runs.mkdir(parents=True) + initialize_canonical_source_fixture(runtime, GOAL, state_path=state_file) if not declared: assert ( cli_main( @@ -218,13 +236,26 @@ def state(metadata: str | None) -> str: assert cli_main(args) == 0 missing = json.loads(capsys.readouterr().out) assert missing["effective_action"] == "autonomous_replan_required" + runtime_revision = 0 + + def reload_source_authority() -> None: + nonlocal runtime, runtime_revision + next_runtime = tmp_path / f"runtime-source-{runtime_revision}" + shutil.copytree(runtime / "goals" / GOAL / "runs", next_runtime / "goals" / GOAL / "runs") + runtime = next_runtime + runtime_revision += 1 + initialize_canonical_source_fixture(runtime, GOAL, state_path=state_file) + args[args.index("--runtime-root") + 1] = str(runtime) + state_file.write_text(state("status=open claimed_by=worker")) + reload_source_authority() assert cli_main(args) == 0 continued = json.loads(capsys.readouterr().out) assert continued["selected_todo"]["todo_id"] == "todo_source_b" state_file.write_text( state("status=deferred claimed_by=worker resume_when=todo_done:todo_recovery") ) + reload_source_authority() assert cli_main(args) == 0 waiting = json.loads(capsys.readouterr().out) assert waiting["execution_obligation"]["must_attempt_work"] is False @@ -353,3 +384,49 @@ def test_monitor_and_excluded_blocker_cannot_cover_causal_work() -> None: result = decision(state=source, runs=[vision_run(declared=False)]) assert not result["goal_frontier_projection"].get("vision_wait_state") assert result["goal_frontier_projection"]["acceptance_gaps"] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("priority", ["P0", "P2"]) +def test_cli_current_agent_wait_preserves_canonical_blocker_priority( + tmp_path: Path, capsys, provider: str, priority: str, +) -> None: + state_file = tmp_path / "ACTIVE_GOAL_STATE.md" + state_file.write_text( + f"# Active Goal State\n\n## Agent Todo\n\n" + f"- [ ] [{priority}] Await authorized source recovery.\n" + " \n" + ) + registry = tmp_path / "registry.json" + registry.write_text(json.dumps({"goals": [{ + "id": GOAL, "status": "active", "domain": "engineering", + "waiting_on": "codex", "state_file": str(state_file), "repo": str(tmp_path), + "adapter": {"kind": "fixture_connected_delivery_v0", "status": "connected-delivery"}, + "quota": {"compute": 1.0, "window_hours": 24, "slot_minutes": 1}, + "coordination": COORDINATION, + }]})) + runtime = tmp_path / "runtime" + initialize_canonical_source_fixture(runtime, GOAL, state_path=state_file, provider=provider) + runs = runtime / "goals" / GOAL / "runs" + runs.mkdir(parents=True, exist_ok=True) + run = vision_run(declared=False) + run["agent_vision"]["todo_delta"] = ["retain:todo_source_a"] + json_path, markdown_path = runs / "checkpoint.json", runs / "checkpoint.md" + json_path.write_text(json.dumps(run)) + markdown_path.write_text("# Source checkpoint\n") + (runs / "index.jsonl").write_text(json.dumps({ + **run, "json_path": str(json_path), "markdown_path": str(markdown_path), + }) + "\n") + + assert cli_main([ + "--format", "json", "--registry", str(registry), "--runtime-root", str(runtime), + "quota", "should-run", "--verbose", "--goal-id", GOAL, "--agent-id", AGENT, + "--runtime-profile", "ark_managed_agent_goal", + ]) == 0 + result = json.loads(capsys.readouterr().out) + wait = result["goal_frontier_projection"]["vision_wait_state"] + assert wait["reason_code"] == "current_agent_blocker" + assert wait["selected_todo_id"] == "todo_source_a" + assert wait["selected_todo_priority"] == priority + assert wait["automatic_resume"] is False