From edb0e4c759363ad5ea167188b28199f9d46a03bf Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 19 Sep 2026 02:38:37 +0800 Subject: [PATCH 1/4] refactor(todos): capture follow-ups in one typed authority transaction Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/cli_commands/todo.py | 2 + .../cli_commands/todo_argument_validation.py | 7 +- loopx/cli_commands/todo_registration.py | 4 + .../coordination/followup_capture_runtime.ts | 37 ++++ .../coordination/todo_followup_capture.ts | 83 ++++++++ .../control_plane/effect_runtime_handlers.ts | 4 + .../todos/active_state_editing.py | 3 +- .../control_plane/todos/completion_policy.ts | 10 +- loopx/control_plane/todos/followup_capture.ts | 107 ++++++++++ .../control_plane/todos/provider_followups.py | 60 ++++++ loopx/control_plane/todos/todo_block_codec.py | 5 +- loopx/control_plane/todos/todo_summary.py | 4 +- loopx/todo_followups.py | 186 ++++++------------ 13 files changed, 378 insertions(+), 134 deletions(-) create mode 100644 loopx/control_plane/coordination/followup_capture_runtime.ts create mode 100644 loopx/control_plane/coordination/todo_followup_capture.ts create mode 100644 loopx/control_plane/todos/followup_capture.ts create mode 100644 loopx/control_plane/todos/provider_followups.py diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py index 83c9b4786b..3a2f3d7b60 100644 --- a/loopx/cli_commands/todo.py +++ b/loopx/cli_commands/todo.py @@ -599,6 +599,8 @@ def handle_todo_command( evidence=args.evidence or "", task_class=args.task_class, action_kind=args.action_kind, + continuation_policy=args.continuation_policy, + capture_operation_id=args.capture_operation_id, required_write_scopes=args.required_write_scopes, required_capabilities=args.required_capabilities, target_capabilities=args.target_capabilities, diff --git a/loopx/cli_commands/todo_argument_validation.py b/loopx/cli_commands/todo_argument_validation.py index 40becb7dc0..30ea70ce8e 100644 --- a/loopx/cli_commands/todo_argument_validation.py +++ b/loopx/cli_commands/todo_argument_validation.py @@ -10,6 +10,7 @@ ("--text", "text"), ("--follow-up", "followups"), ("--todo-id", "todo_id"), + ("--capture-operation-id", "capture_operation_id"), ("--claim-operation-id", "claim_operation_id"), ("--update-operation-id", "update_operation_id"), ("--update-expected-provider-revision", "update_expected_provider_revision"), @@ -491,10 +492,10 @@ def validate_todo_capture_followups_options(args: argparse.Namespace) -> None: { "text", "followups", "evidence", "task_class", "action_kind", "continuation_policy", "required_write_scopes", "required_capabilities", - "target_capabilities", "required_decision_scopes", "state_file", + "target_capabilities", "required_decision_scopes", "state_file", "capture_operation_id", }, "todo capture-followups only accepts --goal-id, --follow-up, optional " - "--text shorthand, --evidence, routing metadata, --project, --state-file, " + "--text shorthand, --evidence, routing metadata, --capture-operation-id, --project, --state-file, " "and --dry-run; unsupported: ", ) @@ -523,6 +524,8 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None: raise ValueError( "--turn-instance-id is supported only by todo complete settlement" ) + if getattr(args, "capture_operation_id", None) is not None and args.todo_command != "capture-followups": + raise ValueError("--capture-operation-id is supported only by todo capture-followups") if getattr(args, "update_operation_id", None) is not None and args.todo_command != "update": raise ValueError("--update-operation-id is supported only by todo update") if getattr(args, "update_expected_provider_revision", None) is not None and args.todo_command != "update": diff --git a/loopx/cli_commands/todo_registration.py b/loopx/cli_commands/todo_registration.py index a9d51172e9..dbabec4dc3 100644 --- a/loopx/cli_commands/todo_registration.py +++ b/loopx/cli_commands/todo_registration.py @@ -65,6 +65,10 @@ def register_todo_command( help="For capture-followups, append one public-safe agent follow-up todo. Repeat up to the requested batch.", ) todo_parser.add_argument("--todo-id", help="Structured todo id from status/quota, such as todo_ab12cd34ef56.") + todo_parser.add_argument( + "--capture-operation-id", + help="For promoted capture-followups, reuse the same id and intent to recover the original batch; omitted ids are fresh per invocation.", + ) todo_parser.add_argument( "--update-operation-id", help=("For promoted text/note, planning or User completion update, reuse this operation id after a lost response; " diff --git a/loopx/control_plane/coordination/followup_capture_runtime.ts b/loopx/control_plane/coordination/followup_capture_runtime.ts new file mode 100644 index 0000000000..3728442d23 --- /dev/null +++ b/loopx/control_plane/coordination/followup_capture_runtime.ts @@ -0,0 +1,37 @@ +/** Bind batch capture to the selected provider and existing maintenance fence. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject, requireBoolean} from "../runtime_decode.ts"; +import {requireAuthorityStoreId} from "./authority_store_codec.ts"; +import {registryAuthoritySourceCheck} from "./authority_source.ts"; +import {openLocalAuthorityStore, localAuthorityOpenFailure, type LocalAuthorityProviderDependencies} from "./local_authority_provider.ts"; +import {runtimeRoot, sourceAuthorityFor} from "./local_authority_runtime.ts"; +import {withCanonicalWriter} from "./local_authority_write.ts"; +import {ShadowManagementError} from "./shadow_management.ts"; +import {COORDINATION_FOLLOWUP_CAPTURE_SCHEMA, COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA, + executeCoordinationFollowupCapture} from "./todo_followup_capture.ts"; + +export async function captureLocalFollowups(value: unknown, dependencies: LocalAuthorityProviderDependencies = {}): Promise { + const evidence = {source_authority: "file_v0", decision_read_from_provider: true, legacy_fallback_used: false}; + try { + const input = requireJsonObject(value, "follow-up capture request"); + if (input.schema_version !== COORDINATION_FOLLOWUP_CAPTURE_SCHEMA) throw new Error("follow-up capture request schema mismatch"); + for (const field of Object.keys(input)) { + if (!["schema_version", "runtime_root", "goal_id", "operation_id", "intent", "dry_run", "registry_source"].includes(field)) throw new Error(`unsupported capture request field: ${field}`); + } + const root = runtimeRoot(input.runtime_root), goal = requireAuthorityStoreId(input.goal_id, "goal id"); + const dryRun = requireBoolean(input.dry_run, "dry_run"); + const operation = requireAuthorityStoreId(input.operation_id, "operation id"); + const intent = requireJsonObject(input.intent, "capture intent"); + const current = registryAuthoritySourceCheck(input, true); + return await withCanonicalWriter(root, goal, dryRun, async () => { + const store = await openLocalAuthorityStore(root, goal, dependencies); + evidence.source_authority = sourceAuthorityFor(store); + return {...await executeCoordinationFollowupCapture(store, {goal_id: goal, operation_id: operation, + intent, dry_run: dryRun, now: new Date()}, current), ...evidence}; + }); + } catch (error) { + return {schema_version: COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA, status: "failed", changed: false, + reason_code: error instanceof ShadowManagementError ? error.reason_code : "followup_capture_unavailable", + reason: error instanceof Error ? error.message : String(error), ...evidence, ...localAuthorityOpenFailure(error)}; + } +} diff --git a/loopx/control_plane/coordination/todo_followup_capture.ts b/loopx/control_plane/coordination/todo_followup_capture.ts new file mode 100644 index 0000000000..b6a51b2dd1 --- /dev/null +++ b/loopx/control_plane/coordination/todo_followup_capture.ts @@ -0,0 +1,83 @@ +/** Batch selection, record admission and one CAS/receipt over the complete head. */ +import type {JsonObject} from "../effect_program.ts"; +import type {AuthorityStore} from "./authority_store.ts"; +import {canonicalAuthorityObject, canonicalAuthoritySha256, requireAuthorityStoreId} from "./authority_store_codec.ts"; +import {CoordinationCommandReceipt, commandReceiptResult} from "./command_receipt.ts"; +import {AUTHORITY_SOURCE_CHANGED, uncheckedAuthoritySource, type AuthoritySourceCheck} from "./authority_source.ts"; +import {indexCoordinationProjection, prepareCoordinationProjectionCommit, validateCoordinationTodoReadModel} from "./coordination_projection.ts"; +import {planCoordinationTodoCreate} from "./todo_create.ts"; +import {TODO_DOMAIN_ITEM_SCHEMA} from "./coordination_state_contract.ts"; +import {FOLLOWUP_CAPTURE_PLAN_SCHEMA, normalizeFollowupCaptureIntent, planFollowupCapture} from "../todos/followup_capture.ts"; +import {requireBoolean} from "../runtime_decode.ts"; + +export const COORDINATION_FOLLOWUP_CAPTURE_SCHEMA = "loopx_coordination_followup_capture_request_v0"; +export const COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA = "loopx_coordination_followup_capture_result_v0"; +const RECEIPT_SCHEMA = "loopx_coordination_followup_capture_receipt_v0"; + +export interface CoordinationFollowupCaptureInput { + goal_id: string; + operation_id: string; + intent: JsonObject; + dry_run: boolean; + now: Date; +} + +function failure(reason_code: string, reason: string): JsonObject & {schema_version: typeof COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA} { + return {schema_version: COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA, status: "failed", changed: false, reason_code, reason}; +} + +export async function executeCoordinationFollowupCapture(store: AuthorityStore, + raw: CoordinationFollowupCaptureInput, + authoritySourcesCurrent: AuthoritySourceCheck = uncheckedAuthoritySource): Promise { + let input: CoordinationFollowupCaptureInput; + try { + input = {goal_id: requireAuthorityStoreId(raw.goal_id, "goal id"), + operation_id: requireAuthorityStoreId(raw.operation_id, "operation id"), + intent: normalizeFollowupCaptureIntent(raw.intent), dry_run: requireBoolean(raw.dry_run, "dry_run"), now: raw.now}; + if (!(input.now instanceof Date) || Number.isNaN(input.now.valueOf())) throw new Error("now must be a valid Date"); + } catch (error) { return failure("invalid_followup_capture_request", error instanceof Error ? error.message : String(error)); } + // Clock and source witness are observations, not a caller's retry identity. + const identity = {schema_version: RECEIPT_SCHEMA, goal_id: input.goal_id, operation_id: input.operation_id, + request_sha256: canonicalAuthoritySha256({goal_id: input.goal_id, intent: input.intent, dry_run: input.dry_run})}; + const receipt = new CoordinationCommandReceipt({result_schema: COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA, + identity, failure, decode: commandReceiptResult}); + const previous = await receipt.read(store); + if (previous) return previous; + if (!await authoritySourcesCurrent()) return failure(AUTHORITY_SOURCE_CHANGED.code, AUTHORITY_SOURCE_CHANGED.reason); + const head = await store.loadAuthority(); + if (head.status !== "loaded") return {schema_version: COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA, ...head, changed: false}; + let result: JsonObject; + const mutations: {kind: "todo_upsert"; todo: JsonObject}[] = []; + try { + const indexed = indexCoordinationProjection(head.head, input.goal_id); + validateCoordinationTodoReadModel(head.head, input.goal_id); + const model = canonicalAuthorityObject(head.head.todo_read_model, "Todo read model"); + result = planFollowupCapture({schema_version: FOLLOWUP_CAPTURE_PLAN_SCHEMA, goal_id: input.goal_id, + operation_id: input.operation_id, intent: input.intent, dry_run: input.dry_run, + updated_at: input.now.toISOString().replace(/\.\d{3}Z$/u, "Z"), + // Done and deferred active rows still suppress repeated capture. Archive + // and User rows do not. The full head, never its display slice, owns this. + existing_texts: [...indexed.todos.values()].filter(todo => todo.role === "agent" && todo.archive_state === "active").map(todo => todo.text)}); + for (const rawItem of result.items as JsonObject[]) { + if (!rawItem.added) continue; + const item = canonicalAuthorityObject(rawItem, "captured Todo"); + const metadata = canonicalAuthorityObject(input.intent.metadata, "capture metadata"); + const created = planCoordinationTodoCreate({...input, actor_agent_id: null, registered_agents: [], + todo: {schema_version: TODO_DOMAIN_ITEM_SCHEMA, ...metadata, todo_id: item.todo_id, + text: item.todo, role: "agent", status: "open", done: false, archive_state: "active", + evidence: input.intent.evidence}}, indexed.todos, model.schema_version, "operation_lane"); + if (created.status !== "planned") throw new Error(String(created.reason)); + mutations.push({kind: "todo_upsert", todo: canonicalAuthorityObject(created.todo, "captured record")}); + } + } catch (error) { return failure("followup_capture_rejected", error instanceof Error ? error.message : String(error)); } + if (!await authoritySourcesCurrent()) return failure(AUTHORITY_SOURCE_CHANGED.code, AUTHORITY_SOURCE_CHANGED.reason); + if (input.dry_run) return {...result, schema_version: COORDINATION_FOLLOWUP_CAPTURE_RESULT_SCHEMA, + status: "planned", provider_revision: head.provider_revision}; + // Even a no-op seals its original decision. Replaying it after later edits + // must never become permission to create work that was previously skipped. + const commit = mutations.length ? prepareCoordinationProjectionCommit({goal_id: input.goal_id, operation_id: input.operation_id, + expected_provider_revision: head.provider_revision, projection: head.head, mutations}) : {operation_id: input.operation_id, + expected_provider_revision: head.provider_revision, next_projection: head.head, events: [], receipts: []}; + commit.receipts = [{...identity, result}]; + return receipt.commit(store, commit); +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index f1a2760aec..f654077873 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -1,3 +1,5 @@ +import {planFollowupCapture} from "./todos/followup_capture.ts"; +import {captureLocalFollowups} from "./coordination/followup_capture_runtime.ts"; import {selectDelegationBinding, transitionDelegationObservation} from "./collaboration/delegation.ts"; import {resolveConversationScope} from "./collaboration/conversation_scope.ts"; import {previewTeamPlan, planTeamTransaction, teamTransactionIdentity} from "./work_items/team_plan.ts"; @@ -505,6 +507,8 @@ export function createEffectRuntimeHandlers( ["coordination.local_authority.todo_continuation", continueLocalTodo], ["coordination.local_authority.todo_claim", claimLocalCoordinationTodo], ["coordination.local_authority.todo_create", createLocalCoordinationTodo], + ["todos.followup_capture.plan", planFollowupCapture], + ["coordination.local_authority.followup_capture", captureLocalFollowups], ["work_items.team_plan.preview", previewTeamPlan], ["work_items.team_plan.plan", planTeamTransaction], ["work_items.team_plan.identity", value => teamTransactionIdentity(requiredObject(value, "team plan request"))], diff --git a/loopx/control_plane/todos/active_state_editing.py b/loopx/control_plane/todos/active_state_editing.py index 0dbfa6ca52..582810d261 100644 --- a/loopx/control_plane/todos/active_state_editing.py +++ b/loopx/control_plane/todos/active_state_editing.py @@ -250,8 +250,9 @@ def todo_blocks( *, role: str | None = None, source_section: str | None = None, + text_limit: int | None = 500, ) -> list[dict[str, Any]]: - blocks = decode_todo_blocks(lines, start, end, visible=visible_markdown_lines(lines)) + blocks = decode_todo_blocks(lines, start, end, visible=visible_markdown_lines(lines), text_limit=text_limit) for block in blocks: ensure_block_identity(block, role=role, source_section=source_section) return blocks diff --git a/loopx/control_plane/todos/completion_policy.ts b/loopx/control_plane/todos/completion_policy.ts index bca5723765..806055eafd 100644 --- a/loopx/control_plane/todos/completion_policy.ts +++ b/loopx/control_plane/todos/completion_policy.ts @@ -19,7 +19,7 @@ export const TODO_COMPLETION_POLICY_REQUEST_SCHEMA = export const TODO_COMPLETION_POLICY_RESULT_SCHEMA = "loopx_todo_completion_policy_result_v0"; -const CONTINUATION_POLICIES = [ +export const TODO_CONTINUATION_POLICIES = [ "independent_handoff", "same_agent_non_delivery", ] as const; @@ -188,12 +188,12 @@ function requireExcludedAgents( function continuationPolicy( value: string | null, -): typeof CONTINUATION_POLICIES[number] { +): typeof TODO_CONTINUATION_POLICIES[number] { const candidate = stripPythonWhitespace(String(value ?? "")).toLowerCase(); - return CONTINUATION_POLICIES.includes( - candidate as typeof CONTINUATION_POLICIES[number], + return TODO_CONTINUATION_POLICIES.includes( + candidate as typeof TODO_CONTINUATION_POLICIES[number], ) - ? candidate as typeof CONTINUATION_POLICIES[number] + ? candidate as typeof TODO_CONTINUATION_POLICIES[number] : "independent_handoff"; } diff --git a/loopx/control_plane/todos/followup_capture.ts b/loopx/control_plane/todos/followup_capture.ts new file mode 100644 index 0000000000..95c1080be5 --- /dev/null +++ b/loopx/control_plane/todos/followup_capture.ts @@ -0,0 +1,107 @@ +/** One batch selection rule for legacy Markdown and canonical authority. + * Capture declares unclaimed work; it never grants execution or closes a Todo. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject, requireStringArray, requireNonEmptyString, requireBoolean} from "../runtime_decode.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import {compactPythonWhitespace} from "../coordination/todo_agents.ts"; +import {canonicalAuthoritySha256, requireAuthorityStoreId} from "../coordination/authority_store_codec.ts"; +import {AGENT_TODO_TASK_CLASSES} from "./authoring_scope.ts"; +import {normalizeTodoWorkRequirements} from "./work_requirements.ts"; +import {normalizeTodoRequiredDecisionScopes} from "./decision_metadata.ts"; +import {TODO_CONTINUATION_POLICIES} from "./completion_policy.ts"; + +export const FOLLOWUP_CAPTURE_PLAN_SCHEMA = "todo_followup_capture_plan_request_v0"; +export const FOLLOWUP_CAPTURE_RESULT_SCHEMA = "todo_followup_capture_result_v0"; +export const MAX_CAPTURED_FOLLOWUPS = 2; + +type UnsafeReason = "local_absolute_path" | "local_state_path" | "credential_literal" | "internal_only_marker"; +type SkipReason = "empty" | "duplicate" | "max_items_exceeded" | `unsafe_boundary:${UnsafeReason}`; +// Compatibility heuristics for this public-safe capture command, not a DLP +// classifier or authorization rule. Keep reasons typed and limitations explicit. +const UNSAFE_PATTERNS: readonly [UnsafeReason, RegExp][] = [ + ["local_absolute_path", /(?:\/Users\/|\/private\/|\/var\/folders\/|file:\/\/)/iu], + ["local_state_path", /(?:^|[\s"'`])(?:\.local\/|\.codex\/|\.loopx\/)/iu], + ["credential_literal", /(? pattern.test(text))?.[0] ?? null; +} + +export interface FollowupCaptureIntent extends JsonObject { + followups: string[]; + evidence: string; + metadata: JsonObject; +} + +/** Validate the entire request before selection, including a malformed tail or + * metadata on an all-duplicate batch. Invalid members must not be dropped. */ +export function normalizeFollowupCaptureIntent(value: unknown): FollowupCaptureIntent { + const raw = requireJsonObject(value, "follow-up capture intent"); + for (const key of Object.keys(raw)) { + if (!["followups", "evidence", "metadata"].includes(key)) throw new EffectRuntimeRequestError(`unsupported capture field: ${key}`); + } + const followups = requireStringArray(raw.followups, "followups").map(compactPythonWhitespace); + if (!followups.length) throw new EffectRuntimeRequestError("todo capture-followups requires at least one --follow-up"); + const evidence = compactPythonWhitespace(typeof raw.evidence === "string" ? raw.evidence : ""); + if (!evidence) throw new EffectRuntimeRequestError("todo capture-followups requires --evidence with a public-safe pointer"); + const unsafe = unsafeReason(evidence); + if (unsafe) throw new EffectRuntimeRequestError(`todo capture-followups evidence is not public-safe: ${unsafe}`); + const input = requireJsonObject(raw.metadata, "follow-up metadata"); + for (const key of Object.keys(input)) { + if (!["task_class", "action_kind", "continuation_policy", "required_write_scopes", "required_capabilities", + "target_capabilities", "required_decision_scopes"].includes(key)) throw new EffectRuntimeRequestError(`unsupported capture metadata: ${key}`); + } + const taskClass = input.task_class == null ? "advancement_task" + : compactPythonWhitespace(requireNonEmptyString(input.task_class, "task_class")).toLowerCase(); + if (!AGENT_TODO_TASK_CLASSES.has(taskClass)) throw new EffectRuntimeRequestError("capture-followups requires an agent task_class"); + const metadata: JsonObject = {task_class: taskClass, ...normalizeTodoWorkRequirements(input)}; + if (input.continuation_policy != null) { + const policy = compactPythonWhitespace(requireNonEmptyString(input.continuation_policy, "continuation_policy")).toLowerCase(); + if (!(TODO_CONTINUATION_POLICIES as readonly string[]).includes(policy)) throw new EffectRuntimeRequestError("unsupported continuation_policy"); + metadata.continuation_policy = policy; + } + const scopes = normalizeTodoRequiredDecisionScopes(input.required_decision_scopes); + if (scopes !== null) metadata.required_decision_scopes = scopes; + return {followups, evidence, metadata}; +} + +export function planFollowupCapture(value: unknown): JsonObject { + const request = requireJsonObject(value, "follow-up capture plan request"); + if (request.schema_version !== FOLLOWUP_CAPTURE_PLAN_SCHEMA) throw new EffectRuntimeRequestError("follow-up capture plan schema mismatch"); + for (const key of Object.keys(request)) { + if (!["schema_version", "goal_id", "operation_id", "updated_at", "dry_run", "intent", "existing_texts"].includes(key)) throw new EffectRuntimeRequestError(`unsupported capture plan field: ${key}`); + } + const goal = requireAuthorityStoreId(request.goal_id, "goal id"); + const operation = requireAuthorityStoreId(request.operation_id, "operation id"); + const updatedAt = requireNonEmptyString(request.updated_at, "updated_at"); + if (Number.isNaN(Date.parse(updatedAt))) throw new EffectRuntimeRequestError("updated_at must be a timestamp"); + const dryRun = requireBoolean(request.dry_run, "dry_run"); + const intent = normalizeFollowupCaptureIntent(request.intent); + const existing = new Set(requireStringArray(request.existing_texts, "existing_texts").map(compactPythonWhitespace)); + const items: JsonObject[] = []; + let recorded = 0; + for (const [index, text] of intent.followups.entries()) { + let reason: SkipReason | null = null; + const unsafe = unsafeReason(text); + if (!text) reason = "empty"; + else if (unsafe) reason = `unsafe_boundary:${unsafe}`; + else if (existing.has(text)) reason = "duplicate"; + else if (recorded >= MAX_CAPTURED_FOLLOWUPS) reason = "max_items_exceeded"; + const item: JsonObject = {todo: text, added: reason === null, already_exists: reason === "duplicate", + skipped: reason !== null, skipped_reason: reason}; + if (reason === null) { + recorded += 1; + existing.add(text); + Object.assign(item, intent.metadata, {todo_id: `todo_${canonicalAuthoritySha256({goal, operation, index}).slice(0, 24)}`, + role: "agent", section: "Agent Todo", status: "open", changed: true, + metadata_updated: false, status_changed: false, evidence: intent.evidence, updated_at: updatedAt}); + } + items.push(item); + } + return {schema_version: FOLLOWUP_CAPTURE_RESULT_SCHEMA, ok: true, dry_run: dryRun, changed: recorded > 0, + goal_id: goal, role: "agent", section: "Agent Todo", max_items: MAX_CAPTURED_FOLLOWUPS, + requested_count: items.length, recorded_count: recorded, skipped_count: items.length - recorded, + evidence: intent.evidence, items, updated_at: recorded ? updatedAt : null}; +} diff --git a/loopx/control_plane/todos/provider_followups.py b/loopx/control_plane/todos/provider_followups.py new file mode 100644 index 0000000000..f70996e834 --- /dev/null +++ b/loopx/control_plane/todos/provider_followups.py @@ -0,0 +1,60 @@ +"""Capture transport and current-head delivery; TypeScript owns the batch.""" +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from ...history import load_registry +from ...state_refresh import resolve_goal_state +from ..coordination.authority_source_capture import authority_registry_source +from ..coordination.local_authority import ( + LOCAL_AUTHORITY_SOURCES, + LocalCoordinationAuthorityUnavailable, + local_authority_is_promoted, +) +from ..effect_runtime import effect_runtime_result +from .provider_projection import settle_canonical_todo_projection + + +def capture_canonical_followups_if_promoted( + *, registry_path: Path, runtime_root: Path, goal_id: str, + operation_id: str, intent: dict[str, Any], dry_run: bool, + project: Path | None, state_file: Path | None, +) -> dict[str, Any] | None: + if not local_authority_is_promoted(runtime_root=runtime_root, goal_id=goal_id): + return None + with authority_registry_source(registry_path) as source: + registry_source = dict(source) + goal, resolved_project, resolved_state = resolve_goal_state( + registry=load_registry(registry_path), goal_id=goal_id, + project_override=project, state_file_override=state_file, + ) + if goal is None: + raise ValueError(f"goal {goal_id!r} is not present in the registry") + result = effect_runtime_result("coordination.local_authority.followup_capture", { + "schema_version": "loopx_coordination_followup_capture_request_v0", + "runtime_root": str(runtime_root.resolve()), "goal_id": goal_id, + "operation_id": operation_id, "intent": intent, "dry_run": dry_run, + "registry_source": registry_source, + }) + if (not isinstance(result, dict) + or result.get("status") not in {"applied", "recovered", "replayed", "no_change", "planned"} + or result.get("source_authority") not in LOCAL_AUTHORITY_SOURCES + or result.get("decision_read_from_provider") is not True + or result.get("legacy_fallback_used") is not False + or not isinstance(result.get("items"), list)): + payload = result if isinstance(result, dict) else {} + raise LocalCoordinationAuthorityUnavailable( + str(payload.get("reason") or "canonical follow-up capture unavailable"), + code=str(payload.get("reason_code") or "followup_capture_unavailable"), + payload={**payload, "capture_operation_id": operation_id}, + ) + payload = {**result, "ok": True, "dry_run": dry_run, + "capture_operation_id": operation_id, + "state_file": str(resolved_state), + "project": str(resolved_project) if resolved_project else None, + "idempotent_replay": result["status"] == "replayed"} + return settle_canonical_todo_projection( + payload, registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, + project=project, state_file=state_file, + ) diff --git a/loopx/control_plane/todos/todo_block_codec.py b/loopx/control_plane/todos/todo_block_codec.py index a21aa08ea5..9894771c84 100644 --- a/loopx/control_plane/todos/todo_block_codec.py +++ b/loopx/control_plane/todos/todo_block_codec.py @@ -7,6 +7,7 @@ def decode_todo_blocks( lines: list[str], start: int, end: int, *, visible: frozenset[int], + text_limit: int | None = 500, ) -> list[dict[str, Any]]: blocks: list[dict[str, Any]] = [] current: dict[str, Any] | None = None @@ -24,12 +25,12 @@ def decode_todo_blocks( status = todo_status_from_marker(marker) current = {"start": index, "end": end, "index": len(blocks) + 1, "done": todo_done_for_status(status), "status": status, - "text": normalize_todo_text(text)} + "text": normalize_todo_text(text, limit=text_limit)} blocks.append(current) elif current is not None and lines[index].startswith((" ", "\t")): metadata = parse_todo_metadata_line(lines[index]) if metadata: current.update(metadata) elif continuation := lines[index].strip(): - current["text"] = normalize_todo_text(f"{current['text']} {continuation}") + current["text"] = normalize_todo_text(f"{current['text']} {continuation}", limit=text_limit) return blocks diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py index a6deeb2a14..0af01988bc 100644 --- a/loopx/control_plane/todos/todo_summary.py +++ b/loopx/control_plane/todos/todo_summary.py @@ -152,9 +152,9 @@ class _TodoGroupLanes: ) -def normalize_todo_text(text: str, *, limit: int = 500) -> str: +def normalize_todo_text(text: str, *, limit: int | None = 500) -> str: compact = " ".join(str(text or "").strip().split()) - if len(compact) <= limit: + if limit is None or len(compact) <= limit: return compact return compact[: limit - 1].rstrip() + "…" diff --git a/loopx/todo_followups.py b/loopx/todo_followups.py index 8d3a0a5ffb..0c42f429b2 100644 --- a/loopx/todo_followups.py +++ b/loopx/todo_followups.py @@ -1,6 +1,6 @@ from __future__ import annotations -import re +from uuid import uuid4 from pathlib import Path from typing import Any @@ -12,51 +12,19 @@ settle_todo_runtime_shadow_capture, ) from .state_refresh import now_local -from .control_plane.todos.contract import TODO_TASK_CLASS_ADVANCEMENT -from .control_plane.todos.todo_summary import normalize_todo_text -from .todos import ( - TODO_SECTION_HEADINGS, - add_todo_to_lines, +from .control_plane.todos.contract import TODO_METADATA_FIELDS, format_todo_metadata_line +from .control_plane.todos.path_resolution import resolve_todo_state_path +from .control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result +from .control_plane.todos.provider_followups import capture_canonical_followups_if_promoted +from .control_plane.todos.active_state_editing import ( + insert_into_existing_section, + insert_new_section, replace_updated_at, - resolve_todo_state_path, section_bounds, todo_blocks, ) -MAX_CAPTURED_FOLLOWUP_TODOS = 2 - -_UNSAFE_FOLLOWUP_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( - ("local_absolute_path", re.compile(r"(?i)(?:/Users/|/private/|/var/folders/|file://)")), - ("local_state_path", re.compile(r"(?i)(?:^|[\s\"'`])(?:\.local/|\.codex/|\.loopx/)")), - ("credential_literal", re.compile(r"(?i)\b(?:api[_-]?key|secret|password|token)\s*[:=]")), - ("internal_only_marker", re.compile(r"(?i)\binternal[-_\s]?only\b")), -) - - -def _unsafe_followup_reason(value: str) -> str | None: - for reason, pattern in _UNSAFE_FOLLOWUP_PATTERNS: - if pattern.search(value): - return reason - return None - - -def _compact_text(value: Any) -> str: - return " ".join(str(value or "").strip().split()) - - -def _existing_agent_todo_texts(lines: list[str]) -> set[str]: - bounds = section_bounds(lines, "agent") - if not bounds: - return set() - start, end, section = bounds - return { - normalize_todo_text(str(block.get("text") or "")) - for block in todo_blocks(lines, start, end, role="agent", source_section=section) - if block.get("text") - } - - def capture_followup_todos( *, registry_path: Path, @@ -65,6 +33,8 @@ def capture_followup_todos( evidence: str, task_class: str | None = None, action_kind: str | None = None, + continuation_policy: str | None = None, + capture_operation_id: str | None = None, required_write_scopes: list[str] | None = None, required_capabilities: list[str] | None = None, target_capabilities: list[str] | None = None, @@ -74,14 +44,27 @@ def capture_followup_todos( dry_run: bool = False, runtime_root_arg: str | None = None, ) -> dict[str, Any]: - if not followups: - raise ValueError("todo capture-followups requires at least one --follow-up") - evidence_text = _compact_text(evidence) - if not evidence_text: - raise ValueError("todo capture-followups requires --evidence with a public-safe pointer") - evidence_reason = _unsafe_followup_reason(evidence_text) - if evidence_reason: - raise ValueError(f"todo capture-followups evidence is not public-safe: {evidence_reason}") + runtime_root = effective_runtime_root(registry_path, runtime_root_arg) + operation_id = capture_operation_id if capture_operation_id is not None else f"followup-capture:{uuid4().hex}" + intent = {"followups": followups, "evidence": evidence, "metadata": { + key: value for key, value in { + "task_class": task_class, "action_kind": action_kind, + "continuation_policy": continuation_policy, + "required_write_scopes": required_write_scopes, + "required_capabilities": required_capabilities, + "target_capabilities": target_capabilities, + "required_decision_scopes": required_decision_scopes, + }.items() if value is not None + }} + canonical = capture_canonical_followups_if_promoted( + registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, + operation_id=operation_id, intent=intent, dry_run=dry_run, + project=project, state_file=state_file, + ) + if canonical is not None: + return canonical + if capture_operation_id is not None: + raise ValueError("--capture-operation-id requires promoted canonical authority; legacy capture has no durable command receipt") resolved_project, resolved_state_file = resolve_todo_state_path( registry_path=registry_path, @@ -90,74 +73,48 @@ def capture_followup_todos( state_file=state_file, ) - items: list[dict[str, Any]] = [] - runtime_root = effective_runtime_root(registry_path, runtime_root_arg) with legacy_todo_write_transaction( registry_path, goal_id, resolved_state_file, None, "todo_capture_followups", dry_run, runtime_root=runtime_root, ): original = resolved_state_file.read_text(encoding="utf-8") + lines = original.splitlines() + bounds = section_bounds(lines, "agent") + existing = todo_blocks(lines, bounds[0], bounds[1], role="agent", + source_section=bounds[2], text_limit=None) if bounds else [] + updated_at = now_local() + try: + result = effect_runtime_result("todos.followup_capture.plan", { + "schema_version": "todo_followup_capture_plan_request_v0", + "goal_id": goal_id, "operation_id": operation_id, + "updated_at": updated_at, "dry_run": dry_run, "intent": intent, + "existing_texts": [block["text"] for block in existing], + }) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from None + if not isinstance(result, dict) or result.get("schema_version") != "todo_followup_capture_result_v0": + raise ValueError("TypeScript follow-up capture plan shape mismatch") + changed = result["changed"] + recorded_count = result["recorded_count"] capture = begin_todo_runtime_shadow_capture( registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, state_path=resolved_state_file, write_class="todo_capture_followups", original_text=original, ) - lines = original.splitlines() - existing_texts = _existing_agent_todo_texts(lines) - seen_texts: set[str] = set() - updated_at = now_local() - changed = False - recorded_count = 0 - - for raw_followup in followups: - todo_text = _compact_text(raw_followup) - item: dict[str, Any] = { - "todo": todo_text, - "added": False, - "already_exists": False, - "skipped": False, - "skipped_reason": None, - } - if not todo_text: - item.update({"skipped": True, "skipped_reason": "empty"}) - items.append(item) - continue - - unsafe_reason = _unsafe_followup_reason(todo_text) - if unsafe_reason: - item.update({"skipped": True, "skipped_reason": f"unsafe_boundary:{unsafe_reason}"}) - items.append(item) - continue - - normalized = normalize_todo_text(todo_text) - if normalized in existing_texts or normalized in seen_texts: - item.update({"already_exists": True, "skipped": True, "skipped_reason": "duplicate"}) - items.append(item) + # This adapter renders accepted rows only. Do not call single-add here: + # its duplicate/admission decisions would recreate a second batch owner. + for item in result["items"]: + if not item["added"]: continue - - if recorded_count >= MAX_CAPTURED_FOLLOWUP_TODOS: - item.update({"skipped": True, "skipped_reason": "max_items_exceeded"}) - items.append(item) - continue - - add_result = add_todo_to_lines( - lines, - role="agent", - text=todo_text, - task_class=task_class or TODO_TASK_CLASS_ADVANCEMENT, - action_kind=action_kind, - required_write_scopes=required_write_scopes, - required_capabilities=required_capabilities, - target_capabilities=target_capabilities, - required_decision_scopes=required_decision_scopes, - evidence=evidence_text, - updated_at=updated_at, - ) - changed = changed or bool(add_result.get("added")) or bool(add_result.get("metadata_updated")) - recorded_count += 1 - seen_texts.add(normalized) - item.update(add_result) - items.append(item) + metadata = format_todo_metadata_line(**{ + key: value for key, value in item.items() if key in TODO_METADATA_FIELDS + }) + row = f"- [ ] {item['todo']}\n{metadata}" + bounds = section_bounds(lines, "agent") + if bounds: + insert_into_existing_section(lines, bounds[0], bounds[1], row) + else: + insert_new_section(lines, "agent", row) if changed: new_text = "\n".join(lines) + ("\n" if original.endswith("\n") else "") @@ -166,23 +123,8 @@ def capture_followup_todos( write_captured_todo_state(capture, runtime_root=runtime_root, goal_id=goal_id, state_path=resolved_state_file, text=new_text) - result = { - "ok": True, - "dry_run": dry_run, - "changed": changed, - "goal_id": goal_id, - "role": "agent", - "section": TODO_SECTION_HEADINGS["agent"], - "state_file": str(resolved_state_file), - "project": str(resolved_project) if resolved_project else None, - "max_items": MAX_CAPTURED_FOLLOWUP_TODOS, - "requested_count": len(followups), - "recorded_count": recorded_count, - "skipped_count": sum(1 for item in items if item.get("skipped")), - "evidence": evidence_text, - "items": items, - "updated_at": updated_at if changed else None, - } + result.update(state_file=str(resolved_state_file), + project=str(resolved_project) if resolved_project else None) if changed and not dry_run: from .control_plane.coordination.local_authority_shadow_observation import observe_local_authority_commit From 4bb4d74e117f085126b4ccbeb7ddebda4cba216e Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 19 Sep 2026 02:38:54 +0800 Subject: [PATCH 2/4] test(todos): qualify atomic capture across real providers and recovery Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../authority-followup-capture-rehearsal.py | 169 +++++++++++++++++ .../test_shadow_writer_boundaries.py | 10 +- .../test_todo_followup_capture.py | 176 ++++++++++++++++++ .../authority_source_conformance.ts | 2 +- .../authority_store_conformance.ts | 2 + .../coordination_command_fixture.ts | 6 +- .../coordination_receipt_conformance.ts | 2 +- .../control_plane_ts/followup_capture.test.ts | 24 +++ .../followup_capture_conformance.ts | 113 +++++++++++ tests/test_cli_argument_diagnostics.py | 2 +- 10 files changed, 497 insertions(+), 9 deletions(-) create mode 100644 examples/control_plane/authority-followup-capture-rehearsal.py create mode 100644 tests/control_plane/test_todo_followup_capture.py create mode 100644 tests/control_plane_ts/followup_capture.test.ts create mode 100644 tests/control_plane_ts/followup_capture_conformance.ts diff --git a/examples/control_plane/authority-followup-capture-rehearsal.py b/examples/control_plane/authority-followup-capture-rehearsal.py new file mode 100644 index 0000000000..6057d1bc0f --- /dev/null +++ b/examples/control_plane/authority-followup-capture-rehearsal.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +"""Rehearse batch capture on a read-only Goal snapshot and disposable providers. + +The source is never promoted or rewritten. Output contains counts and digests; +raw source, identifiers, paths and diagnostics stay outside the public report. +PostgreSQL must be an explicitly isolated server, not an active service tenant. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile + +REPOSITORY = Path(__file__).resolve().parents[2] +if str(REPOSITORY) not in sys.path: + sys.path.insert(0, str(REPOSITORY)) + +from loopx.control_plane.coordination.runtime_shadow import build_runtime_shadow_source_snapshot # noqa: E402 +from loopx.history import load_registry # noqa: E402 +from loopx.paths import resolve_runtime_root # noqa: E402 +from loopx.state_refresh import resolve_goal_state # noqa: E402 +from loopx.todo_followups import capture_followup_todos # noqa: E402 + +NODE_REHEARSAL = r""" +import assert from 'node:assert/strict'; +import {createHash,randomUUID} from 'node:crypto'; +import {mkdtemp,mkdir,writeFile,rm} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; +import {pathToFileURL} from 'node:url'; +import {Pool} from 'pg'; +let raw=''; for await(const chunk of process.stdin) raw+=chunk; +const input=JSON.parse(raw), goal=input.goal_id; +const at=path=>import(pathToFileURL(join(input.repo,'loopx/control_plane',path)).href); +const {captureLocalFollowups:capture}=await at('coordination/followup_capture_runtime.ts'); +const {openLocalAuthorityStore,selectLocalSqliteAuthority}=await at('coordination/local_authority_provider.ts'); +const {PostgreSqlAuthorityStore,installPostgreSqlAuthorityStoreSchema}=await at('coordination/postgresql_authority_store.ts'); +const {PostgreSqlAuthorityService}=await at('coordination/postgresql_authority_service.ts'); +const {engageLegacyCoordinationWriterFence}=await at('coordination/legacy_writer_fence.ts'); +const {canonicalAuthoritySha256:digest}=await at('coordination/authority_store_codec.ts'); +const root=await mkdtemp(join(tmpdir(),'loopx-capture-rehearsal-')); +const pool=new Pool({connectionString:process.env.LOOPX_TEST_POSTGRES_URL,max:4}); +const database={connect:async()=>{const c=await pool.connect();return {query:(t,v)=>c.query(t,v),release:e=>c.release(e)};}}; +const tenant=`capture-rehearsal-${randomUUID()}`, reports={}, semantics={}; +try { + await installPostgreSqlAuthorityStoreSchema(database,`postgresql:${'b'.repeat(32)}`); + for(const arm of ['file','sqlite','postgresql']) { + const runtime=join(root,arm);await mkdir(runtime,{recursive:true}); + const registry=join(runtime,'registry.json'), display=join(runtime,'state.md'); + await writeFile(registry,'{}'); await writeFile(display,'# Disposable display\n'); + let dependencies={}; + if(arm==='sqlite') assert.equal((await selectLocalSqliteAuthority(runtime,goal,true)).ok,true); + if(arm==='postgresql') { + const store=new PostgreSqlAuthorityStore(database,{tenant_id:tenant,goal_id:goal}); + const identity=await store.storeIdentity();assert.equal(identity.status,'available'); + await mkdir(join(runtime,'authority'),{recursive:true}); + await writeFile(join(runtime,'authority',`provider-${createHash('sha256').update(goal).digest('hex')}.json`),JSON.stringify({ + schema_version:'loopx_local_authority_provider_v0',provider:'postgresql',goal_id:goal,tenant_id:tenant,store_identity:identity.store_identity})); + const service=new PostgreSqlAuthorityService({database, + authenticatePrincipal:()=>({status:'authenticated',principal:{principal_id:'isolated-capture'}}), + authorizeTenant:(principal,selected)=>principal==='isolated-capture'&&selected===tenant?{status:'allowed'}: + {status:'denied',reason_code:'wrong_tenant',reason:'outside disposable tenant'}}); + dependencies={openPostgresqlStore:async selected=>{ + const opened=await service.openStore({credential:null,...selected});assert.equal(opened.status,'opened');return opened.store;}}; + } + const store=await openLocalAuthorityStore(runtime,goal,dependencies); + const seed=await store.commitAuthority({operation_id:'seed',expected_provider_revision:null,next_projection:input.projection,events:[],receipts:[]}); + assert.equal(seed.status,'applied'); + assert.equal((await engageLegacyCoordinationWriterFence({schema_version:'loopx_legacy_coordination_writer_fence_engage_request_v0', + runtime_root:runtime,goal_id:goal,state_path:display,fence:{schema_version:'loopx_legacy_coordination_writer_fence_v0',state:'engaged', + goal_id:goal,fence_id:'rehearsal',source_version:'snapshot',source_projection_sha256:digest(input.projection), + expected_shadow_provider_revision:seed.provider_revision}})).status,'applied'); + await rm(display); + const request={schema_version:'loopx_coordination_followup_capture_request_v0',runtime_root:runtime,goal_id:goal, + operation_id:'rehearsal-batch',intent:input.intent,dry_run:false, + registry_source:{path:registry,sha256:createHash('sha256').update('{}').digest('hex')}}; + const before=await store.loadAuthority(); + assert.equal((await capture({...request,dry_run:true},dependencies)).status,'planned'); + assert.deepEqual(await store.loadAuthority(),before); + const applied=await capture(request,dependencies);assert.equal(applied.status,'applied'); + assert.equal(applied.recorded_count,2);assert.equal(applied.skipped_count,2); + const after=await store.loadAuthority();assert.equal(after.status,'loaded'); + const originalIds=new Set(input.projection.todos.map(t=>t.todo_id)); + assert.deepEqual(after.head.todos.filter(t=>originalIds.has(t.todo_id)),input.projection.todos); + assert.deepEqual(after.head.leases,input.projection.leases); + const added=after.head.todos.filter(t=>!originalIds.has(t.todo_id)); + assert.equal(added.length,2);assert(added.every(t=>!t.claimed_by)); + const replay=await capture(request,dependencies);assert.equal(replay.status,'replayed'); + assert.deepEqual(replay.original_receipt,applied.original_receipt); + assert.deepEqual(await store.loadAuthority(),after); + const noop=await capture({...request,operation_id:'duplicate-batch',intent:{...input.intent,followups:input.intent.followups.slice(0,2)}},dependencies); + assert.equal(noop.status,'no_change');assert.equal(noop.recorded_count,0); + semantics[arm]=added.map(t=>({text:t.text,continuation_policy:t.continuation_policy,required_capabilities:t.required_capabilities})).sort((a,b)=>a.text.localeCompare(b.text)); + reports[arm]={added:added.length,skipped:applied.skipped_count,preview_no_write:true,replayed:true,no_op_sealed:true,existing_records_unchanged:true}; + } + assert.deepEqual(semantics.file,semantics.sqlite);assert.deepEqual(semantics.file,semantics.postgresql); + assert.deepEqual(semantics.file,input.legacy_semantics); + process.stdout.write(JSON.stringify({schema_version:'authority_followup_capture_rehearsal_v0', + source_todos:input.projection.todos.length,source_leases:input.projection.leases.length, + semantic_sha256:digest(semantics.file),provider_semantics_equal:true,arms:reports})); +} finally { + for(const table of ['authority_receipts','authority_events','authority_commits','authority_heads']) + await pool.query(`DELETE FROM loopx_control_plane.${table} WHERE tenant_id=$1`,[tenant]); + await pool.end();await rm(root,{recursive:true,force:true}); +} +""" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--registry", type=Path, required=True) + parser.add_argument("--goal-id", required=True) + parser.add_argument("--execute-isolated-postgresql", action="store_true") + parser.add_argument("--private-diagnostics", type=Path) + args = parser.parse_args() + if not args.execute_isolated_postgresql or not os.environ.get("LOOPX_TEST_POSTGRES_URL"): + raise SystemExit("an explicitly isolated PostgreSQL server is required") + registry_path = args.registry.resolve() + registry_bytes = registry_path.read_bytes() + registry = load_registry(registry_path) + goal = next(g for g in registry["goals"] if g["id"] == args.goal_id) + runtime = resolve_runtime_root(registry, None, registry_path=registry_path) + _, _, state = resolve_goal_state(registry=registry, goal_id=args.goal_id, project_override=None, state_file_override=None) + projection, snapshot = build_runtime_shadow_source_snapshot(goal=goal, runtime_root=runtime, state_path=state, registry_path=registry_path) + source_text = state.read_text() + # Public synthetic deltas are intentionally independent of source wording. + intent = {"followups": ["Validate isolated capture atomicity", "Validate isolated capture recovery", + "Validate isolated capture atomicity", "Inspect file://rehearsal"], + "evidence": "validation://isolated-capture", "metadata": { + "continuation_policy": "same_agent_non_delivery", "required_capabilities": ["code_review"]}} + with tempfile.TemporaryDirectory(prefix="loopx-legacy-capture-") as tmp: + root = Path(tmp) + cloned_state = root / "state.md" + cloned_state.write_text(source_text) + cloned_registry = root / "registry.json" + cloned_registry.write_text(json.dumps({"common_runtime_root": str(root / "runtime"), "goals": [ + {"id": args.goal_id, "repo": str(root), "state_file": cloned_state.name}]})) + legacy = capture_followup_todos(registry_path=cloned_registry, goal_id=args.goal_id, + followups=intent["followups"], evidence=intent["evidence"], **intent["metadata"]) + assert legacy["recorded_count"] == 2 and legacy["skipped_count"] == 2 + semantics = sorted(({"text": item["todo"], "continuation_policy": item["continuation_policy"], + "required_capabilities": item["required_capabilities"]} for item in legacy["items"] if item["added"]), key=lambda t: t["text"]) + child = subprocess.run(["node", "--no-warnings", "--experimental-sqlite", "--experimental-strip-types", "--input-type=module", "-e", NODE_REHEARSAL], + input=json.dumps({"repo": str(REPOSITORY), "goal_id": args.goal_id, "projection": projection, "intent": intent, "legacy_semantics": semantics}), + cwd=REPOSITORY, capture_output=True, text=True, timeout=180, check=False) + if child.returncode: + if args.private_diagnostics: + descriptor = os.open(args.private_diagnostics, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "w") as stream: + stream.write(child.stderr) + raise SystemExit("isolated capture rehearsal failed; diagnostics remain private") + after_projection, after_snapshot = build_runtime_shadow_source_snapshot(goal=goal, runtime_root=runtime, state_path=state, registry_path=registry_path) + if projection != after_projection or snapshot != after_snapshot or registry_path.read_bytes() != registry_bytes or state.read_text() != source_text: + raise SystemExit("live source changed during rehearsal; retry from a stable snapshot") + result = json.loads(child.stdout) + result.update(source_unchanged=True, source_snapshot_sha256=hashlib.sha256(json.dumps(snapshot, sort_keys=True).encode()).hexdigest(), + legacy={"added": 2, "skipped": 2, "semantics_equal": True}) + print(json.dumps(result, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/control_plane/test_shadow_writer_boundaries.py b/tests/control_plane/test_shadow_writer_boundaries.py index bfdc48b11e..35f693cc1c 100644 --- a/tests/control_plane/test_shadow_writer_boundaries.py +++ b/tests/control_plane/test_shadow_writer_boundaries.py @@ -43,7 +43,7 @@ def fixture(tmp_path: Path) -> tuple[Path, Path, Path]: @pytest.mark.parametrize("writer", ["handoff", "followups"]) -def test_omitted_writers_refuse_a_fence_before_primary( +def test_writers_refuse_invalid_authority_before_primary( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, writer: str, ) -> None: registry, state, root = fixture(tmp_path) @@ -57,7 +57,9 @@ def test_omitted_writers_refuse_a_fence_before_primary( }, ) before = state.read_bytes() - with pytest.raises(LegacyCoordinationWriterFenced): + from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable + expected = LegacyCoordinationWriterFenced if writer == "handoff" else LocalCoordinationAuthorityUnavailable + with pytest.raises(expected): if writer == "handoff": set_goal_handoff_mode(registry_path=registry, goal_id=GOAL, mode="soft_claim") else: @@ -638,15 +640,13 @@ def refuse_prepare(path: Path, record: object) -> None: assert list(directory.glob("*.committed.json")) == [] -def test_public_preview_does_not_require_primary_write_permission(tmp_path: Path) -> None: +def test_legacy_preview_does_not_require_primary_write_permission(tmp_path: Path) -> None: from loopx.control_plane.coordination.shadow_management import shadow_management_state_path registry, state, root = fixture(tmp_path) before = state.read_bytes() management = shadow_management_state_path(root, GOAL) management.parent.mkdir(parents=True) management.write_text("{}") - fence = legacy_coordination_writer_fence_path(runtime_root=root, goal_id=GOAL) - fence.write_text("{invalid") preview = cli(registry, "todo", "capture-followups", "--goal-id", GOAL, "--follow-up", "Preview remains read-only.", "--evidence", "preview fixture", "--dry-run") assert preview["dry_run"] is True diff --git a/tests/control_plane/test_todo_followup_capture.py b/tests/control_plane/test_todo_followup_capture.py new file mode 100644 index 0000000000..de0aaf445b --- /dev/null +++ b/tests/control_plane/test_todo_followup_capture.py @@ -0,0 +1,176 @@ +"""Public batch capture regressions, including real canonical stores.""" +from pathlib import Path +import runpy + +import pytest + +from loopx.todo_followups import capture_followup_todos +from loopx.control_plane.todos.active_state_todo_parser import parse_todo_source + +_SMOKE = runpy.run_path(str(Path(__file__).resolve().parents[2] / "examples/control_plane/todo-capture-followups-smoke.py")) +fixture = _SMOKE["write_fixture"] +cli = _SMOKE["run_cli"] +GOAL = _SMOKE["GOAL_ID"] + + +def test_capture_cli_preserves_continuation_policy(tmp_path): + registry, state = fixture(tmp_path) + result = cli(registry, "todo", "capture-followups", "--goal-id", GOAL, + "--follow-up", "Validate public follow-up capture", "--evidence", "validation://capture", + "--continuation-policy", "same_agent_non_delivery") + assert result["recorded_count"] == 1 + items, _, _ = parse_todo_source(state.read_text()) + assert items["agent"][-1].get("continuation_policy") == "same_agent_non_delivery" + + +def test_capture_uses_full_text_identity(tmp_path): + registry, state = fixture(tmp_path) + prefix = "[P1] " + "long text " * 60 + one, two = prefix + "first invariant", prefix + "second invariant" + args = dict(registry_path=registry, goal_id=GOAL, followups=[one, two], evidence="validation://capture") + result = capture_followup_todos(**args) + assert result["recorded_count"] == 2 + assert all(item["added"] for item in result["items"]) + assert one in state.read_text() and two in state.read_text() + replay = capture_followup_todos(**args) + assert replay["recorded_count"] == 0 + assert all(item["skipped_reason"] == "duplicate" for item in replay["items"]) + + +def test_legacy_capture_characterization(tmp_path): + registry, state = fixture(tmp_path) + original = state.read_bytes() + args = dict(registry_path=registry, goal_id=GOAL, + followups=["", " First successor ", "First successor", "Inspect file://raw", "Second successor", "Third successor"], + evidence="validation://capture") + preview = capture_followup_todos(**args, dry_run=True) + assert preview["recorded_count"] == 2 + assert [x["skipped_reason"] for x in preview["items"]] == ["empty", None, "duplicate", "unsafe_boundary:local_absolute_path", None, "max_items_exceeded"] + assert state.read_bytes() == original + captured = capture_followup_todos(**args) + assert captured["recorded_count"] == 2 + assert "claimed_by" not in state.read_text() + with pytest.raises(ValueError, match="public-safe"): + capture_followup_todos(**{**args, "evidence": "password=redacted-fixture"}) + + +@pytest.fixture(params=["file", "sqlite"]) +def canonical(tmp_path, monkeypatch, request): + from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted + from loopx.todos import list_goal_todos + if request.param == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, state = fixture(tmp_path) + runtime = tmp_path / "runtime" + projection = build_todo_runtime_shadow_projection(goal_id=GOAL, handoff_mode="hard_lease", + todos=list_goal_todos(registry_path=registry, goal_id=GOAL)["todos"]) + initialize_canonical_authority(runtime, GOAL, projection, state_path=state, provider=request.param) + def read(): + return read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL, include_leases=True) + return registry, state, runtime, read + + +def test_canonical_cli_batch_retry_and_missing_display(canonical): + registry, state, _, read = canonical + initial = read() + args = ("todo", "capture-followups", "--goal-id", GOAL, "--follow-up", "First canonical task", + "--follow-up", "Second canonical task", "--evidence", "validation://capture-cli", + "--continuation-policy", "same_agent_non_delivery", "--capture-operation-id", "capture-cli") + state.unlink() + preview = cli(registry, *args, "--dry-run") + assert preview["status"] == "planned" and preview["recorded_count"] == 2 + assert read() == initial and not state.exists() + applied = cli(registry, *args) + assert applied["status"] == "applied", applied + assert applied["projection_delivery"] == "delivered", applied + assert applied["recorded_count"] == 2 + first = read() + assert len(first["todos"]) == len(initial["todos"]) + 2 + assert first["leases"] == initial["leases"] + added = [x for x in first["todos"] if x["text"] in {"First canonical task", "Second canonical task"}] + assert len(added) == 2 and all(x.get("claimed_by") is None for x in added) + assert all(x["continuation_policy"] == "same_agent_non_delivery" for x in added) + later = capture_followup_todos(registry_path=registry, goal_id=GOAL, + followups=["A later task"], evidence="validation://later") + assert later["recorded_count"] == 1 + latest = read() + state.unlink() + replay = cli(registry, *args) + assert replay["status"] == "replayed" and replay["changed"] is False + assert replay["original_receipt"] == applied["original_receipt"] + assert replay["recorded_count"] == 2 # Historical accepted batch, not new insertions. + assert read() == latest + assert "A later task" in state.read_text() # Display drains latest, not receipt head. + mismatch = cli(registry, *args, "--follow-up", "Changed request", check=False) + assert mismatch["ok"] is False and mismatch["error_code"] == "coordination_operation_identity_mismatch" + assert read() == latest + + +def test_canonical_projection_failure_is_recoverable(canonical, monkeypatch): + from loopx.control_plane.todos import provider_projection + registry, state, _, read = canonical + original = state.read_bytes() + render = provider_projection.project_current_canonical_todos + def fail(**_kwargs): + raise OSError("synthetic display unavailable") + monkeypatch.setattr(provider_projection, "project_current_canonical_todos", fail) + args = dict(registry_path=registry, goal_id=GOAL, followups=["Committed before rendering"], + evidence="validation://capture", capture_operation_id="pending-display") + applied = capture_followup_todos(**args) + assert applied["ok"] and applied["projection_delivery"] == "pending" + after = read() + assert any(x["text"] == "Committed before rendering" for x in after["todos"]) + assert state.read_bytes() == original + monkeypatch.setattr(provider_projection, "project_current_canonical_todos", render) + replay = capture_followup_todos(**args) + assert replay["status"] == "replayed" and replay["projection_delivery"] == "delivered" + assert read() == after + + +def test_canonical_provider_outage_never_writes_legacy(canonical, monkeypatch): + from loopx.control_plane.todos import provider_followups + from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable + registry, state, _, read = canonical + before, original = read(), state.read_bytes() + monkeypatch.setattr(provider_followups, "effect_runtime_result", lambda *_args: { + "status": "unavailable", "reason_code": "synthetic_outage", "reason": "provider unavailable"}) + with pytest.raises(LocalCoordinationAuthorityUnavailable) as error: + capture_followup_todos(registry_path=registry, goal_id=GOAL, + followups=["Must not fall back"], evidence="validation://capture", capture_operation_id="retry-this") + assert error.value.payload["capture_operation_id"] == "retry-this" + assert state.read_bytes() == original and read() == before + + +def test_legacy_rejects_durable_operation_identity(tmp_path): + registry, state = fixture(tmp_path) + before = state.read_bytes() + with pytest.raises(ValueError, match="requires promoted"): + capture_followup_todos(registry_path=registry, goal_id=GOAL, + followups=["Requires canonical history"], evidence="validation://capture", capture_operation_id="durable") + assert state.read_bytes() == before + + +@pytest.mark.parametrize("metadata", [{"required_capabilities": ["valid", "bad/token"]}, + {"required_write_scopes": ["src/**", "../escape"]}, {"continuation_policy": "removed-policy"}]) +def test_legacy_invalid_metadata_is_atomic(tmp_path, metadata): + registry, state = fixture(tmp_path) + before = state.read_bytes() + with pytest.raises(ValueError): + capture_followup_todos(registry_path=registry, goal_id=GOAL, + followups=["First task", "Second task"], evidence="validation://capture", **metadata) + assert state.read_bytes() == before + + +def test_preview_refuses_missing_promoted_provider(tmp_path): + from loopx.control_plane.coordination.legacy_writer_fence import legacy_coordination_writer_fence_path + registry, state = fixture(tmp_path) + fence = legacy_coordination_writer_fence_path(runtime_root=tmp_path / "runtime", goal_id=GOAL) + fence.parent.mkdir(parents=True) + fence.write_text("{invalid") + before = state.read_bytes() + result = cli(registry, "todo", "capture-followups", "--goal-id", GOAL, "--follow-up", "No fake preview", + "--evidence", "validation://capture", "--dry-run", check=False) + assert result["ok"] is False and result["legacy_fallback_used"] is False + assert state.read_bytes() == before diff --git a/tests/control_plane_ts/authority_source_conformance.ts b/tests/control_plane_ts/authority_source_conformance.ts index b5c122175e..5d0ba70ae7 100644 --- a/tests/control_plane_ts/authority_source_conformance.ts +++ b/tests/control_plane_ts/authority_source_conformance.ts @@ -9,7 +9,7 @@ import {registryAuthoritySourceCheck} from "../../loopx/control_plane/coordinati import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; import {coordinationCommandFixture, type Command} from "./coordination_command_fixture.ts"; -const commands: readonly Command[] = ["create", "claim", "update", "complete", "supersede", "monitor"]; +const commands: readonly Command[] = ["create", "claim", "update", "complete", "supersede", "monitor", "capture"]; async function sourceFixture(t: test.TestContext) { const root = await mkdtemp(join(tmpdir(), "loopx-source-")); diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts index cae48387d1..0c9f0913c1 100644 --- a/tests/control_plane_ts/authority_store_conformance.ts +++ b/tests/control_plane_ts/authority_store_conformance.ts @@ -1,3 +1,4 @@ +import {registerFollowupCaptureConformance} from "./followup_capture_conformance.ts"; import {registerUserCompletionUpdateConformance} from "./user_completion_update_conformance.ts"; import {registerLeaseAcquisitionConformance} from "./lease_acquisition_conformance.ts"; import {registerClaimTransferConformance} from "./claim_transfer_conformance.ts"; @@ -215,6 +216,7 @@ export function registerAuthorityStoreConformance( providerName: string, factory: AuthorityStoreConformanceFactory, ): void { + registerFollowupCaptureConformance(providerName, factory); registerLeaseLifecycleConformance(providerName, factory); registerClaimTransferConformance(providerName, factory); registerLeaseAcquisitionConformance(providerName, factory); diff --git a/tests/control_plane_ts/coordination_command_fixture.ts b/tests/control_plane_ts/coordination_command_fixture.ts index f2bfce6fa5..2227a7e204 100644 --- a/tests/control_plane_ts/coordination_command_fixture.ts +++ b/tests/control_plane_ts/coordination_command_fixture.ts @@ -1,3 +1,4 @@ +import {executeCoordinationFollowupCapture} from "../../loopx/control_plane/coordination/todo_followup_capture.ts"; import {executeCoordinationTodoArchiveCompleted} from "../../loopx/control_plane/coordination/todo_archive.ts"; /** Shared real command fixture over the complete production-scale head. */ import assert from "node:assert/strict"; @@ -13,7 +14,7 @@ import {productionScaleCoordinationFixture, PRODUCTION_SCALE_VALIDATION_DECLARAT import type {AuthoritySourceCheck} from "../../loopx/control_plane/coordination/authority_source.ts"; -export type Command = "create" | "claim" | "update" | "complete" | "supersede" | "archive" | "monitor"; +export type Command = "create" | "claim" | "update" | "complete" | "supersede" | "archive" | "monitor" | "capture"; export async function coordinationCommandFixture(store: AuthorityStore, command: Command) { const goal_id = "goal-a"; @@ -39,6 +40,9 @@ export async function coordinationCommandFixture(store: AuthorityStore, command: } = {}): Promise => { const request = {...common, operation_id: options.identity ?? operation_id, dry_run: options.dryRun ?? false}; const sourceCheck = options.authoritySourcesCurrent; + if (command === "capture") return executeCoordinationFollowupCapture(target, {...request, + intent: {followups: ["Recover first captured task", "Recover second captured task"], + evidence: "validation://capture", metadata: {}}}, sourceCheck); if (command === "create") return executeCoordinationTodoCreate(target, {...request, actor_agent_id: "agent-a", todo: {schema_version: TODO_DOMAIN_ITEM_SCHEMA, todo_id: "todo_recovery_created", role: "agent", status: "open", done: false, archive_state: "active", text: "Recover the accepted create"}}, sourceCheck); diff --git a/tests/control_plane_ts/coordination_receipt_conformance.ts b/tests/control_plane_ts/coordination_receipt_conformance.ts index 84fcc9e8a5..fa950c3cfe 100644 --- a/tests/control_plane_ts/coordination_receipt_conformance.ts +++ b/tests/control_plane_ts/coordination_receipt_conformance.ts @@ -7,7 +7,7 @@ import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/a import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; import {coordinationCommandFixture, type Command} from "./coordination_command_fixture.ts"; -const commands: readonly Command[] = ["create", "claim", "update", "complete", "supersede", "archive", "monitor"]; +const commands: readonly Command[] = ["create", "claim", "update", "complete", "supersede", "archive", "monitor", "capture"]; type Fault = "none" | "lost_response" | "unreadable_receipt" | "ambiguous_unreadable" | "thrown_response"; const faults: readonly Fault[] = ["none", "lost_response", "unreadable_receipt", "ambiguous_unreadable", "thrown_response"]; diff --git a/tests/control_plane_ts/followup_capture.test.ts b/tests/control_plane_ts/followup_capture.test.ts new file mode 100644 index 0000000000..bf4fcba800 --- /dev/null +++ b/tests/control_plane_ts/followup_capture.test.ts @@ -0,0 +1,24 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import {planFollowupCapture, FOLLOWUP_CAPTURE_PLAN_SCHEMA} from "../../loopx/control_plane/todos/followup_capture.ts"; +const input = {schema_version: FOLLOWUP_CAPTURE_PLAN_SCHEMA, goal_id: "goal-a", operation_id: "batch", + updated_at: "2026-09-07T07:00:00Z", dry_run: true, existing_texts: [], + intent: {followups: ["One task"], evidence: "validation://public", metadata: {}}}; + +test("capture retains Python whitespace and Unicode word boundaries without lossy identity", () => { + const result = planFollowupCapture({...input, existing_texts: ["Already captured"], intent: {...input.intent, + followups: ["\u001c", "Already\u0085captured", "αtoken" + "=prose", "internal-onlyβ", "token" + "=fixture", "internal only"]}}); + assert.deepEqual((result.items as JsonObject[]).map(i => i.skipped_reason), + ["empty", "duplicate", null, null, "unsafe_boundary:credential_literal", "unsafe_boundary:internal_only_marker"]); +}); + +test("capture validates malformed tails and unknown authority fields before any plan", () => { + for (const invalid of [ + {...input, actor_agent_id: "agent-a"}, + {...input, intent: {...input.intent, followups: ["One", "Two", null]}}, + {...input, intent: {...input.intent, metadata: {claimed_by: "agent-a"}}}, + {...input, intent: {...input.intent, metadata: {task_class: "user_gate"}}}, + {...input, intent: {...input.intent, metadata: {required_decision_scopes: ["direction:goal:valid", "bad"]}}}, + ]) assert.throws(() => planFollowupCapture(invalid)); +}); diff --git a/tests/control_plane_ts/followup_capture_conformance.ts b/tests/control_plane_ts/followup_capture_conformance.ts new file mode 100644 index 0000000000..4119362b1b --- /dev/null +++ b/tests/control_plane_ts/followup_capture_conformance.ts @@ -0,0 +1,113 @@ +/** Batch semantics are identical over every real AuthorityStore implementation. */ +import assert from "node:assert/strict"; +import test from "node:test"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts"; +import {executeCoordinationFollowupCapture as capture} from "../../loopx/control_plane/coordination/todo_followup_capture.ts"; +import {prepareCoordinationProjectionCommit} from "../../loopx/control_plane/coordination/coordination_projection.ts"; +import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; +import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts"; +import {authorityProjectionFixture} from "./authority_projection_fixture.ts"; + +const request = {goal_id: "capture-goal", operation_id: "capture-batch", dry_run: false, + now: new Date("2026-09-07T07:00:00Z"), intent: {followups: ["First new task", "Second new task"], + evidence: "validation://batch", metadata: {continuation_policy: "same_agent_non_delivery", + required_capabilities: ["Code-Review"], required_write_scopes: ["src/**"]}}}; +async function head(store: AuthorityStore) { + const result = await store.loadAuthority(); + assert.equal(result.status, "loaded"); + if (result.status !== "loaded") throw new Error("missing fixture"); + return result; +} +export function registerFollowupCaptureConformance(provider: string, factory: AuthorityStoreConformanceFactory) { + for (const schema of ["native", "legacy"] as const) test(`${provider}: capture batch preserves complete ${schema} state and seals no-op history`, async t => { + const {store} = await factory(t); + const fixture = productionScaleCoordinationFixture(request.goal_id, schema); + const initial = fixture.projection; + const old = initial.todos as JsonObject[]; + const done = old.find(row => row.role === "agent" && row.status === "done" && row.archive_state === "active")!; + assert.ok(done); + const prefix = "Long identity ".repeat(45); + const input = {...request, intent: {...request.intent, + followups: [String(done.text), prefix + "A", prefix + "B", "Third new task"]}}; + assert.equal((await store.commitAuthority({operation_id: "seed", expected_provider_revision: null, + next_projection: initial, events: [], receipts: []})).status, "applied"); + const before = await head(store); + const preview = await capture(store, {...input, dry_run: true}); + assert.equal(preview.status, "planned"); + assert.equal(preview.recorded_count, 2); + assert.deepEqual(await head(store), before); + assert.equal((await store.readReceipt(input.operation_id)).status, "missing"); + const result = await capture(store, input); + assert.equal(result.status, "applied", JSON.stringify(result)); + assert.deepEqual((result.items as JsonObject[]).map(i => i.skipped_reason), ["duplicate", null, null, "max_items_exceeded"]); + const after = await head(store); + const added = (after.head.todos as JsonObject[]).filter(row => !old.some(o => o.todo_id === row.todo_id)); + assert.equal(added.length, 2); + assert.deepEqual((after.head.todos as JsonObject[]).filter(row => old.some(o => o.todo_id === row.todo_id)), old); + assert.deepEqual(after.head.leases, initial.leases); + for (const row of added) { + assert.equal(row.claimed_by, undefined); + assert.equal(row.continuation_policy, "same_agent_non_delivery"); + assert.deepEqual(row.required_capabilities, ["code_review"]); + } + const noOp = {...input, operation_id: "all-duplicates", intent: {...input.intent, followups: [prefix + "A", prefix + "B"]}}; + const noChange = await capture(store, noOp); + assert.equal(noChange.status, "no_change", JSON.stringify(noChange)); + assert.equal(noChange.changed, false); + assert.equal(noChange.recorded_count, 0); + const sealed = await head(store); + assert.deepEqual(sealed.head, after.head, "receipt-only write preserves domain state"); + await store.commitAuthority(prepareCoordinationProjectionCommit({goal_id: request.goal_id, operation_id: "retire-captured", + expected_provider_revision: sealed.provider_revision, projection: sealed.head, + mutations: added.map(todo => ({kind: "todo_upsert" as const, todo: {...todo, archive_state: "archive"}}))})); + const later = await head(store); + for (const original of [input, noOp]) { + const replay = await capture(store, {...original, now: new Date("2030-01-01")}, async () => { throw new Error("replay must precede admission"); }); + assert.equal(replay.status, "replayed"); + assert.equal(replay.changed, false); + assert.deepEqual(await head(store), later); + } + const mismatch = await capture(store, {...input, intent: {...input.intent, evidence: "validation://different"}}); + assert.equal(mismatch.reason_code, "coordination_operation_identity_mismatch"); + }); + + test(`${provider}: capture rejects invalid whole batches and stale CAS without partial rows`, async t => { + const {store, contender} = await factory(t); + const projection = authorityProjectionFixture(request.goal_id, []); + await store.commitAuthority({operation_id: "seed", expected_provider_revision: null, next_projection: projection, events: [], receipts: []}); + const before = await head(store); + for (const intent of [ + {...request.intent, followups: ["Valid first item", 42]}, + {...request.intent, metadata: {required_capabilities: ["valid", "bad/token"]}}, + {...request.intent, metadata: {claimed_by: "agent-a"}}, + {...request.intent, metadata: {task_class: "user_gate"}}, + ]) { + assert.equal((await capture(store, {...request, intent})).status, "failed"); + assert.deepEqual(await head(store), before); + assert.equal((await store.readReceipt(request.operation_id)).status, "missing"); + } + // The competitor commits between our validated read and CAS. It is a real + // provider transaction, not a fake conflict or a pair of partial inserts. + let raced = false; + const racing: AuthorityStore = { + storeIdentity: () => store.storeIdentity(), loadAuthority: () => store.loadAuthority(), + readReceipt: id => store.readReceipt(id), scanCommitted: (cursor, limit) => store.scanCommitted(cursor, limit), + commitAuthority: async commit => { + raced = true; + assert.equal((await capture(contender, {...request, operation_id: "winner"})).status, "applied"); + return store.commitAuthority(commit); + }, + }; + const lost = await capture(racing, request); + assert.equal(raced, true); + assert.equal(lost.status, "conflict"); + assert.equal((await store.readReceipt(request.operation_id)).status, "missing"); + const after = await head(store); + assert.equal((after.head.todos as JsonObject[]).length, 2); + const retry = await capture(store, request); + assert.equal(retry.status, "no_change"); + assert.equal(retry.recorded_count, 0); + assert.deepEqual((await head(store)).head, after.head); + }); +} diff --git a/tests/test_cli_argument_diagnostics.py b/tests/test_cli_argument_diagnostics.py index 3ae8f61ab9..5ece266128 100644 --- a/tests/test_cli_argument_diagnostics.py +++ b/tests/test_cli_argument_diagnostics.py @@ -1047,7 +1047,7 @@ def test_todo_suggest_validation_accepts_suggestion_scope_options() -> None: ( ["--todo-id", "todo_example", "--note", "not accepted"], "todo capture-followups only accepts --goal-id, --follow-up, optional " - "--text shorthand, --evidence, routing metadata, --project, --state-file, " + "--text shorthand, --evidence, routing metadata, --capture-operation-id, --project, --state-file, " "and --dry-run; unsupported: --todo-id, --note", ), ], From 9c3450fdb164937a38b25d1a769d8768971f87d5 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 19 Sep 2026 02:38:54 +0800 Subject: [PATCH 3/4] docs(todos): document capture recovery and migration checkpoint Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...shared-goal-authority-state-provider-v0.md | 2 +- ...-goal-authority-state-provider-v0.zh-CN.md | 2 +- .../typescript-control-plane-migration-v0.md | 6 + ...script-control-plane-migration-v0.zh-CN.md | 9 ++ docs/reference/canonical-followup-capture.md | 119 ++++++++++++++++++ 5 files changed, 136 insertions(+), 2 deletions(-) create mode 100644 docs/reference/canonical-followup-capture.md 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 de7e65bd5f..dd4f197777 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -3071,7 +3071,7 @@ or moving a helper is not by itself a package exit. | Wave / package | Reviewable delivery and TS ownership payoff | Dependencies and exit evidence | | --- | --- | --- | | A / L1: Monitor configuration (this slice) | Existing `todo update` config enters the TS planner/CAS/receipt; delete Python's duplicate intent field catalog. Separate authoring from observed hashes, times and generations. | Ordinary CLI/API, clear/omission, active lease proof, no-op/replay, failed display delivery, complete fixture and real providers. This does not complete delegated Chat or leased polling. | -| A / L2: Complete public mutation admission | User completion updates now share the TS edit/terminal transaction and reviewed Chat recovery. Continue the actual CLI/Turn/Chat inventory for remaining effect-owned decisions, delegated owner actions and Monitor lifecycle transitions; [caller contract](../../reference/canonical-todo-completion-update.md). | Build on merged T1 owners, not a generic raw patch. Prove permission rejection and exact caller response; remove replaced Python admission and name every remaining unsupported command. | +| A / L2: Complete public mutation admission | User completion updates share the TS edit/terminal transaction and reviewed Chat recovery. [Follow-up capture](../../reference/canonical-followup-capture.md) now uses one TS batch plan and provider CAS/receipt, with a shared legacy adapter; it no longer falls through to the fenced writer. Continue the actual CLI/Turn/Chat inventory for remaining effect-owned decisions, delegated owner actions and Monitor lifecycle transitions; [caller contract](../../reference/canonical-todo-completion-update.md). | Build on merged T1 owners, not a generic raw patch. Prove permission rejection and exact caller response; remove replaced Python admission and name every remaining unsupported command. | | A / L3: Canonical lease lifecycle | Standalone acquire/takeover, atomic claim lease admission and maintenance reuse TS facts/decision/materialization and one provider opening fence. Explicit claimed-work transfer now commits source-authorized Todo ownership and the new lease generation together; canonical request types exclude legacy held-fence fields. Acquire success verifies current execution proof; canonical completion can recover missing display. | Full-head scope conflict, archived/ineffective holders, exact create-CAS retry, stale execution, process loss and real CLI/four-arm rehearsal are covered. [Operation and remaining callers](../../reference/canonical-lease-renew.md). Executor-held external-effect fences remain explicit work; D1–D3/default holds remain. | | B / L4: Leased Monitor poll and settlement | Current execution proof now binds CLI intent, observation/generation/independent-successor CAS and historical business receipt. Quota pending admission is frozen before the business write; recovery preserves that decision after lease retirement. | Existing L3 lease lifecycle, real File/SQLite/PostgreSQL, mixed fixtures, process death between business/quota commits, competing renewal and unchanged polling. [Operation and snapshot rehearsal](../../reference/protocols/quota-monitor-observation-receipt-v0.md). No lease lifecycle effects or quota spend; separate authorities stay separate. Event callers, wider L2 admission and D1–D3/default remain open. | | B / L5: Consumer and display closure | Reconcile #4316, audit Turn/quota/Dashboard/Chat source reads, and finish D1 freshness/recovery through the existing projection outbox. | CLI, Lark/Chat and packaged frontend read back their affected interactions; absent/stale display, empty canonical state, pending projection and data beyond UI limits. Delete post-promotion legacy fallbacks with each consumer. | 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 33c3aa55b6..4545a0ef3e 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 @@ -2434,7 +2434,7 @@ canonical renew 候选，#4328 是 SQLite D2 首批测量／恢复候选；它 | 波次／PR 包 | 完整交付内容与 TS 归属收益 | 依赖与退出证据 | | --- | --- | --- | | A／L1：Monitor 配置（本切片） | 现有 `todo update` 配置进入 TS planner／CAS／receipt，删除 Python 重复 intent 字段表；区分配置与观察 hash、时间、代数。 | 普通 CLI/API、清除／省略、active lease proof、no-op／replay、展示失败恢复、完整 fixture 和真实 provider。不宣称完成委托 Chat 或 leased polling。 | -| A／L2：公共 mutation admission 闭合 | 用户 completion update 已共用 TS 编辑／terminal 事务与 Chat 审阅后恢复；继续盘点剩余 effect-owned 决策、委托 owner 动作和 Monitor lifecycle 的 CLI／Turn／Chat caller。见[调用合同](../../reference/canonical-todo-completion-update.md)。 | 复用已合并 T1 owner，不开通通用 raw patch；验证权限拒绝和 caller 响应，删除替代的 Python admission，列全未支持命令。 | +| A／L2：公共 mutation admission 闭合 | 用户 completion update 共用 TS 编辑／terminal 事务与 Chat 审阅后恢复；[后续任务批量捕获](../../reference/canonical-followup-capture.md) 已用同一 TS 计划与 provider CAS／回执，legacy 仅适配共享计划，不再进入被 fence 的旧 writer。继续盘点剩余 effect-owned 决策、委托 owner 动作和 Monitor lifecycle 的 CLI／Turn／Chat caller。见[调用合同](../../reference/canonical-todo-completion-update.md)。 | 复用已合并 T1 owner，不开通通用 raw patch；验证权限拒绝和 caller 响应，删除替代的 Python admission，列全未支持命令。 | | A／L3：canonical lease 生命周期 | 独立 acquire/接管、原子 claim 的 lease 准入及维护复用 TS facts/decision/materializer 与同一 provider opening fence。显式联合交接由源持有者授权，一次提交 Todo 归属与新租约 generation；canonical 请求类型不再携带 legacy 持锁字段。Acquire 成功必须校验当前执行 proof；canonical 完成可恢复缺失展示。 | 已覆盖完整 head scope 冲突、归档/失效 holder、创建 CAS 原样重试、旧执行、进程中断、真实 CLI 与四臂演练。[操作及剩余 caller](../../reference/canonical-lease-renew.md)。跨外部 effect 的 executor 持锁 fence 仍为明确工作；保留 D1–D3/default hold。 | | B／L4：leased Monitor poll 与 settlement | 当前 execution proof 贯穿 CLI intent、观察／generation／独立 successor CAS 和历史业务回执；业务写入前冻结 quota 准入，租约结束后仍按原决策恢复结算。 | 既有 L3 lease lifecycle、真实 File／SQLite／PostgreSQL、混合 fixture、业务与 quota 间真实进程退出、并发 renewal 和 unchanged poll；见[操作与快照演练](../../reference/protocols/quota-monitor-observation-receipt-v0.md)。不操作 lease lifecycle、不消耗 quota，不把两个 authority 假装成同一事务。Event caller、更广 L2 准入及 D1–D3/default 仍开放。 | | B／L5：consumer 与展示闭合 | 核对 #4316，审计 Turn／quota／Dashboard／Chat 的来源，复用 projection outbox 完成 D1 新鲜度和恢复。 | 验证 CLI、Lark／Chat、打包 frontend 的受影响交互；缺失／陈旧展示、权威空状态、pending 投影及超过 UI 上限的数据。逐个删除晋升后的 legacy fallback。 | diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index bbd8e37082..c20d9a8882 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1282,6 +1282,12 @@ requires differential proof; record its deletion trigger when introduced. Current implementation status: Stage 1, the bounded Stage 2A proofs, and the shipped Stage 2B cutovers are in place: +- [Follow-up capture](../../reference/canonical-followup-capture.md): TypeScript owns + complete batch selection, full-text duplicate identity, metadata and canonical + CAS/receipt recovery. Python retains one locked renderer/shadow write for + legacy Goals and provider transport/outbox for promoted Goals. This deletes + Python selection and per-item add decisions; it is T1/L2 command closure, + not L7 capture qualification or a new-Goal default change. - Turn settlement/commit: TypeScript owns preflight authorization, ordered-prefix and replay validation, provider failure classification, receipt construction, terminal closeout joining, and the canonical result. 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 98d69fabd3..a8c285a9b8 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 @@ -103,6 +103,15 @@ coordination 路径使用同一份语言中立的 `coordination_state_contract_v 仅将 typed read result 适配为兼容 summary。这是 contract 检查点，不是已经完成的 CLI lifecycle cutover。 +### 后续任务批量捕获收敛 + +[`capture-followups`](../../reference/canonical-followup-capture.md) 的完整批次筛选、 +全文去重、metadata 和 canonical CAS／历史回执由 TS 持有。Python 删除筛选与逐条 +add 决策，保留 legacy 锁／渲染／shadow 写入及 promoted provider transport／outbox。 +修复 500 字符展示截断误作身份及 continuation policy 丢失；覆盖完整混合图、竞争 +CAS、no-op 封存、lost acknowledgement 和真实 provider／只读快照演练。 +这是 T1／L2 的一个完整命令，不是 L7 shadow capture 连续性或新 Goal 默认切换。 + ### Lease 领取与生命周期收敛（2026-09-18） 独立 acquire/接管和维护共用 local provider/source fence。`task_lease_acquire_decision.ts` diff --git a/docs/reference/canonical-followup-capture.md b/docs/reference/canonical-followup-capture.md new file mode 100644 index 0000000000..e1c7303c73 --- /dev/null +++ b/docs/reference/canonical-followup-capture.md @@ -0,0 +1,119 @@ +# Canonical follow-up capture + +`todo capture-followups` records a bounded batch of **unclaimed Agent Todos**. +It does not complete another Todo, acquire a lease, authorize an executor, or +perform the captured work. The built-in Todo planner owns selection and metadata; +coordination owns the canonical transaction. No new capability or provider is +introduced. + +## Use and recover + +On an already promoted Goal: + +```bash +loopx --format json todo capture-followups --goal-id example \ + --capture-operation-id followups-review-1 \ + --follow-up 'Validate the recovery boundary' \ + --follow-up 'Document the operator recovery command' \ + --evidence 'validation://reviewed-capture' \ + --continuation-policy same_agent_non_delivery \ + --required-capability code_review --dry-run +``` + +Remove `--dry-run` to commit. Reuse the same operation id **and intent** after a +lost response; changing the evidence, metadata, order or follow-ups under that +id is rejected. Omit the id for a fresh operation on every invocation. Preview +writes no business receipt, so its id can be used for the subsequent commit. +An explicit id requires canonical authority; legacy Markdown cannot promise a +durable command receipt and rejects that option before writing. + +Read back through `loopx --format json todo list --goal-id example`. File and +SQLite use the existing local selector. PostgreSQL uses the same transaction +through the existing service-owned store factory, with its own authentication +and tenant admission; this command does not introduce standalone PostgreSQL CLI +configuration or silently fall back when a selected provider is unavailable. + +## Selection and atomicity + +One TypeScript plan processes the whole request in order: + +1. Empty text is skipped. +2. Existing public-safe boundary heuristics reject local paths, credential-like + literals and internal-only markers. Unsafe **evidence rejects the batch**; + an unsafe follow-up is reported as skipped. These are bounded heuristics, + not comprehensive secret detection or a grant to publish captured text. +3. Duplicate full text is skipped after Python-compatible whitespace compaction. + All Agent Todos in the active section count, including done and deferred + records. User and archived records do not suppress capture. +4. At most two new Todos are accepted; skipped items do not consume that limit. + +The old 500-character **display** limit no longer defines capture identity. +Distinct long texts remain distinct, and rereading the legacy source uses an +explicit lossless decoder mode. Other display callers retain their limits. +`--continuation-policy`, previously accepted but dropped by the CLI, now reaches +both writers. Invalid metadata or a malformed tail rejects the entire request; +invalid requirements are not silently removed. This applies even to a batch +whose texts would all be duplicates. Optional routing metadata is returned +when supplied, rather than as an unrelated catalog of empty fields. + +The legacy adapter holds its existing lock, renders only the accepted plan, +and writes once through shadow capture. It no longer owns regex classification, +selection, cap, duplicate identity or per-item add decisions. The canonical +adapter reads the full head, shares native Todo creation admission/materialization, +then commits all accepted rows and one receipt under one provider CAS. A stale +CAS cannot leave the first row committed without the second. Original Todo, +lease and standing-decision records remain untouched. + +A no-op batch also seals a receipt while leaving the domain head unchanged; +its provider revision can advance even though `changed=false`. After later +archive/edit operations, retrying that id still returns the original no-op. +On replay, `changed=false` describes this invocation, while `recorded_count`, +`items[].added` and `original_receipt` describe the historical batch. A receipt +never grants current execution authority. + +## Display delivery and boundaries + +Markdown is delivered separately through the existing committed-authority outbox. +A display failure returns successful business state with +`projection_delivery=pending`; it does not roll the batch back. Retry the same +operation, or use the existing `todo project-markdown` recovery command for the +current provider revision. Delivery renders the **latest head**, not a stale +receipt snapshot. Missing Markdown does not prevent canonical capture or preview. +A missing/unavailable canonical provider also blocks preview; it must not invent +success from legacy Markdown. + +The affected entry point is the CLI/Python capture API and its shared TS runtime. +There is no dedicated capture-followups frontend or Lark editor; existing Todo +readers consume its ordinary unclaimed records and the normal projection outbox. +No UI setting or new default is added. Reverting this code requires keeping +promoted Goals fenced and withholding this command until its transaction is +restored; do not re-enable the old Markdown writer as a rollback shortcut. + +## Validation and roadmap checkpoint + +Native conformance runs over File, SQLite, NoKV and real isolated PostgreSQL, +including complete synthetic graphs, legacy/native records, no-op/replay, +competing CAS, invalid input, source changes and lost acknowledgements. Python +coverage exercises the public CLI on real File/SQLite, missing display, +projection failure/recovery and provider failure without legacy fallback. + +The read-only snapshot rehearsal compares legacy, File, SQLite and the +PostgreSQL service runtime. Supply a disposable PostgreSQL server explicitly: + +```bash +uv run --extra test python examples/control_plane/authority-followup-capture-rehearsal.py \ + --registry --goal-id --execute-isolated-postgresql \ + --private-diagnostics +``` + +`LOOPX_TEST_POSTGRES_URL` selects that isolated server. All effects target temporary +copies and a disposable tenant. Reports contain bounded counts/digests, and the +source is checked unchanged afterward. This is a command qualification, not a +whole-Goal promotion or a long-duration soak. + +This closes capture-followups within shared-authority **L2** and TS **T1**. +It does not close the separate **L7 shadow-capture continuity** package. Remaining +public mutation/effect admission, consumers, D1/D2/D3, whole-Goal rollback, +new-Goal defaults and final legacy-writer retirement retain their existing +owners and acceptance gates. Consult the [shared-authority program](../architecture/rfcs/shared-goal-authority-state-provider-v0.md) +and [TS migration roadmap](../architecture/rfcs/typescript-control-plane-migration-v0.md). From 7de9b7525ba6aa863e2f6f54c671006706a5d42d Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 19 Sep 2026 03:04:47 +0800 Subject: [PATCH 4/4] fix(todos): keep canonical machine projection lossless Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- docs/reference/canonical-followup-capture.md | 5 +++- .../todos/active_state_todo_parser.py | 3 +- .../todos/machine_section_projection.py | 28 +++++++++++-------- loopx/control_plane/todos/todo_summary.py | 3 +- .../test_todo_followup_capture.py | 12 +++++++- .../test_todo_machine_section_projection.py | 17 +++++++++++ 6 files changed, 53 insertions(+), 15 deletions(-) diff --git a/docs/reference/canonical-followup-capture.md b/docs/reference/canonical-followup-capture.md index e1c7303c73..667c70a005 100644 --- a/docs/reference/canonical-followup-capture.md +++ b/docs/reference/canonical-followup-capture.md @@ -49,7 +49,10 @@ One TypeScript plan processes the whole request in order: The old 500-character **display** limit no longer defines capture identity. Distinct long texts remain distinct, and rereading the legacy source uses an -explicit lossless decoder mode. Other display callers retain their limits. +explicit lossless decoder mode. Machine projection also validates through lossless source/metadata codecs rather +than status summaries; long records can be delivered instead of remaining pending. +Missing native priority/title annotations are derived for display, while explicit +contradictions still fail parity. Other display callers retain their limits. `--continuation-policy`, previously accepted but dropped by the CLI, now reaches both writers. Invalid metadata or a malformed tail rejects the entire request; invalid requirements are not silently removed. This applies even to a batch diff --git a/loopx/control_plane/todos/active_state_todo_parser.py b/loopx/control_plane/todos/active_state_todo_parser.py index e97b208980..87f5bc1193 100644 --- a/loopx/control_plane/todos/active_state_todo_parser.py +++ b/loopx/control_plane/todos/active_state_todo_parser.py @@ -25,6 +25,7 @@ def parse_todo_source( *, goal: dict[str, Any] | None = None, state_path: Path | None = None, + text_limit: int | None = 500, ) -> tuple[dict[str, list[dict[str, Any]]], list[dict[str, Any]], dict[str, str | None]]: """Decode active and archived source rows without inventing archive roles.""" source_sections: dict[str, str | None] = {"user": None, "agent": None} @@ -37,7 +38,7 @@ def parse_todo_source( target = archive_items if archive else items[region.role] if not archive and source_sections[region.role] is None: source_sections[region.role] = region.heading - for block in decode_todo_blocks(lines, region.start, region.body_end, visible=visible): + for block in decode_todo_blocks(lines, region.start, region.body_end, visible=visible, text_limit=text_limit): todo = {"archive_state": "archive" if archive else "active", "source_section": region.heading if archive else source_sections[region.role], **({} if archive else {"role": region.role}), diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py index 4182ffd574..1673808ab5 100644 --- a/loopx/control_plane/todos/machine_section_projection.py +++ b/loopx/control_plane/todos/machine_section_projection.py @@ -38,7 +38,7 @@ completion_validation_declaration_sha256, project_completion_validation_authority, ) -from .active_state_todo_parser import parse_active_state_todos +from .active_state_todo_parser import parse_todo_source from .contract import ( TODO_DECISION_SCOPE_SCHEMA_VERSION, TODO_METADATA_FIELDS, @@ -49,7 +49,7 @@ require_todo_decision_scope, todo_marker_for_status, ) -from .todo_summary import canonical_todo_read_record, todo_priority_parts, normalize_todo_text +from .todo_summary import canonical_todo_read_record, todo_priority_parts, normalize_todo_text, structured_todo_item TODO_SECTION_PROJECTION_SCHEMA_VERSION = "loopx_todo_section_projection_v0" @@ -260,15 +260,13 @@ def _render_section( def _parsed_active_records(markdown: str) -> list[dict[str, Any]]: - fields = parse_active_state_todos(markdown, item_limit=None) - records: list[dict[str, Any]] = [] - for role in TODO_SECTION_HEADINGS: - summary = fields.get(f"{role}_todos") - items = summary.get("items") if isinstance(summary, dict) else [] - for item in sorted(items or [], key=_record_sort_key): - if isinstance(item, dict) and item.get("archive_state") == "active": - records.append(canonical_todo_read_record(item, reject_unknown=False)) - return records + # Machine round-trip validation needs full text, not status display slices + # or live eligibility evaluation. Reuse the source and metadata codecs. + items, _, sections = parse_todo_source(markdown, text_limit=None) + return [canonical_todo_read_record(structured_todo_item(item, role=role, + source_section=sections[role], text_limit=None), reject_unknown=False) + for role in TODO_SECTION_HEADINGS + for item in sorted(items[role], key=_record_sort_key)] def _parsed_archive_records(markdown: str) -> list[dict[str, Any]]: @@ -282,6 +280,7 @@ def _parsed_archive_records(markdown: str) -> list[dict[str, Any]]: bounds[0], bounds[1], source_section=COMPLETED_WORK_ARCHIVE_HEADING, + text_limit=None, ): if item.get("role") not in TODO_SECTION_HEADINGS: raise TodoSectionProjectionError( @@ -323,6 +322,13 @@ def _projection_record( }, reject_unknown=True, )) + # Native records need no redundant priority/title fields. Derive their + # display values here, while retaining explicit values so contradictions + # still fail parity instead of silently rewriting canonical metadata. + priority, title = todo_priority_parts(str(projected["text"])) + if priority: + projected.setdefault("priority", priority) + projected.setdefault("title", normalize_todo_text(title)) if "decision_scope" in projected: # The Markdown codec spells out the optional scope schema version on # readback. Normalize that display representation, never the provider diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py index 0af01988bc..cf3c272c24 100644 --- a/loopx/control_plane/todos/todo_summary.py +++ b/loopx/control_plane/todos/todo_summary.py @@ -319,8 +319,9 @@ def structured_todo_item( role: str | None, source_section: str | None, archive_state: str = "active", + text_limit: int | None = 500, ) -> dict[str, Any]: - text = normalize_todo_text(str(item.get("text") or "")) + text = normalize_todo_text(str(item.get("text") or ""), limit=text_limit) priority, title = todo_priority_parts(text) index = item.get("index") explicit_status = normalize_todo_status(item.get("status")) diff --git a/tests/control_plane/test_todo_followup_capture.py b/tests/control_plane/test_todo_followup_capture.py index de0aaf445b..763de287dd 100644 --- a/tests/control_plane/test_todo_followup_capture.py +++ b/tests/control_plane/test_todo_followup_capture.py @@ -23,8 +23,18 @@ def test_capture_cli_preserves_continuation_policy(tmp_path): assert items["agent"][-1].get("continuation_policy") == "same_agent_non_delivery" -def test_capture_uses_full_text_identity(tmp_path): +@pytest.mark.parametrize("provider", [None, "file", "sqlite"]) +def test_capture_uses_full_text_identity(tmp_path, monkeypatch, provider): registry, state = fixture(tmp_path) + if provider is not None: + from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.todos import list_goal_todos + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + projection = build_todo_runtime_shadow_projection(goal_id=GOAL, handoff_mode="hard_lease", + todos=list_goal_todos(registry_path=registry, goal_id=GOAL)["todos"]) + initialize_canonical_authority(tmp_path / "runtime", GOAL, projection, state_path=state, provider=provider) prefix = "[P1] " + "long text " * 60 one, two = prefix + "first invariant", prefix + "second invariant" args = dict(registry_path=registry, goal_id=GOAL, followups=[one, two], evidence="validation://capture") diff --git a/tests/control_plane/test_todo_machine_section_projection.py b/tests/control_plane/test_todo_machine_section_projection.py index 2fb97c1bc7..f481701e79 100644 --- a/tests/control_plane/test_todo_machine_section_projection.py +++ b/tests/control_plane/test_todo_machine_section_projection.py @@ -706,3 +706,20 @@ def run(revision: str, *extra: str) -> tuple[int, dict]: assert code == 0 and replay["changed"] is False assert state.read_bytes() == published assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") == before + + +@pytest.mark.parametrize("archive", [False, True]) +def test_projection_round_trip_keeps_full_native_text_and_derives_display_priority(archive): + text = "[P1] " + "Full authority text " * 40 + "distinct suffix" + record = {"schema_version": "todo_domain_record_v0", "todo_id": "todo_full_text", "role": "agent", + "status": "open", "done": False, "text": text, + "archive_state": "archive" if archive else "active", "task_class": "advancement_task"} + before = deepcopy(record) + projection = render_canonical_todo_sections(SOURCE, [record], provider_revision="rev-full-text") + assert text in projection.markdown + assert record == before # Display derivation cannot add fields to authority. + replay = render_canonical_todo_sections(projection.markdown, [record], provider_revision="rev-full-text") + assert replay.changed is False + with pytest.raises(TodoSectionProjectionError, match="parity mismatch"): + render_canonical_todo_sections(SOURCE, [{**record, "priority": "P4", "title": "Conflicting metadata"}], + provider_revision="rev-conflicting-priority")