From 786a53f563b2d368431e619dd84c709a87519c9b Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Mon, 21 Sep 2026 02:19:48 +0800 Subject: [PATCH 1/2] fix(authority): confirm durable Todo projections against current reads Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../coordination/handoff_mode_runtime.ts | 3 +- .../coordination/local_authority.py | 12 + .../coordination/local_authority_provider.ts | 21 ++ .../coordination/local_authority_read.ts | 139 +++++++++ .../coordination/local_authority_runtime.ts | 150 +--------- .../control_plane/effect_runtime_handlers.ts | 3 +- .../todos/machine_section_projection.py | 8 + .../todos/projection_delivery.ts | 27 ++ .../todos/provider_projection.py | 141 ++++++---- .../test_local_coordination_authority.py | 7 + .../control_plane/test_team_plan_authority.py | 15 +- .../test_todo_machine_section_projection.py | 39 ++- .../test_todo_projection_concurrency.py | 263 ++++++++++++++++++ .../test_todo_provider_projection.py | 19 +- .../authority_store_conformance.ts | 6 +- .../coordination_projection.test.ts | 6 +- .../goal_acceptance_runtime.test.ts | 2 +- .../local_authority_provider.test.ts | 7 +- .../local_authority_runtime.test.ts | 3 +- .../projection_confirmation_conformance.ts | 55 ++++ .../projection_delivery.test.ts | 15 + 21 files changed, 708 insertions(+), 233 deletions(-) create mode 100644 loopx/control_plane/coordination/local_authority_read.ts create mode 100644 tests/control_plane/test_todo_projection_concurrency.py create mode 100644 tests/control_plane_ts/projection_confirmation_conformance.ts diff --git a/loopx/control_plane/coordination/handoff_mode_runtime.ts b/loopx/control_plane/coordination/handoff_mode_runtime.ts index fad914321a..a909e5c227 100644 --- a/loopx/control_plane/coordination/handoff_mode_runtime.ts +++ b/loopx/control_plane/coordination/handoff_mode_runtime.ts @@ -3,7 +3,8 @@ import type {JsonObject} from "../effect_program.ts"; import {requireJsonObject} from "../runtime_decode.ts"; import {requireAuthorityStoreId} from "./authority_store_codec.ts"; import {openLocalAuthorityStore, localAuthorityOpenFailure} from "./local_authority_provider.ts"; -import {runtimeRoot, sourceAuthorityFor} from "./local_authority_runtime.ts"; +import {requireLocalAuthorityRuntimeRoot as runtimeRoot} from "./local_authority_provider.ts"; +import {authorityStoreSourceAuthority as sourceAuthorityFor} from "./authority_store.ts"; import {withCanonicalWriter} from "./local_authority_write.ts"; import {ShadowManagementError} from "./shadow_management.ts"; import {executeHandoffModeSet, HANDOFF_MODE_SET_SCHEMA} from "./handoff_mode_transaction.ts"; diff --git a/loopx/control_plane/coordination/local_authority.py b/loopx/control_plane/coordination/local_authority.py index 8b4f884bd4..c1b9c7b655 100644 --- a/loopx/control_plane/coordination/local_authority.py +++ b/loopx/control_plane/coordination/local_authority.py @@ -184,6 +184,7 @@ def claim_canonical_todo_if_promoted( def read_canonical_todos_if_promoted( *, runtime_root: Path, goal_id: str, include_leases: bool = False, + projection_readback: Mapping[str, Any] | None = None, ) -> dict[str, Any] | None: """Return canonical Todos after cutover, or ``None`` before cutover. @@ -202,6 +203,7 @@ def read_canonical_todos_if_promoted( "runtime_root": str(runtime_root.expanduser().resolve(strict=False)), "goal_id": goal_id, **({"include_leases": True} if include_leases else {}), + **({"projection_readback": dict(projection_readback)} if projection_readback is not None else {}), }, ) if not isinstance(result, Mapping): @@ -245,6 +247,16 @@ def read_canonical_todos_if_promoted( "canonical Todo/lease snapshot is incomplete", code="local_authority_snapshot_incomplete", payload=payload, ) + if projection_readback is not None: + confirmation = payload.get("projection_readback") + if (not isinstance(confirmation, Mapping) + or confirmation.get("provider_revision") != projection_readback["provider_revision"] + or confirmation.get("observed_provider_revision") != payload.get("provider_revision") + or confirmation.get("status") not in {"pending", "delivered", "current"}): + raise LocalCoordinationAuthorityUnavailable( + "canonical projection confirmation is missing or invalid", + code="local_authority_projection_confirmation_invalid", payload=payload, + ) return payload diff --git a/loopx/control_plane/coordination/local_authority_provider.ts b/loopx/control_plane/coordination/local_authority_provider.ts index e6eda7f733..ec27e6ddfb 100644 --- a/loopx/control_plane/coordination/local_authority_provider.ts +++ b/loopx/control_plane/coordination/local_authority_provider.ts @@ -48,6 +48,8 @@ export type LocalPostgreSqlAuthorityFactory = ( ) => Promise | AuthorityStore; export interface LocalAuthorityProviderDependencies { + /** Existing injected runtime store seam; production uses the configured provider. */ + createStore?: (directory: string, goalId: string) => AuthorityStore; /** Service-owned hook for the medium-term PostgreSQL profile. */ openPostgresqlStore?: LocalPostgreSqlAuthorityFactory; } @@ -265,6 +267,25 @@ export async function selectLocalSqliteAuthority(root: string, goalId: string, e }); } +/** One runtime seam owns provider construction for every local command. */ +export async function openRuntimeAuthorityStore( + root: string, + goalId: string, + dependencies: LocalAuthorityProviderDependencies, +): Promise { + if (dependencies.createStore !== undefined) { + return dependencies.createStore(join(root, "authority", "file-v0"), goalId); + } + return await openLocalAuthorityStore(root, goalId, dependencies); +} + +export function requireLocalAuthorityRuntimeRoot(value: unknown): string { + if (typeof value !== "string" || value.trim() !== value || !isAbsolute(value)) { + throw new Error("runtime_root must be an absolute path"); + } + return value; +} + // Narrow administrative entrypoint; business writes continue through loopx todo. if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { try { diff --git a/loopx/control_plane/coordination/local_authority_read.ts b/loopx/control_plane/coordination/local_authority_read.ts new file mode 100644 index 0000000000..ba87f25ab9 --- /dev/null +++ b/loopx/control_plane/coordination/local_authority_read.ts @@ -0,0 +1,139 @@ +/** Canonical Todo reads and projection confirmation share one provider snapshot. */ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject} from "../runtime_decode.ts"; +import {acceptanceWorkGuard, projectGoalAcceptance} from "../goals/acceptance_contract.ts"; +import {authorityStoreSourceAuthority as sourceAuthorityFor} from "./authority_store.ts"; +import {requireAuthorityStoreId} from "./authority_store_codec.ts"; +import {openRuntimeAuthorityStore as openRuntimeStore, requireLocalAuthorityRuntimeRoot as runtimeRoot, + localAuthorityOpenFailure, type LocalAuthorityProviderDependencies} from "./local_authority_provider.ts"; +import {indexCoordinationProjection, indexCoordinationProjectionTodos, validateCoordinationTodoReadModel} from "./coordination_projection.ts"; +import {LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA, LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, + LOCAL_COORDINATION_TODO_READ_REQUEST_SCHEMA, LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA} from "./coordination_state_contract.generated.ts"; +import {decodeProjectionReadback, confirmProjectionReadback} from "../todos/projection_delivery.ts"; + +/** Provider-first exact Todo read. Missing/unavailable state never falls back. */ +export async function readLocalCoordinationTodo( + value: unknown, + dependencies: LocalAuthorityProviderDependencies = {}, +): Promise { + let sourceAuthority = "file_v0"; + try { + const input = requireJsonObject(value, "local coordination Todo read request"); + if (input.schema_version !== LOCAL_COORDINATION_TODO_READ_REQUEST_SCHEMA) { + throw new Error("local coordination Todo read request schema mismatch"); + } + const root = runtimeRoot(input.runtime_root); + const goalId = requireAuthorityStoreId(input.goal_id, "goal id"); + const todoId = requireAuthorityStoreId(input.todo_id, "todo id"); + const store = await openRuntimeStore(root, goalId, dependencies); + sourceAuthority = sourceAuthorityFor(store); + const head = await store.loadAuthority(); + if (head.status !== "loaded") { + return { + schema_version: LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA, + ...head, + source_authority: sourceAuthority, + decision_read_from_provider: true, + legacy_fallback_used: false, + }; + } + const projection = indexCoordinationProjectionTodos(head.head, goalId); + validateCoordinationTodoReadModel(head.head, goalId); + const todo = projection.todos.get(todoId); + const acceptance = todo === undefined ? null : acceptanceWorkGuard(head.head, goalId, todoId); + return { + schema_version: LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA, + status: todo === undefined ? "missing" : "found", + todo_id: todoId, + ...(todo === undefined ? {} : { todo }), + ...(acceptance === null ? {} : {goal_acceptance_guard: acceptance}), + todo_ids: projection.todo_ids, + provider_revision: head.provider_revision, + cursor: head.cursor, + source_authority: sourceAuthority, + decision_read_from_provider: true, + legacy_fallback_used: false, + }; + } catch (error) { + return { + schema_version: LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA, + status: "failed", + reason_code: "invalid_local_coordination_todo_read_request", + reason: error instanceof Error ? error.message : "invalid Todo read request", + source_authority: sourceAuthority, + decision_read_from_provider: true, + legacy_fallback_used: false, + ...localAuthorityOpenFailure(error), + }; + } +} + +/** Provider-first Todo collection read. Missing/unavailable state never falls back. */ +export async function listLocalCoordinationTodos( + value: unknown, + dependencies: LocalAuthorityProviderDependencies = {}, +): Promise { + let sourceAuthority = "file_v0"; + try { + const input = requireJsonObject(value, "local coordination Todo list request"); + if (input.schema_version !== LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA) { + throw new Error("local coordination Todo list request schema mismatch"); + } + const readback = input.projection_readback === undefined ? null : decodeProjectionReadback(input.projection_readback); + if (input.include_leases !== undefined && typeof input.include_leases !== "boolean") { + throw new Error("include_leases must be a boolean"); + } + const root = runtimeRoot(input.runtime_root); + const goalId = requireAuthorityStoreId(input.goal_id, "goal id"); + const store = await openRuntimeStore(root, goalId, dependencies); + sourceAuthority = sourceAuthorityFor(store); + const head = await store.loadAuthority(); + if (head.status !== "loaded") { + return { + schema_version: LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, + ...head, + source_authority: sourceAuthority, + decision_read_from_provider: true, + legacy_fallback_used: false, + }; + } + const projection = indexCoordinationProjectionTodos(head.head, goalId); + const todoReadModel = validateCoordinationTodoReadModel(head.head, goalId); + const leaseIndex = input.include_leases === true + ? indexCoordinationProjection(head.head, goalId) : null; + const acceptance = projectGoalAcceptance(head.head, goalId); + return { + schema_version: LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, + status: "loaded", + ...(readback === null ? {} : {projection_readback: confirmProjectionReadback(readback, head.provider_revision)}), + todos: projection.todo_ids.map((todoId) => projection.todos.get(todoId)!), + todo_ids: projection.todo_ids, + todo_read_model: todoReadModel, + ...(acceptance.enabled !== true ? {} : {goal_acceptance_contract: acceptance, + goal_acceptance_work_guards: Object.fromEntries(projection.todo_ids.flatMap(id => { + const guard = acceptanceWorkGuard(head.head, goalId, id); + return guard === null ? [] : [[id, guard]]; + }))}), + ...(leaseIndex === null ? {} : { + leases: leaseIndex.lease_todo_ids.map((id) => leaseIndex.leases.get(id)!), + handoff_mode: head.head.handoff_mode ?? "legacy", + }), + provider_revision: head.provider_revision, + cursor: head.cursor, + source_authority: sourceAuthority, + decision_read_from_provider: true, + legacy_fallback_used: false, + }; + } catch (error) { + return { + schema_version: LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, + status: "failed", + reason_code: "invalid_local_coordination_todo_list_request", + reason: error instanceof Error ? error.message : "invalid Todo list request", + source_authority: sourceAuthority, + decision_read_from_provider: true, + legacy_fallback_used: false, + ...localAuthorityOpenFailure(error), + }; + } +} diff --git a/loopx/control_plane/coordination/local_authority_runtime.ts b/loopx/control_plane/coordination/local_authority_runtime.ts index 3e28835696..c3fd0f12d0 100644 --- a/loopx/control_plane/coordination/local_authority_runtime.ts +++ b/loopx/control_plane/coordination/local_authority_runtime.ts @@ -10,7 +10,6 @@ import {readFile, realpath} from "node:fs/promises"; import {createHash} from "node:crypto"; import type { JsonObject } from "../effect_program.ts"; -import {acceptanceWorkGuard, projectGoalAcceptance} from "../goals/acceptance_contract.ts"; import {decodeMonitorPollObservation} from "../todos/monitor_metadata.ts"; import {executeCoordinationMonitorPoll, COORDINATION_MONITOR_POLL_REQUEST_SCHEMA, COORDINATION_LEASED_MONITOR_POLL_REQUEST_SCHEMA, COORDINATION_WITNESSED_MONITOR_POLL_REQUEST_SCHEMA, COORDINATION_MONITOR_POLL_RESULT_SCHEMA} from "./todo_monitor_poll.ts"; @@ -28,8 +27,6 @@ import { } from "./coordination_state_contract.generated.ts"; import { indexCoordinationProjection, - indexCoordinationProjectionTodos, - validateCoordinationTodoReadModel, } from "./coordination_projection.ts"; import { authorityStoreSourceAuthority, type AuthorityStore, type AuthorityStoreReceiptResult } from "./authority_store.ts"; import { @@ -42,6 +39,8 @@ import { import { FileAuthorityStore } from "./file_authority_store.ts"; import { openLocalAuthorityStore, + openRuntimeAuthorityStore as openRuntimeStore, + requireLocalAuthorityRuntimeRoot as runtimeRoot, localAuthorityOpenFailure, type LocalAuthorityProviderDependencies, } from "./local_authority_provider.ts"; @@ -445,30 +444,10 @@ export async function pollLocalCoordinationMonitor(value: unknown, } interface LocalAuthorityRuntimeDependencies extends LocalAuthorityProviderDependencies { - createStore?: (directory: string, goalId: string) => AuthorityStore; createShadowStore?: (directory: string, goalId: string) => AuthorityStore; createCanonicalStore?: (directory: string, goalId: string) => AuthorityStore; } -/** One runtime seam owns provider construction for every local command. */ -async function openRuntimeStore( - root: string, - goalId: string, - dependencies: LocalAuthorityRuntimeDependencies, -): Promise { - if (dependencies.createStore !== undefined) { - return dependencies.createStore(authorityDirectory(root), goalId); - } - return await openLocalAuthorityStore(root, goalId, dependencies); -} - -export function runtimeRoot(value: unknown): string { - if (typeof value !== "string" || value.trim() !== value || !isAbsolute(value)) { - throw new Error("runtime_root must be an absolute path"); - } - return value; -} - function claimAgentValue(value: unknown, label: string): string { if (typeof value !== "string" || value.trim().length === 0) { throw new Error(`${label} must be a non-empty string`); @@ -1368,131 +1347,6 @@ export async function acknowledgeLocalCoordinationTodoArchive( } } -/** Provider-first exact Todo read. Missing/unavailable state never falls back. */ -export async function readLocalCoordinationTodo( - value: unknown, - dependencies: LocalAuthorityRuntimeDependencies = {}, -): Promise { - let sourceAuthority = "file_v0"; - try { - const input = requireJsonObject(value, "local coordination Todo read request"); - if (input.schema_version !== LOCAL_COORDINATION_TODO_READ_REQUEST_SCHEMA) { - throw new Error("local coordination Todo read request schema mismatch"); - } - const root = runtimeRoot(input.runtime_root); - const goalId = requireAuthorityStoreId(input.goal_id, "goal id"); - const todoId = requireAuthorityStoreId(input.todo_id, "todo id"); - const store = await openRuntimeStore(root, goalId, dependencies); - sourceAuthority = sourceAuthorityFor(store); - const head = await store.loadAuthority(); - if (head.status !== "loaded") { - return { - schema_version: LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA, - ...head, - source_authority: sourceAuthority, - decision_read_from_provider: true, - legacy_fallback_used: false, - }; - } - const projection = indexCoordinationProjectionTodos(head.head, goalId); - validateCoordinationTodoReadModel(head.head, goalId); - const todo = projection.todos.get(todoId); - const acceptance = todo === undefined ? null : acceptanceWorkGuard(head.head, goalId, todoId); - return { - schema_version: LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA, - status: todo === undefined ? "missing" : "found", - todo_id: todoId, - ...(todo === undefined ? {} : { todo }), - ...(acceptance === null ? {} : {goal_acceptance_guard: acceptance}), - todo_ids: projection.todo_ids, - provider_revision: head.provider_revision, - cursor: head.cursor, - source_authority: sourceAuthority, - decision_read_from_provider: true, - legacy_fallback_used: false, - }; - } catch (error) { - return { - schema_version: LOCAL_COORDINATION_TODO_READ_RESULT_SCHEMA, - status: "failed", - reason_code: "invalid_local_coordination_todo_read_request", - reason: error instanceof Error ? error.message : "invalid Todo read request", - source_authority: sourceAuthority, - decision_read_from_provider: true, - legacy_fallback_used: false, - ...localAuthorityOpenFailure(error), - }; - } -} - -/** Provider-first Todo collection read. Missing/unavailable state never falls back. */ -export async function listLocalCoordinationTodos( - value: unknown, - dependencies: LocalAuthorityRuntimeDependencies = {}, -): Promise { - let sourceAuthority = "file_v0"; - try { - const input = requireJsonObject(value, "local coordination Todo list request"); - if (input.schema_version !== LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA) { - throw new Error("local coordination Todo list request schema mismatch"); - } - if (input.include_leases !== undefined && typeof input.include_leases !== "boolean") { - throw new Error("include_leases must be a boolean"); - } - const root = runtimeRoot(input.runtime_root); - const goalId = requireAuthorityStoreId(input.goal_id, "goal id"); - const store = await openRuntimeStore(root, goalId, dependencies); - sourceAuthority = sourceAuthorityFor(store); - const head = await store.loadAuthority(); - if (head.status !== "loaded") { - return { - schema_version: LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, - ...head, - source_authority: sourceAuthority, - decision_read_from_provider: true, - legacy_fallback_used: false, - }; - } - const projection = indexCoordinationProjectionTodos(head.head, goalId); - const todoReadModel = validateCoordinationTodoReadModel(head.head, goalId); - const leaseIndex = input.include_leases === true - ? indexCoordinationProjection(head.head, goalId) : null; - const acceptance = projectGoalAcceptance(head.head, goalId); - return { - schema_version: LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, - status: "loaded", - todos: projection.todo_ids.map((todoId) => projection.todos.get(todoId)!), - todo_ids: projection.todo_ids, - todo_read_model: todoReadModel, - ...(acceptance.enabled !== true ? {} : {goal_acceptance_contract: acceptance, - goal_acceptance_work_guards: Object.fromEntries(projection.todo_ids.flatMap(id => { - const guard = acceptanceWorkGuard(head.head, goalId, id); - return guard === null ? [] : [[id, guard]]; - }))}), - ...(leaseIndex === null ? {} : { - leases: leaseIndex.lease_todo_ids.map((id) => leaseIndex.leases.get(id)!), - handoff_mode: head.head.handoff_mode ?? "legacy", - }), - provider_revision: head.provider_revision, - cursor: head.cursor, - source_authority: sourceAuthority, - decision_read_from_provider: true, - legacy_fallback_used: false, - }; - } catch (error) { - return { - schema_version: LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA, - status: "failed", - reason_code: "invalid_local_coordination_todo_list_request", - reason: error instanceof Error ? error.message : "invalid Todo list request", - source_authority: sourceAuthority, - decision_read_from_provider: true, - legacy_fallback_used: false, - ...localAuthorityOpenFailure(error), - }; - } -} - /** The explicit local CLI continuation uses the existing promoted writer fence. */ export async function continueLocalTodo( value: unknown, diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 531ad1a3a4..22eb747e0f 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -155,12 +155,11 @@ import { createLocalCoordinationTodo, updateLocalCoordinationTodo, pollLocalCoordinationMonitor, - listLocalCoordinationTodos, promoteLocalCoordinationAuthority, reviewLocalCoordinationAuthorityPromotion, - readLocalCoordinationTodo, terminalLifecycleLocalCoordinationTodo, } from "./coordination/local_authority_runtime.ts"; +import {listLocalCoordinationTodos, readLocalCoordinationTodo} from "./coordination/local_authority_read.ts"; import { evaluateCoordinationTodoClaimDecision } from "./coordination/todo_claim.ts"; import { evaluateCoordinationTodoTerminalDecision, diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py index 74661638c3..5f9dc4cda7 100644 --- a/loopx/control_plane/todos/machine_section_projection.py +++ b/loopx/control_plane/todos/machine_section_projection.py @@ -45,6 +45,7 @@ TODO_STATUS_OPEN, format_todo_metadata_line, normalize_todo_id, + normalize_todo_generation, normalize_todo_status, require_todo_decision_scope, todo_marker_for_status, @@ -287,6 +288,13 @@ def _parsed_archive_records(markdown: str) -> list[dict[str, Any]]: raise TodoSectionProjectionError( f"archived Todo {item.get('todo_id')!r} omits its source role" ) + # Metadata is textual; canonical capture and active reads normalize this + # counter. Archive readback must use the same codec, not compare "12" to 12. + if "material_change_generation" in item: + generation = normalize_todo_generation(item["material_change_generation"]) + if generation is None: + raise TodoSectionProjectionError("invalid archived Monitor material generation") + item["material_change_generation"] = generation priority, title = todo_priority_parts(str(item.get("text") or "")) if priority: item.update(priority=priority, title=normalize_todo_text(title)) diff --git a/loopx/control_plane/todos/projection_delivery.ts b/loopx/control_plane/todos/projection_delivery.ts index eea6577409..d882241434 100644 --- a/loopx/control_plane/todos/projection_delivery.ts +++ b/loopx/control_plane/todos/projection_delivery.ts @@ -20,3 +20,30 @@ export function parseProjectionDelivery(value: unknown): TodoProjectionDelivery export function isProjectionDelivery(value: unknown): value is TodoProjectionDelivery { return typeof value === "string" && PROJECTION_DELIVERY_VALUES.has(value as TodoProjectionDelivery); } + +/** Host attests durable file readback; the provider owns revision comparison. + * Confirmation describes one observed head, never a lock on future commits. */ +export interface ProjectionReadback { + provider_revision: string; + changed: boolean; +} + +export function decodeProjectionReadback(value: unknown): ProjectionReadback { + if (value === null || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError("projection_readback must be an object"); + } + const row = value as Record; + if (Object.keys(row).length !== 2 || typeof row.provider_revision !== "string" || + !row.provider_revision.trim() || row.provider_revision !== row.provider_revision.trim() || + typeof row.changed !== "boolean") throw new TypeError("invalid projection_readback"); + return {provider_revision: row.provider_revision, changed: row.changed}; +} + +export function confirmProjectionReadback(readback: ProjectionReadback, observedRevision: string) { + return { + provider_revision: readback.provider_revision, + observed_provider_revision: observedRevision, + status: readback.provider_revision === observedRevision + ? (readback.changed ? "delivered" : "current") : "pending", + } satisfies {provider_revision: string; observed_provider_revision: string; status: TodoProjectionDelivery}; +} diff --git a/loopx/control_plane/todos/provider_projection.py b/loopx/control_plane/todos/provider_projection.py index 40de2bb161..45140be462 100644 --- a/loopx/control_plane/todos/provider_projection.py +++ b/loopx/control_plane/todos/provider_projection.py @@ -103,70 +103,95 @@ def project_current_canonical_todos( raise ValueError( "Todo Markdown projection requires promoted canonical authority" ) - provider_revision = authority_read.get("provider_revision") - if not isinstance(provider_revision, str) or not provider_revision: - raise ValueError("canonical Todo authority omitted provider revision") - if ( - expected_provider_revision is not None - and provider_revision != expected_provider_revision - ): - raise ValueError( - "Todo Markdown projection provider revision does not match the " - "canonical read head" - ) recovered_missing = False - try: - source = _read_text_exact(state_path) - except FileNotFoundError: - recovered_missing = True - source = ( - f"---\ngoal_id: {json.dumps(goal_id, ensure_ascii=False)}\n---\n\n" - "# Recovered Todo projection\n\n" - "> Regenerated from canonical Todo authority. Non-Todo sections " - "are not in this provider snapshot and were not recovered. " - "This is a Todo projection, not a complete Goal-state restore.\n\n" - "## Agent Todo\n" - ) - projection = render_canonical_todo_sections( - source, - authority_read["todos"], - provider_revision=provider_revision, - private_validation_declarations=load_completion_validation_declarations( - runtime_root=runtime_root, - goal_id=goal_id, - todos=authority_read["todos"], - ), - ) - if execute and projection.changed: - # The projection is a primary-state write: it must respect the - # same source-ownership and shadow-management fences as every - # other Markdown writer, checked under the state lock it holds. - from ..coordination.legacy_writer_fence import ( - require_registry_source_write_allowed, + changed = False + confirmation: dict[str, Any] | None = None + for attempt in range(1, 4): + provider_revision = authority_read.get("provider_revision") + if not isinstance(provider_revision, str) or not provider_revision: + raise ValueError("canonical Todo authority omitted provider revision") + if ( + expected_provider_revision is not None + and provider_revision != expected_provider_revision + ): + raise ValueError( + "Todo Markdown projection provider revision does not match the " + "canonical read head" + ) + missing_this_attempt = False + try: + source = _read_text_exact(state_path) + except FileNotFoundError: + recovered_missing = True + missing_this_attempt = True + source = ( + f"---\ngoal_id: {json.dumps(goal_id, ensure_ascii=False)}\n---\n\n" + "# Recovered Todo projection\n\n" + "> Regenerated from canonical Todo authority. Non-Todo sections " + "are not in this provider snapshot and were not recovered. " + "This is a Todo projection, not a complete Goal-state restore.\n\n" + "## Agent Todo\n" + ) + projection = render_canonical_todo_sections( + source, + authority_read["todos"], + provider_revision=provider_revision, + private_validation_declarations=load_completion_validation_declarations( + runtime_root=runtime_root, + goal_id=goal_id, + todos=authority_read["todos"], + ), ) + if execute and projection.changed: + # The projection is a primary-state write: it must respect the + # same source-ownership and shadow-management fences as every + # other Markdown writer, checked under the state lock it holds. + from ..coordination.legacy_writer_fence import ( + require_registry_source_write_allowed, + ) - require_registry_source_write_allowed( - registry_path=registry_path, - runtime_root=runtime_root, - goal_id=goal_id, - state_file=state_path, + require_registry_source_write_allowed( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + state_file=state_path, + ) + if missing_this_attempt: + atomic_write_state_text(state_path, projection.markdown, create_only=True) + else: + atomic_write_state_text(state_path, projection.markdown) + if _read_text_exact(state_path) != projection.markdown: + raise RuntimeError("Todo Markdown projection readback mismatch") + elif execute: + verify_state_text_durable(state_path, projection.markdown) + changed = changed or projection.changed + if not execute: + break + confirmed = read_canonical_todos_if_promoted( + runtime_root=runtime_root, goal_id=goal_id, + projection_readback={"provider_revision": provider_revision, "changed": changed}, ) - if recovered_missing: - atomic_write_state_text(state_path, projection.markdown, create_only=True) - else: - atomic_write_state_text(state_path, projection.markdown) - if _read_text_exact(state_path) != projection.markdown: - raise RuntimeError("Todo Markdown projection readback mismatch") - elif execute: - verify_state_text_durable(state_path, projection.markdown) + if not isinstance(confirmed, dict) or not isinstance(confirmed.get("projection_readback"), dict): + raise ValueError("canonical projection confirmation is missing") + confirmation = confirmed["projection_readback"] + if parse_projection_delivery(confirmation["status"]) != ProjectionDeliveryStatus.PENDING: + break + # A pinned command must not silently render a different revision. + # Unpinned recovery reuses this complete read for its next attempt. + if expected_provider_revision is not None or attempt == 3: + break + authority_read = confirmed return { "schema_version": TODO_PROJECTION_DELIVERY_SCHEMA, - "status": ( - "delivered" if execute and projection.changed else - "current" if execute else - "planned" - ), + "status": confirmation["status"] if confirmation is not None else "planned", + **({"observed_provider_revision": confirmation["observed_provider_revision"]} + if confirmation is not None else {}), + "delivery_attempts": attempt, + **({"reason_code": "todo_projection_revision_advanced", "retryable": True, + "retry_business_mutation": False, + "recommended_action": "Read the current provider revision with todo list, then retry todo project-markdown for that revision."} + if confirmation is not None and confirmation["status"] == "pending" else {}), "source": "committed_authority_journal", "goal_id": goal_id, "state_file": str(state_path), @@ -174,7 +199,7 @@ def project_current_canonical_todos( "source_authority": authority_read.get("source_authority"), "provider_revision": projection.provider_revision, "todo_count": projection.todo_count, - "changed": projection.changed, + "changed": changed, "executed": execute, "source_sha256": projection.source_sha256, "rendered_sha256": projection.rendered_sha256, diff --git a/tests/control_plane/test_local_coordination_authority.py b/tests/control_plane/test_local_coordination_authority.py index 1b12abd3d7..8468bbe21f 100644 --- a/tests/control_plane/test_local_coordination_authority.py +++ b/tests/control_plane/test_local_coordination_authority.py @@ -1318,6 +1318,7 @@ def count_authority_runtime_call( "coordination.local_authority.todo_terminal", "coordination.local_authority.todo_terminal", "coordination.local_authority.todo_list", + "coordination.local_authority.todo_list", ] successor_id = completed["generated_successor_todo_ids"][0] @@ -1341,6 +1342,7 @@ def count_authority_runtime_call( "coordination.local_authority.todo_list", "coordination.local_authority.todo_terminal", "coordination.local_authority.todo_list", + "coordination.local_authority.todo_list", ] canonical = read_canonical_todos_if_promoted( @@ -1371,6 +1373,7 @@ def count_authority_runtime_call( "coordination.local_authority.todo_list", "coordination.local_authority.todo_archive", "coordination.local_authority.todo_list", + "coordination.local_authority.todo_list", "coordination.local_authority.todo_archive_ack", ] canonical_after_archive = read_canonical_todos_if_promoted( @@ -1403,6 +1406,7 @@ def count_authority_runtime_call( "coordination.local_authority.todo_list", "coordination.local_authority.todo_archive", "coordination.local_authority.todo_list", + "coordination.local_authority.todo_list", ] unchanged = read_canonical_todos_if_promoted( runtime_root=runtime_root, @@ -1854,6 +1858,7 @@ def _crash_projection(*_args: object, **_kwargs: object) -> dict[str, object]: "coordination.local_authority.todo_list", "coordination.local_authority.todo_terminal", "coordination.local_authority.todo_list", + "coordination.local_authority.todo_list", ] canonical = read_canonical_todos_if_promoted( runtime_root=runtime_root, goal_id="goal-a" @@ -2041,6 +2046,8 @@ def test_real_canonical_provider_preserves_complete_complex_todo_semantics( item for item in corrected["todos"] if item["todo_id"] == "todo_claimable" ) assert corrected_item["text"] == "[P0] Corrected before claiming" + assert corrected_item["priority"] == "P0" + assert corrected_item["title"] == "Corrected before claiming" assert not corrected_item.get("claimed_by") assert corrected_item["last_actor_agent_id"] == "agent-b" assert not state_file.exists() diff --git a/tests/control_plane/test_team_plan_authority.py b/tests/control_plane/test_team_plan_authority.py index f5cd4a669f..ad27739202 100644 --- a/tests/control_plane/test_team_plan_authority.py +++ b/tests/control_plane/test_team_plan_authority.py @@ -39,17 +39,26 @@ def canonical_team(tmp_path): return runtime, state, service, preview -def test_canonical_chat_commit_replays_after_projection_failure(canonical_team, monkeypatch): +@pytest.mark.parametrize("failure_point", ["write", "confirmation"]) +def test_canonical_chat_commit_replays_after_projection_failure(canonical_team, monkeypatch, failure_point): runtime, state, service, preview = canonical_team def fail(*args, **kwargs): raise OSError("display unavailable") - monkeypatch.setattr(provider_projection, "atomic_write_state_text", fail) + if failure_point == "write": + monkeypatch.setattr(provider_projection, "atomic_write_state_text", fail) + else: + read_authority = provider_projection.read_canonical_todos_if_promoted + def read_after_write(**kwargs): + if kwargs.get("projection_readback") is not None: + raise OSError("confirmation unavailable") + return read_authority(**kwargs) + monkeypatch.setattr(provider_projection, "read_canonical_todos_if_promoted", read_after_write) failed = service.apply(preview["proposal_id"])["proposal"] assert failed["status"] == "failed" read = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") assert len(read["todos"]) == 2 assert len({row["todo_id"] for row in read["todos"]}) == 2 - assert "Same work" not in state.read_text() + assert ("Same work" not in state.read_text()) is (failure_point == "write") monkeypatch.undo() applied = service.apply(preview["proposal_id"])["proposal"] assert applied["status"] == "applied", applied diff --git a/tests/control_plane/test_todo_machine_section_projection.py b/tests/control_plane/test_todo_machine_section_projection.py index 1af7d204df..576fc1c7f8 100644 --- a/tests/control_plane/test_todo_machine_section_projection.py +++ b/tests/control_plane/test_todo_machine_section_projection.py @@ -92,6 +92,17 @@ def _records() -> list[dict[str, object]]: ] +def _confirmed_payload(payload, request): + if payload is None or "projection_readback" not in request: + return payload + witness = request["projection_readback"] + return {**payload, "projection_readback": { + "provider_revision": witness["provider_revision"], + "observed_provider_revision": payload["provider_revision"], + "status": "delivered" if witness["changed"] else "current", + }} + + @pytest.mark.parametrize("newline", ["\n", "\r\n"]) def test_projection_replaces_only_machine_sections_and_is_idempotent(newline: str) -> None: projected = render_canonical_todo_sections( @@ -394,7 +405,7 @@ def run( monkeypatch.setattr( provider_projection, "read_canonical_todos_if_promoted", - lambda **_kwargs: payload, + lambda **kwargs: _confirmed_payload(payload, kwargs), ) monkeypatch.setattr( provider_projection, @@ -527,11 +538,11 @@ def test_project_markdown_cli_publishes_with_atomic_replace( monkeypatch.setattr( provider_projection, "read_canonical_todos_if_promoted", - lambda **_kwargs: { + lambda **kwargs: _confirmed_payload({ "todos": _records(), "source_authority": "file_v0", "provider_revision": "rev-1", - }, + }, kwargs), ) monkeypatch.setattr( provider_projection, @@ -623,11 +634,11 @@ def test_project_markdown_cli_preserves_narrative_boundaries( monkeypatch.setattr( provider_projection, "read_canonical_todos_if_promoted", - lambda **_kwargs: { + lambda **kwargs: _confirmed_payload({ "todos": _records(), "source_authority": "file_v0", "provider_revision": "rev-1", - }, + }, kwargs), ) monkeypatch.setattr( provider_projection, @@ -724,3 +735,21 @@ 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('generation', [0, 1, 12]) +@pytest.mark.parametrize('native', [False, True]) +def test_archived_monitor_retains_numeric_material_generation(generation, native): + record = {**_records()[0], 'status': 'done', 'done': True, + 'archive_state': 'archive', 'source_section': 'Completed Work Archive', + 'task_class': 'continuous_monitor', 'material_change_generation': generation} + if native: + record['schema_version'] = 'todo_domain_record_v0' + record.pop('index') + record.pop('source_section') + before = deepcopy(record) + rendered = render_canonical_todo_sections(SOURCE, [record], provider_revision='revision:monitor') + from loopx.control_plane.todos.machine_section_projection import _parsed_archive_records + assert _parsed_archive_records(rendered.markdown)[0]['material_change_generation'] == generation + assert record == before + assert render_canonical_todo_sections(rendered.markdown, [record], provider_revision='revision:monitor').changed is False diff --git a/tests/control_plane/test_todo_projection_concurrency.py b/tests/control_plane/test_todo_projection_concurrency.py new file mode 100644 index 0000000000..90fd944bb2 --- /dev/null +++ b/tests/control_plane/test_todo_projection_concurrency.py @@ -0,0 +1,263 @@ +"""Real canonical projection delivery under overlapping commits and retry.""" + +import json +import subprocess +from pathlib import Path + +import pytest +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.todos import provider_projection + +REPO = Path(__file__).resolve().parents[2] + + +@pytest.fixture(params=["file", "sqlite"]) +def canonical_projection(tmp_path, monkeypatch, request): + isolate_sqlite_runtime(tmp_path, monkeypatch) + runtime, state, registry = ( + tmp_path / "runtime", + tmp_path / "state.md", + tmp_path / "registry.json", + ) + state.write_text("# Recovery\n\nHuman narrative.\n\n## Agent Todo\n") + registry.write_text( + json.dumps( + { + "common_runtime_root": str(runtime), + "goals": [ + { + "id": "projection-goal", + "repo": str(tmp_path), + "state_file": state.name, + } + ], + } + ) + ) + projection = build_todo_runtime_shadow_projection( + goal_id="projection-goal", + todos=[ + { + "schema_version": "todo_item_v0", + "todo_id": "todo_work", + "role": "agent", + "status": "open", + "done": False, + "text": "Canonical work", + "archive_state": "active", + "source_section": "Agent Todo", + "index": 1, + "task_class": "advancement_task", + } + ], + leases=[], + handoff_mode="soft_claim", + ) + first = initialize_canonical_authority( + runtime, "projection-goal", projection, state_path=state, provider=request.param + ) + args = { + "registry_path": registry, + "runtime_root": runtime, + "goal_id": "projection-goal", + } + + def advance(): + # A real concurrent authority commit; the display lock is not its CAS. + script = """import {openLocalAuthorityStore} from './loopx/control_plane/coordination/local_authority_provider.ts'; +const store=await openLocalAuthorityStore(process.argv[1],'projection-goal'); +const h=await store.loadAuthority(); +const r=await store.commitAuthority({expected_provider_revision:h.provider_revision, + operation_id:'overlap-'+h.cursor,next_projection:h.head,events:[],receipts:[]}); +if(r.status!=='applied')throw new Error(JSON.stringify(r));console.log(JSON.stringify(r));""" + child = subprocess.run( + [ + "node", + "--no-warnings", + "--experimental-strip-types", + "--input-type=module", + "-e", + script, + str(runtime), + ], + cwd=REPO, + capture_output=True, + text=True, + timeout=45, + check=True, + ) + return json.loads(child.stdout) + + return args, state, first, advance + + +def test_delivery_catches_up_after_real_overlapping_commit( + canonical_projection, monkeypatch +): + args, state, first, advance = canonical_projection + write = provider_projection.atomic_write_state_text + commits = [] + + def overlap(*a, **kw): + write(*a, **kw) + if not commits: + commits.append(advance()) + + monkeypatch.setattr(provider_projection, "atomic_write_state_text", overlap) + result = provider_projection.project_current_canonical_todos(**args) + assert result["provider_revision"] == commits[0]["provider_revision"] + assert result["status"] == "delivered" + assert result["delivery_attempts"] == 2 + assert "Human narrative." in state.read_text() + assert commits[0]["provider_revision"] in state.read_text() + assert first["provider_revision"] not in state.read_text() + + +def test_pinned_projection_reports_overlap_without_retargeting( + canonical_projection, monkeypatch +): + args, state, first, advance = canonical_projection + write = provider_projection.atomic_write_state_text + commits = [] + + def overlap(*a, **kw): + write(*a, **kw) + commits.append(advance()) + + monkeypatch.setattr(provider_projection, "atomic_write_state_text", overlap) + result = provider_projection.project_current_canonical_todos( + **args, expected_provider_revision=first["provider_revision"] + ) + assert result["status"] == "pending" + assert result["provider_revision"] == first["provider_revision"] + assert result["observed_provider_revision"] == commits[0]["provider_revision"] + assert result["retry_business_mutation"] is False + assert result["delivery_attempts"] == 1 + assert first["provider_revision"] in state.read_text() + + +def test_continuous_commits_stay_pending_and_recover_without_business_replay( + canonical_projection, monkeypatch +): + args, state, _, advance = canonical_projection + write = provider_projection.atomic_write_state_text + commits = [] + + def overlap(*a, **kw): + write(*a, **kw) + commits.append(advance()) + + with monkeypatch.context() as patch: + patch.setattr(provider_projection, "atomic_write_state_text", overlap) + committed = provider_projection.settle_canonical_todo_projection( + { + "status": "applied", + "changed": True, + "operation_id": "original-business-operation", + }, + **args, + ) + assert committed["status"] == "applied" + assert committed["projection_delivery"] == "pending" + assert committed["projection_outbox"]["delivery_attempts"] == 3 + assert committed["projection_outbox"]["retry_business_mutation"] is False + assert len(commits) == 3 + replay = provider_projection.settle_canonical_todo_projection( + { + "status": "replayed", + "changed": False, + "operation_id": "original-business-operation", + }, + **args, + ) + assert replay["projection_delivery"] == "delivered" + assert ( + replay["projection_outbox"]["provider_revision"] + == commits[-1]["provider_revision"] + ) + assert len(commits) == 3 + assert "Human narrative." in state.read_text() + + +def test_missing_display_can_catch_up_without_create_only_collision( + canonical_projection, monkeypatch +): + args, state, _, advance = canonical_projection + state.unlink() + write = provider_projection.atomic_write_state_text + commits = [] + + def overlap(*a, **kw): + write(*a, **kw) + if not commits: + commits.append(advance()) + + monkeypatch.setattr(provider_projection, "atomic_write_state_text", overlap) + result = provider_projection.project_current_canonical_todos(**args) + assert result["status"] == "delivered" and result["delivery_attempts"] == 2 + assert result["recovery_scope"] == "todo_sections_only" + assert result["narrative_preserved"] is False + assert commits[-1]["provider_revision"] in state.read_text() + + +def test_confirmation_outage_preserves_committed_business_and_retries( + canonical_projection, monkeypatch +): + args, state, _, _ = canonical_projection + read = provider_projection.read_canonical_todos_if_promoted + + def fail_confirmation(**kwargs): + if kwargs.get("projection_readback") is not None: + raise provider_projection.LocalCoordinationAuthorityUnavailable( + "synthetic read outage", code="unavailable", payload={} + ) + return read(**kwargs) + + with monkeypatch.context() as patch: + patch.setattr( + provider_projection, "read_canonical_todos_if_promoted", fail_confirmation + ) + result = provider_projection.settle_canonical_todo_projection( + {"status": "applied", "changed": True}, **args + ) + assert result["status"] == "applied" and result["projection_delivery"] == "pending" + assert result["projection_outbox"]["retry_business_mutation"] is False + assert "Canonical work" in state.read_text() + replay = provider_projection.settle_canonical_todo_projection( + {"status": "replayed", "changed": False}, **args + ) + assert replay["projection_delivery"] == "current" + + +def test_downlevel_runtime_cannot_acknowledge_delivery( + canonical_projection, monkeypatch +): + from loopx.control_plane.coordination import local_authority + + args, state, first, _ = canonical_projection + invoke = local_authority.effect_runtime_result + + def without_confirmation(method, payload): + result = invoke(method, payload) + if payload.get("projection_readback") is not None: + result.pop("projection_readback", None) + return result + + with monkeypatch.context() as patch: + patch.setattr(local_authority, "effect_runtime_result", without_confirmation) + result = provider_projection.settle_canonical_todo_projection( + {"status": "applied", "changed": True}, **args + ) + assert result["status"] == "applied" + assert result["projection_delivery"] == "pending" + assert result["projection_outbox"]["retry_business_mutation"] is False + assert first["provider_revision"] in state.read_text() + replay = provider_projection.project_current_canonical_todos(**args) + assert replay["status"] == "current" + assert replay["provider_revision"] == first["provider_revision"] diff --git a/tests/control_plane/test_todo_provider_projection.py b/tests/control_plane/test_todo_provider_projection.py index 34b8a3512f..1d8bf2c0bd 100644 --- a/tests/control_plane/test_todo_provider_projection.py +++ b/tests/control_plane/test_todo_provider_projection.py @@ -42,8 +42,8 @@ def _registry(tmp_path: Path) -> tuple[Path, Path, Path]: return registry, runtime_root, state_file -def _authority_read() -> dict[str, object]: - return { +def _authority_read(**kwargs) -> dict[str, object]: + result = { "status": "loaded", "source_authority": "file_v0", "provider_revision": "file:7:abc", @@ -71,6 +71,15 @@ def _authority_read() -> dict[str, object]: ], } + if "projection_readback" in kwargs: + witness = kwargs["projection_readback"] + result["projection_readback"] = { + "provider_revision": witness["provider_revision"], + "observed_provider_revision": result["provider_revision"], + "status": "delivered" if witness["changed"] else "current", + } + return result + def test_project_current_canonical_todos_is_durable_and_idempotent( monkeypatch: pytest.MonkeyPatch, @@ -80,7 +89,7 @@ def test_project_current_canonical_todos_is_durable_and_idempotent( monkeypatch.setattr( provider_projection, "read_canonical_todos_if_promoted", - lambda **_kwargs: _authority_read(), + _authority_read, ) delivered = provider_projection.project_current_canonical_todos( @@ -116,7 +125,7 @@ def test_settlement_preserves_commit_and_replays_after_projection_failure( monkeypatch.setattr( provider_projection, "read_canonical_todos_if_promoted", - lambda **_kwargs: _authority_read(), + _authority_read, ) real_write = provider_projection.atomic_write_state_text monkeypatch.setattr( @@ -169,7 +178,7 @@ def test_explicit_projection_fences_requested_revision( monkeypatch.setattr( provider_projection, "read_canonical_todos_if_promoted", - lambda **_kwargs: _authority_read(), + _authority_read, ) with pytest.raises(ValueError, match="does not match"): diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts index 6e15c62d3b..8ed66c2b95 100644 --- a/tests/control_plane_ts/authority_store_conformance.ts +++ b/tests/control_plane_ts/authority_store_conformance.ts @@ -1,4 +1,5 @@ import {registerTodoConsumerScopeConformance} from "./todo_consumer_scope_conformance.ts"; +import {registerProjectionConfirmationConformance} from "./projection_confirmation_conformance.ts"; import {registerUserCompletionFollowthroughConformance} from "./user_completion_followthrough_conformance.ts"; import {registerSuccessionReadConformance} from "./succession_read_conformance.ts"; import {registerUserCompletionUpdateConformance} from "./user_completion_update_conformance.ts"; @@ -43,8 +44,8 @@ import { executeCoordinationTodoClaim } from "../../loopx/control_plane/coordina import { executeCoordinationTodoCreate } from "../../loopx/control_plane/coordination/todo_create.ts"; import {executeCoordinationMonitorPoll} from "../../loopx/control_plane/coordination/todo_monitor_poll.ts"; import { executeCoordinationTodoUpdate } from "../../loopx/control_plane/coordination/todo_update.ts"; -import { listLocalCoordinationTodos, LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA } - from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {listLocalCoordinationTodos} from "../../loopx/control_plane/coordination/local_authority_read.ts"; import { sharedGoalWorkFacts } from "../../loopx/control_plane/goals/shared_goal_work.ts"; import {projectStandingDecisions} from "../../loopx/control_plane/todos/standing_decision.ts"; import {evaluateTodoResumeConditions} from "../../loopx/control_plane/todos/resume_condition.ts"; @@ -242,6 +243,7 @@ export function registerAuthorityStoreConformance( providerName: string, factory: AuthorityStoreConformanceFactory, ): void { + registerProjectionConfirmationConformance(providerName, factory); registerLeaseLifecycleConformance(providerName, factory); registerClaimTransferConformance(providerName, factory); registerLeaseAcquisitionConformance(providerName, factory); diff --git a/tests/control_plane_ts/coordination_projection.test.ts b/tests/control_plane_ts/coordination_projection.test.ts index b472df1fd2..64f06013d8 100644 --- a/tests/control_plane_ts/coordination_projection.test.ts +++ b/tests/control_plane_ts/coordination_projection.test.ts @@ -15,10 +15,8 @@ import { TODO_CANONICAL_READ_RECORD_SCHEMA, } from "../../loopx/control_plane/coordination/coordination_projection.ts"; import { FileAuthorityStore } from "../../loopx/control_plane/coordination/file_authority_store.ts"; -import { - listLocalCoordinationTodos, - LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA, -} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {listLocalCoordinationTodos} from "../../loopx/control_plane/coordination/local_authority_read.ts"; import { TODO_DOMAIN_ITEM_SCHEMA, TODO_DOMAIN_READ_RECORD_SCHEMA, diff --git a/tests/control_plane_ts/goal_acceptance_runtime.test.ts b/tests/control_plane_ts/goal_acceptance_runtime.test.ts index c02ded30ce..c85deebd37 100644 --- a/tests/control_plane_ts/goal_acceptance_runtime.test.ts +++ b/tests/control_plane_ts/goal_acceptance_runtime.test.ts @@ -16,7 +16,7 @@ import {executeCoordinationTodoClaim} from "../../loopx/control_plane/coordinati import {executeCoordinationTodoUpdate} from "../../loopx/control_plane/coordination/todo_update.ts"; import {executeCanonicalTaskLeaseAcquire} from "../../loopx/control_plane/coordination/task_lease_acquire.ts"; import {executeCoordinationTodoTerminalLifecycle, type CoordinationTodoTerminalLifecycleInput} from "../../loopx/control_plane/coordination/todo_terminal_lifecycle.ts"; -import {listLocalCoordinationTodos, readLocalCoordinationTodo} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {listLocalCoordinationTodos, readLocalCoordinationTodo} from "../../loopx/control_plane/coordination/local_authority_read.ts"; const now = new Date("2026-09-17T12:00:00Z"); const todo = (extra: JsonObject = {}): JsonObject => ({schema_version: "todo_domain_record_v0", diff --git a/tests/control_plane_ts/local_authority_provider.test.ts b/tests/control_plane_ts/local_authority_provider.test.ts index 271b793fd1..5492284711 100644 --- a/tests/control_plane_ts/local_authority_provider.test.ts +++ b/tests/control_plane_ts/local_authority_provider.test.ts @@ -13,10 +13,13 @@ import { FileAuthorityStore } from "../../loopx/control_plane/coordination/file_ import { createRequire } from "node:module"; import { createHash } from "node:crypto"; import { authorityStoreCommitFixture } from "./authority_store_conformance.ts"; -import * as runtime from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import * as mutations from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import * as reads from "../../loopx/control_plane/coordination/local_authority_read.ts"; +const runtime = {...mutations, ...reads}; import { qualifiedShadow, promotionRequest, engageFence } from "./local_promotion_fixture.ts"; import { loadLegacyCoordinationWriterFence, legacyCoordinationWriterFencePath } from "../../loopx/control_plane/coordination/legacy_writer_fence.ts"; -import { acknowledgeLocalCoordinationTodoArchive, archiveLocalCoordinationTodos, listLocalCoordinationTodos } from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {acknowledgeLocalCoordinationTodoArchive, archiveLocalCoordinationTodos} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {listLocalCoordinationTodos} from "../../loopx/control_plane/coordination/local_authority_read.ts"; for (const [fault, source, reason] of [ ["database_missing", "sqlite_v0", "local_authority_provider_missing"], diff --git a/tests/control_plane_ts/local_authority_runtime.test.ts b/tests/control_plane_ts/local_authority_runtime.test.ts index 179bf43550..fcc235eeb7 100644 --- a/tests/control_plane_ts/local_authority_runtime.test.ts +++ b/tests/control_plane_ts/local_authority_runtime.test.ts @@ -36,13 +36,12 @@ import { LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA, LOCAL_COORDINATION_TODO_TERMINAL_LIFECYCLE_REQUEST_SCHEMA, archiveLocalCoordinationTodos, - listLocalCoordinationTodos, claimLocalCoordinationTodo, promoteLocalCoordinationAuthority, reviewLocalCoordinationAuthorityPromotion, - readLocalCoordinationTodo, terminalLifecycleLocalCoordinationTodo, } from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import {listLocalCoordinationTodos, readLocalCoordinationTodo} from "../../loopx/control_plane/coordination/local_authority_read.ts"; import { COORDINATION_TODO_CLAIM_RESULT_SCHEMA, evaluateCoordinationTodoClaimDecision, diff --git a/tests/control_plane_ts/projection_confirmation_conformance.ts b/tests/control_plane_ts/projection_confirmation_conformance.ts new file mode 100644 index 0000000000..fdb867a1b9 --- /dev/null +++ b/tests/control_plane_ts/projection_confirmation_conformance.ts @@ -0,0 +1,55 @@ +import {coordinationTodoReadModel} from "../../loopx/control_plane/coordination/coordination_projection.ts"; +import assert from "node:assert/strict"; +import test from "node:test"; +import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; +import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts"; +import {listLocalCoordinationTodos, readLocalCoordinationTodo} from "../../loopx/control_plane/coordination/local_authority_read.ts"; +import {LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA, LOCAL_COORDINATION_TODO_READ_REQUEST_SCHEMA} from "../../loopx/control_plane/coordination/coordination_state_contract.generated.ts"; + +export function registerProjectionConfirmationConformance(name: string, factory: AuthorityStoreConformanceFactory): void { + for (const shape of ["native", "legacy"] as const) test(`${name}: projection confirmation observes current full source (${shape})`, async context => { + const {store, contender} = await factory(context); + const goal = "projection-confirmation", fixture = productionScaleCoordinationFixture(goal, shape); + // Retained archived Monitor generations previously stranded Markdown recovery. + const monitor = (fixture.projection.todos as Record[]).find(row => row.task_class === "continuous_monitor")!; + Object.assign(monitor, {archive_state: "archive", status: "done", done: true, material_change_generation: 12, + ...(shape === "legacy" ? {source_section: "Completed Work Archive"} : {})}); + fixture.projection.todo_read_model = coordinationTodoReadModel(fixture.projection.todos as Record[], + (fixture.projection.todo_read_model as Record).schema_version as string); + const seeded = await store.commitAuthority({operation_id: "projection-source", expected_provider_revision: null, + next_projection: fixture.projection, events: [], receipts: []}); + assert.equal(seeded.status, "applied"); + const dependencies = {createStore: () => store}; + const request = {schema_version: LOCAL_COORDINATION_TODO_LIST_REQUEST_SCHEMA, goal_id: goal, runtime_root: "/synthetic-runtime"}; + const plain = await listLocalCoordinationTodos(request, dependencies); + assert.equal(plain.status, "loaded"); + assert.equal(Object.hasOwn(plain, "projection_readback"), false); + for (const changed of [false, true]) { + const confirmed = await listLocalCoordinationTodos({...request, + projection_readback: {provider_revision: seeded.provider_revision, changed}}, dependencies); + const {projection_readback, ...unchanged} = confirmed; + assert.deepEqual(unchanged, plain, "confirmation cannot change full-source semantics"); + assert.deepEqual(projection_readback, {status: changed ? "delivered" : "current", + provider_revision: seeded.provider_revision, observed_provider_revision: seeded.provider_revision}); + } + const before = await contender.loadAuthority(); assert.equal(before.status, "loaded"); + if (before.status !== "loaded") return; + const committed = await contender.commitAuthority({operation_id: "overlapping-revision", expected_provider_revision: before.provider_revision, + next_projection: before.head, events: [], receipts: []}); + assert.equal(committed.status, "applied"); + const after = await store.loadAuthority(); + const stale = await listLocalCoordinationTodos({...request, include_leases: true, + projection_readback: {provider_revision: seeded.provider_revision, changed: true}}, dependencies); + assert.deepEqual(stale.projection_readback, {status: "pending", provider_revision: seeded.provider_revision, + observed_provider_revision: committed.provider_revision}); + assert.equal((stale.todos as unknown[]).length, fixture.expected_initial_todo_count); + assert.equal((stale.leases as unknown[]).length, fixture.expected_current_lease_count); + assert.equal(stale.provider_revision, committed.provider_revision); + const exact = await readLocalCoordinationTodo({schema_version: LOCAL_COORDINATION_TODO_READ_REQUEST_SCHEMA, + runtime_root: request.runtime_root, goal_id: goal, todo_id: fixture.completion_todo_id}, dependencies); + assert.equal(exact.status, "found"); assert.equal(exact.provider_revision, stale.provider_revision); + const invalid = await listLocalCoordinationTodos({...request, projection_readback: {changed: true}}, dependencies); + assert.equal(invalid.status, "failed"); + assert.deepEqual(await store.loadAuthority(), after, "confirmation and rejected reads never write authority"); + }); +} diff --git a/tests/control_plane_ts/projection_delivery.test.ts b/tests/control_plane_ts/projection_delivery.test.ts index bc5c1a0a1c..734ce7f3a9 100644 --- a/tests/control_plane_ts/projection_delivery.test.ts +++ b/tests/control_plane_ts/projection_delivery.test.ts @@ -33,3 +33,18 @@ test("end-to-end fixture preserves delivery causal chain", async () => { item.readback === undefined ? projectionDelivery(item.changed === true) : parseProjectionDelivery(item.readback)); assert.deepEqual(observed, ["pending", "delivered", "current", "not_required", "pending"]); }); + +test("projection confirmation binds durable host readback to one observed revision", async () => { + const {decodeProjectionReadback, confirmProjectionReadback} = await import("../../loopx/control_plane/todos/projection_delivery.ts"); + for (const changed of [true, false]) { + const readback = decodeProjectionReadback({provider_revision: "revision-a", changed}); + assert.equal(confirmProjectionReadback(readback, "revision-a").status, changed ? "delivered" : "current"); + assert.deepEqual(confirmProjectionReadback(readback, "revision-b"), { + status: "pending", provider_revision: "revision-a", observed_provider_revision: "revision-b", + }); + } + for (const bad of [null, [], {}, {provider_revision: "", changed: false}, + {provider_revision: "a", changed: "true"}, {provider_revision: "a", changed: false, verified: true}]) { + assert.throws(() => decodeProjectionReadback(bad)); + } +}); From 4ec4de488a10c2308cebe4bd46510d6125e175f7 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Mon, 21 Sep 2026 02:20:01 +0800 Subject: [PATCH 2/2] docs(authority): define bounded projection confirmation and recovery Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...shared-goal-authority-state-provider-v0.md | 8 +++ ...-goal-authority-state-provider-v0.zh-CN.md | 5 ++ .../typescript-control-plane-migration-v0.md | 12 +++++ .../active-state-structured-projection-v0.md | 52 ++++++++++++++++--- 4 files changed, 69 insertions(+), 8 deletions(-) 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 c1988411ce..ae383674dc 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -3079,6 +3079,14 @@ provider conformance cover the consumer family. See [operation and semantic changes](../../reference/todo-continuation-readback.md). This closes a bounded L5/L7 gap; permanent projection delivery/recovery, D2 and D3 are still open. +D1 delivery confirmation now follows durable Markdown readback with a typed +canonical revision check. Unpinned settlement retries up to three times using +the returned complete snapshot; pinned projection never silently retargets. +Overlap, churn and confirmation outage remain pending without repeating business +commits. This qualifies the bounded delivery/retry boundary, not permanent +freshness, a background drainer, all L5 consumers or D2/D3. See the +[projection contract](../../reference/protocols/active-state-structured-projection-v0.md). + **D2 — qualify exactly one local profile; independent of PostgreSQL deployment.** - Reconcile the SQLite candidate #4121 with Section 7.2 before adding code. 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 820569e3be..cbe53e14db 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 @@ -2429,6 +2429,11 @@ route planner 本身仍不授予权限。CLI 将已提交回执交给既有 jour - caller 迁走后才删除旧 projection repair/receipt 路径。退出条件是可复核的 freshness/readback 和可操作修复路径，不能只证明成功渲染过一次。 +D1 交付确认现于 Markdown 耐久读回后核对 canonical revision。未固定版本的结算 +最多追赶三次，复用返回的完整快照；固定版本不擅自换目标。并发、持续变化及确认故障 +保留 pending，不重做业务提交。这闭合有界交付／重试，不代表永久新鲜度、后台 drain、 +全部 L5 或 D2/D3；见[投影合同](../../reference/protocols/active-state-structured-projection-v0.md)。 + **D2 — 资格化一个本地 profile，不等待 PostgreSQL 部署。** - 写代码前对齐 #4121 SQLite 候选与第 7.2 节；资格化及批准前保持 opt-in。 diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 7cb7f305d1..a0ade6a850 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1766,3 +1766,15 @@ new capability/provider, or Python storage migration is introduced. Python keeps input normalization and rendering until their actual host consumers migrate. See [the read contract](../../reference/todo-work-counts.md); broader L5/D1 and local-default qualifications remain open. + +### T2 canonical read and display confirmation boundary + +Canonical single-Todo and full-source reads now have one read-only TypeScript +module, separate from mutation orchestration and sharing the provider opening +boundary. Projection delivery composes a revision confirmation with the existing +full-source read; ordinary callers retain their response shape. Python owns +physical Markdown durability/retry, not the current-head comparison. Three-attempt +recovery and pinned-intent preservation use the existing journal-backed path; +no new RPC method, durable ACK or provider default. The stronger confirmation +costs one additional read on a stable delivery. Full L5/D1 qualification, D2 and +cutover remain open; see the [projection contract](../../reference/protocols/active-state-structured-projection-v0.md). diff --git a/docs/reference/protocols/active-state-structured-projection-v0.md b/docs/reference/protocols/active-state-structured-projection-v0.md index 845340bc11..e79c0fe760 100644 --- a/docs/reference/protocols/active-state-structured-projection-v0.md +++ b/docs/reference/protocols/active-state-structured-projection-v0.md @@ -219,11 +219,14 @@ freshness guarantee. Each section includes a compact `loopx:todo-section-projection-v0` marker with the canonical provider revision and a SHA-256 digest of the complete canonical records for that role. The marker is lineage evidence, not a write API. -The command proves that the rendered records came from the exact provider head -observed at read time. It does not claim that the revision remains the current -head after that read; a later canonical mutation makes the Markdown projection -stale until the journal-backed delivery replays. Consumers must always read the -provider, never the Markdown marker, when they need current authority state. +The command proves that the rendered records came from an exact provider head. +Execution now also reads authority **after** durable file readback: `delivered` +and `current` require the rendered revision to match that observed head. This +strengthens the previous read-time provenance contract; a successful file write +alone no longer acknowledges delivery when an overlapping commit is observed. +`observed_provider_revision` names the confirmation point, not a lock on future +commits. Later mutations can still make the display stale. Consumers must always +read the provider, never the Markdown marker, for current authority state. Rollback is intentionally asymmetric. Before promotion, the existing shadow rollback quarantines the candidate provider lineage and Markdown remains @@ -278,11 +281,44 @@ concurrently restored document. When bytes already match, execution still syncs the file and parent directory before reporting `current`: a previous failure may have occurred after rename but before directory durability. A failed barrier keeps delivery `pending` and does not acknowledge or repeat the business mutation. -Preview remains read-only. +Preview remains read-only and does not request a delivery confirmation. + +Unpinned mutation settlement makes at most three delivery attempts under the +existing display lock, reusing a newer complete read for the next attempt. A +pinned `project-markdown --provider-revision` checks its basis before writing and +never silently retargets another revision. An overlap after its write returns +`pending`, the rendered and observed revisions, `delivery_attempts`, and +`retry_business_mutation=false`. Persistent churn also returns pending rather +than looping indefinitely. A confirmation outage preserves the successful +business commit and remains retryable through the existing projection path. +Archived Monitor material generations use the same numeric decoder as active +reads and capture. Textual metadata such as `material_change_generation=12` +round-trips to the canonical integer; zero remains present and mismatched values +still fail parity. This fixes full-document recovery rejected by retained archived +Monitors without rewriting their authority records. + +No new queue, persistent ACK, background worker, authority write or provider +default is introduced. Ordinary list/exact reads keep their response shape; +only the internal projection readback request opts into confirmation metadata. + +The TypeScript read owner validates complete canonical data and compares the +host's durable readback revision with the same loaded head. Python retains +Markdown ownership, durability, bounded IO retry and rendering. A missing +confirmation from a downlevel runtime cannot be treated as delivery success. +The normal successful execution adds one provider read; each caught-up attempt +reuses the already returned full snapshot. This is a freshness cost, not a +latency improvement or atomic transaction across the database and filesystem. 中文：普通状态与投影共用原子落盘；缺失展示通过仅创建方式发布，避免覆盖并发恢复。 -字节相同的执行重试也重新完成文件和目录耐久化，之后才报告 `current`；失败继续 -保留“业务已提交、展示 pending”，不确认投影交付、不重执行业务。预览不写入。 +字节相同的执行重试也重新完成文件和目录耐久化。现在还必须在落盘后重新读取 authority， +由 TS 核对版本，才能确认 `current/delivered`。这加强了旧的“读取时来源正确”合同； +确认只对应一次观察点，不承诺之后永不变旧。未固定版本的交付最多尝试三次，复用较新 +完整快照；显式 `--provider-revision` 不自动换目标。持续并发或确认失败保留业务提交， +展示返回 pending，重试只恢复展示，不重复业务。缺失文件的第二次追赶使用普通原子 +替换，不能继续误用仅创建写入。预览不写入，也不确认交付。未新增队列、持久 ACK、 +后台任务或默认 provider；普通读取形状不变，正常交付增加一次真实 provider 读取。 +归档 Monitor 的代数元数据复用现有整数解码，修复字符串与整数比较造成的整份恢复失败； +零值仍保留，语义不一致仍拒绝，不改写 canonical 记录。 Supported non-Monitor Agent updates include action/domain/repository and required write scopes, required/target capabilities and Explore node references. These