diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index 660ac77277..4bfc84557d 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -2561,6 +2561,11 @@ source paths, authorize monitor writeback, or change provider/promotion holds. **D1 — qualify permanent projection delivery; may overlap T1/T2.** +The T2 monitor successor route owner is now shared across preflight, the legacy +effect adapter and receipt checks. Its result proves only normalized intent, +not actor authority, provider commit or atomic monitor-plus-successor durability. +Keep the monitor writer fence and promotion hold until that transaction closes. + - Start from `loopx/control_plane/todos/provider_projection.py`, the existing Todo-section renderer and canonical journal/outbox. #4097 already recovers missing Todo sections with `recovery_scope=todo_sections_only`; reuse it. diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index 971dbb5828..acb69de349 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -2034,6 +2034,11 @@ summary,之前消费 legacy summary;真实 CLI 覆盖容量变化和 promote **D1 — 资格化永久投影交付,可与 T1/T2 重叠推进。** +T2 monitor successor 的路由 owner 已由 preflight、legacy effect adapter 和回执校验 +共享。其结果仅证明规范化 intent,不证明 actor authority、provider commit 或 +monitor-plus-successor 原子持久化;事务闭合前继续保留 monitor writer fence 与 +promotion hold。 + - 从 `loopx/control_plane/todos/provider_projection.py`、既有 Todo-section renderer、 canonical journal/outbox 入手。复用 #4097 已有的缺失 Todo section 恢复及 `recovery_scope=todo_sections_only`;它不能恢复丢失的独立 Goal 正文。 diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 854c8b0032..127e423fdd 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -380,6 +380,20 @@ the shared plan, not another per-agent checklist database. **T2 — close monitor writeback and its atomic follow-up.** +Bounded prerequisite delivered: `scheduler/monitor_successor.ts` owns successor +route validation and normalization for quota preflight, legacy writeback and +receipt verification. The Python route guard/resolver and the separate TS +receipt-default/capability interpretation are removed. Invalid capability entries, +malformed successor claims and follow-ups without material change fail before +the observation write; valid action/claim/capability aliases and Git transports +are compared as the same route at readback. The original wire observation still +owns the v0 replay digest; normalization must not silently invalidate pending +receipts. The node-independent repository/bootstrap codec remains separately +characterized, not replaced by a runtime dependency. +This is **not** the T2 atomic transaction: monitor mutation and successor writes +still use existing fenced effects. Cross-effect crash recovery, native writer +closure and whole-Goal promotion remain held; do not infer them from a route plan. + - Inventory `monitor_poll_writeback.py` and its event/Todo/lease callers. Reuse existing monitor generation, independent-successor and settlement owners. Compose one transaction rather than adding a second monitor engine. 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 7d46b754ac..4f26f45c45 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 @@ -294,6 +294,16 @@ commit。#4121(SQLite 候选)和 #4101(投影 receipt 保留)是独立 **T2 — 闭合 monitor 写回及原子后续动作。** +已交付有边界前置项:`scheduler/monitor_successor.ts` 统一 quota preflight、legacy +writeback 与 receipt verification 的后续路由校验和规范化,删除 Python route +guard/resolver 及 TS 回执端独立的默认值/capability 解释。非法 capability 项、非法 +后续 claim 和未声明 material change 的 follow-up 在 observation 写入前拒绝;合法 +action/claim/capability 别名和 Git transport 在回执核对时指向同一路由。v0 replay +digest 仍绑定原始 wire observation,不能因规范化而悄悄使 pending receipt 失效。 +无需 Node 的 repository/bootstrap codec 暂留并做跨运行时对照,不引入启动依赖。 +这**不是** T2 原子事务:monitor mutation 和 successor 写入仍通过既有 fenced effect +执行;跨 effect crash 恢复、native writer 闭合及整 Goal promotion 仍未放行。 + - 盘点 `monitor_poll_writeback.py` 及 event/Todo/lease caller,复用 monitor generation、独立 successor 和 settlement owner,组成一笔事务,不建第二套引擎。 - 保持 unchanged poll/reschedule、generation fence、material-change successor diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md index 67f2b95e16..2f37b8851a 100644 --- a/docs/project-agent-todo-contract.md +++ b/docs/project-agent-todo-contract.md @@ -301,6 +301,19 @@ Relevant command results expose the compact `monitor_advancement_authoring_v0` contract so an Agent can recover this sequence without parsing documentation prose. +Monitor successor routing uses one typed plan for preflight, writeback and +receipt verification. Common Git transport URLs resolve to the same canonical +repository identity, and action/claim/capability aliases are normalized before +comparison. Repository routes must be representable as canonical `git:/` +identities; control characters, backslashes and percent-encoded paths are rejected. +Every supplied capability must be valid: an invalid entry is not silently dropped +from a partly valid list. Follow-ups require `--material-change`; assignment or +other agent-route flags without `--next-agent-todo` are rejected before writeback. +User follow-ups still require explicit `user_action` or `user_gate`, never an +implicit global gate. A route plan is not a claim, approval or atomic commit. +Replay identity continues to bind the original observation, not a rewritten +canonical spelling; retry the same logical observation with the same arguments. + Open todos may also carry `resume_when` when they are visible but not yet executable. Until the parsed `resume_condition.satisfied` value is true, status and quota keep the todo out of `first_executable_items`, diff --git a/examples/shared-goal-authority-e2e/mutants.py b/examples/shared-goal-authority-e2e/mutants.py index 755ce95762..33a9bf91da 100644 --- a/examples/shared-goal-authority-e2e/mutants.py +++ b/examples/shared-goal-authority-e2e/mutants.py @@ -57,6 +57,16 @@ def command(self) -> list[str]: Case('todo_successor_scope_unbound', (('loopx/control_plane/todos/authoring_scope.ts', replacement( 'if (blocks && (goal || !bound || bound !== blocks)) return "agent_binding_conflict";', '')),), 'tests/control_plane_ts/todo_authoring_scope.test.ts', 'resolved successor scope'), + Case('monitor_route_drops_invalid_capability', (('loopx/control_plane/scheduler/monitor_successor.ts', replacement( + ' throw new EffectRuntimeRequestError(`${label} must contain public-safe capability tokens; invalid entries cannot be dropped`);', + ' continue;')),), + 'tests/control_plane_ts/monitor_successor.test.ts', 'invalid successor intent is rejected'), + Case('monitor_route_material_guard_removed', (('loopx/control_plane/scheduler/monitor_successor.ts', replacement( + 'if ((agentTodo || userTodo) && !material)', 'if (false)')),), + 'tests/control_plane_ts/monitor_successor.test.ts', 'invalid successor intent is rejected'), + Case('monitor_route_rewrites_fingerprint', (('loopx/control_plane/quota/monitor_poll_commit.ts', replacement( + ' monitorSuccessorIntent(result);', ' Object.assign(result, monitorSuccessorIntent(result));')),), + 'tests/control_plane_ts/quota_monitor_poll_commit.test.ts', 'preserves the legacy pending observation fingerprint'), Case('delivery_wait_target_unbound', (('loopx/control_plane/todos/resume_condition.ts', replacement( 'condition.target_todo_id !== spec.target || ', '')),), 'tests/control_plane_ts/delivery_response.test.ts', 'exact dependency identity'), diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 99e1c86296..0393a210bb 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -48,6 +48,7 @@ import { evaluateQuotaVoidCommit } from "./quota/void_commit.ts"; import { readQuotaSettlement } from "./quota/settlement_readback.ts"; import { evaluateTurnEnvelope } from "./quota/turn_envelope.ts"; import { evaluateQuotaMonitorPollCommit } from "./quota/monitor_poll_commit.ts"; +import { planMonitorSuccessor } from "./scheduler/monitor_successor.ts"; import { evaluateDeliveryWorkspace } from "./agents/delivery_workspace.ts"; import { interpretTurnJournal, @@ -461,6 +462,7 @@ export function createEffectRuntimeHandlers( ], ["task_lease.write_scopes.overlap", evaluateTaskLeaseWriteScopesOverlap], ["quota.monitor_poll.commit", evaluateQuotaMonitorPollCommit], + ["scheduler.monitor_successor.plan", planMonitorSuccessor], ["coordination.local_authority_shadow.record", recordLocalAuthorityShadow], ["coordination.runtime_shadow.commit_entry", commitLocalAuthorityShadowEntry], ["coordination.runtime_shadow.outbox_read", readLocalAuthorityShadow], diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index 3539204984..e5c05b81ad 100644 --- a/loopx/control_plane/quota/monitor_poll_commit.ts +++ b/loopx/control_plane/quota/monitor_poll_commit.ts @@ -1,6 +1,7 @@ import { createHash } from "node:crypto"; import { access, readFile, rm } from "node:fs/promises"; import { basename, dirname, join, resolve } from "node:path"; +import { monitorSuccessorIntent, monitorSuccessorRoute, monitorSuccessorCapabilities } from "../scheduler/monitor_successor.ts"; import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; @@ -363,42 +364,9 @@ function observationObject(value: unknown): MonitorObservation { "`quota monitor-poll --material-change` requires --todo-id or --target-key", ); } - if ((result.next_agent_todo || result.next_user_todo) && !materialChange) { - throw new EffectRuntimeRequestError( - "`--next-agent-todo` and `--next-user-todo` require --material-change", - ); - } - if (result.next_agent_todo && !result.next_action_kind) { - throw new EffectRuntimeRequestError( - "`quota monitor-poll --next-agent-todo` requires explicit successor action semantics via --next-action-kind", - ); - } - const agentRoute = result.next_action_kind || result.next_task_repository || - result.next_required_capabilities.length || result.next_continuation_policy || - result.next_target_key; - if (!result.next_agent_todo && agentRoute) { - throw new EffectRuntimeRequestError( - "monitor successor routing options require --next-agent-todo", - ); - } - if (result.next_user_todo && !result.next_user_task_class) { - throw new EffectRuntimeRequestError( - "--next-user-todo requires explicit --next-user-task-class user_action|user_gate", - ); - } - if (!result.next_user_todo && result.next_user_task_class) { - throw new EffectRuntimeRequestError( - "--next-user-task-class requires --next-user-todo", - ); - } - if ( - result.next_user_task_class && - !["user_action", "user_gate"].includes(result.next_user_task_class) - ) { - throw new EffectRuntimeRequestError( - "--next-user-task-class must be user_action or user_gate", - ); - } + // Validate the route without rewriting the persisted observation fingerprint. + // Pending receipts from earlier versions must remain replayable. + monitorSuccessorIntent(result); return result; } @@ -883,15 +851,8 @@ function requireProviderCapabilityMatch( expected: readonly string[], label: string, ): void { - const canonical = (items: readonly string[]): string[] => [ - ...new Set( - items - .map((item) => item.trim().toLowerCase().replace(/[-\s]+/g, "_")) - .filter(Boolean), - ), - ]; - const actualCapabilities = canonical(requireStringArray(actual, label)); - const expectedCapabilities = canonical(expected); + const actualCapabilities = monitorSuccessorCapabilities(actual, label); + const expectedCapabilities = monitorSuccessorCapabilities(expected, label); if (pythonJson(actualCapabilities) !== pythonJson(expectedCapabilities)) { throw new EffectRuntimeRequestError(`${label} must match provider plan`); } @@ -915,19 +876,9 @@ function requireProviderTodoText( requireProviderMatch(actual, compactExpected, label); } -function derivedMonitorSuccessorTargetKey(todoId: string, resultHash: string): string { - return `monitor-successor:${todoId}:${sha256Hex(resultHash).slice(0, 16)}`; -} - function requireCanonicalSuccessorRoute( value: JsonObject, - expected: { - task_repository: string | null; - required_capabilities: readonly string[]; - continuation_policy: string; - target_key: string; - claimed_by: string | null; - }, + expected: JsonObject, label: string, ): void { requireProviderMatch( @@ -937,7 +888,7 @@ function requireCanonicalSuccessorRoute( ); requireProviderCapabilityMatch( value.required_capabilities ?? [], - expected.required_capabilities, + requireStringArray(expected.required_capabilities, "expected capabilities"), `${label} required_capabilities`, ); requireProviderMatch( @@ -978,6 +929,7 @@ function validateSuccessorReceipts( } let offset = 0; if (plan.material_change && plan.next_agent_todo) { + const canonicalRoute = monitorSuccessorRoute(monitorSuccessorIntent(plan), todoId, plan.result_hash); const receipt = receipts[offset]; const nextTodo = nextTodos[offset++]; requireProviderMatch(receipt.role, "agent", "agent successor role"); @@ -994,12 +946,12 @@ function validateSuccessorReceipts( ); requireProviderMatch( receipt.action_kind, - plan.next_action_kind, + canonicalRoute.action_kind, "agent successor action_kind", ); requireProviderMatch( nextTodo.action_kind, - plan.next_action_kind, + canonicalRoute.action_kind, "agent next_todo action_kind", ); requireProviderMatch( @@ -1022,13 +974,6 @@ function validateSuccessorReceipts( requiredProviderTodoId(nextTodo.todo_id, "agent next_todo todo_id"), "agent successor todo_id", ); - const canonicalRoute = { - task_repository: plan.next_task_repository, - required_capabilities: plan.next_required_capabilities, - continuation_policy: plan.next_continuation_policy ?? "independent_handoff", - target_key: plan.next_target_key ?? derivedMonitorSuccessorTargetKey(todoId, plan.result_hash), - claimed_by: plan.next_claimed_by, - }; requireCanonicalSuccessorRoute(receipt, canonicalRoute, "agent successor"); requireCanonicalSuccessorRoute(nextTodo, canonicalRoute, "agent next_todo"); } diff --git a/loopx/control_plane/scheduler/monitor_poll_writeback.py b/loopx/control_plane/scheduler/monitor_poll_writeback.py index f060c67209..fc634cbd05 100644 --- a/loopx/control_plane/scheduler/monitor_poll_writeback.py +++ b/loopx/control_plane/scheduler/monitor_poll_writeback.py @@ -1,6 +1,5 @@ from __future__ import annotations -import hashlib from pathlib import Path from typing import Any @@ -8,109 +7,13 @@ TODO_TASK_CLASS_ADVANCEMENT, TODO_TASK_CLASS_MONITOR, TODO_TASK_CLASS_USER_GATE, - normalize_required_capabilities, - normalize_todo_action_kind, - normalize_todo_continuation_policy, normalize_todo_id, - normalize_todo_task_repository, - resolve_next_user_task_class, - resolve_todo_continuation_policy, ) +from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from ..todos.monitor_metadata import MonitorPollObservation from .monitor_todo import monitor_todo_task_class -def require_monitor_successor_route( - *, - next_agent_todo: str | None, - next_action_kind: str | None, - next_task_repository: str | None, - next_required_capabilities: list[str] | None, - next_continuation_policy: str | None, - next_target_key: str | None, -) -> None: - route_fields = { - "--next-action-kind": next_action_kind, - "--next-task-repository": next_task_repository, - "--next-continuation-policy": next_continuation_policy, - "--next-target-key": next_target_key, - } - if not next_agent_todo: - if any(route_fields.values()) or next_required_capabilities: - raise ValueError("monitor successor routing options require --next-agent-todo") - return - if not str(next_action_kind or "").strip(): - raise ValueError( - "`quota monitor-poll --next-agent-todo` requires explicit successor " - "action semantics via --next-action-kind" - ) - - -def _derived_monitor_successor_target_key(*, todo_id: str, result_hash: str) -> str: - digest = hashlib.sha256(result_hash.encode("utf-8")).hexdigest()[:16] - return f"monitor-successor:{todo_id}:{digest}" - - -def _resolve_monitor_successor_route( - *, - next_agent_todo: str | None, - next_action_kind: str | None, - next_task_repository: str | None, - next_required_capabilities: list[str] | None, - next_continuation_policy: str | None, - next_target_key: str | None, - source_task_repository: str | None, - todo_id: str, - result_hash: str, -) -> dict[str, Any]: - if not next_agent_todo: - return {} - action_kind = normalize_todo_action_kind(next_action_kind) - if not action_kind: - raise ValueError( - "--next-action-kind must be a public-safe token: lowercase letters, " - "digits, '_' or '-'" - ) - task_repository = normalize_todo_task_repository(next_task_repository) - if next_task_repository and not task_repository: - raise ValueError( - "--next-task-repository must be a credential-free Git remote or " - "canonical git:/ identity" - ) - if source_task_repository and not task_repository: - raise ValueError( - "repository-bound monitor successors require explicit " - "--next-task-repository so same-repository and cross-repository " - "routing cannot be confused" - ) - required_capabilities = normalize_required_capabilities( - next_required_capabilities - ) - if next_required_capabilities and not required_capabilities: - raise ValueError( - "--next-required-capability must contain public-safe capability tokens" - ) - if next_continuation_policy and not normalize_todo_continuation_policy( - next_continuation_policy - ): - raise ValueError( - "--next-continuation-policy must be a supported todo continuation policy" - ) - return { - "action_kind": action_kind, - "task_repository": task_repository, - "required_capabilities": required_capabilities, - "continuation_policy": resolve_todo_continuation_policy( - next_continuation_policy - ).value, - "target_key": str(next_target_key or "").strip() - or _derived_monitor_successor_target_key( - todo_id=todo_id, - result_hash=result_hash, - ), - } - - def resolve_monitor_todo_item( *, registry_path: Path, @@ -222,18 +125,6 @@ def write_monitor_poll_todo_state( safe_result_hash = str(result_hash or "").strip() if not safe_result_hash: raise ValueError("monitor todo writeback requires --result-hash") - effective_next_user_task_class = resolve_next_user_task_class( - next_user_todo, - next_user_task_class, - ) - require_monitor_successor_route( - next_agent_todo=next_agent_todo, - next_action_kind=next_action_kind, - next_task_repository=next_task_repository, - next_required_capabilities=next_required_capabilities, - next_continuation_policy=next_continuation_policy, - next_target_key=next_target_key, - ) item = resolve_monitor_todo_item( registry_path=registry_path, goal_id=goal_id, @@ -244,18 +135,32 @@ def write_monitor_poll_todo_state( resolved_todo_id = normalize_todo_id(item.get("todo_id")) if not resolved_todo_id: raise ValueError("resolved monitor todo has no stable todo_id") - successor_route = _resolve_monitor_successor_route( - next_agent_todo=next_agent_todo, - next_action_kind=next_action_kind, - next_task_repository=next_task_repository, - next_required_capabilities=next_required_capabilities, - next_continuation_policy=next_continuation_policy, - next_target_key=next_target_key, - source_task_repository=str(item.get("task_repository") or "").strip() - or None, - todo_id=resolved_todo_id, - result_hash=safe_result_hash, - ) + # The typed plan validates the complete successor intent before the first + # write. Source lookup and actual effects remain in this adapter. + try: + plan = effect_runtime_result("scheduler.monitor_successor.plan", { + "schema_version": "loopx_monitor_successor_plan_request_v0", + "todo_id": resolved_todo_id, "result_hash": safe_result_hash, + "source_task_repository": item.get("task_repository"), + "intent": { + "material_change": material_change, + "next_agent_todo": next_agent_todo, "next_action_kind": next_action_kind, + "next_task_repository": next_task_repository, + "next_required_capabilities": next_required_capabilities or [], + "next_continuation_policy": next_continuation_policy, + "next_target_key": next_target_key, "next_claimed_by": next_claimed_by, + "next_user_todo": next_user_todo, "next_user_task_class": next_user_task_class, + }, + }) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from exc + if not isinstance(plan, dict) or plan.get("schema_version") != "loopx_monitor_successor_plan_result_v0": + raise TypeError("TypeScript monitor successor plan shape mismatch") + successor_route = plan["agent_route"] + intent = plan["intent"] + next_agent_todo = intent["next_agent_todo"] + next_user_todo = intent["next_user_todo"] + effective_next_user_task_class = intent["next_user_task_class"] safe_target_key = str(target_key or "").strip() update_result = update_goal_todo( registry_path=registry_path, @@ -301,7 +206,7 @@ def write_monitor_poll_todo_state( task_repository=successor_route["task_repository"], continuation_policy=successor_route["continuation_policy"], required_capabilities=successor_route["required_capabilities"], - claimed_by=next_claimed_by, + claimed_by=successor_route["claimed_by"], unblocks_todo_id=resolved_todo_id, monitor_metadata={"target_key": successor_route["target_key"]}, dry_run=not execute, diff --git a/loopx/control_plane/scheduler/monitor_successor.ts b/loopx/control_plane/scheduler/monitor_successor.ts new file mode 100644 index 0000000000..292ee29587 --- /dev/null +++ b/loopx/control_plane/scheduler/monitor_successor.ts @@ -0,0 +1,146 @@ +/** Monitor follow-up intent, shared by preflight, legacy writeback and receipt + * verification. This read-only plan is not authorization or a commit receipt. */ +import { createHash } from "node:crypto"; +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { optionalNonEmptyString, requireBoolean, requireJsonObject, requireStringArray } from "../runtime_decode.ts"; +import { compactPythonWhitespace, normalizeTodoAgent, stripPythonWhitespace } from "../coordination/todo_agents.ts"; + +export const MONITOR_SUCCESSOR_REQUEST_SCHEMA = "loopx_monitor_successor_plan_request_v0"; +export const MONITOR_SUCCESSOR_RESULT_SCHEMA = "loopx_monitor_successor_plan_result_v0"; + +function text(value: unknown, field: string): string | null { + const raw = optionalNonEmptyString(value, field); + return raw === null ? null : stripPythonWhitespace(raw) || null; +} + +// The node-independent repository/bootstrap codec remains in repository_identity.py. +// This pure transport codec is characterized against that public contract; do +// not use WHATWG's normalized pathname, which silently removes dot segments. +function repository(value: unknown): string | null { + let raw = text(value, "next_task_repository"); + if (!raw) return null; + if (/[\\\s\u0000-\u001f\u007f]/u.test(raw)) { + throw new EffectRuntimeRequestError("--next-task-repository must be a credential-free Git remote without control characters or backslashes"); + } + let host: string, path: string; + const canonical = /^git:([a-z0-9.-]+(?::[0-9]{1,5})?)\/([A-Za-z0-9._~+/-]+)$/.exec(raw); + if (canonical) [host, path] = [canonical[1], canonical[2]]; + else { + const scp = /^(?:[^@/]+@)?([^:/]+):(.+)$/.exec(raw); + if (scp && !raw.includes("://")) raw = `ssh://${scp[1]}/${scp[2]}`; + try { + const url = new URL(raw); + if (!["git:", "http:", "https:", "ssh:"].includes(url.protocol) || + !url.hostname || url.password || url.search || url.hash) throw new Error(); + host = url.hostname.toLowerCase().replace(/^\[|\]$/g, ""); + const port = Number(url.port); + if (port && !((["http:", "git:"].includes(url.protocol) && port === 80) || + (["https:", "ssh:"].includes(url.protocol) && [22, 443].includes(port)))) host += `:${port}`; + const pathMatch = /^[^:]+:\/\/[^/?#]*([^?#]*)/.exec(raw); + if (!pathMatch) throw new Error(); + path = pathMatch[1]; + } catch { + throw new EffectRuntimeRequestError("--next-task-repository must be a credential-free Git remote or canonical git:/ identity"); + } + } + path = path.replace(/\/+/g, "/").replace(/^\/+|\/+$/g, "").replace(/\.git$/, ""); + if (!/^[A-Za-z0-9._~+/-]+$/.test(path) || !/^[a-z0-9.-]+(?::[0-9]{1,5})?$/.test(host) || + path.split("/").some(part => part === "." || part === "..")) { + throw new EffectRuntimeRequestError("--next-task-repository must include a safe repository path"); + } + return `git:${host}/${path}`; +} + +export function monitorSuccessorCapabilities(value: unknown, label: string): string[] { + const result: string[] = []; + for (const raw of requireStringArray(value ?? [], label)) { + const token = compactPythonWhitespace(raw).toLowerCase().replaceAll("-", "_").replaceAll(" ", "_"); + if (!/^[a-z][a-z0-9_:-]{0,63}$/.test(token)) { + throw new EffectRuntimeRequestError(`${label} must contain public-safe capability tokens; invalid entries cannot be dropped`); + } + if (!result.includes(token)) result.push(token); + } + return result; +} + +export interface MonitorSuccessorIntent extends JsonObject { + next_agent_todo: string | null; + next_action_kind: string | null; + next_task_repository: string | null; + next_required_capabilities: string[]; + next_continuation_policy: string | null; + next_target_key: string | null; + next_claimed_by: string | null; + next_user_todo: string | null; + next_user_task_class: "user_action" | "user_gate" | null; +} + +export function monitorSuccessorIntent(value: unknown): MonitorSuccessorIntent { + const input = requireJsonObject(value, "monitor successor intent"); + const material = requireBoolean(input.material_change, "material_change"); + const agentTodo = text(input.next_agent_todo, "next_agent_todo"); + const userTodo = text(input.next_user_todo, "next_user_todo"); + if ((agentTodo || userTodo) && !material) { + throw new EffectRuntimeRequestError("`--next-agent-todo` and `--next-user-todo` require --material-change"); + } + const action = text(input.next_action_kind, "next_action_kind")?.toLowerCase() ?? null; + const policy = text(input.next_continuation_policy, "next_continuation_policy")?.toLowerCase() ?? null; + const target = text(input.next_target_key, "next_target_key"); + const claim = text(input.next_claimed_by, "next_claimed_by"); + const repo = repository(input.next_task_repository); + const capabilities = monitorSuccessorCapabilities(input.next_required_capabilities, "--next-required-capability"); + if (!agentTodo && (action || policy || target || claim || repo || capabilities.length)) { + throw new EffectRuntimeRequestError("monitor successor routing options require --next-agent-todo"); + } + if (agentTodo && !action) { + throw new EffectRuntimeRequestError("`quota monitor-poll --next-agent-todo` requires explicit successor action semantics via --next-action-kind"); + } + if (action && !/^[a-z][a-z0-9_-]{0,63}$/.test(action)) { + throw new EffectRuntimeRequestError("--next-action-kind must be a public-safe token: lowercase letters, digits, '_' or '-'"); + } + if (policy && !["independent_handoff", "same_agent_non_delivery"].includes(policy)) { + throw new EffectRuntimeRequestError("--next-continuation-policy must be a supported todo continuation policy"); + } + const userClass = text(input.next_user_task_class, "next_user_task_class"); + if (userTodo && !userClass) throw new EffectRuntimeRequestError("--next-user-todo requires explicit --next-user-task-class user_action|user_gate"); + if (!userTodo && userClass) throw new EffectRuntimeRequestError("--next-user-task-class requires --next-user-todo"); + if (userClass !== null && userClass !== "user_action" && userClass !== "user_gate") { + throw new EffectRuntimeRequestError("--next-user-task-class must be user_action or user_gate"); + } + let owner: string | null = null; + if (claim) { + try { owner = normalizeTodoAgent(claim, "next_claimed_by"); } + catch { throw new EffectRuntimeRequestError("--next-claimed-by must be a public-safe agent id"); } + } + return {next_agent_todo: agentTodo, next_action_kind: action, + next_task_repository: repo, next_required_capabilities: capabilities, + next_continuation_policy: policy, next_target_key: target, next_claimed_by: owner, + next_user_todo: userTodo, next_user_task_class: userClass}; +} + +export function monitorSuccessorRoute(intent: MonitorSuccessorIntent, todoId: string, resultHash: string): JsonObject { + if (!intent.next_agent_todo) return {}; + return {action_kind: intent.next_action_kind, task_repository: intent.next_task_repository, + required_capabilities: intent.next_required_capabilities, + continuation_policy: intent.next_continuation_policy ?? "independent_handoff", + target_key: intent.next_target_key ?? `monitor-successor:${todoId}:${createHash("sha256").update(resultHash).digest("hex").slice(0, 16)}`, + claimed_by: intent.next_claimed_by}; +} + +export function planMonitorSuccessor(value: unknown): JsonObject { + const request = requireJsonObject(value, "monitor successor plan"); + if (request.schema_version !== MONITOR_SUCCESSOR_REQUEST_SCHEMA) throw new EffectRuntimeRequestError("monitor successor plan schema mismatch"); + const intent = monitorSuccessorIntent(request.intent); + const todoId = text(request.todo_id, "todo_id"); + const resultHash = text(request.result_hash, "result_hash"); + if (!todoId || !/^todo_[a-z0-9_-]{3,64}$/.test(todoId) || !resultHash) { + throw new EffectRuntimeRequestError("monitor successor plan requires a stable todo_id and result_hash"); + } + const sourceRepository = text(request.source_task_repository, "source_task_repository"); + if (intent.next_agent_todo && sourceRepository && !intent.next_task_repository) { + throw new EffectRuntimeRequestError("repository-bound monitor successors require explicit --next-task-repository so same-repository and cross-repository routing cannot be confused"); + } + return {schema_version: MONITOR_SUCCESSOR_RESULT_SCHEMA, intent, + agent_route: monitorSuccessorRoute(intent, todoId, resultHash)}; +} diff --git a/tests/control_plane/test_monitor_successor_validation.py b/tests/control_plane/test_monitor_successor_validation.py new file mode 100644 index 0000000000..b8509c822e --- /dev/null +++ b/tests/control_plane/test_monitor_successor_validation.py @@ -0,0 +1,43 @@ +"""Monitor routing errors must be rejected before any observation writeback.""" +import pytest + +from test_monitor_followthrough_contract import _write_fixture, _add_monitor, GOAL_ID, AGENT_ID +from loopx.control_plane.scheduler.monitor_poll_writeback import write_monitor_poll_todo_state +from loopx.control_plane.testing.canary_harness import run_json_cli + + +@pytest.mark.parametrize("invalid", [ + {"next_required_capabilities": ["filesystem-write", "not/a/capability"]}, + {"next_claimed_by": "invalid/actor"}, + {"material_change": False}, +]) +def test_invalid_successor_intent_does_not_partially_update_monitor(tmp_path, invalid): + registry, runtime, state = _write_fixture(tmp_path) + monitor = _add_monitor(registry, text="Watch a public target.", target_key="public-target") + before = state.read_bytes() + with pytest.raises(ValueError): + write_monitor_poll_todo_state(registry_path=registry, runtime_root=runtime, + goal_id=GOAL_ID, todo_id=monitor["todo_id"], execute=True, + generated_at="2026-09-09T12:00:00Z", result_hash="revision-a", agent_id=AGENT_ID, + next_agent_todo="Validate the material change.", next_action_kind="validate", + **{"material_change": True, **invalid}) + assert state.read_bytes() == before + + +def test_public_monitor_poll_accepts_repository_and_action_aliases(tmp_path): + registry, runtime, _state = _write_fixture(tmp_path) + monitor = _add_monitor(registry, text="Watch a public target.", target_key="public-target", + next_due_at="2000-01-01T00:00:00Z") + result = run_json_cli("quota", "monitor-poll", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--runtime-profile", "generic_cli", + "--todo-id", monitor["todo_id"], "--result-hash", "revision-a", "--material-change", + "--next-agent-todo", "Validate the transition.", "--next-action-kind", "VALIDATE", + "--next-task-repository", "https://github.com/example/repo.git", + "--next-required-capability", "file--write", "--execute", + registry_path=registry, runtime_root=runtime) + successor = result["todo_writeback"]["next_todos"][0] + assert successor["action_kind"] == "validate" + assert successor["task_repository"] == "git:github.com/example/repo" + assert successor["required_capabilities"] == ["file__write"] + assert successor["task_class"] == "advancement_task" + assert successor["unblocks_todo_id"] == monitor["todo_id"] diff --git a/tests/control_plane_ts/monitor_successor.test.ts b/tests/control_plane_ts/monitor_successor.test.ts new file mode 100644 index 0000000000..4b9126490c --- /dev/null +++ b/tests/control_plane_ts/monitor_successor.test.ts @@ -0,0 +1,83 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { spawnSync } from "node:child_process"; +import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts"; +import { monitorSuccessorIntent, planMonitorSuccessor, MONITOR_SUCCESSOR_REQUEST_SCHEMA } from "../../loopx/control_plane/scheduler/monitor_successor.ts"; + +const intent = {material_change: true, next_agent_todo: "Validate the transition.", next_action_kind: "validate"}; +const plan = (extra = {}, source_task_repository: string | null = null) => planMonitorSuccessor({ + schema_version: MONITOR_SUCCESSOR_REQUEST_SCHEMA, todo_id: "todo_monitor", result_hash: "revision-a", + source_task_repository, intent: {...intent, ...extra}, +}); + +test("one successor plan preserves explicit execution and user-decision semantics", () => { + const result = plan({next_action_kind: "VALIDATE", next_claimed_by: "Agent A", + next_required_capabilities: ["file--write", "network read", "file--write"]}); + assert.deepEqual(result.agent_route, {action_kind: "validate", task_repository: null, + required_capabilities: ["file__write", "network_read"], continuation_policy: "independent_handoff", + claimed_by: "agent-a", target_key: "monitor-successor:todo_monitor:5189357b78fc8ba5"}); + for (const taskClass of ["user_action", "user_gate"]) { + const normalized = monitorSuccessorIntent({material_change: true, next_user_todo: "Review the transition.", next_user_task_class: taskClass}); + assert.equal(normalized.next_user_task_class, taskClass); + assert.equal(normalized.next_agent_todo, null); + } +}); + +test("invalid successor intent is rejected as a whole, not partly normalized away", () => { + for (const invalid of [ + {next_required_capabilities: ["filesystem_write", "bad/capability"]}, + {next_claimed_by: "bad/actor"}, {next_action_kind: "not an action"}, + {next_continuation_policy: "primary_review"}, {material_change: false}, + {next_user_todo: "Approve"}, {next_user_task_class: "user_gate"}, + ]) assert.throws(() => plan(invalid)); + assert.throws(() => monitorSuccessorIntent({material_change: false, next_claimed_by: "agent-a"}), /require --next-agent-todo/); + assert.throws(() => plan({}, "git:github.com/example/source"), /require explicit --next-task-repository/); +}); + +test("repository transport aliases match the retained node-independent codec", () => { + const inputs = ["git:github.com/example/repo", "https://github.com/example/repo.git", + "git@github.com:example/repo.git", "ssh://git@github.com/example/repo.git", + "http://github.com:80/example/repo/", "https://github.com:22/example/repo", + "ssh://git@github.com:443/example/repo", "ssh://git@github.com:8022/example//repo.git", + "git://github.com:80/example/repo", "https://GITHUB.com/example/repo.git"]; + const python = spawnSync("python", ["-c", "import json,sys; from loopx.repository_identity import normalize_repository_identity; print(json.dumps([normalize_repository_identity(x) for x in json.load(sys.stdin)]))"], + {input: JSON.stringify(inputs), encoding: "utf8"}); + assert.equal(python.status, 0, python.stderr); + const expected = JSON.parse(python.stdout); + for (const [index, input] of inputs.entries()) { + const actual = monitorSuccessorIntent({...intent, next_task_repository: input}).next_task_repository; + assert.equal(actual, expected[index]); + assert.equal(monitorSuccessorIntent({...intent, next_task_repository: actual}).next_task_repository, actual); + } +}); + +test("unsafe repository routes cannot be silently repaired by URL parsing", () => { + for (const repo of ["https://github.com/example/../other", "git:github.com/example/./repo", + "https://user:password@example.invalid/repo", "https://example.invalid/repo?credential=value", + "https://example.invalid/repo#fragment", "file:///repo", "https://example.invalid/", "not a repository", + "https://example.invalid\\other/repo", "https://example.invalid/a%2fb", "https://example.invalid/a b"]) { + assert.throws(() => plan({next_task_repository: repo})); + } +}); + +test("production-scale monitor route planning is read-only and preserves target separation", () => { + const fixture = productionScaleCoordinationFixture("goal-route-fixture"); + const original = structuredClone(fixture.projection); + const todos = fixture.projection.todos as Record[]; + const monitors = todos.filter(todo => todo.task_class === "continuous_monitor"); + assert.equal(monitors.length, 63); + const targets = new Set(); + for (const monitor of monitors) { + // A route can be described for historical context, but is never a lifecycle grant. + const result = planMonitorSuccessor({schema_version: MONITOR_SUCCESSOR_REQUEST_SCHEMA, + todo_id: monitor.todo_id, result_hash: "same-material-hash", + source_task_repository: monitor.task_repository ?? null, + intent: {...intent, next_task_repository: monitor.task_repository ?? null, + next_required_capabilities: monitor.required_capabilities ?? []}}); + targets.add((result.agent_route as Record).target_key); + } + assert.equal(targets.size, monitors.length); + assert.equal(todos.length, fixture.expected_initial_todo_count); + assert.equal((fixture.projection.leases as unknown[]).length, fixture.expected_current_lease_count); + assert.deepEqual(fixture.projection, original); +}); diff --git a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts index 048202f948..8a87da1ed0 100644 --- a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts +++ b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts @@ -1,5 +1,6 @@ import assert from "node:assert/strict"; import { createHash } from "node:crypto"; +import { spawnSync } from "node:child_process"; import { mkdir, mkdtemp, readFile, rm, unlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -511,12 +512,12 @@ test("Todo commit fences the full material successor receipt", async (t) => { result_hash: "approved-42", material_change: true, next_agent_todo: "Advance the approved release.", - next_action_kind: "advance_release", - next_task_repository: "git:github.com/owner/repo", + next_action_kind: "ADVANCE_RELEASE", + next_task_repository: "https://github.com/owner/repo.git", next_required_capabilities: ["filesystem-write"], - next_continuation_policy: "same_agent_non_delivery", + next_continuation_policy: "SAME_AGENT_NON_DELIVERY", next_target_key: "public-release:42:advance", - next_claimed_by: "codex-main-control", + next_claimed_by: "Codex Main Control", }), }); await evaluateQuotaMonitorPollCommit(params); @@ -596,6 +597,51 @@ test("Todo commit fences the full material successor receipt", async (t) => { (written.payload.todo_writeback as Record).successor_receipts, [successor], ); + assert.equal((await evaluateQuotaMonitorPollCommit({...params, phase: "commit", provider_receipt: providerReceipt})).status, "replayed"); +}); + +test("successor normalization preserves the legacy pending observation fingerprint", async (t) => { + const runtimeRoot = await tempRuntime(t); + const params = request({phase: "preflight", runtime_root: runtimeRoot, execute: true, + effect_id: "quota-monitor-poll:legacy-route-fingerprint", + observation: observation({todo_id: "todo_public_monitor", result_hash: "revision-a", + material_change: true, next_agent_todo: "Validate.", next_action_kind: "VALIDATE", + next_task_repository: "git@github.com:example/repo.git", next_required_capabilities: ["file--write"], + next_continuation_policy: "SAME_AGENT_NON_DELIVERY", next_claimed_by: "Agent A"})}); + // The shipped v0 identity recipe hashes wire observation, not its normalized route. + const legacyEnvelope = Object.fromEntries(["schema_version", "effect_id", "runtime_root", "goal_id", + "source", "turn_instance_id", "observation"].map(key => [key, params[key]])); + const oracle = spawnSync("python", ["-c", "import hashlib,json,sys; print('sha256:'+hashlib.sha256(json.dumps(json.load(sys.stdin),ensure_ascii=False,sort_keys=True).encode()).hexdigest())"], + {input: JSON.stringify(legacyEnvelope), encoding: "utf8"}); + assert.equal(oracle.status, 0, oracle.stderr); + const first = await evaluateQuotaMonitorPollCommit(params); + assert.equal(first.status, "provider_required"); + assert.equal(first.request_digest, oracle.stdout.trim()); + assert.equal(first.provider_plan?.next_action_kind, "VALIDATE"); + const retry = await evaluateQuotaMonitorPollCommit({...params, generated_at: "2026-09-10T12:00:00Z"}); + assert.equal(retry.status, "provider_required"); + assert.equal(retry.request_digest, first.request_digest); + assert.deepEqual(retry.provider_plan, first.provider_plan); + const conflict = await evaluateQuotaMonitorPollCommit({...params, + observation: {...params.observation as object, next_task_repository: "git:github.com/example/other"}}); + assert.equal(conflict.status, "conflict"); + assert.equal(conflict.written, false); +}); + +test("invalid successor routes fail before a pending provider effect is saved", async (t) => { + const runtimeRoot = await tempRuntime(t); + const effectId = "quota-monitor-poll:invalid-route"; + for (const invalid of [{next_required_capabilities: ["file_write", "bad/token"]}, + {next_claimed_by: "bad/actor"}, {next_continuation_policy: "primary_review"}, + {next_task_repository: "https://example.invalid/project/../other"}]) { + await assert.rejects(() => evaluateQuotaMonitorPollCommit(request({phase: "preflight", + runtime_root: runtimeRoot, execute: true, effect_id: effectId, + observation: observation({todo_id: "todo_public_monitor", result_hash: "revision-a", + material_change: true, next_agent_todo: "Validate.", next_action_kind: "validate", ...invalid})}))); + } + const path = join(runtimeRoot, "goals", goalId, "runs", ".transactions", "quota-monitor-poll", + `${createHash("sha256").update(effectId).digest("hex").slice(0, 24)}.json`); + await assert.rejects(() => readFile(path), {code: "ENOENT"}); }); test("Todo commit rejects injected defaults in the material successor route", async (t) => { diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index 3789c32bc1..9f6d424b8b 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -88,6 +88,7 @@ "tests/control_plane_ts/settlement_workspace_causality.test.ts", "tests/control_plane_ts/quota_settlement_readback.test.ts", "tests/control_plane_ts/quota_monitor_poll_commit.test.ts", + "tests/control_plane_ts/monitor_successor.test.ts", "tests/control_plane_ts/quota_spend_commit.test.ts", "tests/control_plane_ts/quota_void_commit.test.ts", "tests/control_plane_ts/scheduler_state_store.test.ts",