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 bac0923c6..0fe99fd3e 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -3269,6 +3269,35 @@ semantic corrections. A green unit suite, a canonical selector, or a new config field alone cannot advance a package to default readiness. Planned integration, soak, release, merge and live promotion retain their respective authorization. +### Reviewed cutover checkpoint + +The saved-plan/recovery slice closes a concrete operator gap: execution can be +bound to the reviewed source/provider/policy, and a fenced cutover can be +completed or read back without reconstructing intent from legacy Markdown. +The TS owner shares durable qualification and exact receipt proof between both +paths. See [operation and acceptance](../../reference/reviewed-coordination-promotion.md). +This stage does not authorize an active Goal migration or flip a default. + +For an existing claimed Goal, integrate the claim-preserving migration in #4870 +with this slice, qualify the exact combined head and resolve its existing CI and +review holds. Preserve the registered owners, existing claims and leases; do not +clear ownership to make storage migration appear ready. The saved-plan carrier +must retain migration strategy, registered-agent facts and target digest when +that extension is integrated. + +The remaining default-on program is still approximately **5–8 cohesive PR +packages**, with scope rather than line counts determining the split: caller / +external-effect fencing (1–2), consumer/projection closure (1), contributor-owned +SQLite D2 (#4224, 1–2), integrated capture/whole-Goal acceptance (1–2), then default +onboarding plus bounded Python retirement (1). This slice contributes to the +integrated migration package; it does not count an entire package complete. +Actual elapsed soak cannot be compressed into a promised number of PRs. +PostgreSQL service admission and operations remain a separate medium-term lane. + +现有 Goal 的可审核晋升与恢复、所有新 Goal 默认选用 provider、删除全部 Python， +是三个不同完成条件。先交付一条能保留状态、能读回、能恢复的真实迁移路径，再按调用方 +闭合程度删除旧实现。不要用已合入 PR 数量替代端到端验收。 + ### Parallel delivery plan | Lane | May start | Scope and exit condition | Dependency | diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 9fb2ce0df..24cb668dd 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1830,3 +1830,17 @@ 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). + +### Reviewed coordination cutover ownership + +Saved-plan execution and fenced recovery now share the TypeScript promotion +owner. Fresh-source qualification wraps durable lineage qualification; recovery +uses that same lineage rule after exact fence verification. The Python CLI loads +a reviewed JSON carrier and transports fresh observations, without recreating +plan hashes, recovery decisions or receipt proof. Both commit paths share one +receipt/first-transaction readback contract. + +This is a migration orchestration checkpoint, not completion of Stage 3 or a +default-provider flip. Integrate claim-preserving migration separately, retain +real-backend and captured-source qualification, and retire Python only where its +actual callers have moved. [Operator contract](../../reference/reviewed-coordination-promotion.md). 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 8b1428e72..c5dc54fa3 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 @@ -1405,3 +1405,17 @@ transaction 只能靠削弱既有行为才能通过 invariant/recovery/performan 实测交付记录存于[逐条 ledger](ledger/typescript-control-plane-migration-v0/)。 每条记录说明已交付边界及剩余验收缺口；上方 T1–T4 检查点仍是当前迁移计划。 + +### Reviewed coordination cutover ownership + +Saved-plan execution and fenced recovery now share the TypeScript promotion +owner. Fresh-source qualification wraps durable lineage qualification; recovery +uses that same lineage rule after exact fence verification. The Python CLI loads +a reviewed JSON carrier and transports fresh observations, without recreating +plan hashes, recovery decisions or receipt proof. Both commit paths share one +receipt/first-transaction readback contract. + +This is a migration orchestration checkpoint, not completion of Stage 3 or a +default-provider flip. Integrate claim-preserving migration separately, retain +real-backend and captured-source qualification, and retire Python only where its +actual callers have moved. [Operator contract](../../reference/reviewed-coordination-promotion.md). diff --git a/docs/reference/local-authority-provider-selection.md b/docs/reference/local-authority-provider-selection.md index fcfd26f1f..d35881fce 100644 --- a/docs/reference/local-authority-provider-selection.md +++ b/docs/reference/local-authority-provider-selection.md @@ -56,3 +56,5 @@ selected-provider failure without fallback, PostgreSQL factory identity fencing, and the factory's rejection of a different provider. File, SQLite, and PostgreSQL continue to share the provider-neutral transaction conformance contract; PostgreSQL's real-server qualification remains a separate gate. + +See [reviewed promotion and recovery](reviewed-coordination-promotion.md) for the explicit saved-plan CLI journey. diff --git a/docs/reference/local-authority-provider-selection.zh-CN.md b/docs/reference/local-authority-provider-selection.zh-CN.md index 0ae9bac67..cbce17215 100644 --- a/docs/reference/local-authority-provider-selection.zh-CN.md +++ b/docs/reference/local-authority-provider-selection.zh-CN.md @@ -52,3 +52,5 @@ provider selection matrix 使用 production-scale synthetic coordination fixture PostgreSQL factory identity fencing，以及 factory 返回其他 provider 时的拒绝。 File、SQLite 和 PostgreSQL 继续共享 provider-neutral transaction conformance contract；PostgreSQL 的真实服务器 qualification 仍是独立 gate。 + +保存计划、执行和断点恢复的操作见[审核后的晋升与恢复](reviewed-coordination-promotion.zh-CN.md)。 diff --git a/docs/reference/reviewed-coordination-promotion.md b/docs/reference/reviewed-coordination-promotion.md new file mode 100644 index 000000000..69fb6de85 --- /dev/null +++ b/docs/reference/reviewed-coordination-promotion.md @@ -0,0 +1,142 @@ +# Reviewed coordination promotion and recovery + +A promotion moves a Goal's Todo/lease coordination authority from the legacy +source to its selected canonical provider. Preview, writer fencing, provider +commit, and acknowledgement are distinct steps. A successful preview is neither +a grant nor evidence that cutover has happened. + +The operator can now save the exact preview, execute that plan, and recover its +original transaction without reconstructing intent from a later Markdown view. +The TypeScript coordination boundary owns plan validation, qualification, +fencing and receipt proof; Python only loads the file and transports the request. + +## Preview and execute + +Use an explicitly enabled, bootstrapped and qualified runtime shadow. Its +qualification must cover real mutations and required event classes; an empty +shadow or a saved JSON file cannot substitute for that evidence. Existing v0 +promotion still requires `hard_lease`. Provider selection and migration approval +remain separate from these commands. + +```bash +loopx --format json coordination-shadow promote \ + --goal-id example-goal \ + --minimum-operations 3 \ + --require-event-kind todo_update > reviewed-promotion.json + +loopx --format json coordination-shadow promote \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json + +loopx --format json coordination-shadow promote \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json --execute +``` + +Inspect `ok`, `promotion.status`, the plan's target provider, source revision, +projection digest and qualification policy before execution. The saved file may +be the entire successful CLI preview or its `promotion.plan.reviewed_plan` +envelope. Keep it in operator-owned local storage: it carries a runtime path and +Goal identity, so it is not a public collaboration artifact. + +`--reviewed-plan` owns the operation id and qualification policy. Combining it +with `--minimum-operations` or `--require-event-kind` is an error. A normal +`promote` command without a saved plan retains its existing defaults. + +Execution captures and qualifies the source again under the existing locks. If +the computed plan digest differs, it returns +`local_authority_reviewed_plan_changed` before engaging a writer fence. Review a +new preview after legitimate source changes; do not edit the old digest to force +acceptance. The digest detects changed intent; the durable fence and provider +state establish whether that intent may proceed. + +## Recover the original cutover + +```bash +loopx --format json coordination-shadow recover-promotion \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json + +loopx --format json coordination-shadow recover-promotion \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json --execute +``` + +Recovery resolves the registered Goal and runtime but does not read legacy +Markdown or require the transient shadow opt-in. It requires the exact existing +writer fence. It never creates a missing fence, selects another provider, or +falls back to a legacy source. + +| Durable state | Preview | With `--execute` | +| --- | --- | --- | +| No matching writer fence | Reject | Reject | +| Matching fence, no canonical commit, exact qualified shadow retained | `recovery_ready` | Commit and read back | +| Original promotion committed, including a later canonical head | `replayed` | `replayed`; no business write | +| Different canonical initialization or inconsistent receipt lineage | Reject | Reject | +| Provider unavailable | Report provider failure | Report provider failure | + +For an uncommitted recovery, the original shadow revision, projection, capture +binding, complete transaction lineage, outbox settlement, operation count and +event coverage must still qualify. Recovery validates these durable facts under +the same maintenance guard used by canonical writers. It does not pretend to +observe fresh source parity after the source has ceased to be authority. + +A thrown commit acknowledgement can mean that the provider already committed. +Both promotion paths therefore share one commit/readback implementation. It +attempts the business commit once, then checks the persisted receipt and first +transaction. The receipt body, operation id, cursor, provider revision and +initial projection must agree. A matching proof reports success/recovery even +if later work has advanced the head. A missing or conflicting proof remains a +failure; an unavailable proof read is not silently treated as absence. + +The returned promotion revision and cursor identify the original cutover, not +the current head. `executed=false` on a replay means this invocation performed no +business write. Inspect `legacy_writer_fenced` and reconciliation evidence when +an execution fails; do not infer that a failure left legacy writers usable. + +## Bounded capture proof transport + +A long Goal can exceed the existing 2 MiB RPC response budget before promotion: +sequence recovery used to return a full head and full projections for retained +transactions. The `outbox_read` proof read model now keeps full lineage validation +inside TypeScript, while returning progress, receipts, projection digests and +partition markers. Sequence allocation requests no transaction rows; drain uses +the compact rows. Existing full diagnostic reads retain their default contract. +No transport limit, stored population or transaction validation is weakened. +A pending outbox still blocks promotion; use the existing bounded +`authority-shadow drain --goal-id example-goal --budget-seconds 60` operation +and inspect its result before retrying preview. + +## Product and rollout boundary + +This is an operator CLI administration journey. It adds no dashboard, Lark or +managed-Turn automatic migration trigger, settings editor, capability grant or +new provider selector. Those surfaces continue to consume canonical data through +the existing routing/projection contracts after a separately authorized cutover. + +File and SQLite use their existing local stores. PostgreSQL follows the same +transaction/readback contract through its service-owned factory; a local CLI +selector alone does not provide a PostgreSQL connection or tenant authority. + +The claim-preserving migration work in PR #4870 is a complementary prerequisite +for Goals that need explicit `preserve` or a claim-preserving `hard_lease` +transition. The two changes overlap the promotion orchestration and must be +integrated and tested together; this saved-plan feature alone does not enable +that policy conversion on a v0-only checkout. + +Default-on promotion, SQLite long-duration qualification, post-promotion export +or rollback, and retirement of remaining Python callers retain their RFC gates. +Recovery is a forward completion/readback operation, not rollback. Do not remove +a live fence, reset canonical storage, or replace the source to make recovery +pass. Before execution, abandoning a saved preview needs no runtime mutation. + +## Validation contract + +Durable tests cover real File/SQLite CLI preview, saved-plan execution, source +drift, policy override rejection, later canonical writes and recovery after +legacy deletion. The provider conformance suite uses the shared production-scale +fixture in legacy and native record shapes, preserving the complete Todo/lease +population through File, SQLite and a real isolated PostgreSQL server. + +Negative receipt tests independently corrupt the receipt index and first +transaction. Interrupted-commit tests distinguish failure before commit from a +lost acknowledgement after commit. These are synthetic fault injections, not a +claim of arbitrary process-death or elapsed-soak coverage. Real local rehearsals +must use read-only captured sources and disposable copies; never promote an +active Goal merely to validate this refactor. diff --git a/docs/reference/reviewed-coordination-promotion.zh-CN.md b/docs/reference/reviewed-coordination-promotion.zh-CN.md new file mode 100644 index 000000000..557e90c12 --- /dev/null +++ b/docs/reference/reviewed-coordination-promotion.zh-CN.md @@ -0,0 +1,99 @@ +# 审核后的协调状态晋升与恢复 + +晋升会把 Goal 的 Todo／lease 权威来源从旧路径切换到选定的 canonical provider。 +预览、封住旧写者、提交新存储、收到成功响应是不同的步骤。预览成功并不代表已经切换。 + +现在可以保存完整预览，执行这份计划，再用同一份计划恢复原事务。计划校验、准入、 +fence 与 receipt 证明由 TypeScript 协调边界负责；Python 只读文件、传输请求。 + +## 操作 + +先显式启用并 bootstrap runtime shadow，让它捕获真实变更并通过资格校验。 +现有 v0 晋升仍要求 Goal 已处于 `hard_lease`；保存 JSON 不会降低这个条件。 + +```bash +loopx --format json coordination-shadow promote \ + --goal-id example-goal --minimum-operations 3 \ + --require-event-kind todo_update > reviewed-promotion.json + +loopx --format json coordination-shadow promote \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json + +loopx --format json coordination-shadow promote \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json --execute +``` + +执行前审核 `ok`、`promotion.status`、目标 provider、源 revision、projection digest +和资格策略。文件可以是完整 CLI 成功预览，也可以是其中的 +`promotion.plan.reviewed_plan`。它包含 runtime 路径与 Goal 身份，应保存在本地， +不要贴到公开 PR。 + +保存的计划决定 operation id 和资格策略，不能再叠加 `--minimum-operations` 或 +`--require-event-kind`。不传计划文件的旧命令继续沿用原默认值。 + +执行会在现有锁内重新捕获、校验源状态。计划变了，就在 fencing 前返回 +`local_authority_reviewed_plan_changed`。此时重新预览并审核；不要修改旧 digest +来强行通过。digest 说明“执行的是哪份意图”，持久 fence 和 provider 状态说明 +“这份意图现在能否执行”。 + +## 断点恢复 + +```bash +loopx --format json coordination-shadow recover-promotion \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json + +loopx --format json coordination-shadow recover-promotion \ + --goal-id example-goal --reviewed-plan reviewed-promotion.json --execute +``` + +恢复仍要找到注册的 Goal 与 runtime，但不读取旧 Markdown，也不依赖临时 shadow +开关。它必须看到完全相同的持久 writer fence，不能创建缺失的 fence、换 provider， +也不能回退到旧来源。 + +| 状态 | 仅预览 | 加 `--execute` | +| --- | --- | --- | +| 没有匹配的 fence | 拒绝 | 拒绝 | +| 有 fence、尚未提交、保留的 shadow 仍完全合格 | `recovery_ready` | 提交并读回 | +| 原晋升已提交，甚至 canonical 已继续变更 | `replayed` | `replayed`，不重写 | +| 已被其他事务初始化，或 receipt 与事务链矛盾 | 拒绝 | 拒绝 | +| provider 不可用 | 报告 provider 错误 | 报告 provider 错误 | + +尚未提交的恢复仍核对原 shadow revision、projection、capture binding、完整事务链、 +outbox 是否结清、操作次数和事件覆盖；校验与提交共用 canonical writer 的维护锁。 +它不会把已经失去权威地位的旧文件当作必须重新观察的来源。 + +提交调用抛错，也可能是“数据已落盘，但响应丢了”。两条晋升路径现在共用一次提交、 +持久读回的实现，不盲目重做业务事务。receipt 与第一笔事务的 operation id、cursor、 +provider revision、receipt 内容和初始 projection 必须全部一致。 + +返回的 revision 与 cursor 指向原晋升事务，不一定是当前最新 head。重放返回 +`executed=false`，表示本次没有业务写入。失败时还要看 `legacy_writer_fenced` 和 +恢复提示；失败不等于旧写者一定还能继续工作。 + +## 大 Goal 的捕获证明 + +序号恢复以前会返回完整 head 和历史事务的完整 projection，大 Goal 可能因此超过 +现有 2 MiB RPC 响应上限，甚至不能产生晋升所需的 shadow 变更。 +现在 `outbox_read` 的 proof read model 在 TS 内完整校验事务链，只传回进度、receipt、 +projection digest 和 partition marker；分配序号不返回事务行，drain 使用紧凑行。 +原有完整诊断读取保持默认合同，不提高传输上限、不减少持久数据，也不省略链校验。 +若 outbox 尚未结清，晋升仍会拒绝。先执行并检查既有的有界 drain： +`authority-shadow drain --goal-id example-goal --budget-seconds 60`。 + +## 交付边界 + +这是运维 CLI 的显式管理操作，没有新增 Dashboard／飞书自动迁移按钮、设置项或 +capability grant。它们在获授权的切换后继续使用既有 canonical 路由和展示合同。 +PostgreSQL 仍需要服务持有的 factory 与租户权限，不能仅靠本地 selector 接通数据库。 + +PR #4870 提供保留 claim 的 `preserve`／`hard_lease` 转换，属于互补前置工作； +两者涉及同一个晋升编排，需要组合验证。本功能单独合入不会让 v0 checkout 自动获得 +这些模式转换。默认切换、SQLite 长时资格、晋升后导出／回退、剩余 Python 删除，仍 +遵守 RFC 的独立门槛。 + +恢复用于向前补齐或确认原切换，不是 rollback。不要删除活跃 fence、重置 canonical +存储或替换源文件来绕过拒绝。执行前放弃一份预览，只需停止使用该文件。 + +验证包括真实 File／SQLite CLI、完整合成状态的多 provider conformance、receipt +索引与事务链分别被破坏的负例，以及提交前中断、提交后丢响应的故障注入。故障注入 +不等于任意进程崩溃或长时 soak 已验证。真实项目演练只在只读快照的可丢弃副本上执行。 diff --git a/loopx/cli_commands/coordination_shadow.py b/loopx/cli_commands/coordination_shadow.py index 95902f837..b43e665ff 100644 --- a/loopx/cli_commands/coordination_shadow.py +++ b/loopx/cli_commands/coordination_shadow.py @@ -23,6 +23,7 @@ resolve_coordination_runtime_shadow_config, rollback_coordination_runtime_shadow, ) +from ..control_plane.coordination.promotion_review import execute_reviewed_coordination_promotion from ..history import load_registry from ..paths import resolve_runtime_root from ..registry import find_registry_goal @@ -66,17 +67,22 @@ def register_coordination_shadow_command( "Import the current legacy projection into an empty file shadow.", ), ("rollback", "Quarantine one exact pre-promotion file shadow lineage."), + ("recover-promotion", "Read back or recover an already-fenced saved promotion without reading legacy Markdown."), ): action = actions.add_parser(name, help=help_text) action.add_argument("--goal-id", required=True) - action.add_argument("--project", type=Path) - action.add_argument("--state-file", type=Path) - if name in {"bootstrap", "rollback", "promote"}: + if name != "recover-promotion": + action.add_argument("--project", type=Path) + action.add_argument("--state-file", type=Path) + if name in {"bootstrap", "rollback", "promote", "recover-promotion"}: action.add_argument( "--execute", action="store_true", help="Execute the administrative effect; otherwise preview only.", ) + if name in {"promote", "recover-promotion"}: + action.add_argument("--reviewed-plan", type=Path, required=name == "recover-promotion", + help="Saved successful promotion preview or its exact reviewed_plan envelope. The plan owns operation and qualification policy.") if name == "rollback": selector = action.add_mutually_exclusive_group(required=True) selector.add_argument( @@ -101,7 +107,7 @@ def register_coordination_shadow_command( action.add_argument( "--minimum-operations", type=int, - default=3, + default=None, help="Minimum verified primary mutations in the selected lineage (default: 3).", ) action.add_argument( @@ -213,6 +219,33 @@ def handle_coordination_shadow_command( goal = find_registry_goal(registry, args.goal_id) if goal is None: raise ValueError(f"goal {args.goal_id!r} is not present in the registry") + runtime_root = resolve_runtime_root(registry, runtime_root_arg, registry_path=registry_path) + reviewed_path = getattr(args, "reviewed_plan", None) + reviewed_plan = None + if reviewed_path is not None: + reviewed_plan = json.loads(reviewed_path.read_text(encoding="utf-8")) + if not isinstance(reviewed_plan, dict): + raise ValueError("reviewed promotion plan must be a JSON object") + if args.coordination_shadow_command == "recover-promotion": + if reviewed_plan is None: + raise ValueError("promotion recovery requires --reviewed-plan") + promotion = execute_reviewed_coordination_promotion( + reviewed_plan=reviewed_plan, runtime_root=runtime_root, goal_id=args.goal_id, + action="recover", execute=bool(args.execute), + ) + payload = { + "ok": promotion.get("status") in {"recovery_ready", "applied", "replayed", "recovered"}, + "schema_version": "loopx_coordination_shadow_admin_v0", + "action": "recover-promotion", "goal_id": args.goal_id, + "executed": promotion.get("executed") is True, "promotion": promotion, + "decision_read_from_shadow": False, + } + print_payload(payload, output_format(args), _render) + return 0 if payload["ok"] else 1 + if reviewed_plan is not None and ( + args.minimum_operations is not None or args.require_event_kind + ): + raise ValueError("--reviewed-plan owns qualification policy; omit --minimum-operations and --require-event-kind") config = resolve_coordination_runtime_shadow_config(goal) if not config.enabled: payload = { @@ -232,11 +265,6 @@ def handle_coordination_shadow_command( } print_payload(payload, output_format(args), _render) return 1 - runtime_root = resolve_runtime_root( - registry, - runtime_root_arg, - registry_path=registry_path, - ) _, _, state_path = resolve_goal_state(registry=registry, goal_id=args.goal_id, project_override=args.project, state_file_override=args.state_file) if args.coordination_shadow_command == "rollback": @@ -340,47 +368,54 @@ def handle_coordination_shadow_command( and read_candidate.get("decision_read_from_shadow") is False ) if args.coordination_shadow_command == "promote": - registered_agents = registered_agent_ids_for_goal(goal) - operation_digest = _projection_version( - { - "goal_id": args.goal_id, - "projection": projection, - "minimum_operations": args.minimum_operations, - "required_event_kinds": args.require_event_kind, - **( - { - "handoff_mode_migration": args.handoff_mode_migration, - "registered_agents": registered_agents, - } + # A saved plan owns qualification policy, so the flag defaults to + # `None`; normalize it only for the freshly reviewed path. + minimum_operations = args.minimum_operations if args.minimum_operations is not None else 3 + if reviewed_plan is not None: + promotion = execute_reviewed_coordination_promotion( + reviewed_plan=reviewed_plan, runtime_root=runtime_root, goal_id=args.goal_id, + action="apply", execute=bool(args.execute), + projection=projection, source_snapshot=source_snapshot, + ) + else: + registered_agents = registered_agent_ids_for_goal(goal) + operation_digest = _projection_version( + { + "goal_id": args.goal_id, + "projection": projection, + "minimum_operations": minimum_operations, + "required_event_kinds": args.require_event_kind, + **( + { + "handoff_mode_migration": args.handoff_mode_migration, + "registered_agents": registered_agents, + } + if args.handoff_mode_migration is not None + else {} + ), + } + ) + promotion = review_local_coordination_authority_promotion( + goal=goal, + runtime_root=runtime_root, + goal_id=args.goal_id, + operation_id=f"promote:{args.goal_id}:{operation_digest}", + projection=projection, + source_snapshot=source_snapshot, + minimum_operations=minimum_operations, + required_event_kinds=args.require_event_kind, + handoff_mode_migration=args.handoff_mode_migration, + registered_agents=( + registered_agents if args.handoff_mode_migration is not None - else {} + else None ), - } - ) - promotion = review_local_coordination_authority_promotion( - goal=goal, - runtime_root=runtime_root, - goal_id=args.goal_id, - operation_id=f"promote:{args.goal_id}:{operation_digest}", - projection=projection, - source_snapshot=source_snapshot, - minimum_operations=args.minimum_operations, - required_event_kinds=args.require_event_kind, - handoff_mode_migration=args.handoff_mode_migration, - registered_agents=( - registered_agents - if args.handoff_mode_migration is not None - else None - ), - execute=bool(args.execute), - ) - payload["executed"] = bool(args.execute) + execute=bool(args.execute), + ) + payload["executed"] = promotion.get("executed") is True payload["promotion"] = promotion payload["ok"] = promotion.get("status") in { - "preview_ready", - "applied", - "replayed", - "recovered", + "preview_ready", "applied", "replayed", "recovered", } if args.coordination_shadow_command == "rollback": provider_revision = getattr(args, "provider_revision", None) diff --git a/loopx/control_plane/coordination/local_authority_runtime.ts b/loopx/control_plane/coordination/local_authority_runtime.ts index 1a3865817..68c0d877b 100644 --- a/loopx/control_plane/coordination/local_authority_runtime.ts +++ b/loopx/control_plane/coordination/local_authority_runtime.ts @@ -1,3 +1,5 @@ +import {readPromotionReceipt, commitPromotionAndReadBack} from './promotion_receipt.ts'; +import {reviewedPromotionPlan, promotionPlanDigest, decodeReviewedPromotionOperation, REVIEWED_PROMOTION_OPERATION_RESULT_SCHEMA} from './reviewed_promotion_plan.ts'; import {registryAuthoritySourceCheck} from "./authority_source.ts"; import {decodeTaskLeaseProof} from "./task_lease_proof.ts"; import {COORDINATION_TODO_ARCHIVE_RESULT_SCHEMA} from "./todo_archive.ts"; @@ -29,7 +31,7 @@ import { import { indexCoordinationProjection, } from "./coordination_projection.ts"; -import { authorityStoreSourceAuthority, type AuthorityStore, type AuthorityStoreReceiptResult } from "./authority_store.ts"; +import { authorityStoreSourceAuthority, type AuthorityStore } from "./authority_store.ts"; import { authorityUnicodeCompare, canonicalAuthorityBytes, @@ -52,10 +54,9 @@ import { loadLegacyCoordinationWriterFence, } from "./legacy_writer_fence.ts"; import { - COORDINATION_RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA, decodeRuntimeShadowRequest, qualifyCoordinationRuntimeShadowUnderLocks, - qualifyCoordinationRuntimeShadow, + qualifyCoordinationShadowLineageUnderLocks, withShadowSourceLocks, } from "./runtime_shadow.ts"; import { @@ -149,7 +150,7 @@ export async function reviewLocalCoordinationAuthorityPromotion( value, LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, ["operation_id", "minimum_operations", "required_event_kinds", "execute", - "handoff_mode_migration", "registered_agents"], + "handoff_mode_migration", "registered_agents", "expected_promotion_plan_sha256"], ); const operationId = requireAuthorityStoreId(input.operation_id, "operation id"); const minimumOperations = requiredPositiveSafeInteger( @@ -171,6 +172,8 @@ export async function reviewLocalCoordinationAuthorityPromotion( const registeredAgents = input.registered_agents === undefined || input.registered_agents === null ? [] : normalizeRegisteredTodoAgents(input.registered_agents as string[]); + const expectedPlan = input.expected_promotion_plan_sha256 === undefined + ? null : promotionPlanDigest(input.expected_promotion_plan_sha256); const statePath = await realpath(String(input.source_snapshot.state_path)); const shadow = dependencies.createShadowStore?.( shadowDirectory(input.runtime_root), @@ -242,6 +245,14 @@ export async function reviewLocalCoordinationAuthorityPromotion( expected_target_projection_sha256: migration.target_projection_sha256, } : {}), }); + if (expectedPlan !== null && expectedPlan !== promotionPlanSha256) return { + schema_version:schema, status:"not_ready", executed:false, + reason_code:"local_authority_reviewed_plan_changed", + reason:"The current promotion differs from the reviewed plan; preview and review the new plan.", + expected_promotion_plan_sha256:expectedPlan, + observed_promotion_plan_sha256:promotionPlanSha256, + legacy_writer_fenced:false, legacy_fallback_used:false, + }; const fence = canonicalAuthorityObject({ schema_version: LEGACY_COORDINATION_WRITER_FENCE_SCHEMA, state: "engaged", @@ -325,6 +336,7 @@ export async function reviewLocalCoordinationAuthorityPromotion( }; } const plan = { + reviewed_plan: reviewedPromotionPlan({schema_version:LOCAL_COORDINATION_PROMOTION_REQUEST_SCHEMA,...request}, promotionPlanSha256), operation_id: operationId, promotion_plan_sha256: promotionPlanSha256, canonical_authority: canonicalAuthority, @@ -370,26 +382,21 @@ export async function reviewLocalCoordinationAuthorityPromotion( writerFenceVerified = true; } const identity = promotionIdentity(request); - const committed = await canonical.commitAuthority({ - expected_provider_revision: null, - operation_id: operationId, - events: [{ - ...identity, - schema_version: "loopx_local_coordination_promotion_event_v0", - mode_transition: `legacy_canonical_to_${canonicalAuthority}`, - handoff_mode_transition: `${migration.previous_mode}_to_${migration.target_mode}`, - }], - next_projection: migration.target_projection, - receipts: [identity], - }); - const readback = await promotionReadback(canonical, request); + const attempted = await commitPromotionAndReadBack(canonical, { + operation_id:operationId, receipt:identity, + projection_sha256:promotionTargetProjectionSha256(request), + }, migration.target_projection, {...identity, + schema_version:"loopx_local_coordination_promotion_event_v0", + mode_transition:`legacy_canonical_to_${canonicalAuthority}`, + handoff_mode_transition:`${migration.previous_mode}_to_${migration.target_mode}`}); + const {commit:committed,readback}=attempted; if (readback.matched) return { schema_version: schema, ...promotionResult( request, - committed.status === "applied" + committed?.status === "applied" ? recoveringFromFence ? "recovered" : "applied" - : committed.status === "ambiguous" ? "recovered" : "replayed", + : attempted.interrupted || committed?.status === "ambiguous" ? "recovered" : "replayed", readback, canonicalAuthority, ), @@ -403,7 +410,7 @@ export async function reviewLocalCoordinationAuthorityPromotion( executed: true, reason_code: readback.reason_code ?? "local_authority_promotion_readback_mismatch", reason: "promotion did not produce an exact canonical readback", - reconciliation_required: committed.status === "ambiguous", + reconciliation_required: attempted.interrupted || committed?.status === "ambiguous", qualification: publicQualification, legacy_writer_fenced: true, legacy_fallback_used: false, @@ -698,51 +705,18 @@ function promotionTargetProjectionSha256(request: LocalCoordinationPromotionRequ return request.expected_target_projection_sha256 ?? request.expected_shadow_projection_sha256; } -function matchingReceipt( - result: AuthorityStoreReceiptResult, - expected: JsonObject, -): Extract | null { - return result.status === "found" && result.receipts.length === 1 && - canonicalAuthorityBytes(result.receipts[0]).equals(canonicalAuthorityBytes(expected)) - ? result - : null; -} - -async function promotionReadback( - store: AuthorityStore, - request: LocalCoordinationPromotionRequest, -): Promise<{ matched: boolean; provider_revision?: string; cursor?: string; reason_code?: string }> { - const receiptResult = await store.readReceipt(request.operation_id); - const receipt = matchingReceipt(receiptResult, promotionIdentity(request)); - if (receipt === null) { - return { - matched: false, - reason_code: receiptResult.status === "found" - ? "local_authority_promotion_identity_mismatch" - : "local_authority_promotion_receipt_missing", - }; - } - const lineage = await store.scanCommitted(null, 1); - const promotion = lineage.status === "page" ? lineage.transactions[0] : undefined; - if ( - promotion === undefined || - promotion.cursor !== "1" || - promotion.operation_id !== request.operation_id || - canonicalAuthoritySha256(promotion.projection) !== promotionTargetProjectionSha256(request) - ) { - return { matched: false, reason_code: "local_authority_promotion_lineage_mismatch" }; - } - return { - matched: true, - provider_revision: receipt.provider_revision, - cursor: receipt.cursor, - }; +/** One receipt/lineage readback owner for reviewed and already-fenced promotion. */ +function promotionReadback(store: AuthorityStore, request: LocalCoordinationPromotionRequest) { + return readPromotionReceipt(store, { + operation_id:request.operation_id, receipt:promotionIdentity(request), + projection_sha256:promotionTargetProjectionSha256(request), + }); } function promotionResult( request: LocalCoordinationPromotionRequest, status: "applied" | "replayed" | "recovered", - readback: Awaited>, + readback: Extract>, {matched:true}>, canonicalAuthority: string, ): JsonObject { return { @@ -777,8 +751,14 @@ export async function promoteLocalCoordinationAuthority( dependencies: LocalAuthorityRuntimeDependencies = {}, ): Promise { let request: LocalCoordinationPromotionRequest; + let execute = true; try { request = decodePromotionRequest(value); + const raw=requireJsonObject(value,"promotion request"); + if(raw.execute !== undefined) { + if(typeof raw.execute !== "boolean") throw new TypeError("execute must be a JSON boolean"); + execute=raw.execute; + } } catch (error) { return { schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, @@ -816,153 +796,151 @@ export async function promoteLocalCoordinationAuthority( // Provider opening can fail before durable fence readback. Report only // evidence this invocation actually verified, including in the outer catch. let writerFenceVerified = false; + let commitAttempted = false; try { - const shadow = dependencies.createShadowStore?.( - shadowDirectory(request.runtime_root), - request.goal_id, - ) ?? new FileAuthorityStore(shadowDirectory(request.runtime_root), request.goal_id); - const canonical = dependencies.createCanonicalStore?.( - authorityDirectory(request.runtime_root), - request.goal_id, - ) ?? await openRuntimeStore(request.runtime_root, request.goal_id, dependencies); - const canonicalAuthority = sourceAuthorityFor(canonical); - if (request.canonical_authority !== canonicalAuthority) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: "local_authority_promotion_provider_mismatch", - reason: "promotion request is not bound to the selected canonical authority provider", - expected_canonical_authority: request.canonical_authority, - observed_canonical_authority: canonicalAuthority, - legacy_writer_fenced: false, - legacy_fallback_used: false, - }; - const promotionPlanSha256 = localCoordinationPromotionPlanSha256(request); - if (request.writer_fence.promotion_plan_sha256 !== promotionPlanSha256) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: "local_authority_writer_fence_plan_mismatch", - reason: "writer fence is not bound to the complete reviewed promotion plan", - promotion_plan_sha256: promotionPlanSha256, - legacy_writer_fenced: false, - legacy_fallback_used: false, - }; - const persistedFence = await loadLegacyCoordinationWriterFence( - request.runtime_root, - request.goal_id, - ); - if ( - persistedFence.status !== "loaded" || - !canonicalAuthorityBytes(persistedFence.fence).equals( - canonicalAuthorityBytes(request.writer_fence), - ) - ) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: persistedFence.status === "failed" - ? persistedFence.reason_code - : "local_authority_writer_fence_not_verified", - reason: persistedFence.status === "failed" - ? persistedFence.reason - : "exact durable legacy writer fence must be engaged before promotion", - legacy_writer_fenced: false, - legacy_fallback_used: false, - }; - writerFenceVerified = true; - const existing = await canonical.loadAuthority(); - if (existing.status === "loaded") { - const readback = await promotionReadback(canonical, request); - return readback.matched - ? promotionResult(request, "replayed", readback, canonicalAuthority) - : { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: readback.reason_code ?? "local_authority_already_initialized", - reason: "canonical local authority is already initialized by different content", - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; - } - if (existing.status !== "missing") return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - ...existing, - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; + return await withCanonicalWriter(request.runtime_root, request.goal_id, false, async () => { + const shadow = dependencies.createShadowStore?.( + shadowDirectory(request.runtime_root), + request.goal_id, + ) ?? new FileAuthorityStore(shadowDirectory(request.runtime_root), request.goal_id); + const canonical = dependencies.createCanonicalStore?.( + authorityDirectory(request.runtime_root), + request.goal_id, + ) ?? await openRuntimeStore(request.runtime_root, request.goal_id, dependencies); + const canonicalAuthority = sourceAuthorityFor(canonical); + if (request.canonical_authority !== canonicalAuthority) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: "local_authority_promotion_provider_mismatch", + reason: "promotion request is not bound to the selected canonical authority provider", + expected_canonical_authority: request.canonical_authority, + observed_canonical_authority: canonicalAuthority, + legacy_writer_fenced: false, + legacy_fallback_used: false, + }; + const promotionPlanSha256 = localCoordinationPromotionPlanSha256(request); + if (request.writer_fence.promotion_plan_sha256 !== promotionPlanSha256) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: "local_authority_writer_fence_plan_mismatch", + reason: "writer fence is not bound to the complete reviewed promotion plan", + promotion_plan_sha256: promotionPlanSha256, + legacy_writer_fenced: false, + legacy_fallback_used: false, + }; + const persistedFence = await loadLegacyCoordinationWriterFence( + request.runtime_root, + request.goal_id, + ); + if ( + persistedFence.status !== "loaded" || + !canonicalAuthorityBytes(persistedFence.fence).equals( + canonicalAuthorityBytes(request.writer_fence), + ) + ) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: persistedFence.status === "failed" + ? persistedFence.reason_code + : "local_authority_writer_fence_not_verified", + reason: persistedFence.status === "failed" + ? persistedFence.reason + : "exact durable legacy writer fence must be engaged before promotion", + legacy_writer_fenced: false, + legacy_fallback_used: false, + }; + writerFenceVerified = true; + const existing = await canonical.loadAuthority(); + if (existing.status === "loaded") { + const readback = await promotionReadback(canonical, request); + return readback.matched + ? promotionResult(request, "replayed", readback, canonicalAuthority) + : { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: readback.reason_code ?? "local_authority_already_initialized", + reason: "canonical local authority is already initialized by different content", + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; + } + if (existing.status !== "missing") return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + ...existing, + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; - const shadowHead = await shadow.loadAuthority(); - if (shadowHead.status !== "loaded") return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: shadowHead.status === "missing" - ? "local_authority_shadow_missing" - : shadowHead.reason_code, - reason: shadowHead.status === "missing" ? "qualified shadow authority is missing" : shadowHead.reason, - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; - const observedDigest = canonicalAuthoritySha256(shadowHead.head); - if ( - shadowHead.provider_revision !== request.expected_shadow_provider_revision || - observedDigest !== request.expected_shadow_projection_sha256 - ) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: "local_authority_shadow_fence_mismatch", - reason: "shadow revision or projection changed before promotion", - observed_shadow_provider_revision: shadowHead.provider_revision, - observed_shadow_projection_sha256: observedDigest, - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; - indexCoordinationProjection(shadowHead.head, request.goal_id); - const qualification = await qualifyCoordinationRuntimeShadow({ - schema_version: COORDINATION_RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA, - runtime_root: request.runtime_root, - goal_id: request.goal_id, - projection: shadowHead.head, - minimum_operations: request.minimum_operations, - required_event_kinds: request.required_event_kinds, - }, { - createStore: () => shadow, - }); - if (qualification.status !== "qualified" || qualification.qualified !== true) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: "local_authority_shadow_not_qualified", - reason: "shadow parity evidence does not satisfy the promotion policy", - qualification_status: qualification.status, - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; + const shadowHead = await shadow.loadAuthority(); + if (shadowHead.status !== "loaded") return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: shadowHead.status === "missing" + ? "local_authority_shadow_missing" + : shadowHead.reason_code, + reason: shadowHead.status === "missing" ? "qualified shadow authority is missing" : shadowHead.reason, + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; + const observedDigest = canonicalAuthoritySha256(shadowHead.head); + if ( + shadowHead.provider_revision !== request.expected_shadow_provider_revision || + observedDigest !== request.expected_shadow_projection_sha256 + ) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: "local_authority_shadow_fence_mismatch", + reason: "shadow revision or projection changed before promotion", + observed_shadow_provider_revision: shadowHead.provider_revision, + observed_shadow_projection_sha256: observedDigest, + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; + indexCoordinationProjection(shadowHead.head, request.goal_id); - const migration = planPromotionHandoffMigration( - shadowHead.head, - request.goal_id, - request.handoff_mode_migration, - request.registered_agents ?? [], - new Date(), - ); - if (!migration.ready) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: migration.reason_code ?? "handoff_mode_migration_conflict", - reason: migration.reason ?? "handoff-mode migration is not ready", - handoff_mode_migration: publicPromotionHandoffMigrationPlan(migration), - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; - if (migration.target_projection_sha256 !== promotionTargetProjectionSha256(request)) return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: "local_authority_promotion_target_projection_mismatch", - reason: "handoff-mode migration target differs from the reviewed promotion plan", - observed_target_projection_sha256: migration.target_projection_sha256, - expected_target_projection_sha256: promotionTargetProjectionSha256(request), - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; + const qualification = await qualifyCoordinationShadowLineageUnderLocks({ + runtime_root: request.runtime_root, + goal_id: request.goal_id, + projection: shadowHead.head, + }, {createStore: () => shadow}, request.minimum_operations, request.required_event_kinds); + if (qualification.status !== "qualified" || qualification.qualified !== true) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: "local_authority_shadow_not_qualified", + reason: "shadow parity evidence does not satisfy the promotion policy", + qualification_status: qualification.status, + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; + // A default reviewed promotion still requires the shadow's own + // hard_lease policy; an explicit migration request supplies the target. + const migration = planPromotionHandoffMigration( + shadowHead.head, + request.goal_id, + request.handoff_mode_migration, + request.registered_agents ?? [], + new Date(), + ); + if (!migration.ready) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: migration.reason_code ?? "handoff_mode_migration_conflict", + reason: migration.reason ?? "handoff-mode migration is not ready", + handoff_mode_migration: publicPromotionHandoffMigrationPlan(migration), + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; + if (migration.target_projection_sha256 !== promotionTargetProjectionSha256(request)) return { + schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status: "failed", + reason_code: "local_authority_promotion_target_projection_mismatch", + reason: "handoff-mode migration target differs from the reviewed promotion plan", + observed_target_projection_sha256: migration.target_projection_sha256, + expected_target_projection_sha256: promotionTargetProjectionSha256(request), + legacy_writer_fenced: true, + legacy_fallback_used: false, + }; - return await withCanonicalWriter(request.runtime_root, request.goal_id, false, async () => { const finalShadowHead = await shadow.loadAuthority(); if ( finalShadowHead.status !== "loaded" || @@ -995,53 +973,39 @@ export async function promoteLocalCoordinationAuthority( legacy_fallback_used: false, }; + if(!execute) return { + schema_version:LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + status:"recovery_ready", executed:false, operation_id:request.operation_id, + promotion_plan_sha256:promotionPlanSha256, canonical_authority:canonicalAuthority, + legacy_writer_fenced:true, legacy_fallback_used:false, + }; const identity = promotionIdentity(request); - const committed = await canonical.commitAuthority({ - expected_provider_revision: null, - operation_id: request.operation_id, - events: [{ - ...identity, - schema_version: "loopx_local_coordination_promotion_event_v0", - mode_transition: `legacy_canonical_to_${canonicalAuthority}`, - handoff_mode_transition: `${finalMigration.previous_mode}_to_${finalMigration.target_mode}`, - }], - next_projection: finalMigration.target_projection, - receipts: [identity], - }); - if (committed.status === "applied") { - const readback = await promotionReadback(canonical, request); - return readback.matched - ? promotionResult(request, "applied", readback, canonicalAuthority) - : { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", - reason_code: readback.reason_code ?? "local_authority_promotion_readback_mismatch", - reason: "promotion commit lacks an exact durable readback", - legacy_writer_fenced: true, - legacy_fallback_used: false, - }; - } - const readback = await promotionReadback(canonical, request); - if (readback.matched) { - return promotionResult( - request, - committed.status === "ambiguous" ? "recovered" : "replayed", - readback, - canonicalAuthority, - ); - } + commitAttempted = true; + const attempted=await commitPromotionAndReadBack(canonical, { + operation_id:request.operation_id,receipt:identity, + projection_sha256:promotionTargetProjectionSha256(request), + }, finalMigration.target_projection, {...identity, + schema_version:"loopx_local_coordination_promotion_event_v0", + mode_transition:`legacy_canonical_to_${canonicalAuthority}`, + handoff_mode_transition:`${finalMigration.previous_mode}_to_${finalMigration.target_mode}`}); + const {commit:committed,readback}=attempted; + if(readback.matched) return promotionResult(request, + committed?.status === "applied" ? "applied" + : attempted.interrupted || committed?.status === "ambiguous" ? "recovered" : "replayed", + readback,canonicalAuthority); return { - schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - ...committed, - ...(committed.status === "ambiguous" ? { reconciliation_required: true } : {}), - legacy_writer_fenced: true, - legacy_fallback_used: false, + schema_version:LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, + ...(committed ?? {}), status:"failed", executed:true, + reason_code:readback.reason_code, + reason:"promotion did not produce an exact durable readback", + reconciliation_required:attempted.interrupted || committed?.status === "ambiguous", + legacy_writer_fenced:true, legacy_fallback_used:false, }; }); } catch (error) { return { schema_version: LOCAL_COORDINATION_PROMOTION_RESULT_SCHEMA, - status: "failed", + status: "failed", executed:commitAttempted, reason_code: error instanceof ShadowManagementError ? error.reason_code : "local_authority_promotion_unavailable", reason: error instanceof Error ? error.message : "promotion unavailable", legacy_writer_fenced: writerFenceVerified, @@ -1547,3 +1511,35 @@ export async function observeLocalCoordinationOwnership(value: unknown): Promise decision_read_from_provider: true, legacy_fallback_used: false, ...localAuthorityOpenFailure(error)}; } } + +/** Saved review data cannot grant a fence or authorize another Goal. */ +export async function executeReviewedCoordinationPromotion( + value: unknown, dependencies: LocalAuthorityRuntimeDependencies = {}, +): Promise { + try { + const input=decodeReviewedPromotionOperation(value); + const request=decodePromotionRequest(input.request); + const digest=localCoordinationPromotionPlanSha256(request); + if(digest !== input.expected_plan_sha256 || request.writer_fence.promotion_plan_sha256 !== digest) { + throw new TypeError("reviewed promotion request does not match its plan digest"); + } + const result=input.action === "recover" + ? await promoteLocalCoordinationAuthority({...input.request,execute:input.execute},dependencies) + : await reviewLocalCoordinationAuthorityPromotion({ + schema_version:LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, + runtime_root:input.runtime_root,goal_id:input.goal_id, + operation_id:request.operation_id,minimum_operations:request.minimum_operations, + required_event_kinds:request.required_event_kinds,execute:input.execute, + expected_promotion_plan_sha256:input.expected_plan_sha256, + projection:input.projection,source_snapshot:input.source_snapshot, + },dependencies); + return {...result, reviewed_plan_sha256:input.expected_plan_sha256, + reviewed_action:input.action, executed:result.executed === true || (input.execute && + ["applied","recovered"].includes(String(result.status)))}; + } catch(error) { + return {schema_version:REVIEWED_PROMOTION_OPERATION_RESULT_SCHEMA, + status:"failed",executed:false,reason_code:"invalid_reviewed_promotion_plan", + reason:error instanceof Error ? error.message : "reviewed promotion is unavailable", + legacy_writer_fenced:false,legacy_fallback_used:false}; + } +} diff --git a/loopx/control_plane/coordination/local_authority_shadow.ts b/loopx/control_plane/coordination/local_authority_shadow.ts index 15ad4e57a..a60e197c8 100644 --- a/loopx/control_plane/coordination/local_authority_shadow.ts +++ b/loopx/control_plane/coordination/local_authority_shadow.ts @@ -410,6 +410,7 @@ const ENTRY_FIELDS = new Set([ "resolution", ]); const READ_REQUEST_FIELDS = new Set([ + "read_model", "receipt_operation_id", "schema_version", "runtime_root", @@ -489,6 +490,7 @@ export interface LocalAuthorityShadowCommitEntryResult extends JsonObject { } interface ReadRequest { + read_model: "full" | "proof"; receipt_operation_id: string | null; runtime_root: string; goal_id: string; @@ -651,6 +653,9 @@ function decodeReadRequest(value: unknown): ReadRequest { throw new EffectRuntimeRequestError(`scan_limit must be between 0 and ${MAX_SCAN_LIMIT}`); } return { + read_model: request.read_model === undefined || request.read_model === "full" ? "full" + : request.read_model === "proof" ? "proof" + : (() => {throw new EffectRuntimeRequestError("read_model must be full or proof");})(), receipt_operation_id: optionalString(request.receipt_operation_id, "receipt_operation_id"), runtime_root: requireNonEmptyString(request.runtime_root, "runtime_root"), goal_id: requireGoalId(request.goal_id), @@ -1405,12 +1410,13 @@ function loadedReadResult( base: JsonObject, storeIdentity: string, loaded: Extract, + includeHead: boolean, ): JsonObject { const result: JsonObject = { ...base, status: loaded.status, store_identity: storeIdentity }; if (loaded.status === "loaded") { result.provider_revision = loaded.provider_revision; result.cursor = loaded.cursor; - result.head = structuredClone(loaded.head); + result.head = includeHead ? structuredClone(loaded.head) : null; result.head_digest = localAuthorityShadowHeadDigest(loaded.head); result.partitions = partitionsOf(loaded.head); } @@ -1485,7 +1491,7 @@ export async function readLocalAuthorityShadow( store_identity: identity.store_identity, }; } - const result = loadedReadResult(base, identity.store_identity, loaded); + const result = loadedReadResult(base, identity.store_identity, loaded, request.read_model === "full"); if (request.store_kind === "runtime_shadow" && loaded.status === "loaded" && loaded.head.capture_profile !== "file_outbox_v1") { result.eligible = false; result.reason_code = "legacy_lineage_ineligible"; @@ -1499,13 +1505,16 @@ export async function readLocalAuthorityShadow( bootstrap_provider_revision: binding.bootstrap_provider_revision, last_sequences: lineage.last_sequences, last_applied_sequences: lineage.last_applied_sequences, - transactions: structuredClone(lineage.transactions.filter((transaction) => + transactions: lineage.transactions.filter((transaction) => request.scan_after_cursor === null || Number(transaction.cursor) > Number(request.scan_after_cursor) - ).slice(0, request.scan_limit)) as unknown as JsonObject[], - receipt: receipt === null ? null : structuredClone(receipt) as unknown as JsonObject, + ).slice(0, request.scan_limit).map(transaction => request.read_model === "proof" + ? scanTransactionView(transaction) : structuredClone(transaction) as unknown as JsonObject), + receipt: receipt === null ? null : request.read_model === "proof" + ? scanTransactionView(receipt) : structuredClone(receipt) as unknown as JsonObject, }; } - return request.scan_limit > 0 ? await appendScanPage(store, request, result) : result; + return request.read_model === "full" && request.scan_limit > 0 + ? await appendScanPage(store, request, result) : result; } catch (error) { const raw = error as { reason_code?: string; code?: string }; return { ...base, status: "failed", reason_code: raw.reason_code ?? raw.code ?? "provider_call_failed" }; diff --git a/loopx/control_plane/coordination/local_authority_shadow_adapter.py b/loopx/control_plane/coordination/local_authority_shadow_adapter.py index 4b10ce3a5..4f973a0ba 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_adapter.py +++ b/loopx/control_plane/coordination/local_authority_shadow_adapter.py @@ -417,6 +417,7 @@ def read_local_authority_shadow( scan_after_cursor: str | None = None, scan_limit: int = 0, receipt_operation_id: str | None = None, + read_model: str = "full", ) -> dict[str, Any]: """Read-only candidate view through the TypeScript store boundary.""" @@ -430,6 +431,7 @@ def read_local_authority_shadow( "scan_after_cursor": scan_after_cursor, "scan_limit": scan_limit, "receipt_operation_id": receipt_operation_id, + "read_model": read_model, }, timeout=15.0, ) @@ -520,6 +522,7 @@ def _proof(self) -> tuple[dict[str, Any], list[dict[str, Any]]]: runtime_root=self._runtime_root, goal_id=self._goal_id, scan_limit=10_000, + read_model="proof", ) proof = view.get("proof") if view.get("status") != "loaded" or not isinstance(proof, dict): @@ -724,7 +727,7 @@ def _reconcile( def _cursor_digest(self, transaction: dict[str, Any]) -> str | None: # The native history validator owns this applied-mutation marker. # Settled no-ops advance position but never synthesize a baseline digest. - marker = transaction["projection"]["partitions"][self._partition] + marker = transaction["projection_partitions"][self._partition] return None if marker is None else marker["partition_digest"] def _resolve( diff --git a/loopx/control_plane/coordination/local_authority_shadow_outbox.py b/loopx/control_plane/coordination/local_authority_shadow_outbox.py index 620d43ae7..43455c80e 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_outbox.py +++ b/loopx/control_plane/coordination/local_authority_shadow_outbox.py @@ -580,7 +580,8 @@ def _proved_sequence(runtime_root: Path, goal_id: str, partition: str) -> int: "schema_version": LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, "runtime_root": str(runtime_root), "goal_id": goal_id, - "scan_limit": 10_000, + "scan_limit": 0, + "read_model": "proof", }, timeout=15.0, ) diff --git a/loopx/control_plane/coordination/promotion_receipt.ts b/loopx/control_plane/coordination/promotion_receipt.ts new file mode 100644 index 000000000..e765b56ea --- /dev/null +++ b/loopx/control_plane/coordination/promotion_receipt.ts @@ -0,0 +1,83 @@ +/** One durable readback contract for reviewed and already-fenced promotion. + * The original transaction proves cutover even after later canonical writes. */ +import type { JsonObject } from "../effect_program.ts"; +import type { AuthorityStore, AuthorityStoreCommitResult } from "./authority_store.ts"; +import { canonicalAuthorityBytes, canonicalAuthoritySha256 } from "./authority_store_codec.ts"; + +export interface PromotionCommitIdentity { + operation_id: string; + receipt: JsonObject; + projection_sha256: string; +} +export type PromotionReadback = + | { matched: true; provider_revision: string; cursor: string } + | { matched: false; reason_code: string }; + +export async function readPromotionReceipt( + store: AuthorityStore, + identity: PromotionCommitIdentity, +): Promise { + const found = await store.readReceipt(identity.operation_id); + if (found.status === "failed" || found.status === "unavailable") { + return {matched: false, reason_code: found.reason_code}; + } + if ( + found.status !== "found" || + found.receipts.length !== 1 || + !canonicalAuthorityBytes(found.receipts[0]).equals(canonicalAuthorityBytes(identity.receipt)) + ) { + return { + matched: false, + reason_code: + found.status === "found" + ? "local_authority_promotion_identity_mismatch" + : "local_authority_promotion_receipt_missing", + }; + } + const lineage = await store.scanCommitted(null, 1); + if (lineage.status !== "page") return {matched: false, reason_code: lineage.reason_code}; + const first = lineage.transactions[0]; + if ( + first === undefined || + first.cursor !== "1" || + first.operation_id !== identity.operation_id || + first.provider_revision !== found.provider_revision || + first.cursor !== found.cursor || + first.receipts.length !== 1 || + !canonicalAuthorityBytes(first.receipts[0]).equals(canonicalAuthorityBytes(identity.receipt)) || + canonicalAuthoritySha256(first.projection) !== identity.projection_sha256 + ) { + return { matched: false, reason_code: "local_authority_promotion_lineage_mismatch" }; + } + return { matched: true, provider_revision: found.provider_revision, cursor: found.cursor }; +} + +export interface PromotionCommitReadback { + commit: AuthorityStoreCommitResult | null; + readback: PromotionReadback; + interrupted: boolean; +} + +/** A thrown transport response does not establish that the commit failed. + * Read the durable identity once; never issue a second business commit here. */ +export async function commitPromotionAndReadBack( + store: AuthorityStore, + identity: PromotionCommitIdentity, + projection: JsonObject, + event: JsonObject, +): Promise { + let commit: AuthorityStoreCommitResult | null = null; + let interrupted = false; + try { + commit = await store.commitAuthority({ + expected_provider_revision: null, + operation_id: identity.operation_id, + events: [event], + next_projection: projection, + receipts: [identity.receipt], + }); + } catch { + interrupted = true; + } + return { commit, interrupted, readback: await readPromotionReceipt(store, identity) }; +} diff --git a/loopx/control_plane/coordination/promotion_review.py b/loopx/control_plane/coordination/promotion_review.py new file mode 100644 index 000000000..304eff3ba --- /dev/null +++ b/loopx/control_plane/coordination/promotion_review.py @@ -0,0 +1,43 @@ +"""Host IO adapter for the typed, saved-plan promotion operation. + +This module deliberately does not interpret the review carrier, invent an +operation id, or decide whether a persisted fence permits recovery. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from ..effect_runtime import effect_runtime_result + + +def execute_reviewed_coordination_promotion( + *, + reviewed_plan: Mapping[str, Any], + runtime_root: Path, + goal_id: str, + action: str, + execute: bool, + projection: Mapping[str, Any] | None = None, + source_snapshot: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + request: dict[str, Any] = { + "schema_version": "loopx_reviewed_coordination_promotion_operation_v0", + "action": action, + "runtime_root": str(runtime_root.expanduser().absolute()), + "goal_id": goal_id, + "reviewed_plan": dict(reviewed_plan), + "execute": execute, + } + if projection is not None: + request["projection"] = dict(projection) + if source_snapshot is not None: + request["source_snapshot"] = dict(source_snapshot) + result = effect_runtime_result( + "coordination.local_authority.promotion_reviewed", request, timeout=30.0 + ) + if not isinstance(result, dict): + raise ValueError("reviewed promotion runtime returned an invalid result") + return result diff --git a/loopx/control_plane/coordination/reviewed_promotion_plan.ts b/loopx/control_plane/coordination/reviewed_promotion_plan.ts new file mode 100644 index 000000000..26f0748d8 --- /dev/null +++ b/loopx/control_plane/coordination/reviewed_promotion_plan.ts @@ -0,0 +1,102 @@ +/** Portable operator input, not a grant or a second migration-state store. */ +import type { JsonObject } from "../effect_program.ts"; +import { requireJsonObject, requireBoolean } from "../runtime_decode.ts"; +import { canonicalAuthorityObject, requireAuthorityStoreId } from "./authority_store_codec.ts"; + +export const REVIEWED_PROMOTION_PLAN_SCHEMA = "loopx_reviewed_coordination_promotion_v0"; +export const REVIEWED_PROMOTION_OPERATION_SCHEMA = "loopx_reviewed_coordination_promotion_operation_v0"; +export const REVIEWED_PROMOTION_OPERATION_RESULT_SCHEMA = + "loopx_reviewed_coordination_promotion_operation_result_v0"; +export type ReviewedPromotionAction = "apply" | "recover"; + +export function promotionPlanDigest(value: unknown): string { + if (typeof value !== "string" || !/^[a-f0-9]{64}$/u.test(value)) { + throw new TypeError("reviewed promotion plan digest must be a lowercase SHA-256"); + } + return value; +} + +export function reviewedPromotionPlan(request: unknown, digest: string): JsonObject { + return { + schema_version: REVIEWED_PROMOTION_PLAN_SCHEMA, + promotion_plan_sha256: promotionPlanDigest(digest), + request: canonicalAuthorityObject(request, "reviewed promotion request"), + }; +} + +interface ReviewedPromotionIdentity { + runtime_root: string; + goal_id: string; + execute: boolean; + request: JsonObject; + expected_plan_sha256: string; +} +export type ReviewedPromotionOperation = ReviewedPromotionIdentity & + ( + | { action: "apply"; projection: JsonObject; source_snapshot: JsonObject } + | { action: "recover"; projection?: never; source_snapshot?: never } + ); + +function exactKeys(row: JsonObject, required: string[], optional: string[], label: string): void { + if ( + required.some((key) => !(key in row)) || + Object.keys(row).some((key) => ![...required, ...optional].includes(key)) + ) { + throw new TypeError(`${label} has unsupported or missing fields`); + } +} + +/** Accept the exact envelope or an unchanged public CLI preview response. + * Never recursively search arbitrary JSON for a plausible plan. */ +export function decodeReviewedPromotionOperation(value: unknown): ReviewedPromotionOperation { + const input = requireJsonObject(value, "reviewed promotion operation"); + exactKeys( + input, + ["schema_version", "action", "runtime_root", "goal_id", "execute", "reviewed_plan"], + ["projection", "source_snapshot"], + "reviewed promotion operation", + ); + if (input.schema_version !== REVIEWED_PROMOTION_OPERATION_SCHEMA) + throw new TypeError("reviewed promotion operation schema mismatch"); + if (input.action !== "apply" && input.action !== "recover") + throw new TypeError("reviewed promotion action must be apply or recover"); + const raw = requireJsonObject(input.reviewed_plan, "reviewed promotion input"); + let envelope = raw; + if (raw.schema_version === "loopx_coordination_shadow_admin_v0") { + if (raw.action !== "promote" || raw.ok !== true || raw.goal_id !== input.goal_id) + throw new TypeError("reviewed CLI preview does not match this Goal"); + const promotion = requireJsonObject(raw.promotion, "promotion preview"); + const plan = requireJsonObject(promotion.plan, "promotion plan"); + envelope = requireJsonObject(plan.reviewed_plan, "reviewed promotion plan"); + } + exactKeys(envelope, ["schema_version", "promotion_plan_sha256", "request"], [], "reviewed promotion plan"); + if (envelope.schema_version !== REVIEWED_PROMOTION_PLAN_SCHEMA) + throw new TypeError("reviewed promotion plan schema mismatch"); + const request = canonicalAuthorityObject(envelope.request, "reviewed promotion request"); + const goal = requireAuthorityStoreId(input.goal_id, "goal id"); + const root = requireAuthorityStoreId(input.runtime_root, "runtime root"); + if (request.goal_id !== goal || request.runtime_root !== root) + throw new TypeError("reviewed promotion plan addresses a different Goal or runtime root"); + if ("execute" in request) throw new TypeError("reviewed promotion plan cannot carry execution authority"); + const projection = + input.projection === undefined + ? undefined + : canonicalAuthorityObject(input.projection, "current projection"); + const snapshot = + input.source_snapshot === undefined + ? undefined + : canonicalAuthorityObject(input.source_snapshot, "current source snapshot"); + if (input.action === "recover" && (projection !== undefined || snapshot !== undefined)) + throw new TypeError("promotion recovery cannot use a new legacy source"); + const identity = { + runtime_root: root, + goal_id: goal, + execute: requireBoolean(input.execute, "execute"), + request, + expected_plan_sha256: promotionPlanDigest(envelope.promotion_plan_sha256), + }; + if (input.action === "recover") return { ...identity, action: "recover" }; + if (projection === undefined || snapshot === undefined) + throw new TypeError("reviewed promotion apply requires a fresh source observation"); + return { ...identity, action: "apply", projection, source_snapshot: snapshot }; +} diff --git a/loopx/control_plane/coordination/runtime_shadow.ts b/loopx/control_plane/coordination/runtime_shadow.ts index 0286e27f7..563b58a97 100644 --- a/loopx/control_plane/coordination/runtime_shadow.ts +++ b/loopx/control_plane/coordination/runtime_shadow.ts @@ -252,7 +252,21 @@ export async function qualifyCoordinationRuntimeShadowUnderLocks( minimum: number, required: string[], ): Promise { - await verifyShadowSourceSnapshot(request); + await verifyShadowSourceSnapshot(request); + const result = await qualifyCoordinationShadowLineageUnderLocks(request, dependencies, minimum, required); + await verifyShadowSourceSnapshot(request); + return result; +} + +/** Revalidate durable capture lineage under the maintenance lock. Callers must + * either verify a fresh source snapshot or first verify the exact durable fence. + * A fenced recovery cannot require a legacy document that is no longer authority. */ +export async function qualifyCoordinationShadowLineageUnderLocks( + request: Pick, + dependencies: RuntimeShadowDependencies, + minimum: number, + required: string[], +): Promise { const store = dependencies.createStore?.(join(request.runtime_root, "authority-shadow", "file-v0"), request.goal_id) ?? new FileAuthorityStore(join(request.runtime_root, "authority-shadow", "file-v0"), request.goal_id, { existingOnly: true }); const initial = await store.loadAuthority(); @@ -260,7 +274,6 @@ export async function qualifyCoordinationRuntimeShadowUnderLocks( const binding = await requireShadowCaptureBinding(request.runtime_root, request.goal_id); const lineage = await loadValidatedShadowLineage(store, request.runtime_root, request.goal_id, binding); const pending = await pendingOutbox(request.runtime_root, request.goal_id, binding, lineage.transactions); - await verifyShadowSourceSnapshot(request); const matched = localAuthorityShadowHeadDigest(request.projection) === localAuthorityShadowHeadDigest(lineage.head.head); const missing = required.filter((kind) => !lineage.write_classes.includes(kind)); const operations = lineage.transactions.slice(1).filter((transaction) => transaction.receipts[0]?.no_op === false).length; diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 4d509c101..2fa4ff5c4 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -161,6 +161,7 @@ import { pollLocalCoordinationMonitor, promoteLocalCoordinationAuthority, reviewLocalCoordinationAuthorityPromotion, + executeReviewedCoordinationPromotion, terminalLifecycleLocalCoordinationTodo, } from "./coordination/local_authority_runtime.ts"; import {listLocalCoordinationTodos, readLocalCoordinationTodo} from "./coordination/local_authority_read.ts"; @@ -532,6 +533,7 @@ export function createEffectRuntimeHandlers( ["coordination.runtime_shadow.rollback", rollbackCoordinationRuntimeShadow], ["coordination.local_authority.promote", promoteLocalCoordinationAuthority], ["coordination.local_authority.promotion_review", reviewLocalCoordinationAuthorityPromotion], + ["coordination.local_authority.promotion_reviewed", executeReviewedCoordinationPromotion], ["coordination.local_authority.todo_continuation", continueLocalTodo], ["coordination.local_authority.todo_claim", claimLocalCoordinationTodo], ["coordination.local_authority.todo_create", createLocalCoordinationTodo], diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index ce41e9b7e..b62b5b8e6 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson. | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair | | --- | --- | --- | --- | --- | +| `shadow_proof_transport_amplification` | A large Goal cannot capture its first mutation or finish drain although the same tiny Goal succeeds; RPC rejects an oversized response. | Same source population, base/head serialized response sizes, actual sequence/drain callers, qualified lineage and cursor readback. | A consumer needing progress or partition markers received the full head and every historical projection across the language boundary. | Keep full history verification in the typed owner and return a purpose-specific compact proof. Preserve receipts, sequence and lineage checks; do not raise transport limits, truncate source records or weaken qualification to make the test pass. Cover the old oversized response and real CLI capture, drain and reviewed cutover on a disposable snapshot. | | `qualification_host_contract_mismatch` | A model stops on ordinary inspection, shell composition or draft correction and the result is reported as a semantic-control failure. | Actual synthetic operation, advertised tool contract, OS isolation, subprocess exit status, returned diagnostics and durable writeback attempts. | A shell-labelled host imposed a separate command language or ended execution without returning normal tool errors. | Use a normal shell inside an isolated execution environment, with real CLI effects supervised at their existing authority boundary. Observe source evidence and durable outcomes instead of requiring a command spelling or read ritual. Return errors within a disclosed scenario budget; keep original inputs and authority stores protected. Separate host rejection, budget exhaustion and core semantic admission, retaining earlier failures. Do not insert model answers, waive evidence or grow a command whitelist one failed trajectory at a time. | | `replan_projection_discharge_mismatch` | A projected refresh cannot satisfy its own vision obligation, or a required-vision qualification executes an ordinary repeated-progress fixture. | Typed trigger kinds, required outcomes, projected input/authoring contract, actual actor fixture, bound durable writeback, checkpoint and post-settlement readback. | Projection and admission encoded separate outcome rules; a successful narrow action was mistaken for a complete journey. | Share outcome matching and executable input projection at the typed owner. Preserve original-Turn recovery, authoring limits and legal successor/terminal exits. Match qualification scope to the fixture; distinguish accepted writeback, checkpoint and one-spend settlement. A new successor duty is not a failed closeout. Do not add ACK rituals, weaken acceptance or replay until green. Full closeout and one action need not share a call cap: an authorized, cost-disclosed budget revision is a new qualification and must retain old failures, not relabel them. | | `delegation_runtime_discovery_split` | A coordinator knows a child-count or model preference but cannot discover requester-authorized managed routes, so a direct SDK experiment is mistaken for formal LoopX delegation or the task stays unnecessarily serial. | Goal orchestration readback, requester-scoped binding directory, Turn host/profile readiness, operation journal, signed `before_plan`/`before_delegate`/`after_delegate_result` context, and parent artifact validation. | Execution grants and runtime facts existed only behind the dispatch CLI/MCP while planning consumed a separate prompt or model-preference snapshot; adding provider names to one skill would create another scheduler/config owner. | Store only an ignored Goal-local pointer to the existing operator binding file, project bounded public-safe requester routes through the generic capability context, and keep runtime readiness separate from route selection and task adoption. Recheck runtime/model/budget at dispatch, forbid silent substitution, and reconcile original operation receipts after return. Do not require every heartbeat to use every route or copy credentials/host arguments into registry, frontend, Lark or prompts. | diff --git a/tests/control_plane/test_reviewed_promotion_cli.py b/tests/control_plane/test_reviewed_promotion_cli.py new file mode 100644 index 000000000..761d1bdbe --- /dev/null +++ b/tests/control_plane/test_reviewed_promotion_cli.py @@ -0,0 +1,230 @@ +"""Use the installed CLI transport and real local stores through cutover and recovery.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess + +import pytest + +from tests.control_plane.test_local_authority_shadow_cli_e2e import ( + _workspace, + _cli, + _command, + _env, + REPO_ROOT, +) + + +def command_result(registry, root, *args): + process = subprocess.run( + _command(registry, root, *args), + cwd=REPO_ROOT, + env=_env(), + capture_output=True, + text=True, + timeout=60, + ) + return process.returncode, json.loads(process.stdout) + + +def prepare(tmp_path: Path, provider: str): + registry, state, root = _workspace(tmp_path, goal_id="goal-a") + if provider == "sqlite": + subprocess.run( + [ + os.environ.get("LOOPX_CONTROL_PLANE_NODE", "node"), + "--no-warnings", + "--experimental-strip-types", + "--experimental-sqlite", + "loopx/control_plane/coordination/local_authority_provider.ts", + "--runtime-root", + str(root), + "--goal-id", + "goal-a", + "--execute", + ], + cwd=REPO_ROOT, + check=True, + capture_output=True, + text=True, + ) + _cli( + registry, + root, + "configure-goal", + "--goal-id", + "goal-a", + "--coordination-runtime-shadow-file", + "--execute", + ) + bootstrap = _cli( + registry, + root, + "coordination-shadow", + "bootstrap", + "--goal-id", + "goal-a", + "--execute", + ) + assert bootstrap["bootstrap"]["status"] == "applied" + for index in range(3): + _cli( + registry, + root, + "todo", + "add", + "--goal-id", + "goal-a", + "--role", + "agent", + "--text", + f"Preserve migration record {index}", + ) + preview = _cli( + registry, root, "coordination-shadow", "promote", "--goal-id", "goal-a" + ) + assert preview["promotion"]["status"] == "preview_ready", preview + saved = tmp_path / "reviewed.json" + saved.write_text(json.dumps(preview), encoding="utf-8") + return registry, state, root, saved, preview + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_saved_plan_cutover_and_recovery_after_canonical_write_and_missing_legacy( + tmp_path, provider +): + registry, state, root, saved, preview = prepare(tmp_path, provider) + args = ( + "coordination-shadow", + "promote", + "--goal-id", + "goal-a", + "--reviewed-plan", + str(saved), + ) + dry_run = _cli(registry, root, *args) + assert dry_run["promotion"]["status"] == "preview_ready" + assert dry_run["executed"] is False + applied = _cli(registry, root, *args, "--execute") + assert applied["promotion"]["status"] == "applied", applied + assert applied["promotion"]["canonical_authority"] == f"{provider}_v0" + assert ( + applied["promotion"]["promotion_plan_sha256"] + == preview["promotion"]["plan"]["promotion_plan_sha256"] + ) + _cli( + registry, + root, + "todo", + "add", + "--goal-id", + "goal-a", + "--role", + "agent", + "--text", + "Continue after provider cutover", + ) + state.unlink() + # Recovery belongs to the durable cutover, not the transient shadow opt-in. + config = json.loads(registry.read_text()) + config["goals"][0]["coordination"].pop("runtime_shadow", None) + registry.write_text(json.dumps(config)) + for execution in [(), ("--execute",)]: + replay = _cli( + registry, + root, + "coordination-shadow", + "recover-promotion", + "--goal-id", + "goal-a", + "--reviewed-plan", + str(saved), + *execution, + ) + assert replay["promotion"]["status"] == "replayed", replay + assert replay["promotion"]["cursor"] == "1" + assert ( + replay["promotion"]["provider_revision"] + == applied["promotion"]["provider_revision"] + ) + assert replay["executed"] is False + assert not state.exists() + + +def test_saved_plan_source_drift_does_not_freeze_legacy_writes(tmp_path): + registry, _state, root, saved, _preview = prepare(tmp_path, "file") + _cli( + registry, + root, + "todo", + "add", + "--goal-id", + "goal-a", + "--role", + "agent", + "--text", + "New work before cutover", + ) + code, result = command_result( + registry, + root, + "coordination-shadow", + "promote", + "--goal-id", + "goal-a", + "--reviewed-plan", + str(saved), + "--execute", + ) + assert code == 1 + assert result["promotion"]["reason_code"] == "local_authority_reviewed_plan_changed" + assert result["promotion"]["legacy_writer_fenced"] is False + _cli( + registry, + root, + "todo", + "add", + "--goal-id", + "goal-a", + "--role", + "agent", + "--text", + "Legacy writer remains usable", + ) + + +def test_saved_plan_rejects_policy_override_and_recovery_without_fence(tmp_path): + registry, _state, root, saved, _preview = prepare(tmp_path, "file") + code, result = command_result( + registry, + root, + "coordination-shadow", + "promote", + "--goal-id", + "goal-a", + "--reviewed-plan", + str(saved), + "--minimum-operations", + "1", + "--execute", + ) + assert code == 1 and "owns qualification policy" in result["error"] + code, result = command_result( + registry, + root, + "coordination-shadow", + "recover-promotion", + "--goal-id", + "goal-a", + "--reviewed-plan", + str(saved), + "--execute", + ) + assert code == 1 + assert ( + result["promotion"]["reason_code"] + == "local_authority_writer_fence_not_verified" + ) diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts index 24b946abd..3e0d4ec11 100644 --- a/tests/control_plane_ts/authority_store_conformance.ts +++ b/tests/control_plane_ts/authority_store_conformance.ts @@ -1,5 +1,6 @@ import {registerPeriodicReportConformance} from "./periodic_report_conformance.ts"; import {registerIssueFixMonitorReconciliationConformance} from "./issue_fix_monitor_reconciliation_conformance.ts"; +import {registerPromotionRecoveryConformance} from "./promotion_recovery_conformance.ts"; import {registerTodoConsumerScopeConformance} from "./todo_consumer_scope_conformance.ts"; import {registerProjectionConfirmationConformance} from "./projection_confirmation_conformance.ts"; import {registerUserCompletionFollowthroughConformance} from "./user_completion_followthrough_conformance.ts"; @@ -266,6 +267,7 @@ export function registerAuthorityStoreConformance( registerCoordinationReceiptConformance(providerName, factory); registerAuthoritySourceConformance(providerName, factory); registerHandoffModeConformance(providerName, factory); + registerPromotionRecoveryConformance(providerName, factory); for (const native of [false, true]) test(`${providerName} conformance: standing revocation survives canonical ordering and archive (${native ? "native" : "legacy"})`, async (t) => { const {store} = await factory(t); const goal = "goal-standing"; diff --git a/tests/control_plane_ts/local_authority_provider.test.ts b/tests/control_plane_ts/local_authority_provider.test.ts index 549228471..a7dbb43cd 100644 --- a/tests/control_plane_ts/local_authority_provider.test.ts +++ b/tests/control_plane_ts/local_authority_provider.test.ts @@ -131,6 +131,13 @@ function providerCalls(directory: string, revision: string, dryRun: boolean) { {...witnessed, schema_version: runtime.LOCAL_COORDINATION_TODO_TERMINAL_LIFECYCLE_WITNESSED_REQUEST_SCHEMA}], archiveLocalCoordinationTodos: [{...input, schema_version: runtime.LOCAL_COORDINATION_TODO_ARCHIVE_REQUEST_SCHEMA, max_active_done: 0}], acknowledgeLocalCoordinationTodoArchive: [{...input, schema_version: runtime.LOCAL_COORDINATION_TODO_ARCHIVE_ACK_REQUEST_SCHEMA}], + executeReviewedCoordinationPromotion: [{ + schema_version:"loopx_reviewed_coordination_promotion_operation_v0",action:"recover", + runtime_root:directory,goal_id:"goal-a",execute:!dryRun, + reviewed_plan:{schema_version:"loopx_reviewed_coordination_promotion_v0", + promotion_plan_sha256:runtime.localCoordinationPromotionPlanSha256(promotionRequest(directory,{},"file:synthetic:1")), + request:promotionRequest(directory,{},"file:synthetic:1")}, + }], promoteLocalCoordinationAuthority: [promotionRequest(directory, {}, "file:synthetic:1")], reviewLocalCoordinationAuthorityPromotion: [{ schema_version: runtime.LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, @@ -348,7 +355,7 @@ for (const provider of ["file", "sqlite"] as const) { const directory = await root(t); if (provider === "sqlite") await selectLocalSqliteAuthority(directory, "goal-a", true); const canonical = await openLocalAuthorityStore(directory, "goal-a"); - const shadow = await qualifiedShadow(directory); + const shadow = await qualifiedShadow(directory, "hard_lease"); const shadowStore = new FileAuthorityStore(join(directory, "authority-shadow", "file-v0"), "goal-a"); const canonicalAuthority = provider === "sqlite" ? "sqlite_v0" : "file_v0"; const request = promotionRequest( @@ -370,6 +377,11 @@ for (const provider of ["file", "sqlite"] as const) { canonicalAuthority, )); } + if (phase === "qualification") { + // An otherwise valid fenced source has only one verified mutation. + request.minimum_operations = 2; + request.writer_fence.promotion_plan_sha256 = runtime.localCoordinationPromotionPlanSha256(request); + } if (phase !== "fence_missing") await engageFence(request); const fencePath = legacyCoordinationWriterFencePath(directory, "goal-a"); if (phase === "fence_corrupt") await writeFile(fencePath, "{"); diff --git a/tests/control_plane_ts/local_authority_runtime.test.ts b/tests/control_plane_ts/local_authority_runtime.test.ts index 25727def3..728a328c1 100644 --- a/tests/control_plane_ts/local_authority_runtime.test.ts +++ b/tests/control_plane_ts/local_authority_runtime.test.ts @@ -188,14 +188,14 @@ test("legacy write guard flips from allowed to fail-closed after the durable fen assert.equal(blocked.authority_mode, "file_v0"); }); -test("new file outbox qualification does not implicitly enable canonical promotion", async () => { +test("fenced v0 recovery cannot bypass reviewed promotion handoff-mode policy", async () => { const root = await mkdtemp(join(tmpdir(), "loopx-local-authority-promote-")); const shadow = await qualifiedShadow(root); const request = promotionRequest(root, shadow.projection, shadow.providerRevision); await engageFence(request); const applied = await promoteLocalCoordinationAuthority(request); assert.equal(applied.status, "failed"); - assert.equal(applied.reason_code, "local_authority_shadow_not_qualified"); + assert.equal(applied.reason_code, "local_authority_promotion_requires_hard_lease"); const canonical = new FileAuthorityStore(join(root, "authority", "file-v0"), "goal-a", {existingOnly: true}); assert.equal((await canonical.loadAuthority()).status, "missing"); }); diff --git a/tests/control_plane_ts/promotion_receipt.test.ts b/tests/control_plane_ts/promotion_receipt.test.ts new file mode 100644 index 000000000..36b4f14ef --- /dev/null +++ b/tests/control_plane_ts/promotion_receipt.test.ts @@ -0,0 +1,137 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import type { + AuthorityStore, + AuthorityStoreCommit, + AuthorityStoreReceiptResult, + AuthorityStoreScanResult, +} from "../../loopx/control_plane/coordination/authority_store.ts"; +import { FileAuthorityStore } from "../../loopx/control_plane/coordination/file_authority_store.ts"; +import { canonicalAuthoritySha256 } from "../../loopx/control_plane/coordination/authority_store_codec.ts"; +import { + commitPromotionAndReadBack, + readPromotionReceipt, +} from "../../loopx/control_plane/coordination/promotion_receipt.ts"; + +async function fixture(t: test.TestContext) { + const root = await mkdtemp(join(tmpdir(), "promotion-receipt-")); + t.after(() => rm(root, { recursive: true, force: true })); + const store = new FileAuthorityStore(root, "goal-a"); + const projection = { goal_id: "goal-a", todos: [], leases: [], handoff_mode: "hard_lease" }; + const receipt = { schema_version: "promotion-test-receipt", operation_id: "promotion", plan: "reviewed" }; + const identity = { + operation_id: "promotion", + receipt, + projection_sha256: canonicalAuthoritySha256(projection), + }; + const committed = await commitPromotionAndReadBack(store, identity, projection, { kind: "promotion" }); + assert.equal(committed.readback.matched, true); + return { store, projection, identity }; +} + +// The receipt index and first transaction are independent proof surfaces. A +// provider bug in either cannot certify cutover just because the other matches. +for (const mutation of [ + "receipt_revision", + "receipt_cursor", + "receipt_content", + "first_revision", + "first_cursor", + "first_receipt", + "first_projection", + "first_operation", + "empty_lineage", +] as const) { + test(`promotion proof rejects ${mutation} divergence`, async (t) => { + const { store, identity } = await fixture(t); + const proxy = new Proxy(store, { + get(target, key) { + if (key === "readReceipt") + return async (operation: string): Promise => { + const found = await target.readReceipt(operation); + if (found.status !== "found") return found; + if (mutation === "receipt_revision") return { ...found, provider_revision: "another-revision" }; + if (mutation === "receipt_cursor") return { ...found, cursor: "2" }; + if (mutation === "receipt_content") return { ...found, receipts: [{ wrong: true }] }; + return found; + }; + if (key === "scanCommitted") + return async (): Promise => { + const page = await target.scanCommitted(null, 1); + if (page.status !== "page") return page; + const first = structuredClone(page.transactions[0]); + if (mutation === "first_revision") first.provider_revision = "another-revision"; + if (mutation === "first_cursor") first.cursor = "2"; + if (mutation === "first_receipt") first.receipts = [{ wrong: true }]; + if (mutation === "first_projection") first.projection = { wrong: true }; + if (mutation === "first_operation") first.operation_id = "another-operation"; + return { ...page, transactions: mutation === "empty_lineage" ? [] : [first] }; + }; + const member = Reflect.get(target, key); + return typeof member === "function" ? member.bind(target) : member; + }, + }); + const before = await store.loadAuthority(); + assert.equal((await readPromotionReceipt(proxy, identity)).matched, false); + assert.deepEqual(await store.loadAuthority(), before); + }); +} + +test("an unavailable proof read is not silently converted into commit absence", async (t) => { + const { store, identity } = await fixture(t); + const proxy = new Proxy(store, { + get(target, key) { + if (key === "readReceipt") + return async () => { + throw new Error("receipt transport unavailable"); + }; + const value = Reflect.get(target, key); + return typeof value === "function" ? value.bind(target) : value; + }, + }); + await assert.rejects(readPromotionReceipt(proxy, identity), /receipt transport unavailable/); +}); + +test("failed commit evidence cannot be upgraded by an unrelated successful receipt", async (t) => { + const { store, identity, projection } = await fixture(t); + let attempts = 0; + const proxy: AuthorityStore = new Proxy(store, { + get(target, key) { + if (key === "commitAuthority") + return async (_request: AuthorityStoreCommit) => { + attempts++; + throw new Error("uncertain transport outcome"); + }; + const value = Reflect.get(target, key); + return typeof value === "function" ? value.bind(target) : value; + }, + }); + const result = await commitPromotionAndReadBack( + proxy, + { ...identity, operation_id: "other-operation" }, + projection, + {}, + ); + assert.equal(result.interrupted, true); + assert.equal(result.commit, null); + assert.deepEqual(result.readback, { + matched: false, + reason_code: "local_authority_promotion_receipt_missing", + }); + assert.equal(attempts, 1, "the helper never blindly retries a business commit"); +}); + +for (const method of ["readReceipt", "scanCommitted"] as const) { + test(`typed ${method} unavailability retains its actionable provider reason`, async t => { + const {store, identity} = await fixture(t); + const proxy = new Proxy(store, {get(target, key) { + if (key === method) return async () => ({status: "unavailable", reason_code: "provider_offline", reason: "offline"}); + const member = Reflect.get(target, key); + return typeof member === "function" ? member.bind(target) : member; + }}); + assert.deepEqual(await readPromotionReceipt(proxy, identity), {matched: false, reason_code: "provider_offline"}); + }); +} diff --git a/tests/control_plane_ts/promotion_recovery_conformance.ts b/tests/control_plane_ts/promotion_recovery_conformance.ts new file mode 100644 index 000000000..e46465258 --- /dev/null +++ b/tests/control_plane_ts/promotion_recovery_conformance.ts @@ -0,0 +1,161 @@ +import { LOCAL_AUTHORITY_SHADOW_TRANSACTION_PROJECTION_SCHEMA } from "../../loopx/control_plane/coordination/coordination_state_contract.generated.ts"; +import assert from "node:assert/strict"; +import test from "node:test"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; +import type { AuthorityStoreConformanceFactory } from "./authority_store_conformance.ts"; +import { authorityStoreSourceAuthority } from "../../loopx/control_plane/coordination/authority_store.ts"; +import { FileAuthorityStore } from "../../loopx/control_plane/coordination/file_authority_store.ts"; +import { + bootstrapCoordinationRuntimeShadow, + COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA, +} from "../../loopx/control_plane/coordination/runtime_shadow.ts"; +import { + reviewLocalCoordinationAuthorityPromotion, + executeReviewedCoordinationPromotion, + LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, +} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import { commitLocalAuthorityShadowEntry } from "../../loopx/control_plane/coordination/local_authority_shadow.ts"; +import { canonicalAuthoritySha256 } from "../../loopx/control_plane/coordination/authority_store_codec.ts"; +import { loadLegacyCoordinationWriterFence } from "../../loopx/control_plane/coordination/legacy_writer_fence.ts"; +import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts"; +import { sourceRequest, pendingEntry, settleFiles } from "./shadow_file_fixture.ts"; + +/** Full retained Todo/lease population, then one actual outbox transaction. + * State bytes are fixture transport; no real Goal or implicit promotion is used. */ +export async function qualifiedPromotionSource(root: string, input: JsonObject) { + const projection: JsonObject = { + ...input, + schema_version: LOCAL_AUTHORITY_SHADOW_TRANSACTION_PROJECTION_SCHEMA, + source_authority: "legacy_markdown_and_task_lease", + partitions: { todos: null, leases: null }, + }; + const statePath = join(root, "ACTIVE_GOAL_STATE.md"); + await writeFile(statePath, "---\ngoal_id: goal-a\nhandoff_mode: hard_lease\n---\n\n## Agent Todo\n\n"); + const store = new FileAuthorityStore(join(root, "authority-shadow", "file-v0"), "goal-a"); + const leaseDirectory = join(root, "goals", "goal-a", "task-leases"); + await mkdir(leaseDirectory, { recursive: true }); + for (const lease of projection.leases as JsonObject[]) { + await writeFile(join(leaseDirectory, `${lease.todo_id}.json`), JSON.stringify(lease)); + } + const fixture = { root, statePath, store, baseline: projection }; + const bootstrapped = await bootstrapCoordinationRuntimeShadow({ + ...(await sourceRequest(fixture, projection)), + schema_version: COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA, + operation_id: "bootstrap:full-source", + source_version: "full-source:0", + }); + assert.equal(bootstrapped.status, "applied", JSON.stringify(bootstrapped)); + const todos = structuredClone(projection.todos) as JsonObject[]; + todos[0].note = "A verified source mutation precedes the migration review"; + const entry = await pendingEntry( + fixture, + 1, + { handoff_mode: "hard_lease", todos }, + { writeClass: "todo_update" }, + ); + const mirrored = await commitLocalAuthorityShadowEntry(entry); + assert.equal(mirrored.outcome, "delivered", JSON.stringify(mirrored)); + await settleFiles(fixture, entry, mirrored); + const loaded = await store.loadAuthority(); + assert.equal(loaded.status, "loaded"); + if (loaded.status !== "loaded") throw new Error("qualified source missing"); + const source: JsonObject = { ...loaded.head, partitions: { todos: null, leases: null } }; + for (const field of ["capture_lineage_id", "capture_profile", "source_root_digest"]) delete source[field]; + const request: JsonObject = { + ...(await sourceRequest(fixture, source)), + schema_version: LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, + operation_id: "promote:full-source", + minimum_operations: 1, + required_event_kinds: ["todo_update"], + execute: false, + }; + return { statePath, store, head: loaded.head, request }; +} + +export function registerPromotionRecoveryConformance( + name: string, + factory: AuthorityStoreConformanceFactory, +): void { + for (const schema of ["native", "legacy"] as const) { + test(`${name}: reviewed full-source migration and recovery retain exact history (${schema})`, async (context) => { + const { store, contender } = await factory(context); + const root = await mkdtemp(join(tmpdir(), "promotion-full-source-")); + context.after(() => rm(root, { recursive: true, force: true })); + const fixture = productionScaleCoordinationFixture("goal-a", schema); + const source = await qualifiedPromotionSource(root, fixture.projection); + const dependencies = { createCanonicalStore: () => store }; + const preview = await reviewLocalCoordinationAuthorityPromotion(source.request, dependencies); + assert.equal(preview.status, "preview_ready", JSON.stringify(preview)); + const envelope = (preview.plan as JsonObject).reviewed_plan as JsonObject; + const operation = { + schema_version: "loopx_reviewed_coordination_promotion_operation_v0", + runtime_root: root, + goal_id: "goal-a", + reviewed_plan: envelope, + }; + const applied = await executeReviewedCoordinationPromotion( + { + ...operation, + action: "apply", + execute: true, + projection: source.request.projection, + source_snapshot: source.request.source_snapshot, + }, + dependencies, + ); + assert.equal(applied.status, "applied", JSON.stringify(applied)); + assert.equal(applied.canonical_authority, authorityStoreSourceAuthority(store)); + const promoted = await contender.loadAuthority(); + assert.equal(promoted.status, "loaded"); + if (promoted.status !== "loaded") return; + assert.deepEqual( + promoted.head, + source.head, + "migration cannot drop unknown metadata, claims, standing decisions or leases", + ); + assert.equal((promoted.head.todos as unknown[]).length, fixture.expected_initial_todo_count); + assert.equal((promoted.head.leases as unknown[]).length, fixture.expected_current_lease_count); + const receipt = await contender.readReceipt(String(applied.operation_id)); + assert.equal(receipt.status, "found"); + // A later writer advances canonical state. Recovery must prove transaction one, + // without replacing later work or interpreting a current revision as its receipt. + const later = { ...promoted.head, recovery_conformance_marker: "later-independent-write" }; + const changed = await contender.commitAuthority({ + expected_provider_revision: promoted.provider_revision, + operation_id: "after-promotion", + next_projection: later, + events: [], + receipts: [], + }); + assert.equal(changed.status, "applied"); + await rm(source.statePath); + const head = await store.loadAuthority(); + const fence = await loadLegacyCoordinationWriterFence(root, "goal-a"); + for (const execute of [false, true]) { + const recovered = await executeReviewedCoordinationPromotion( + { ...operation, action: "recover", execute }, + dependencies, + ); + assert.equal(recovered.status, "replayed", JSON.stringify(recovered)); + assert.equal(recovered.provider_revision, applied.provider_revision); + assert.equal(recovered.cursor, "1"); + assert.equal(recovered.executed, false); + assert.deepEqual(await store.loadAuthority(), head); + assert.deepEqual(await store.readReceipt(String(applied.operation_id)), receipt); + assert.deepEqual(await loadLegacyCoordinationWriterFence(root, "goal-a"), fence); + } + const altered = structuredClone(envelope); + (altered.request as JsonObject).minimum_operations = 2; + const invalid = await executeReviewedCoordinationPromotion( + { ...operation, reviewed_plan: altered, action: "recover", execute: true }, + dependencies, + ); + assert.equal(invalid.reason_code, "invalid_reviewed_promotion_plan"); + assert.deepEqual(await store.loadAuthority(), head); + assert.notEqual(canonicalAuthoritySha256(later), canonicalAuthoritySha256(source.head)); + }); + } +} diff --git a/tests/control_plane_ts/reviewed_promotion.test.ts b/tests/control_plane_ts/reviewed_promotion.test.ts new file mode 100644 index 000000000..4b6b061a1 --- /dev/null +++ b/tests/control_plane_ts/reviewed_promotion.test.ts @@ -0,0 +1,287 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { mkdtemp, rm, readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { qualifiedShadow } from "./local_promotion_fixture.ts"; +import { sourceRequest, pendingEntry, settleFiles } from "./shadow_file_fixture.ts"; +import { FileAuthorityStore } from "../../loopx/control_plane/coordination/file_authority_store.ts"; +import { + reviewLocalCoordinationAuthorityPromotion, + executeReviewedCoordinationPromotion, + LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, +} from "../../loopx/control_plane/coordination/local_authority_runtime.ts"; +import { loadLegacyCoordinationWriterFence } from "../../loopx/control_plane/coordination/legacy_writer_fence.ts"; +import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; + +async function fixture(t: test.TestContext) { + const root = await mkdtemp(join(tmpdir(), "reviewed-promotion-")); + t.after(() => rm(root, { recursive: true, force: true })); + const shadow = await qualifiedShadow(root, "hard_lease"); + const projection = { ...shadow.projection, partitions: { todos: null, leases: null } } as JsonObject; + for (const field of ["capture_lineage_id", "capture_profile", "source_root_digest"]) + delete projection[field]; + const source = await sourceRequest( + { + root, + statePath: join(root, "ACTIVE_GOAL_STATE.md"), + store: new FileAuthorityStore(join(root, "authority-shadow", "file-v0"), "goal-a"), + baseline: projection, + }, + projection, + ); + const request = { + ...source, + schema_version: LOCAL_COORDINATION_PROMOTION_REVIEW_REQUEST_SCHEMA, + operation_id: "reviewed-promotion", + minimum_operations: 1, + required_event_kinds: ["todo_claim"], + execute: false, + }; + return { root, request, projection }; +} + +test("promotion preview exposes an executable exact-plan envelope without fencing", async (t) => { + const { root, request } = await fixture(t); + const preview = await reviewLocalCoordinationAuthorityPromotion(request); + assert.equal(preview.status, "preview_ready"); + const reviewed = (preview.plan as JsonObject).reviewed_plan as JsonObject; + assert.equal(reviewed?.schema_version, "loopx_reviewed_coordination_promotion_v0"); + assert.equal(reviewed.promotion_plan_sha256, (preview.plan as JsonObject).promotion_plan_sha256); + assert.equal((await loadLegacyCoordinationWriterFence(root, "goal-a")).status, "missing"); +}); + +test("a changed reviewed plan is rejected before writer fencing or canonical initialization", async (t) => { + const { root, request } = await fixture(t); + const result = await reviewLocalCoordinationAuthorityPromotion({ + ...request, + execute: true, + expected_promotion_plan_sha256: "0".repeat(64), + }); + assert.equal(result.status, "not_ready"); + assert.equal(result.reason_code, "local_authority_reviewed_plan_changed"); + assert.equal((await loadLegacyCoordinationWriterFence(root, "goal-a")).status, "missing"); + assert.equal( + ( + await new FileAuthorityStore(join(root, "authority", "file-v0"), "goal-a", { + existingOnly: true, + }).loadAuthority() + ).status, + "missing", + ); +}); + +const operation = ( + root: string, + plan: JsonObject, + action: "apply" | "recover", + execute: boolean, + source?: JsonObject, +) => ({ + schema_version: "loopx_reviewed_coordination_promotion_operation_v0", + action, + runtime_root: root, + goal_id: "goal-a", + execute, + reviewed_plan: plan, + ...(action === "apply" ? { projection: source!.projection, source_snapshot: source!.source_snapshot } : {}), +}); +async function reviewed(request: JsonObject, dependencies = {}) { + const result = await reviewLocalCoordinationAuthorityPromotion(request, dependencies); + assert.equal(result.status, "preview_ready", JSON.stringify(result)); + return (result.plan as JsonObject).reviewed_plan as JsonObject; +} + +test("reviewed apply uses the saved operation and recovery reads the original receipt after later writes", async (t) => { + const { root, request } = await fixture(t); + const plan = await reviewed(request); + const applied = await executeReviewedCoordinationPromotion(operation(root, plan, "apply", true, request)); + assert.equal(applied.status, "applied", JSON.stringify(applied)); + const store = new FileAuthorityStore(join(root, "authority", "file-v0"), "goal-a"); + const before = await store.loadAuthority(); + assert.equal(before.status, "loaded"); + if (before.status !== "loaded") return; + const later = await store.commitAuthority({ + operation_id: "later-canonical-operation", + expected_provider_revision: before.provider_revision, + next_projection: before.head, + events: [], + receipts: [{ kind: "later" }], + }); + assert.equal(later.status, "applied"); + await rm(join(root, "ACTIVE_GOAL_STATE.md")); + const head = await store.loadAuthority(); + for (const execute of [false, true]) { + const replay = await executeReviewedCoordinationPromotion(operation(root, plan, "recover", execute)); + assert.equal(replay.status, "replayed", JSON.stringify(replay)); + assert.equal(replay.provider_revision, applied.provider_revision); + assert.equal(replay.cursor, "1"); + assert.equal(replay.executed, false); + assert.deepEqual(await store.loadAuthority(), head); + } +}); + +test("a source change between review and execution rejects the old plan without freezing the Goal", async (t) => { + const { root, request, projection } = await fixture(t); + const plan = await reviewed(request); + const f = { + root, + statePath: join(root, "ACTIVE_GOAL_STATE.md"), + baseline: projection, + store: new FileAuthorityStore(join(root, "authority-shadow", "file-v0"), "goal-a"), + }; + const { commitLocalAuthorityShadowEntry } = await import( + "../../loopx/control_plane/coordination/local_authority_shadow.ts" + ); + const rows = structuredClone(projection.todos) as JsonObject[]; + rows[0].text = "A newer independently committed objective"; + const entry = await pendingEntry( + f, + 2, + { handoff_mode: "hard_lease", todos: rows }, + { writeClass: "todo_update" }, + ); + const committed = await commitLocalAuthorityShadowEntry(entry); + assert.equal(committed.outcome, "delivered"); + await settleFiles(f, entry, committed); + const fresh: JsonObject = { ...projection, todos: rows }; + const { coordinationTodoReadModel } = await import( + "../../loopx/control_plane/coordination/coordination_projection.ts" + ); + fresh.todo_read_model = coordinationTodoReadModel( + rows, + (projection.todo_read_model as JsonObject).schema_version, + ); + const observation = await sourceRequest(f, fresh); + const before = await readFile(f.statePath); + const rejected = await executeReviewedCoordinationPromotion( + operation(root, plan, "apply", true, observation), + ); + assert.equal(rejected.reason_code, "local_authority_reviewed_plan_changed", JSON.stringify(rejected)); + assert.equal((await loadLegacyCoordinationWriterFence(root, "goal-a")).status, "missing"); + assert.deepEqual(await readFile(f.statePath), before); +}); + +test("recovery is effect-free until execute and cannot create a missing writer fence", async (t) => { + const { root, request } = await fixture(t); + const plan = await reviewed(request); + for (const execute of [false, true]) { + const rejected = await executeReviewedCoordinationPromotion(operation(root, plan, "recover", execute)); + assert.equal(rejected.reason_code, "local_authority_writer_fence_not_verified"); + assert.equal((await loadLegacyCoordinationWriterFence(root, "goal-a")).status, "missing"); + } +}); + +test("a fence-to-commit interruption recovers with no legacy document and no new review identity", async (t) => { + const { root, request } = await fixture(t); + class InterruptOnce extends FileAuthorityStore { + interrupt = true; + override async commitAuthority( + value: import("../../loopx/control_plane/coordination/authority_store.ts").AuthorityStoreCommit, + ) { + if (this.interrupt) { + this.interrupt = false; + throw new Error("process stopped before commit"); + } + return super.commitAuthority(value); + } + } + const store = new InterruptOnce(join(root, "authority", "file-v0"), "goal-a"); + const deps = { createCanonicalStore: () => store }; + const plan = await reviewed(request, deps); + const failed = await executeReviewedCoordinationPromotion( + operation(root, plan, "apply", true, request), + deps, + ); + assert.equal(failed.status, "failed"); + assert.equal(failed.legacy_writer_fenced, true); + assert.equal((await store.loadAuthority()).status, "missing"); + await rm(join(root, "ACTIVE_GOAL_STATE.md")); + const preview = await executeReviewedCoordinationPromotion(operation(root, plan, "recover", false), deps); + assert.equal(preview.status, "recovery_ready", JSON.stringify(preview)); + assert.equal(preview.executed, false); + assert.equal((await store.loadAuthority()).status, "missing"); + const recovered = await executeReviewedCoordinationPromotion(operation(root, plan, "recover", true), deps); + assert.equal(recovered.status, "applied", JSON.stringify(recovered)); + assert.equal(recovered.operation_id, request.operation_id); +}); + +test("a thrown acknowledgement after a successful commit is resolved by durable receipt readback", async (t) => { + const { root, request } = await fixture(t); + class LoseResponse extends FileAuthorityStore { + writes = 0; + override async commitAuthority( + value: import("../../loopx/control_plane/coordination/authority_store.ts").AuthorityStoreCommit, + ): Promise { + this.writes++; + await super.commitAuthority(value); + throw new Error("response lost after durable commit"); + } + } + const store = new LoseResponse(join(root, "authority", "file-v0"), "goal-a"); + const deps = { createCanonicalStore: () => store }; + const plan = await reviewed(request, deps); + const recovered = await executeReviewedCoordinationPromotion( + operation(root, plan, "apply", true, request), + deps, + ); + assert.equal(recovered.status, "recovered", JSON.stringify(recovered)); + assert.equal(store.writes, 1); + const replay = await executeReviewedCoordinationPromotion(operation(root, plan, "recover", true), deps); + assert.equal(replay.status, "replayed"); + assert.equal(store.writes, 1); +}); + +for (const mutation of [ + "goal", + "runtime", + "digest", + "policy", + "source", + "operation", + "embedded_execute", + "unknown_envelope", +] as const) { + test(`review carrier rejects changed ${mutation} before any provider opens`, async (t) => { + const { root, request } = await fixture(t); + const plan = await reviewed(request); + const changed = structuredClone(plan); + const row = changed.request as JsonObject; + if (mutation === "goal") row.goal_id = "other"; + if (mutation === "runtime") row.runtime_root = root + "-other"; + if (mutation === "digest") changed.promotion_plan_sha256 = "f".repeat(64); + if (mutation === "policy") row.minimum_operations = 2; + if (mutation === "source") row.expected_shadow_projection_sha256 = "f".repeat(64); + if (mutation === "operation") row.operation_id = "another-operation"; + if (mutation === "embedded_execute") row.execute = true; + if (mutation === "unknown_envelope") changed.allow_unqualified = true; + let opens = 0; + const result = await executeReviewedCoordinationPromotion(operation(root, changed, "recover", true), { + createCanonicalStore: () => { + opens++; + throw new Error("must reject before opening"); + }, + }); + assert.equal(result.reason_code, "invalid_reviewed_promotion_plan"); + assert.equal(opens, 0); + assert.equal((await loadLegacyCoordinationWriterFence(root, "goal-a")).status, "missing"); + }); +} + +test("the unmodified CLI preview envelope is accepted and cross-Goal wrappers are rejected", async (t) => { + const { root, request } = await fixture(t); + const preview = await reviewLocalCoordinationAuthorityPromotion(request); + const cli = { + schema_version: "loopx_coordination_shadow_admin_v0", + action: "promote", + ok: true, + goal_id: "goal-a", + promotion: preview, + }; + const result = await executeReviewedCoordinationPromotion(operation(root, cli, "apply", false, request)); + assert.equal(result.status, "preview_ready"); + const rejected = await executeReviewedCoordinationPromotion( + operation(root, { ...cli, goal_id: "other" }, "apply", true, request), + ); + assert.equal(rejected.reason_code, "invalid_reviewed_promotion_plan"); +}); diff --git a/tests/control_plane_ts/shadow_compact_proof.test.ts b/tests/control_plane_ts/shadow_compact_proof.test.ts new file mode 100644 index 000000000..338567f13 --- /dev/null +++ b/tests/control_plane_ts/shadow_compact_proof.test.ts @@ -0,0 +1,83 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; +import { coordinationTodoReadModel } from "../../loopx/control_plane/coordination/coordination_projection.ts"; +import { readLocalAuthorityShadow } from "../../loopx/control_plane/coordination/local_authority_shadow.ts"; +import { LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA } from "../../loopx/control_plane/coordination/coordination_state_contract.generated.ts"; +import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts"; +import { qualifiedPromotionSource } from "./promotion_recovery_conformance.ts"; + +for (const shape of ["native", "legacy"] as const) { + test(`compact shadow proof keeps full lineage decisions below the RPC response budget (${shape})`, async (t) => { + const root = await mkdtemp(join(tmpdir(), "compact-shadow-proof-")); + t.after(() => rm(root, { recursive: true, force: true })); + const fixture = productionScaleCoordinationFixture("goal-a", shape); + const rows = fixture.projection.todos as JsonObject[]; + // Broad retained records reproduce repeated-projection transport pressure. + // No population, history or stored metadata is removed to meet the budget. + for (const row of rows) row.note = "Retained public fixture metadata. ".repeat(35); + fixture.projection.todo_read_model = coordinationTodoReadModel( + rows, + (fixture.projection.todo_read_model as JsonObject).schema_version as string, + ); + const source = await qualifiedPromotionSource(root, fixture.projection); + const request = { + schema_version: LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, + runtime_root: root, + goal_id: "goal-a", + scan_limit: 10000, + }; + const full = await readLocalAuthorityShadow(request); + const compact = await readLocalAuthorityShadow({ ...request, read_model: "proof" }); + assert.equal(full.status, "loaded"); + assert.equal(compact.status, "loaded"); + assert.ok( + Buffer.byteLength(JSON.stringify(full)) > 2 * 1024 * 1024, + "the former full response must actually reproduce the 2 MiB transport failure", + ); + assert.ok(Buffer.byteLength(JSON.stringify(compact)) < 128 * 1024); + assert.equal(compact.head, null); + assert.equal(compact.scan, null, "do not return a second copy of the same transaction proof"); + for (const key of ["provider_revision", "cursor", "store_identity", "head_digest", "partitions"]) { + assert.deepEqual(compact[key], full[key]); + } + const fullProof = full.proof as JsonObject; + const proof = compact.proof as JsonObject; + for (const key of [ + "capture_lineage_id", + "bootstrap_provider_revision", + "last_sequences", + "last_applied_sequences", + ]) { + assert.deepEqual(proof[key], fullProof[key]); + } + const original = fullProof.transactions as JsonObject[]; + const transactions = proof.transactions as JsonObject[]; + assert.equal(transactions.length, original.length); + for (let i = 0; i < transactions.length; i++) { + for (const key of ["cursor", "provider_revision", "operation_id", "receipts"]) { + assert.deepEqual(transactions[i][key], original[i][key]); + } + assert.deepEqual( + transactions[i].projection_partitions, + (original[i].projection as JsonObject).partitions, + ); + assert.equal("projection" in transactions[i], false); + } + const progress = await readLocalAuthorityShadow({ ...request, read_model: "proof", scan_limit: 0 }); + assert.deepEqual((progress.proof as JsonObject).transactions, []); + assert.deepEqual((progress.proof as JsonObject).last_sequences, { todos: 1, leases: 0 }); + const receipt = await readLocalAuthorityShadow({ + ...request, + read_model: "proof", + scan_limit: 0, + receipt_operation_id: transactions[1].operation_id, + }); + assert.deepEqual((receipt.proof as JsonObject).receipt, transactions[1]); + assert.deepEqual((await source.store.loadAuthority()).status, "loaded"); + await assert.rejects(readLocalAuthorityShadow({ ...request, read_model: "unchecked" }), /read_model/); + }); +} diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index b0acc3d1b..a464d2874 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -96,6 +96,9 @@ "tests/control_plane_ts/coordination_projection.test.ts", "tests/control_plane_ts/coordination_state_contract.test.ts", "tests/control_plane_ts/local_authority_runtime.test.ts", + "tests/control_plane_ts/reviewed_promotion.test.ts", + "tests/control_plane_ts/promotion_receipt.test.ts", + "tests/control_plane_ts/shadow_compact_proof.test.ts", "tests/control_plane_ts/todo_completion_update_runtime.test.ts", "tests/control_plane_ts/todo_summary_lanes.test.ts", "tests/control_plane_ts/local_archive_attempt.test.ts",