From b2c6e8e292171d45fa6f9d5d68cf51979dbc2320 Mon Sep 17 00:00:00 2001 From: wchwawa Date: Thu, 10 Sep 2026 18:56:23 +1000 Subject: [PATCH] test(authority): execute the Stage 2C parity-half ladder rows Replace the nine pending `s2c2.*` declarations with executable ladder rows that drive one explicitly enabled `coordination.runtime_shadow` goal through the public CLI and assert only through `authority-shadow status|drain`, `coordination-shadow bootstrap|inspect|qualify|read-candidate|rollback` and `migrate-state`, reading history through the TypeScript shadow read over the retained store. A tenth row pins the event-only Todo source hold and its recovery. The rows carry the `stage2c_e2e` marker so CI runs them in the stage2c correctness job next to the other real-CLI and process-death suites. Two declarations stay pending: the >=10-day soak owned by RFC section 7.2, and a capture gap the parity row exposed. `todo archive-completed` on a Todo that holds a released lease record keeps that lease in the candidate head while the source projection drops the orphaned lease, so bounded qualification reports `shadow_projection_drift`; the parity row therefore archives nothing and the gap stays visible as `s2c2.archive_after_leased_completion_parity`. Three negative controls remove one operator-visible truth each and turn the corresponding row red. The new module joins the strict mypy file list. Signed-off-by: wchwawa --- ...shared-goal-authority-state-provider-v0.md | 44 +- ...-goal-authority-state-provider-v0.zh-CN.md | 34 +- examples/shared-goal-authority-e2e/README.md | 55 +- .../shared-goal-authority-e2e/correctness.md | 8 +- examples/shared-goal-authority-e2e/mutants.py | 21 + .../testing/authority_e2e_ladder.py | 125 +- .../testing/authority_e2e_rows_stage2c2.py | 1142 +++++++++++++++++ pyproject.toml | 1 + .../test_shared_goal_authority_e2e.py | 36 +- 9 files changed, 1424 insertions(+), 42 deletions(-) create mode 100644 loopx/control_plane/testing/authority_e2e_rows_stage2c2.py 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 4bfc84557d..1381c41726 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -1899,6 +1899,33 @@ Per stage, this increment implements: follow-up capture, and a leased completion in one store identity while the registry root gains neither a candidate lineage nor lease state; and `migrate-state` seeds a fresh lineage without legacy bytes. +- Stage 2C parity half: ten `s2c2.*` rows drive one explicitly enabled + `coordination.runtime_shadow` goal through the public CLI and assert only + through `authority-shadow status|drain`, `coordination-shadow + bootstrap|inspect|qualify|read-candidate|rollback` and `migrate-state`, + reading history through the retained TypeScript store. A Python Todo writer + and a TypeScript lease writer leave prepared records with committed markers + that one drain delivers once; bounded drains are cumulative, an idle drain + changes nothing and a writer replay mints no entry; a SIGKILL around the + primary replace settles as `abandoned` or `committed_proven_by_readback`, + and a SIGKILL inside the inline drain is recovered from exact receipts + without a second delivery; rollback archives pending entries, holds capture + as `bootstrap_required`, rebootstraps a fresh lineage and replays; three + cycles of interleaved writers (add, note update with a no-change repeat, + explicit exclusion set and clear with a no-change repeat, acquire, renew, + transfer, leased complete and supersede with their fence closes, + capture-followups) keep every bounded qualification matched with + `sustained_parity_verdict=not_evaluated`; a + direct primary edit reports `shadow_projection_drift`, a later write holds + on `source_partition_continuity_unproved`, and only rollback plus rebootstrap + recovers; an event-only Todo source holds `inspect`, `qualify` and + `read-candidate` with `event_log_writer_not_bound` while the primary keeps + committing; `migrate-state` refuses an active capture source with + `shadow_source_replacement_requires_rebootstrap` until rollback and disabled + capture, after which the migrated goal bootstraps a fresh lineage that + drains; and ten transactions measure file-v0 history growth with complete + projections retained, per-transaction growth accelerating by at most one + live record, and no capacity horizon claimed. Live rows are environment-gated (`LOOPX_TEST_POSTGRES_URL`; `NOKV_COORDINATION_LIVE=1` plus the `NOKV_*` stack variables; @@ -1917,12 +1944,17 @@ relaxes. Delivery boundary: test-only. No production entry point constructs any store; the ladder adds no product path and reads the candidate only through the -retained TypeScript store. The Stage 2C parity half -(`s2c2.*`: outbox entries, idempotent drain, SIGKILL before and during drain, -rollback with pending entries, parity equal and divergent, -migration seed-and-drain, growth measurement) are declared as pending rows, -not claimed. This subsection records executable evidence for the stages above; -it does not promote any provider or complete the Stage 2C promotion. +retained TypeScript store. The Stage 2C parity half executes through the ten +`s2c2.*` rows above; two declarations stay pending. +`s2c2.archive_after_leased_completion_parity` records a capture gap the parity +row exposed: `todo archive-completed` on a Todo holding a released lease record +keeps that lease in the candidate head while the source projection drops the +orphaned lease, so bounded qualification reports `shadow_projection_drift`. +`s2c2.sustained_parity_soak` is the >=10-day synthetic-goal soak owned by +Section 7.2 and lane L, and bounded qualification keeps reporting +`sustained_parity_verdict=not_evaluated`. This subsection records executable +evidence for the stages above; it does not promote any provider or complete +the Stage 2C promotion. ### 11.3 Remaining qualification and promotion plan diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index acb69de349..adc68d9ca0 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -1512,6 +1512,27 @@ CLI runner、observation-lock 窗口、候选回读）、只读 TypeScript 探 不同时，todo add、task-lease acquire、todo update、follow-up 捕获与带 lease 的 complete 仍落入同一个 store identity，registry root 既不产生候选 lineage 也不 产生 lease 状态；`migrate-state` 在不携带 legacy 字节的前提下建立新 lineage。 +- Stage 2C parity 后半段：十个 `s2c2.*` 行只通过公开 CLI 驱动一个显式开启 + `coordination.runtime_shadow` 的 goal，并且只经 `authority-shadow status|drain`、 + `coordination-shadow bootstrap|inspect|qualify|read-candidate|rollback` 与 + `migrate-state` 断言，历史只经保留的 TypeScript store 读回。Python Todo writer + 与 TypeScript lease writer 各留下 prepared 记录与 committed 标记，一次 drain 恰好 + 投递一次；有界 drain 可累积、空转 drain 不改变任何东西、writer 重放不铸造条目； + 主写 replace 前后的 SIGKILL 结算为 `abandoned` 或 `committed_proven_by_readback`， + inline drain 内的 SIGKILL 由精确 receipt 恢复且不会二次投递；rollback 归档 pending + 条目、把 capture 置于 `bootstrap_required`、重新 bootstrap 出新 lineage 并可重放； + 三轮交错 writer（add、note update 及其无变化重复、显式 exclusion 设置与清除及其 + 无变化重复、acquire、renew、transfer、带 lease 的 complete 与 supersede 及其 + fence close、capture-followups）让每次有界 qualification 都保持 matched，且 + `sustained_parity_verdict=not_evaluated`； + 直接改主文件会报告 `shadow_projection_drift`，其后的写入以 + `source_partition_continuity_unproved` 挂起，只有 rollback 加重新 bootstrap 才能恢复； + event-only Todo 来源让 `inspect`、`qualify`、`read-candidate` 以 + `event_log_writer_not_bound` 失败关闭而主写继续提交；`migrate-state` 对处于 + active capture 的来源以 `shadow_source_replacement_requires_rebootstrap` 拒绝， + 直到 rollback 并关闭 capture 之后才执行，迁移后的 goal 重新 bootstrap 出新 lineage + 并完成 drain；十笔事务度量 file-v0 历史增长：完整投影全部保留、每笔增量最多增加 + 一条 live 记录，不宣称任何容量水平线。 Live 行按环境门控（`LOOPX_TEST_POSTGRES_URL`；`NOKV_COORDINATION_LIVE=1` 加 `NOKV_*` 栈变量；`LOOPX_NOKV_AUTHORITY_LIVE=1` 加 `LOOPX_NOKV_AUTHORITY_*` 输入）。 @@ -1524,11 +1545,14 @@ Live 行按环境门控（`LOOPX_TEST_POSTGRES_URL`；`NOKV_COORDINATION_LIVE=1` `summary.privacy_violations` 阻止 green 退出，任何开关都不能放宽。 交付边界：test-only。没有任何生产入口构造任何 store；ladder 不新增产品路径， -只经保留的 TypeScript store 读取候选。Stage 2C parity 后半段 -（`s2c2.*`：outbox 条目、幂等 drain、drain 前与 drain 中的 SIGKILL、带 pending -条目的 rollback、parity 相等与分歧、迁移 seed-and-drain、增长 -度量）以 pending 行声明，而非宣称已完成。本小节记录的是上述阶段的可执行证据； -它不晋升任何 provider，也不完成 Stage 2C promotion。 +只经保留的 TypeScript store 读取候选。Stage 2C parity 后半段由上述十个 +`s2c2.*` 行执行；仍有两条声明保持 pending。`s2c2.archive_after_leased_completion_parity` +记录 parity 行暴露的一个 capture 缺口：对持有已释放 lease 记录的 Todo 执行 +`todo archive-completed` 后，候选 head 仍保留该 lease，而 source 投影会丢弃这条 +已成孤儿的 lease，于是有界 qualification 报告 `shadow_projection_drift`。 +`s2c2.sustained_parity_soak` 是由 7.2 节与车道 L 负责的 >=10 天合成 goal soak， +有界 qualification 继续报告 `sustained_parity_verdict=not_evaluated`。本小节记录 +的是上述阶段的可执行证据；它不晋升任何 provider，也不完成 Stage 2C promotion。 ### 11.3 剩余验证与晋升计划 diff --git a/examples/shared-goal-authority-e2e/README.md b/examples/shared-goal-authority-e2e/README.md index f698d52def..b70b032213 100644 --- a/examples/shared-goal-authority-e2e/README.md +++ b/examples/shared-goal-authority-e2e/README.md @@ -9,9 +9,10 @@ candidate only through the production TypeScript `FileAuthorityStore`, and never reports green while a selected row is unverified. ```bash -python examples/shared-goal-authority-e2e/ladder.py # exit 1 here: live rows unverified, parity rows pending +python examples/shared-goal-authority-e2e/ladder.py # exit 1 here: live rows unverified, the soak row pending python examples/shared-goal-authority-e2e/ladder.py --allow-unverified --allow-pending python examples/shared-goal-authority-e2e/ladder.py --stage 2c1 --report-json ladder-report.json +python examples/shared-goal-authority-e2e/ladder.py --stage 2c2 --allow-pending --report-json ladder-report.json python examples/shared-goal-authority-e2e/ladder.py --list ``` @@ -22,7 +23,10 @@ skips on Windows. Five `s2c1.*` rows whose assertions through the same product path (configure round trip, default-off isolation, candidate failure, crash gap, dual runtime root) are skipped in the default CI projection to stay within the pytest job budget; `LOOPX_LADDER_FULL=1` runs -them in pytest, and the example runner always runs every row. +them in pytest, and the example runner always runs every row. The ten +`s2c2.*` rows carry the `stage2c_e2e` marker, so CI runs them in the stage2c +correctness job next to the other real-CLI, process-death and recovery +suites rather than in the pytest shards. ## Rows @@ -40,14 +44,38 @@ them in pytest, and the example runner always runs every row. | `s2c1.crash_gap_loses_observation` | 2c1 | real_cli | deterministic (POSIX) | a writer SIGKILLed while the observation lock is held commits its todo but leaves no candidate document; the next write captures the full two-todo snapshot without claiming an outbox or correlation | | `s2c1.dual_runtime_root_consistency` | 2c1 | real_cli | deterministic | with `common_runtime_root` different from `--runtime-root`, todo add, task-lease acquire, todo update, capture-followups, and a leased completion all observe into one store identity; the head holds both todos and the released lease; the registry root gains neither a candidate lineage nor lease state | | `s2c1.migration_seeds_new_lineage` | 2c1 | real_cli | deterministic | `migrate-state` dry run plans the seed without writing; execute seeds one fresh `file:` lineage at cursor `1` that carries no legacy identity, revision, source path, or private byte | +| `s2c2.outbox_prepared_then_committed_entries` | 2c2 | real_cli | deterministic | with the maintenance lock held, `todo add` (Python) and `task-lease acquire` (TypeScript) report `drain_deferred/drain_lock_busy`, `status` shows one `committed_pending` entry per partition with one prepared record and one committed marker on disk; one `drain` delivers both (`delivered=2`), history holds the bootstrap plus two committed receipts from both writer runtimes, and the next write delivers inline at cursor `4` | +| `s2c2.drain_idempotent` | 2c2 | real_cli | deterministic | three deferred entries: `drain --max-entries 1` delivers one (`pending_after=2`, `budget_exhausted`), the next `drain` delivers two, an idle `drain` reports `nothing_pending` with unchanged cursor, `head_digest` and `provider_revision`; receipts settle sequences 1..3; an idempotent same-key re-acquire carries no capture evidence and adds no transaction | +| `s2c2.sigkill_between_primary_write_and_drain` | 2c2 | real_cli | deterministic (POSIX) | `todo add` SIGKILLed at `before_replace`, `after_replace` and `before_marker` leaves one prepared-only entry each; `drain` settles it as `abandoned` (no-op, primary unchanged) or `committed_proven_by_readback`, the projection equals the primary, and `inspect` ends `matched` | +| `s2c2.sigkill_mid_drain` | 2c2 | real_cli | deterministic (POSIX) | `todo add` SIGKILLed at `before_commit`, `after_commit`, `after_cursor` and `between_unlinks`: the next `drain` delivers the uncommitted entry once or replays the committed one (`replayed=1, delivered=0`), history holds exactly one delivery, only the cursor remains, and a further drain is idle | +| `s2c2.rollback_with_pending_entries` | 2c2 | real_cli | deterministic (POSIX) | with one committed-pending and one prepared-only entry, `inspect` reports `outbox_pending` at the exact revision, a rollback preview writes nothing, `rollback --execute` applies and archives the outbox with both entries, the marker, the cursor and the manifest; capture then reports `bootstrap_required` while primary writes continue, a rebootstrap starts a new lineage from the current primary (three todos), and the historical rollback replays against it | +| `s2c2.parity_equal` | 2c2 | real_cli | deterministic | three cycles interleave Python Markdown writers (add, note update plus a no-change repeat, explicit exclusion set and clear plus a no-change repeat, complete, supersede, capture-followups) with TypeScript lease writers (acquire, renew, transfer, and the fence close of a leased complete or supersede); after each cycle `inspect` is `matched`, `qualify` with every required write class is `qualified` with `operation_count` equal to the delivered mutations, `read-candidate` returns the anchor todo, and `sustained_parity_verdict` stays `not_evaluated` | +| `s2c2.parity_divergent_detects_foreign_edit` | 2c2 | real_cli | deterministic | a direct edit of the primary makes `inspect` report `drifted/shadow_projection_drift`, `qualify` and `read-candidate` reject, a later `todo add` commits but its capture holds on `source_partition_continuity_unproved`; restoring the bytes does not requalify (`outbox_pending`), `drain` stays `stopped`, and only `rollback --execute` plus a fresh bootstrap qualifies again | +| `s2c2.event_only_todo_source_holds` | 2c2 | real_cli | deterministic | an event-only Todo appended to the goal's state event log makes `inspect`, `qualify` and `read-candidate` fail closed with `event_log_writer_not_bound`, `status` stays readable, a Markdown write still commits with its capture held, the event log is untouched; removing the event source does not requalify, and rollback plus rebootstrap recovers | +| `s2c2.migration_seeds_and_drains` | 2c2 | real_cli | deterministic | `migrate-state` previews an actively captured goal without writing, refuses `--execute` with `shadow_source_replacement_requires_rebootstrap` (also when capture is merely disabled), and executes only after `rollback`; the migrated goal carries its disabled capture configuration, plans no observation seed, requires its own `bootstrap`, then captures a write to cursor `2` and qualifies on it while the legacy archive is retained | +| `s2c2.growth_measurement_gate` | 2c2 | real_cli | deterministic | ten fixed-size `todo add` writes: the cursor advances by one each time, `store_bytes` grows monotonically, the per-transaction delta accelerates by at most 2048 bytes (one live record), every retained transaction carries its complete projection, `retention_pressure` stays false; the report carries final and cumulative publication bytes and claims no capacity horizon (`capacity_verdict=not_evaluated`) | Pending rows are declared in the report as `pending`, never counted as pass, -and they block a green exit unless `--allow-pending` is passed. The Stage 2C -parity rows are pending: `s2c2.outbox_prepared_then_committed_entries`, `s2c2.drain_idempotent`, -`s2c2.sigkill_between_primary_write_and_drain`, `s2c2.sigkill_mid_drain`, -`s2c2.rollback_with_pending_entries`, `s2c2.parity_equal`, `s2c2.parity_divergent_detects_foreign_edit`, -`s2c2.migration_seeds_and_drains`, `s2c2.growth_measurement_gate` (until the -Stage 2C parity PRs land). +and they block a green exit unless `--allow-pending` is passed. Two +declarations remain. `s2c2.archive_after_leased_completion_parity` records a +capture gap found while building the parity row: `todo archive-completed` on a +Todo that holds a released lease record leaves that lease in the candidate +head while the source projection drops the now-orphaned lease, so `inspect` +reports `shadow_projection_drift`; the parity row therefore archives nothing +and the gap stays visible until the archive writer captures the lease it +orphans. `s2c2.sustained_parity_soak` is the >=10-day synthetic-goal soak of +the selected local profile owned by RFC Section 7.2 (lane L). Bounded +qualification reports `sustained_parity_verdict=not_evaluated`, and no +`s2c2.*` row promotes a provider or completes the Stage 2C promotion. + +The `s2c2.*` rows use two scheduling-only seams outside every product decision: +holding the stable maintenance lock, which makes a writer report +`drain_deferred/drain_lock_busy` and leave its committed entry pending, and a +POSIX crash worker that pauses one real CLI process at a named persistence +window so the row can SIGKILL it there. Neither substitutes a result or edits a +byte; every assertion still goes through `status`, `drain`, `inspect`, +`qualify`, `read-candidate`, `rollback`, `migrate-state` and the retained +TypeScript store read. ## Gates and environment variables @@ -73,7 +101,7 @@ Exit code is `0` iff `fail == 0` and `privacy_violations == 0` and (`unverified == 0` or `--allow-unverified`) and (`pending == 0` or `--allow-pending`): a selected row that never executed, whether gated or declared pending, is an unmet -obligation, so `--row s2c2.parity_equal` exits 1 with zero executions, and a +obligation, so `--row s2c2.sustained_parity_soak` exits 1 with zero executions, and a mixed selection exits 1 even when its executable rows pass. `--list` only prints the registry and never claims verification. A privacy scan runs over the finished report: any occurrence of a temporary root, the home directory, @@ -94,10 +122,11 @@ boundaries. [Installed-package E2E](installed.py) repeats the public lifecycle outside the checkout for both wheel and sdist. [Negative controls](mutants.py) deliberately remove correctness checks in disposable source copies. -These checks do not change the nine pending `s2c2.*` ladder declarations above. -Sustained production parity, the migration/growth gates, and promotion remain -separate obligations. A bounded qualification result reports -`sustained_parity_verdict=not_evaluated`. +The ten `s2c2.*` ladder rows above exercise the same lifecycle through the +public interfaces and read history only through the retained TypeScript store. +Sustained (elapsed-time) parity and promotion remain separate obligations: the +soak and the archive-after-lease capture gap stay pending declarations, and a +bounded qualification result reports `sustained_parity_verdict=not_evaluated`. Future ladder rows must use the actual product interfaces: diff --git a/examples/shared-goal-authority-e2e/correctness.md b/examples/shared-goal-authority-e2e/correctness.md index 24aa6ffd1e..fb61354eb9 100644 --- a/examples/shared-goal-authority-e2e/correctness.md +++ b/examples/shared-goal-authority-e2e/correctness.md @@ -308,12 +308,16 @@ Python/TS/JSON provenance, and reads back through an independent native process. | Fence and maintenance boundaries, source and Goal override races, whole-file durability, paragraph injection, refresh CAS | `test_shadow_writer_boundaries.py`, `test_shadow_writer_variant_e2e.py`, `shadow_native_writer_boundary.test.ts`, `test_shadow_drain_adversarial.py` | | Canonical native/v0 Todo updates through CLI and native RPC, real pending management, M ordering, and unchanged authority on hold | `test_shadow_native_todo_update_e2e.py` | | History flaws despite equal snapshots, legacy mixed profile, source drift, event-only hold, qualified reads | `coordination_runtime_shadow.test.ts`, `file_outbox_qualification.test.ts`, `test_runtime_shadow_bounded_e2e.py` | +| Ladder parity half through the public CLI and management interfaces: deferred entries, bounded and idempotent drain, primary and drain SIGKILL windows, rollback with pending entries, sustained mixed-writer parity, drift, the event-only hold, migration refusal and rebootstrap, growth measurement | `test_shared_goal_authority_e2e.py` (`s2c2.*` rows), `ladder.py --stage 2c2` | | Installed lifecycle and resource provenance in wheel and sdist | `installed.py` | | Missing checks, lock placement, duplicate mirror, early marker, cursor regression | `mutants.py` with unchanged GREEN controls and assertion RED results | The mandatory repair set must have zero failures, skips, pending, or unverified -cases. Broader ladder rows retain their declared pending/environment gates; -these tests grant neither production promotion nor a completed Stage 2C claim. +cases. Broader ladder rows retain their declared pending/environment gates: +`s2c2.archive_after_leased_completion_parity` stays pending until the +archive-completed writer captures the released lease it orphans, and +`s2c2.sustained_parity_soak` until the Section 7.2 soak exists; these tests +grant neither production promotion nor a completed Stage 2C claim. For a caller comparison, run both `test_shadow_observable*_e2e.py` files with `LOOPX_SHADOW_COMPARISON_SOURCE` set to an immutable baseline checkout, then to diff --git a/examples/shared-goal-authority-e2e/mutants.py b/examples/shared-goal-authority-e2e/mutants.py index 33a9bf91da..c4d338aa06 100644 --- a/examples/shared-goal-authority-e2e/mutants.py +++ b/examples/shared-goal-authority-e2e/mutants.py @@ -147,6 +147,27 @@ def command(self) -> list[str]: "tests/control_plane/test_shadow_drain_e2e.py::test_primary_sigkill_preserves_complete_bytes_and_proves_before_marker[before_replace]"), ]) +# Ladder parity-half rows (s2c2.*) drive the same lifecycle through the public +# CLI; these mutants remove one operator-visible truth each and must turn the +# corresponding row red. +LADDER_ROW = "tests/control_plane/test_shared_goal_authority_e2e.py::test_ladder_row_passes_or_is_declared_unverified" +CASES.extend([ + Case("status_hides_prepared_only", ((COORDINATION + "local_authority_shadow_outbox.py", replacement( + ' "committed_pending": sum(1 for entry in entries if entry.is_committed),\n' + ' "prepared_only": sum(1 for entry in entries if not entry.is_committed),', + ' "committed_pending": len(entries),\n' + ' "prepared_only": 0,')),), + LADDER_ROW + "[s2c2.sigkill_between_primary_write_and_drain]"), + Case("qualification_ignores_drift", ((COORDINATION + "runtime_shadow.ts", replacement( + "const matched = localAuthorityShadowHeadDigest(request.projection) === localAuthorityShadowHeadDigest(lineage.head.head);", + "const matched = true;")),), + LADDER_ROW + "[s2c2.parity_divergent_detects_foreign_edit]"), + Case("replay_counted_as_delivery", ((COORDINATION + "local_authority_shadow_adapter.py", replacement( + ' self._result.replayed += 1\n self._result.no_op += int(receipt["no_op"])', + ' self._result.delivered += 1\n self._result.no_op += int(receipt["no_op"])')),), + LADDER_ROW + "[s2c2.sigkill_mid_drain]"), +]) + def remove_fence(source: str) -> str: function = next(node for node in ast.parse(source).body diff --git a/loopx/control_plane/testing/authority_e2e_ladder.py b/loopx/control_plane/testing/authority_e2e_ladder.py index 6d8a9af8c4..29fbd15d07 100644 --- a/loopx/control_plane/testing/authority_e2e_ladder.py +++ b/loopx/control_plane/testing/authority_e2e_ladder.py @@ -49,6 +49,18 @@ row_every_writer_family_captures, row_migration_seeds_new_lineage, ) +from .authority_e2e_rows_stage2c2 import ( + row_drain_idempotent, + row_event_only_todo_source_holds, + row_growth_measurement_gate, + row_migration_seeds_and_drains, + row_outbox_prepared_then_committed_entries, + row_parity_divergent_detects_foreign_edit, + row_parity_equal, + row_rollback_with_pending_entries, + row_sigkill_between_primary_write_and_drain, + row_sigkill_mid_drain, +) from .authority_e2e_fixtures import ( REPO_ROOT, CliOutputError, @@ -139,6 +151,7 @@ Path("loopx") / "control_plane" / "testing" / "authority_e2e_fixtures.py", Path("loopx") / "control_plane" / "testing" / "authority_e2e_row_support.py", Path("loopx") / "control_plane" / "testing" / "authority_e2e_rows_stage2c.py", + Path("loopx") / "control_plane" / "testing" / "authority_e2e_rows_stage2c2.py", ) FILE_MATRIX_ROWS: tuple[str, ...] = ( "same_todo_one_winner", @@ -647,18 +660,112 @@ def _row_postgresql_conformance_live(context: RowContext) -> RowOutcome: posix_only=False, run=row_migration_seeds_new_lineage, ), + LadderRow( + id="s2c2.outbox_prepared_then_committed_entries", + stage="2c2", + title="Python and TypeScript writers leave prepared records with committed markers that one drain delivers once", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_outbox_prepared_then_committed_entries, + ), + LadderRow( + id="s2c2.drain_idempotent", + stage="2c2", + title="Bounded drains are cumulative, an idle drain changes nothing, and a writer replay mints no entry", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_drain_idempotent, + ), + LadderRow( + id="s2c2.sigkill_between_primary_write_and_drain", + stage="2c2", + title="A SIGKILL around the primary replace leaves a prepared-only entry that drain settles from the primary bytes", + product_path="real_cli", + gate="deterministic", + posix_only=True, + run=row_sigkill_between_primary_write_and_drain, + ), + LadderRow( + id="s2c2.sigkill_mid_drain", + stage="2c2", + title="A SIGKILL inside the inline drain is recovered from exact receipts without a second delivery", + product_path="real_cli", + gate="deterministic", + posix_only=True, + run=row_sigkill_mid_drain, + ), + LadderRow( + id="s2c2.rollback_with_pending_entries", + stage="2c2", + title="Rollback archives pending entries, holds capture until rebootstrap, and replays its historical result", + product_path="real_cli", + gate="deterministic", + posix_only=True, + run=row_rollback_with_pending_entries, + ), + LadderRow( + id="s2c2.parity_equal", + stage="2c2", + title="Sustained interleaving of Python and TypeScript writers keeps every bounded qualification matched", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_parity_equal, + ), + LadderRow( + id="s2c2.parity_divergent_detects_foreign_edit", + stage="2c2", + title="A direct primary edit is reported as drift, holds later captures, and recovers only by rollback and rebootstrap", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_parity_divergent_detects_foreign_edit, + ), + LadderRow( + id="s2c2.event_only_todo_source_holds", + stage="2c2", + title="An event-only Todo source holds qualification and candidate reads fail-closed until rollback and rebootstrap", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_event_only_todo_source_holds, + ), + LadderRow( + id="s2c2.migration_seeds_and_drains", + stage="2c2", + title="migrate-state refuses an active capture source; after rollback the migrated goal bootstraps a fresh lineage that drains", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_migration_seeds_and_drains, + ), + LadderRow( + id="s2c2.growth_measurement_gate", + stage="2c2", + title="file-v0 history growth is measured per transaction and gated on retention integrity, claiming no capacity horizon", + product_path="real_cli", + gate="deterministic", + posix_only=False, + run=row_growth_measurement_gate, + ), ) PENDING_ROWS: tuple[PendingRow, ...] = ( - PendingRow("s2c2.outbox_prepared_then_committed_entries", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.drain_idempotent", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.sigkill_between_primary_write_and_drain", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.sigkill_mid_drain", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.rollback_with_pending_entries", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.parity_equal", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.parity_divergent_detects_foreign_edit", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.migration_seeds_and_drains", "2c2", "Stage 2C parity PRs"), - PendingRow("s2c2.growth_measurement_gate", "2c2", "Stage 2C parity PRs"), + PendingRow( + "s2c2.archive_after_leased_completion_parity", + "2c2", + "the archive-completed writer captures the released lease it orphans: archiving a Todo " + "that holds a released lease record leaves that lease in the candidate head while the " + "source projection drops it, so bounded qualification reports shadow_projection_drift", + ), + PendingRow( + "s2c2.sustained_parity_soak", + "2c2", + "a >=10-day synthetic-goal soak of the selected local profile (RFC section 7.2, lane L); " + "bounded qualification reports sustained_parity_verdict=not_evaluated", + ), ) diff --git a/loopx/control_plane/testing/authority_e2e_rows_stage2c2.py b/loopx/control_plane/testing/authority_e2e_rows_stage2c2.py new file mode 100644 index 0000000000..13f965f12d --- /dev/null +++ b/loopx/control_plane/testing/authority_e2e_rows_stage2c2.py @@ -0,0 +1,1142 @@ +"""Stage 2C parity-half rows (``s2c2.*``) of the shared-goal-authority ladder. + +Every row drives the product through the real ``python -m loopx.cli`` against +one goal whose ``coordination.runtime_shadow`` capture is explicitly enabled +and bootstrapped, and asserts only through shipped operator interfaces: +``authority-shadow status|drain``, ``coordination-shadow bootstrap|inspect| +qualify|read-candidate|rollback`` and ``migrate-state``. Candidate history is +read back through the retained TypeScript store (the adapter's read RPC). The +rows add no product path and never read the candidate for a decision. + +Two scheduling-only seams exist, both outside every product decision: + +* holding the stable maintenance lock defers a writer's post-commit drain, so + its committed outbox entry stays pending (``drain_lock_busy``); +* a POSIX crash worker pauses a real CLI process at one persistence window and + the row SIGKILLs it there. It substitutes no result and edits no byte. +""" + +from __future__ import annotations + +import importlib +import json +import select +import subprocess +import sys +from collections.abc import Iterator, Mapping, Sequence +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path + +from ...file_lock import exclusive_file_lock +from ..coordination import local_authority_shadow_outbox as shadow_outbox +from .authority_e2e_fixtures import ( + REPO_ROOT, + GoalWorkspace, + JsonObject, + build_goal_workspace, + cli_env, + node_executable, + parse_json_object, + run_cli, + unique_goal_id, +) +from .authority_e2e_row_support import ( + AGENT_A, + AGENT_B, + RowContext, + RowOutcome, + acquire_lease, + add_todo, + expect, + lease_version, + passed, +) + +RUNTIME_SHADOW_CONFIG_SCHEMA = "loopx_coordination_runtime_shadow_config_v0" +RUNTIME_SHADOW_PROVIDER = "file_v0" +CRASH_BARRIER_TIMEOUT_SECONDS = 30.0 +PRIMARY_CRASH_WINDOWS: tuple[tuple[str, bool], ...] = ( + ("before_replace", False), + ("after_replace", True), + ("before_marker", True), +) +DRAIN_CRASH_WINDOWS: tuple[str, ...] = ("before_commit", "after_commit", "after_cursor", "between_unlinks") +PARITY_CYCLES = 3 +# ``todo archive-completed`` is deliberately absent: archiving a Todo that holds +# a released lease record orphans that lease in the source projection while the +# candidate head keeps it, so the bounded qualification drifts. The ladder +# declares that gap as ``s2c2.archive_after_leased_completion_parity`` instead +# of hiding it inside a passing row. +PARITY_REQUIRED_WRITE_CLASSES: tuple[str, ...] = ( + "todo_add", + "todo_update", + "todo_complete", + "todo_supersede", + "todo_capture_followups", + "task_lease_acquire", + "task_lease_renew", + "task_lease_transfer", + "task_lease_fence_close", +) +GROWTH_TRANSACTIONS = 10 +GROWTH_TEXT_TEMPLATE = "Growth workload todo %02d " + "x" * 160 +# Each file-v0 transaction retains the complete projection, so the per-transaction +# byte delta may grow by about one Todo record per transaction. A larger jump +# means something beyond the live projection is being re-published. +GROWTH_DELTA_ACCELERATION_ENVELOPE_BYTES = 2048 +EVENT_ONLY_HOLD = "event_log_writer_not_bound" +CONTINUITY_HOLD = "source_partition_continuity_unproved" +SHADOW_READ_MODULE = Path("loopx") / "control_plane" / "coordination" / "local_authority_shadow.ts" +SHADOW_READ_REQUEST_SCHEMA = "loopx_coordination_runtime_shadow_outbox_read_v0" +SHADOW_READ_SCAN_LIMIT = 10_000 + +# A real CLI process that pauses at exactly one persistence window. The parent +# waits for the BARRIER line and SIGKILLs the child there. Every hook forwards +# to the production function; nothing decides an outcome or substitutes bytes. +CRASH_WORKER = r""" +import json, pathlib, sys, time +from loopx.cli import main +from loopx.control_plane.coordination import local_authority_shadow_adapter as adapter +from loopx.control_plane.coordination import local_authority_shadow_outbox as outbox +from loopx.control_plane.todos import active_state_editing +window, state = sys.argv[1], pathlib.Path(sys.argv[2]).resolve() +def pause(): + print('BARRIER ' + json.dumps({'window': window}), flush=True) + time.sleep(40) + raise RuntimeError('parent failed to terminate at persistence barrier') +actual_rpc = adapter.effect_runtime_result +def rpc(method, request, **kwargs): + if method == 'coordination.runtime_shadow.commit_entry' and window == 'before_commit': + pause() + result = actual_rpc(method, request, **kwargs) + if method == 'coordination.runtime_shadow.commit_entry' and window == 'after_commit': + pause() + return result +adapter.effect_runtime_result = rpc +actual_cursor = outbox.write_cursor +def cursor(*args, **kwargs): + result = actual_cursor(*args, **kwargs) + if window == 'after_cursor': + pause() + return result +outbox.write_cursor = cursor +actual_json = outbox.durable_write_json +def write_json(path, value): + if window == 'before_marker' and path.name.endswith('.committed.json'): + pause() + return actual_json(path, value) +outbox.durable_write_json = write_json +actual_replace = active_state_editing.os.replace +def replace(source, target): + is_primary = pathlib.Path(target).resolve() == state + if is_primary and window == 'before_replace': + pause() + result = actual_replace(source, target) + if is_primary and window == 'after_replace': + pause() + return result +active_state_editing.os.replace = replace +actual_unlink = pathlib.Path.unlink +def unlink(path, *args, **kwargs): + result = actual_unlink(path, *args, **kwargs) + if window == 'between_unlinks' and path.name.endswith('.prepared.json'): + pause() + return result +pathlib.Path.unlink = unlink +raise SystemExit(main(sys.argv[3:])) +""" + + +# --------------------------------------------------------------------------- +# Workspace, CLI and read-back helpers +# --------------------------------------------------------------------------- + + +def _object(value: object, label: str) -> JsonObject: + expect(isinstance(value, dict), f"{label} must be an object") + assert isinstance(value, dict) + return {str(key): item for key, item in value.items()} + + +def _list(value: object, label: str) -> list[object]: + expect(isinstance(value, list), f"{label} must be a list") + assert isinstance(value, list) + return list(value) + + +def set_runtime_shadow(workspace: GoalWorkspace, *, enabled: bool) -> None: + """The operator step: enable or disable transaction capture in the registry.""" + + registry = parse_json_object(workspace.registry_path.read_text(encoding="utf-8")) + goals = _list(registry.get("goals"), "registry goals") + goal = _object(goals[0], "registry goal") + coordination = _object(goal.get("coordination"), "goal coordination") + coordination["runtime_shadow"] = { + "schema_version": RUNTIME_SHADOW_CONFIG_SCHEMA, + "enabled": enabled, + "provider": RUNTIME_SHADOW_PROVIDER, + } + goal["coordination"] = coordination + goals[0] = goal + registry["goals"] = goals + workspace.registry_path.write_text(json.dumps(registry), encoding="utf-8") + + +def goal_cli(workspace: GoalWorkspace, *args: str, check: bool = True) -> JsonObject: + return run_cli(workspace, *args, "--goal-id", workspace.goal_id, check=check) + + +def bootstrap_capture(workspace: GoalWorkspace) -> JsonObject: + """Bootstrap the capture lineage; returns the whole CLI payload.""" + + payload = goal_cli(workspace, "coordination-shadow", "bootstrap", "--execute") + bootstrap = _object(payload.get("bootstrap"), "bootstrap") + expect(bootstrap.get("status") == "applied", "bootstrap must apply a fresh lineage") + expect(bootstrap.get("cursor") == "1", "bootstrap must be the first transaction") + expect(payload.get("decision_read_from_shadow") is False, "bootstrap must not read the candidate for a decision") + return payload + + +def capture_workspace( + context: RowContext, + prefix: str, + *, + handoff_mode: str = "hard_lease", + bootstrap: bool = True, +) -> GoalWorkspace: + workspace = build_goal_workspace( + context.root, + goal_id=unique_goal_id(prefix), + handoff_mode=handoff_mode, + shadow_enabled=False, + runtime_root_binding="cli_override", + ) + set_runtime_shadow(workspace, enabled=True) + if bootstrap: + bootstrap_capture(workspace) + return workspace + + +def capture_evidence(payload: Mapping[str, object], *, label: str) -> JsonObject: + evidence = _object(payload.get("coordination_runtime_shadow"), f"{label} capture evidence") + expect(evidence.get("primary_writeback_preserved") is True, f"{label} must preserve the primary writeback") + expect(evidence.get("provider_to_local_writes") is False, f"{label} must never write from provider to local state") + expect(evidence.get("candidate_read_for_decision") is False, f"{label} must never read the candidate for a decision") + expect(evidence.get("parity_verdict") == "not_evaluated", f"{label} must not claim parity on the write path") + return evidence + + +def delivered(payload: Mapping[str, object], *, label: str) -> JsonObject: + """A primary write whose outbox entry was drained inline into the candidate.""" + + evidence = capture_evidence(payload, label=label) + expect(evidence.get("outcome") == "delivered", f"{label} must deliver its outbox entry") + expect( + evidence.get("source_transaction_correlated") is True and evidence.get("durable_source_outbox") is True, + f"{label} must correlate a durable outbox entry", + ) + drain = _object(evidence.get("drain"), f"{label} drain") + expect(drain.get("outcome") == "drained" and drain.get("candidate_readback_verified") is True, f"{label} must verify its candidate read-back") + return evidence + + +def no_transaction(payload: Mapping[str, object], *, label: str) -> JsonObject: + evidence = capture_evidence(payload, label=label) + expect(evidence.get("outcome") == "no_transaction", f"{label} must record no transaction") + expect(_object(evidence.get("entry"), f"{label} entry").get("entry_id") is None, f"{label} must mint no outbox entry") + return evidence + + +def deferred(payload: Mapping[str, object], *, label: str) -> JsonObject: + evidence = capture_evidence(payload, label=label) + expect( + evidence.get("outcome") == "drain_deferred" and evidence.get("reason_code") == "drain_lock_busy", + f"{label} must defer its drain behind the held maintenance lock", + ) + expect( + evidence.get("source_transaction_correlated") is True and evidence.get("durable_source_outbox") is True, + f"{label} must still correlate a durable outbox entry", + ) + return evidence + + +def shadow_status(workspace: GoalWorkspace) -> JsonObject: + return goal_cli(workspace, "authority-shadow", "status", check=False) + + +def backlog(status: Mapping[str, object], partition: str) -> JsonObject: + return _object(_object(status.get("outbox"), "status outbox").get(partition), f"{partition} backlog") + + +def candidate(status: Mapping[str, object]) -> JsonObject: + return _object(status.get("candidate"), "status candidate") + + +def management_status(status: Mapping[str, object]) -> str: + return str(_object(status.get("management"), "status management").get("status")) + + +def drain(workspace: GoalWorkspace, *flags: str) -> JsonObject: + return goal_cli(workspace, "authority-shadow", "drain", *flags, check=False) + + +def inspect(workspace: GoalWorkspace) -> JsonObject: + payload = goal_cli(workspace, "coordination-shadow", "inspect", check=False) + expect(payload.get("decision_read_from_shadow") is False, "inspect must not read the candidate for a decision") + return payload + + +def qualify(workspace: GoalWorkspace, *flags: str) -> JsonObject: + payload = goal_cli(workspace, "coordination-shadow", "qualify", *flags, check=False) + expect(payload.get("decision_read_from_shadow") is False, "qualify must not read the candidate for a decision") + return payload + + +def read_candidate(workspace: GoalWorkspace, todo_id: str) -> JsonObject: + payload = goal_cli(workspace, "coordination-shadow", "read-candidate", "--todo-id", todo_id, check=False) + expect(payload.get("decision_read_from_shadow") is False, "read-candidate must not read the candidate for a decision") + return payload + + +def qualified(payload: Mapping[str, object], *, label: str) -> JsonObject: + qualification = _object(payload.get("qualification"), f"{label} qualification") + expect(payload.get("ok") is True and qualification.get("status") == "qualified", f"{label} must qualify") + expect(qualification.get("parity_matches") is True and qualification.get("scope") == "bounded", f"{label} must be a bounded parity match") + expect( + qualification.get("sustained_parity_verified") is False and qualification.get("sustained_parity_verdict") == "not_evaluated", + f"{label} must not claim sustained parity", + ) + return qualification + + +def rejected(payload: Mapping[str, object], key: str, *, label: str) -> JsonObject: + result = _object(payload.get(key), f"{label} result") + expect(payload.get("ok") is False and result.get("qualified") is False, f"{label} must be rejected") + return result + + +def history(workspace: GoalWorkspace) -> list[JsonObject]: + """Complete candidate history through the TypeScript shadow read over the retained store. + + The read runs in an independent node process so the row never imports the + Python adapter; it is the same ``readLocalAuthorityShadow`` the product + uses for status and drain proof. + """ + + node = node_executable() + expect(node is not None, "node is required to read the candidate history") + assert node is not None + request = { + "schema_version": SHADOW_READ_REQUEST_SCHEMA, + "runtime_root": str(workspace.runtime_root), + "goal_id": workspace.goal_id, + "scan_limit": SHADOW_READ_SCAN_LIMIT, + } + request_path = workspace.home / "shadow-read-request.json" + request_path.write_text(json.dumps(request), encoding="utf-8") + script = ( + f"import {{ readLocalAuthorityShadow }} from {json.dumps((REPO_ROOT / SHADOW_READ_MODULE).as_uri())};" + "import { readFile } from 'node:fs/promises';" + "process.stdout.write(JSON.stringify(await readLocalAuthorityShadow(JSON.parse(await readFile(process.argv[1], 'utf8')))));" + ) + completed = subprocess.run( + [node, "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script, str(request_path)], + cwd=REPO_ROOT, + capture_output=True, + text=True, + timeout=60, + check=False, + ) + expect(completed.returncode == 0, f"the TypeScript shadow read exited {completed.returncode}") + view = parse_json_object(completed.stdout) + expect(view.get("status") == "loaded", "candidate history must load through the TypeScript store") + proof = _object(view.get("proof"), "history proof") + return [_object(item, "transaction") for item in _list(proof.get("transactions"), "history transactions")] + + +def single_receipt(transaction: Mapping[str, object], *, label: str) -> JsonObject: + receipts = _list(transaction.get("receipts"), f"{label} receipts") + expect(len(receipts) == 1, f"{label} must carry exactly one receipt") + return _object(receipts[0], f"{label} receipt") + + +def committed_receipt(transaction: Mapping[str, object], *, label: str) -> JsonObject: + receipt = single_receipt(transaction, label=label) + expect(receipt.get("resolution") == "committed" and receipt.get("no_op") is False, f"{label} must be a committed, effective receipt") + return receipt + + +def projected_todo_count(transaction: Mapping[str, object]) -> int: + projection = _object(transaction.get("projection"), "transaction projection") + todos = projection.get("todos") + return len(todos) if isinstance(todos, list) else 0 + + +def partition_files(workspace: GoalWorkspace, partition: str) -> list[str]: + directory = shadow_outbox.partition_directory(workspace.runtime_root, workspace.goal_id, partition) + return sorted(path.name for path in directory.iterdir()) if directory.is_dir() else [] + + +def phase_count(names: Sequence[str], phase: str) -> int: + return sum(1 for name in names if name.endswith(f".{phase}.json")) + + +def outbox_present(workspace: GoalWorkspace) -> bool: + return shadow_outbox.outbox_root(workspace.runtime_root, workspace.goal_id).exists() + + +def store_documents(workspace: GoalWorkspace) -> list[Path]: + return sorted((workspace.runtime_root / "authority-shadow" / "file-v0").glob("authority-store-*.json")) + + +def todo_count(workspace: GoalWorkspace) -> int: + listed = goal_cli(workspace, "todo", "list") + return len(_list(listed.get("todos"), "todo list")) + + +@contextmanager +def hold_drain_lock(workspace: GoalWorkspace) -> Iterator[None]: + """Hold the stable maintenance lock so writers defer their post-commit drain.""" + + with exclusive_file_lock( + shadow_outbox.drain_lock_target(workspace.runtime_root, workspace.goal_id), + operation="e2e_window", + ): + yield + + +def crash_cli(workspace: GoalWorkspace, window: str, *args: str) -> None: + """Run one real CLI command and SIGKILL it at ``window``.""" + + command = [sys.executable, "-c", CRASH_WORKER, window, str(workspace.state_path), *workspace.cli_prefix(), *args, "--goal-id", workspace.goal_id] + process = subprocess.Popen( + command, + cwd=REPO_ROOT, + env=cli_env(workspace), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + line = "" + try: + assert process.stdout is not None + readable, _, _ = select.select([process.stdout], [], [], CRASH_BARRIER_TIMEOUT_SECONDS) + if readable: + line = process.stdout.readline() + finally: + process.kill() + _, stderr = process.communicate(timeout=10) + expect(line.startswith("BARRIER "), f"{window}: the CLI did not reach its persistence window: {stderr[-200:]}") + expect(process.returncode == -9, f"{window}: the crash worker must die by SIGKILL") + + +# --------------------------------------------------------------------------- +# Rows +# --------------------------------------------------------------------------- + + +def row_outbox_prepared_then_committed_entries(context: RowContext) -> RowOutcome: + """A Python Todo writer and a TypeScript lease writer each leave one prepared record and committed marker; drain delivers both once.""" + + workspace = capture_workspace(context, "ladder-outbox") + with hold_drain_lock(workspace): + added = add_todo(workspace, "Prepared and committed while the drain lock is held.") + expect(added.get("added") is True, "the todo must commit while its drain is deferred") + todo_id = str(added["todo_id"]) + acquired = acquire_lease(workspace, todo_id=todo_id, owner=AGENT_A, idempotency_key="ladder-outbox-lease") + expect(acquired.get("acquired") is True, "the lease must commit while its drain is deferred") + entries = [ + _object(deferred(added, label="todo add").get("entry"), "todo entry"), + _object(deferred(acquired, label="task-lease acquire").get("entry"), "lease entry"), + ] + expect([entry.get("partition") for entry in entries] == ["todos", "leases"], "the writers must target their own partitions") + expect(all(entry.get("seq") == 1 for entry in entries), "each partition must start at sequence 1") + status = shadow_status(workspace) + for partition in ("todos", "leases"): + pending = backlog(status, partition) + expect( + pending.get("committed_pending") == 1 and pending.get("prepared_only") == 0 and pending.get("cursor_last_seq") is None, + f"{partition} must hold exactly one committed, undrained entry", + ) + names = partition_files(workspace, partition) + expect( + phase_count(names, "prepared") == 1 and phase_count(names, "committed") == 1, + f"{partition} must retain its prepared record and committed marker", + ) + expect(todo_count(workspace) == 1, "the primary must show the committed todo while its entry is pending") + drained = drain(workspace) + expect( + drained.get("ok") is True and drained.get("outcome") == "drained" and drained.get("delivered") == 2 and drained.get("replayed") == 0, + "one drain must deliver both pending entries exactly once", + ) + settled = [_object(item, "drained entry") for item in _list(drained.get("entries"), "drain entries")] + expect( + all(item.get("outcome") == "delivered" and item.get("resolution") == "committed" for item in settled), + "every drained entry must be a committed delivery", + ) + status = shadow_status(workspace) + for partition in ("todos", "leases"): + pending = backlog(status, partition) + expect(pending.get("committed_pending") == 0 and pending.get("prepared_only") == 0 and pending.get("cursor_last_seq") == 1, f"{partition} must be drained to sequence 1") + expect(partition_files(workspace, partition) == ["drain-cursor.json"], f"{partition} must retain only its cursor after delivery") + transactions = history(workspace) + expect(len(transactions) == 3 and transactions[0].get("receipts") == [] and transactions[0].get("cursor") == "1", "history must be the bootstrap plus two deliveries") + receipts = [committed_receipt(transaction, label=f"transaction {index}") for index, transaction in enumerate(transactions[1:], start=2)] + runtimes = sorted({str(receipt.get("writer_runtime")) for receipt in receipts}) + classes = sorted({str(receipt.get("write_class")) for receipt in receipts}) + expect(runtimes == ["python", "typescript"], "the deliveries must come from both writer runtimes") + expect(classes == ["task_lease_acquire", "todo_add"], "the deliveries must carry their write classes") + inline = delivered(add_todo(workspace, "Delivered inline once the lock is free."), label="todo add (inline)") + expect(_object(inline.get("drain"), "inline drain").get("last_cursor") == "4", "an inline delivery must continue the same lineage") + return passed( + deferred_entries=2, + partitions=["leases", "todos"], + drained=2, + writer_runtimes=runtimes, + write_classes=classes, + inline_cursor="4", + ) + + +def row_drain_idempotent(context: RowContext) -> RowOutcome: + """Bounded drains are cumulative, an idle drain changes nothing, and a writer replay mints no entry.""" + + workspace = capture_workspace(context, "ladder-drain") + with hold_drain_lock(workspace): + for index in range(3): + deferred(add_todo(workspace, f"Deferred todo {index}."), label=f"todo add {index}") + pending = backlog(shadow_status(workspace), "todos") + expect(pending.get("committed_pending") == 3 and pending.get("next_seq") == 3, "three committed entries must be pending") + first = drain(workspace, "--max-entries", "1") + expect( + first.get("ok") is True and first.get("delivered") == 1 and first.get("replayed") == 0 and first.get("pending_after") == 2 and first.get("budget_exhausted") is True, + "a bounded drain must deliver one entry and preserve the rest", + ) + expect(first.get("cursor_after") == "2", "the bounded drain must advance the candidate by one transaction") + rest = drain(workspace) + expect(rest.get("ok") is True and rest.get("delivered") == 2 and rest.get("replayed") == 0 and rest.get("pending_after") == 0, "the next drain must deliver the remaining entries") + expect(rest.get("cursor_after") == "4", "the second drain must finish at cursor 4") + idle = drain(workspace) + expect( + idle.get("ok") is True and idle.get("outcome") == "nothing_pending" and idle.get("delivered") == 0 and idle.get("replayed") == 0, + "an idle drain must deliver and replay nothing", + ) + expect( + idle.get("cursor_before") == "4" and idle.get("cursor_after") == "4" + and idle.get("head_digest") == rest.get("head_digest") and idle.get("provider_revision") == rest.get("provider_revision"), + "an idle drain must leave the candidate head unchanged", + ) + after = backlog(shadow_status(workspace), "todos") + expect(after.get("committed_pending") == 0 and after.get("cursor_last_seq") == 3, "the backlog must be empty at sequence 3") + expect(partition_files(workspace, "todos") == ["drain-cursor.json"], "only the cursor may remain") + transactions = history(workspace) + expect(len(transactions) == 4, "history must hold the bootstrap plus three deliveries") + sequences = [committed_receipt(transaction, label=f"transaction {index}").get("seq") for index, transaction in enumerate(transactions[1:], start=2)] + expect(sequences == [1, 2, 3], "receipts must settle the outbox sequences in order") + listed = _list(goal_cli(workspace, "todo", "list").get("todos"), "todo list") + todo_id = str(_object(listed[0], "listed todo")["todo_id"]) + acquired = delivered( + acquire_lease(workspace, todo_id=todo_id, owner=AGENT_A, idempotency_key="ladder-drain-lease"), + label="task-lease acquire", + ) + replayed = acquire_lease(workspace, todo_id=todo_id, owner=AGENT_A, idempotency_key="ladder-drain-lease") + expect(replayed.get("idempotent") is True, "a same-key re-acquire must be idempotent") + expect("coordination_runtime_shadow" not in replayed, "an idempotent replay must not capture") + expect(len(history(workspace)) == 5, "a writer replay must not add a candidate transaction") + expect(backlog(shadow_status(workspace), "leases").get("committed_pending") == 0, "a writer replay must leave no pending entry") + return passed( + bounded_first_pass=1, + second_pass=2, + idle_pass_delivered=0, + head_digest_stable=True, + lease_partition_cursor=str(_object(acquired.get("drain"), "lease drain").get("last_cursor")), + transactions=5, + ) + + +def row_sigkill_between_primary_write_and_drain(context: RowContext) -> RowOutcome: + """A writer killed around its primary replace leaves a prepared-only entry that drain settles from the primary bytes.""" + + workspace = capture_workspace(context, "ladder-primary-crash") + resolutions: dict[str, str] = {} + visible = 0 + for index, (window, replaced) in enumerate(PRIMARY_CRASH_WINDOWS, start=1): + crash_cli(workspace, window, "todo", "add", "--role", "agent", "--text", f"Primary write killed at {window}.", "--task-class", "advancement_task") + visible += 1 if replaced else 0 + expect(todo_count(workspace) == visible, f"{window}: the primary must show exactly the replaced writes") + pending = backlog(shadow_status(workspace), "todos") + expect(pending.get("committed_pending") == 0 and pending.get("prepared_only") == 1, f"{window}: one prepared-only entry must remain") + expect(phase_count(partition_files(workspace, "todos"), "committed") == 0, f"{window}: no committed marker may exist") + drained = drain(workspace) + expect( + drained.get("ok") is True and drained.get("delivered") == 1 and drained.get("replayed") == 0 and drained.get("no_op") == (0 if replaced else 1), + f"{window}: drain must settle the entry from the primary bytes", + ) + transactions = history(workspace) + expect(len(transactions) == index + 1, f"{window}: every settled entry must be one transaction") + receipt = single_receipt(transactions[-1], label=window) + resolution = str(receipt.get("resolution")) + expect( + resolution == ("committed_proven_by_readback" if replaced else "abandoned") and receipt.get("no_op") is (not replaced), + f"{window}: the receipt must prove the primary outcome", + ) + expect(projected_todo_count(transactions[-1]) == visible, f"{window}: the candidate must equal the primary after recovery") + after = backlog(shadow_status(workspace), "todos") + expect(after.get("prepared_only") == 0 and after.get("cursor_last_seq") == index, f"{window}: the cursor must settle sequence {index}") + resolutions[window] = resolution + inspection = _object(inspect(workspace).get("inspection"), "inspection") + expect(inspection.get("status") == "matched" and inspection.get("parity_matches") is True, "recovered history must match the primary") + return passed(windows=resolutions, primary_todos=visible, transactions=len(PRIMARY_CRASH_WINDOWS) + 1, parity="matched") + + +def row_sigkill_mid_drain(context: RowContext) -> RowOutcome: + """A writer killed inside its inline drain is recovered by the next drain from exact receipts, never delivered twice.""" + + outcomes: dict[str, JsonObject] = {} + for window in DRAIN_CRASH_WINDOWS: + workspace = capture_workspace(context, "ladder-drain-crash") + crash_cli(workspace, window, "todo", "add", "--role", "agent", "--text", f"Drain killed at {window}.", "--task-class", "advancement_task") + expect(todo_count(workspace) == 1, f"{window}: the primary write must be committed") + before = backlog(shadow_status(workspace), "todos") + drained = drain(workspace) + expected_delivered = 1 if window == "before_commit" else 0 + expect( + drained.get("ok") is True and drained.get("outcome") == "drained" + and drained.get("delivered") == expected_delivered and drained.get("replayed") == 1 - expected_delivered, + f"{window}: recovery must deliver an uncommitted entry once and replay a committed one", + ) + transactions = history(workspace) + expect(len(transactions) == 2, f"{window}: the candidate must hold exactly one delivery") + committed_receipt(transactions[1], label=window) + expect(projected_todo_count(transactions[1]) == 1, f"{window}: the delivery must project the committed todo") + after = backlog(shadow_status(workspace), "todos") + expect(after.get("committed_pending") == 0 and after.get("prepared_only") == 0 and after.get("cursor_last_seq") == 1, f"{window}: the backlog must settle at sequence 1") + expect(partition_files(workspace, "todos") == ["drain-cursor.json"], f"{window}: only the cursor may remain") + idle = drain(workspace) + expect(idle.get("outcome") == "nothing_pending" and idle.get("delivered") == 0 and idle.get("replayed") == 0, f"{window}: a further drain must be idle") + outcomes[window] = { + "delivered": drained.get("delivered"), + "replayed": drained.get("replayed"), + "committed_pending_before": before.get("committed_pending"), + "backlog_invalid_before": before.get("invalid"), + } + return passed(windows=outcomes) + + +def row_rollback_with_pending_entries(context: RowContext) -> RowOutcome: + """Rollback archives the outbox with its pending entries, holds capture until an explicit rebootstrap, and replays.""" + + workspace = capture_workspace(context, "ladder-rollback", bootstrap=False) + first = _object(bootstrap_capture(workspace).get("bootstrap"), "first bootstrap") + delivered(add_todo(workspace, "Delivered before rollback."), label="todo add") + with hold_drain_lock(workspace): + deferred(add_todo(workspace, "Committed but not yet drained."), label="todo add (deferred)") + crash_cli(workspace, "before_replace", "todo", "add", "--role", "agent", "--text", "Prepared only.", "--task-class", "advancement_task") + status = shadow_status(workspace) + pending = backlog(status, "todos") + expect(pending.get("committed_pending") == 1 and pending.get("prepared_only") == 1 and pending.get("cursor_last_seq") == 1, "one committed and one prepared-only entry must be pending") + current = candidate(status) + revision = str(current.get("provider_revision")) + expect(current.get("status") == "loaded" and current.get("cursor") == "2", "the candidate must be at cursor 2") + inspection = _object(inspect(workspace).get("inspection"), "inspection") + expect( + inspection.get("qualified") is False and inspection.get("reason_code") == "outbox_pending" and inspection.get("provider_revision") == revision, + "pending entries must block qualification and name the exact revision", + ) + preview = goal_cli(workspace, "coordination-shadow", "rollback", "--provider-revision", revision) + expect(preview.get("ok") is True and preview.get("executed") is False and "rollback" not in preview, "a rollback preview must not execute") + expect(outbox_present(workspace), "a preview must leave the outbox in place") + executed = goal_cli(workspace, "coordination-shadow", "rollback", "--provider-revision", revision, "--execute") + rollback = _object(executed.get("rollback"), "rollback") + expect( + rollback.get("status") == "applied" and rollback.get("active_shadow_removed") is True and rollback.get("archive_retained") is True, + "rollback must retire the active candidate into a retained archive", + ) + expect(rollback.get("archived_cursor") == "2" and rollback.get("archived_provider_revision") == revision, "the archive must bind the exact revision") + archive = rollback.get("outbox_archive_path") + expect(isinstance(archive, str) and Path(archive).is_dir(), "rollback must archive the goal outbox") + assert isinstance(archive, str) + archived = sorted(path.name for path in Path(archive).rglob("*") if path.is_file()) + expect( + phase_count(archived, "prepared") == 2 and phase_count(archived, "committed") == 1 + and "drain-cursor.json" in archived and "manifest.json" in archived, + "the archive must preserve pending entries, the marker, the cursor and the manifest", + ) + expect(not outbox_present(workspace), "the active outbox must be moved, not copied") + status = shadow_status(workspace) + expect(management_status(status) == "inactive" and candidate(status).get("status") == "missing", "management must be inactive with no candidate") + expect(backlog(status, "todos").get("committed_pending") == 0 and backlog(status, "todos").get("prepared_only") == 0, "no backlog may remain") + after = add_todo(workspace, "Primary write after rollback.") + expect(after.get("added") is True, "primary writes must resume after rollback") + held = capture_evidence(after, label="todo add (after rollback)") + expect(held.get("outcome") == "no_transaction" and held.get("reason_code") == "bootstrap_required", "capture must report bootstrap_required, not a transaction") + rebootstrap = bootstrap_capture(workspace) + second = _object(rebootstrap.get("bootstrap"), "second bootstrap") + expect(second.get("capture_lineage_id") != first.get("capture_lineage_id"), "a rebootstrap must start a new lineage") + summary = _object(rebootstrap.get("projection_summary"), "projection summary") + expect(summary.get("todo_count") == 3 and summary.get("lease_count") == 0, "the new baseline must import the current primary") + replay = _object(goal_cli(workspace, "coordination-shadow", "rollback", "--provider-revision", revision, "--execute").get("rollback"), "rollback replay") + expect( + replay.get("status") == "replayed" and replay.get("current_capture_lineage_id") == second.get("capture_lineage_id"), + "a replayed rollback must return its historical result against the new lineage", + ) + resumed = delivered(add_todo(workspace, "Captured in the new lineage."), label="todo add (new lineage)") + expect(_object(resumed.get("drain"), "drain").get("last_cursor") == "2", "the new lineage must capture from cursor 2") + transactions = history(workspace) + expect(len(transactions) == 2 and transactions[0].get("receipts") == [] and projected_todo_count(transactions[0]) == 3, "the new history must start from the imported baseline") + return passed( + pending_before_rollback={"committed": 1, "prepared_only": 1}, + archived_files=len(archived), + archived_prepared=2, + archived_committed=1, + capture_after_rollback="bootstrap_required", + rebootstrap_baseline_todos=3, + new_lineage=True, + replayed_rollback=True, + ) + + +@dataclass +class _MixedWriterLedger: + workspace: GoalWorkspace + mutations: list[tuple[str, JsonObject]] = field(default_factory=list) + deliveries: int = 0 + no_change_writes: int = 0 + + def cli(self, *args: str) -> JsonObject: + return goal_cli(self.workspace, *args) + + def mutate(self, label: str, payload: JsonObject, *, flag: str) -> JsonObject: + expect(payload.get(flag) is True, f"{label} must report {flag}=true") + evidence = delivered(payload, label=label) + # A leased complete or supersede also closes its lease fence, so one + # CLI call may deliver two transactions; count what the drain reports. + count = _object(evidence.get("drain"), f"{label} drain").get("delivered") + expect(isinstance(count, int) and count >= 1, f"{label} must deliver at least one transaction") + assert isinstance(count, int) + self.deliveries += count + self.mutations.append((label, evidence)) + return payload + + def unchanged(self, label: str, payload: JsonObject) -> None: + expect(payload.get("changed") is False, f"{label} must report changed=false") + no_transaction(payload, label=label) + self.no_change_writes += 1 + + +def _mixed_writer_cycle(ledger: _MixedWriterLedger, cycle: int) -> None: + """One cycle of interleaved Python Markdown writers and TypeScript lease writers.""" + + workspace = ledger.workspace + key_a, key_b, key_c = (f"ladder-parity-{cycle}-{suffix}" for suffix in ("a", "b", "c")) + first = ledger.mutate("todo add", add_todo(workspace, f"Cycle {cycle}: deliver one bounded change."), flag="added") + todo_id = str(first["todo_id"]) + acquired = ledger.mutate("task-lease acquire", acquire_lease(workspace, todo_id=todo_id, owner=AGENT_A, idempotency_key=key_a), flag="acquired") + note = ["todo", "update", "--todo-id", todo_id, "--note", f"Cycle {cycle} note.", "--agent-id", AGENT_A] + ledger.mutate("todo update (note)", ledger.cli(*note), flag="changed") + ledger.unchanged("todo update (same note)", ledger.cli(*note)) + renewed = ledger.mutate( + "task-lease renew", + ledger.cli( + "task-lease", "renew", "--todo-id", todo_id, "--owner", AGENT_A, "--idempotency-key", key_a, + "--expected-version", lease_version(acquired, label="acquire"), "--ttl-seconds", "120", + ), + flag="renewed", + ) + transferred = ledger.mutate( + "task-lease transfer", + ledger.cli( + "task-lease", "transfer", "--todo-id", todo_id, "--owner", AGENT_A, "--idempotency-key", key_a, + "--new-owner", AGENT_B, "--new-idempotency-key", key_b, + "--expected-version", lease_version(renewed, label="renew"), "--ttl-seconds", "120", + ), + flag="transferred", + ) + ledger.mutate( + "todo complete", + ledger.cli( + "todo", "complete", "--todo-id", todo_id, "--agent-id", AGENT_B, + "--task-lease-idempotency-key", key_b, "--task-lease-expected-version", lease_version(transferred, label="transfer"), + "--evidence", "validation://ladder-parity", "--no-follow-up", + ), + flag="completed", + ) + second = ledger.mutate("todo add (second)", add_todo(workspace, f"Cycle {cycle}: replace this work with a successor."), flag="added") + second_id = str(second["todo_id"]) + ledger.mutate( + "todo update (explicit exclusion)", + ledger.cli("todo", "update", "--todo-id", second_id, "--excluded-agent", AGENT_B, "--agent-id", AGENT_A), + flag="changed", + ) + clear = ["todo", "update", "--todo-id", second_id, "--clear-excluded-agents", "--agent-id", AGENT_A] + ledger.mutate("todo update (explicit clear)", ledger.cli(*clear), flag="changed") + ledger.unchanged("todo update (clear again)", ledger.cli(*clear)) + held = ledger.mutate("task-lease acquire (second)", acquire_lease(workspace, todo_id=second_id, owner=AGENT_A, idempotency_key=key_c), flag="acquired") + ledger.mutate( + "todo supersede", + ledger.cli( + "todo", "supersede", "--todo-id", second_id, "--agent-id", AGENT_A, "--reason", "Replace obsolete work.", + "--next-agent-todo", f"Cycle {cycle}: carry the bounded work forward.", + "--task-lease-idempotency-key", key_c, "--task-lease-expected-version", lease_version(held, label="acquire (second)"), + ), + flag="superseded", + ) + ledger.mutate( + "todo capture-followups", + ledger.cli( + "todo", "capture-followups", + "--follow-up", f"Cycle {cycle}: verify the captured projection.", + "--evidence", "validation://ladder-parity-followup", + ), + flag="changed", + ) + + +def _assert_bounded_parity(ledger: _MixedWriterLedger, cycle: int, anchor_todo_id: str) -> JsonObject: + workspace = ledger.workspace + inspection = _object(inspect(workspace).get("inspection"), f"cycle {cycle} inspection") + expect(inspection.get("status") == "matched" and inspection.get("parity_matches") is True, f"cycle {cycle}: the candidate head must match the primary") + flags = ["--minimum-operations", str(ledger.deliveries)] + for write_class in PARITY_REQUIRED_WRITE_CLASSES: + flags.extend(["--require-event-kind", write_class]) + qualification = qualified(qualify(workspace, *flags), label=f"cycle {cycle}") + evidence = _object(qualification.get("evidence"), f"cycle {cycle} evidence") + expect(evidence.get("operation_count") == ledger.deliveries, f"cycle {cycle}: every delivered transaction must count as one verified operation") + expect(evidence.get("missing_required_event_kinds") == [] and evidence.get("pending_outbox") is False, f"cycle {cycle}: coverage must be complete with nothing pending") + read = read_candidate(workspace, anchor_todo_id) + result = _object(read.get("read_candidate"), f"cycle {cycle} read-candidate") + expect( + read.get("ok") is True and result.get("status") == "matched" and result.get("read_candidate_qualified") is True, + f"cycle {cycle}: a qualified read must return the anchor todo from the verified head", + ) + return qualification + + +def row_parity_equal(context: RowContext) -> RowOutcome: + """Sustained interleaving of Python and TypeScript writers keeps every bounded qualification matched.""" + + workspace = capture_workspace(context, "ladder-parity") + ledger = _MixedWriterLedger(workspace) + anchor = ledger.mutate("todo add (anchor)", add_todo(workspace, "Anchor todo that stays open for qualified reads."), flag="added") + anchor_todo_id = str(anchor["todo_id"]) + qualification: JsonObject = {} + for cycle in range(1, PARITY_CYCLES + 1): + _mixed_writer_cycle(ledger, cycle) + qualification = _assert_bounded_parity(ledger, cycle, anchor_todo_id) + transactions = history(workspace) + expect(len(transactions) == ledger.deliveries + 1, "history must hold the bootstrap plus one transaction per delivery") + receipts = [committed_receipt(transaction, label=f"transaction {index}") for index, transaction in enumerate(transactions[1:], start=2)] + runtimes = sorted({str(receipt.get("writer_runtime")) for receipt in receipts}) + classes = sorted({str(receipt.get("write_class")) for receipt in receipts}) + expect(runtimes == ["python", "typescript"], "both writer runtimes must appear in one lineage") + expect(set(PARITY_REQUIRED_WRITE_CLASSES) <= set(classes), "every required write class must be captured") + expect(len({str(receipt.get("entry_id")) for receipt in receipts}) == len(receipts), "every receipt must carry a distinct entry id") + return passed( + cycles=PARITY_CYCLES, + mutations=len(ledger.mutations), + deliveries=ledger.deliveries, + no_change_writes=ledger.no_change_writes, + writer_runtimes=runtimes, + write_classes=classes, + final_cursor=str(qualification.get("cursor")), + operation_count=_object(qualification.get("evidence"), "evidence").get("operation_count"), + sustained_parity_verdict=str(qualification.get("sustained_parity_verdict")), + ) + + +def _recover_by_rollback_and_rebootstrap(workspace: GoalWorkspace, *, label: str) -> JsonObject: + """The documented recovery for a held lineage: exact-revision rollback, then a fresh bootstrap.""" + + revision = str(candidate(shadow_status(workspace)).get("provider_revision")) + rollback = _object(goal_cli(workspace, "coordination-shadow", "rollback", "--provider-revision", revision, "--execute").get("rollback"), f"{label} rollback") + expect(rollback.get("status") == "applied" and rollback.get("archive_retained") is True, f"{label}: rollback must retire the held lineage into an archive") + rebootstrap = bootstrap_capture(workspace) + for index in range(3): + delivered(add_todo(workspace, f"{label} recovery write {index}."), label=f"{label} recovery write {index}") + qualified(qualify(workspace), label=f"{label} after recovery") + return _object(rebootstrap.get("projection_summary"), f"{label} rebootstrap summary") + + +def row_parity_divergent_detects_foreign_edit(context: RowContext) -> RowOutcome: + """A direct edit of the primary is detected as drift, holds later captures, and is recovered only by rollback and rebootstrap.""" + + workspace = capture_workspace(context, "ladder-drift") + todo_ids: list[str] = [] + for index in range(3): + added = add_todo(workspace, f"Drift baseline {index}.") + delivered(added, label=f"todo add {index}") + todo_ids.append(str(added["todo_id"])) + qualified(qualify(workspace), label="baseline") + original = workspace.state_path.read_text(encoding="utf-8") + foreign = original.replace("handoff_mode: hard_lease", "handoff_mode: soft_claim") + expect(foreign != original, "the foreign edit must change the primary") + workspace.state_path.write_text(foreign, encoding="utf-8") + inspection = _object(inspect(workspace).get("inspection"), "inspection") + expect( + inspection.get("status") == "drifted" and inspection.get("parity_matches") is False + and inspection.get("reason_code") == "shadow_projection_drift" and inspection.get("qualified") is False, + "inspect must report the drift", + ) + rejected(qualify(workspace), "qualification", label="qualify after drift") + read = read_candidate(workspace, todo_ids[0]) + expect(read.get("ok") is False and rejected(read, "read_candidate", label="read after drift").get("read_candidate_qualified") is False, "no candidate read may qualify after drift") + after = add_todo(workspace, "Public write after the foreign edit.") + expect(after.get("added") is True, "the primary write must still commit") + held = capture_evidence(after, label="todo add (after drift)") + expect(held.get("outcome") == "pending" and held.get("reason_code") == CONTINUITY_HOLD, "the capture must hold on unproven continuity") + expect(_object(held.get("drain"), "held drain").get("outcome") == "stopped", "the inline drain must stop, not deliver") + expect(backlog(shadow_status(workspace), "todos").get("committed_pending") == 1, "the held entry must stay pending") + workspace.state_path.write_text(workspace.state_path.read_text(encoding="utf-8").replace("handoff_mode: soft_claim", "handoff_mode: hard_lease"), encoding="utf-8") + restored = rejected(qualify(workspace), "qualification", label="qualify after restore") + expect(restored.get("reason_code") == "outbox_pending", "restoring the bytes must not requalify a held lineage") + stopped = drain(workspace) + expect( + stopped.get("ok") is False and stopped.get("outcome") == "stopped" and stopped.get("delivered") == 0 and stopped.get("reason_code") == CONTINUITY_HOLD, + "drain must keep holding the entry rather than guess continuity", + ) + summary = _recover_by_rollback_and_rebootstrap(workspace, label="drift") + return passed( + drift_status="drifted", + drift_reason="shadow_projection_drift", + held_write=CONTINUITY_HOLD, + restore_requalifies=False, + recovered_by="rollback_then_bootstrap", + rebootstrap_baseline_todos=summary.get("todo_count"), + recovered_qualification="qualified", + ) + + +def row_event_only_todo_source_holds(context: RowContext) -> RowOutcome: + """An event-only Todo source holds qualification and candidate reads fail-closed; recovery needs rollback and rebootstrap.""" + + workspace = capture_workspace(context, "ladder-event") + todo_ids: list[str] = [] + for index in range(3): + added = add_todo(workspace, f"Markdown baseline {index}.") + delivered(added, label=f"todo add {index}") + todo_ids.append(str(added["todo_id"])) + qualified(qualify(workspace), label="baseline") + log = workspace.state_path.with_name("events.jsonl") + # The product's own state-event store writes the event-only source. It is + # loaded lazily so this strictly typed ladder module does not follow the + # untyped state-event module at type-check time. + state_events = importlib.import_module("loopx.event_sourced_state") + state_events.AppendOnlyStateEventStore(log).append( + state_events.make_state_event( + event_id="ladder-event-only-todo", + goal_id=workspace.goal_id, + event_type=state_events.TODO_ADDED, + refs={"todo_id": "todo_event_only"}, + payload={"role": "agent", "title": "An event-only todo without a Markdown writer.", "task_class": "advancement_task"}, + recorded_at="2026-09-06T00:00:00+00:00", + ) + ) + log_bytes = log.read_bytes() + surfaces = {"inspect": inspect(workspace), "qualify": qualify(workspace), "read-candidate": read_candidate(workspace, todo_ids[0])} + for label, payload in surfaces.items(): + expect(payload.get("ok") is False and payload.get("error") == EVENT_ONLY_HOLD, f"{label} must hold on the unbound event source") + status = shadow_status(workspace) + expect(status.get("ok") is True and management_status(status) == "active", "status must stay readable while the lineage is held") + during = add_todo(workspace, "Markdown write during the event-only hold.") + expect(during.get("added") is True, "the primary write must still commit") + held = capture_evidence(during, label="todo add (during hold)") + expect(held.get("outcome") == "pending" and held.get("reason_code") == CONTINUITY_HOLD, "the capture must hold on unproven continuity") + expect(log.read_bytes() == log_bytes, "the hold must not touch the event log") + expect(backlog(shadow_status(workspace), "todos").get("committed_pending") == 1, "the held entry must stay pending") + log.unlink() + removed = rejected(qualify(workspace), "qualification", label="qualify after removal") + expect(removed.get("reason_code") == "outbox_pending", "removing the event source must not requalify the held lineage") + stopped = drain(workspace) + expect(stopped.get("outcome") == "stopped" and stopped.get("reason_code") == CONTINUITY_HOLD, "drain must keep holding the entry") + summary = _recover_by_rollback_and_rebootstrap(workspace, label="event-only") + return passed( + hold=EVENT_ONLY_HOLD, + held_surfaces=sorted(surfaces), + primary_write_during_hold=CONTINUITY_HOLD, + event_log_untouched=True, + removal_requalifies=False, + recovered_by="rollback_then_bootstrap", + rebootstrap_baseline_todos=summary.get("todo_count"), + ) + + +@dataclass(frozen=True) +class _MigrationTarget: + """The target registry, repository and runtime root of one migration.""" + + registry_path: Path + repo: Path + runtime_root: Path + home: Path + goal_id: str + + def cli_prefix(self) -> list[str]: + return ["--registry", str(self.registry_path), "--runtime-root", str(self.runtime_root), "--format", "json"] + + +MIGRATION_SENTINEL = b'{"schema_version":"existing","goals":[]}\n' + + +def _migration_target(context: RowContext, legacy: GoalWorkspace) -> _MigrationTarget: + root = context.root / "migration-target" + repo = root / "repo" + repo.mkdir(parents=True) + registry_path = root / "registry.json" + registry_path.write_bytes(MIGRATION_SENTINEL) + return _MigrationTarget(registry_path=registry_path, repo=repo, runtime_root=root / "runtime", home=legacy.home, goal_id=unique_goal_id("migrated")) + + +def _migration_arguments(legacy: GoalWorkspace, target: _MigrationTarget) -> list[str]: + return [ + "migrate-state", + "--legacy-registry", str(legacy.registry_path), + "--legacy-runtime-root", str(legacy.runtime_root), + "--target-runtime-root", str(target.runtime_root), + "--goal-id", legacy.goal_id, + "--goal-id-map", f"{legacy.goal_id}={target.goal_id}", + "--path-map", f"{legacy.repo}={target.repo}", + "--copy-active-state", + "--copy-runtime", + "--no-global-sync", + ] + + +def _set_target_runtime_shadow(target: _MigrationTarget, *, enabled: bool) -> JsonObject: + registry = parse_json_object(target.registry_path.read_text(encoding="utf-8")) + goals = [_object(goal, "target goal") for goal in _list(registry.get("goals"), "target goals")] + expect(len(goals) == 1 and goals[0].get("id") == target.goal_id, "the target registry must carry exactly the migrated goal") + coordination = _object(goals[0].get("coordination"), "target coordination") + carried = _object(coordination.get("runtime_shadow"), "carried runtime_shadow") + coordination["runtime_shadow"] = {**carried, "enabled": enabled} + goals[0]["coordination"] = coordination + registry["goals"] = goals + target.registry_path.write_text(json.dumps(registry), encoding="utf-8") + return carried + + +def row_migration_seeds_and_drains(context: RowContext) -> RowOutcome: + """migrate-state refuses an active capture source; after rollback the migrated goal bootstraps a fresh lineage that drains.""" + + legacy = capture_workspace(context, "ladder-legacy", bootstrap=False) + first = _object(bootstrap_capture(legacy).get("bootstrap"), "legacy bootstrap") + delivered(add_todo(legacy, "Captured before migration."), label="legacy todo add") + documents = store_documents(legacy) + expect(len(documents) == 1, "the legacy runtime must hold one candidate document") + legacy_bytes = documents[0].read_bytes() + target = _migration_target(context, legacy) + arguments = _migration_arguments(legacy, target) + preview = run_cli(target, *arguments, check=False) + expect(preview.get("ok") is True and preview.get("dry_run") is True and target.registry_path.read_bytes() == MIGRATION_SENTINEL, "a dry run must plan without writing") + refused = run_cli(target, *arguments, "--execute", check=False) + expect( + refused.get("ok") is False and refused.get("error_code") == "shadow_source_replacement_requires_rebootstrap", + "an active capture source must refuse a generic state rebuild", + ) + expect(target.registry_path.read_bytes() == MIGRATION_SENTINEL and documents[0].read_bytes() == legacy_bytes, "a refused migration must write nothing") + set_runtime_shadow(legacy, enabled=False) + disabled_only = run_cli(target, *arguments, "--execute", check=False) + expect(disabled_only.get("error_code") == "shadow_source_replacement_requires_rebootstrap", "disabling capture alone must not release an active lineage") + set_runtime_shadow(legacy, enabled=True) + revision = str(candidate(shadow_status(legacy)).get("provider_revision")) + rollback = _object(goal_cli(legacy, "coordination-shadow", "rollback", "--provider-revision", revision, "--execute").get("rollback"), "legacy rollback") + expect(rollback.get("status") == "applied" and rollback.get("archive_retained") is True, "the legacy lineage must be retired into an archive first") + set_runtime_shadow(legacy, enabled=False) + executed = run_cli(target, *arguments, "--execute") + expect(executed.get("ok") is True and executed.get("wrote_project_registry") is True, "the migration must execute after rollback") + runtime_goal = _object(_list(executed.get("runtime_goals"), "runtime goals")[0], "runtime goal") + expect(runtime_goal.get("copied") is True, "the runtime goal directory must be copied") + expect(executed.get("authority_shadow_seeds") == [], "a capture-only goal plans no observation seed") + expect((target.repo / legacy.state_path.name).exists(), "the active state must be copied to the mapped repository") + carried = _set_target_runtime_shadow(target, enabled=True) + expect(carried.get("enabled") is False and carried.get("provider") == RUNTIME_SHADOW_PROVIDER, "the migrated goal must carry its disabled capture configuration") + inspection = _object(run_cli(target, "coordination-shadow", "inspect", "--goal-id", target.goal_id, check=False).get("inspection"), "target inspection") + expect(inspection.get("status") == "missing" and inspection.get("bootstrap_required") is True, "the migrated goal must require its own bootstrap") + bootstrap_payload = run_cli(target, "coordination-shadow", "bootstrap", "--goal-id", target.goal_id, "--execute") + bootstrap = _object(bootstrap_payload.get("bootstrap"), "target bootstrap") + expect(bootstrap.get("status") == "applied" and bootstrap.get("capture_lineage_id") != first.get("capture_lineage_id"), "the target must seed a fresh lineage") + summary = _object(bootstrap_payload.get("projection_summary"), "target projection summary") + expect(summary.get("todo_count") == 1 and summary.get("lease_count") == 0, "the fresh lineage must import the migrated baseline") + added = run_cli(target, "todo", "add", "--goal-id", target.goal_id, "--role", "agent", "--text", "Captured in the migrated runtime root.", "--task-class", "advancement_task") + drained = _object(delivered(added, label="target todo add").get("drain"), "target drain") + expect(drained.get("last_cursor") == "2", "the first migrated write must drain to cursor 2") + qualification = _object(run_cli(target, "coordination-shadow", "qualify", "--goal-id", target.goal_id, "--minimum-operations", "1", check=False).get("qualification"), "target qualification") + expect(qualification.get("status") == "qualified" and _object(qualification.get("evidence"), "evidence").get("operation_count") == 1, "the migrated lineage must qualify on its own write") + archive = legacy.runtime_root / "authority-shadow" / "file-v0" / "rollback" + expect(archive.is_dir() and any(archive.iterdir()), "the legacy archive must be retained") + return passed( + preview_dry_run=True, + active_capture_refused="shadow_source_replacement_requires_rebootstrap", + disabled_config_still_refused=True, + executed_after_rollback=True, + observation_seeds=0, + migrated_baseline_todos=1, + new_lineage=True, + drained_cursor="2", + legacy_archive_retained=True, + ) + + +def row_growth_measurement_gate(context: RowContext) -> RowOutcome: + """Measure file-v0 history growth per transaction and gate its integrity; no capacity horizon is claimed.""" + + workspace = capture_workspace(context, "ladder-growth") + status = shadow_status(workspace) + sizes = [int(str(status.get("store_bytes")))] + cursors = [str(candidate(status).get("cursor"))] + for index in range(GROWTH_TRANSACTIONS): + delivered(add_todo(workspace, GROWTH_TEXT_TEMPLATE % index), label=f"growth write {index}") + status = shadow_status(workspace) + expect(status.get("retention_pressure") is False, "the bounded workload must not trip retention pressure") + sizes.append(int(str(status.get("store_bytes")))) + cursors.append(str(candidate(status).get("cursor"))) + expect(cursors == [str(index) for index in range(1, GROWTH_TRANSACTIONS + 2)], "every write must advance the cursor by exactly one") + deltas = [after - before for before, after in zip(sizes, sizes[1:])] + expect(all(delta > 0 for delta in deltas), "every transaction must grow the retained history") + accelerations = [later - earlier for earlier, later in zip(deltas, deltas[1:])] + expect( + max(abs(value) for value in accelerations) <= GROWTH_DELTA_ACCELERATION_ENVELOPE_BYTES, + "per-transaction growth may increase by at most one live record per transaction", + ) + transactions = history(workspace) + expect(len(transactions) == GROWTH_TRANSACTIONS + 1, "history must retain every transaction") + expect( + [projected_todo_count(transaction) for transaction in transactions] == list(range(GROWTH_TRANSACTIONS + 1)), + "every transaction must retain its complete projection", + ) + final = sizes[-1] + cumulative = sum(sizes) + return passed( + transactions=GROWTH_TRANSACTIONS + 1, + final_history_bytes=final, + cumulative_publication_bytes=cumulative, + publication_rewrite_ratio=round(cumulative / final, 2), + first_delta_bytes=deltas[0], + last_delta_bytes=deltas[-1], + delta_acceleration_max_bytes=max(abs(value) for value in accelerations), + delta_acceleration_envelope_bytes=GROWTH_DELTA_ACCELERATION_ENVELOPE_BYTES, + retention_pressure=False, + capacity_verdict="not_evaluated", + qualification_horizon="not_claimed", + ) + + +__all__ = [ + "CRASH_WORKER", + "DRAIN_CRASH_WINDOWS", + "GROWTH_DELTA_ACCELERATION_ENVELOPE_BYTES", + "GROWTH_TRANSACTIONS", + "PARITY_CYCLES", + "PARITY_REQUIRED_WRITE_CLASSES", + "PRIMARY_CRASH_WINDOWS", + "row_drain_idempotent", + "row_event_only_todo_source_holds", + "row_growth_measurement_gate", + "row_migration_seeds_and_drains", + "row_outbox_prepared_then_committed_entries", + "row_parity_divergent_detects_foreign_edit", + "row_parity_equal", + "row_rollback_with_pending_entries", + "row_sigkill_between_primary_write_and_drain", + "row_sigkill_mid_drain", +] diff --git a/pyproject.toml b/pyproject.toml index a099f40e77..4258f9ecf8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -145,6 +145,7 @@ files = [ "loopx/control_plane/testing/authority_e2e_ladder.py", "loopx/control_plane/testing/authority_e2e_row_support.py", "loopx/control_plane/testing/authority_e2e_rows_stage2c.py", + "loopx/control_plane/testing/authority_e2e_rows_stage2c2.py", "loopx/control_plane/work_items/delivery_batch_scale.py", "loopx/control_plane/work_items/delivery_outcome.py", "loopx/control_plane/work_items/lifecycle.py", diff --git a/tests/control_plane/test_shared_goal_authority_e2e.py b/tests/control_plane/test_shared_goal_authority_e2e.py index 009dd9024c..b61f982a21 100644 --- a/tests/control_plane/test_shared_goal_authority_e2e.py +++ b/tests/control_plane/test_shared_goal_authority_e2e.py @@ -31,7 +31,8 @@ "s2a.nokv_live_qualification", "s2b.postgresql_conformance_live", ) -PENDING_ONLY_ROW_ID = "s2c2.parity_equal" +PENDING_ONLY_ROW_ID = "s2c2.sustained_parity_soak" +PENDING_ROW_IDS = ("s2c2.archive_after_leased_completion_parity", PENDING_ONLY_ROW_ID) CHEAP_DETERMINISTIC_ROW_ID = "s0.file_matrix_twelve_rows" FULL_LADDER_VARIABLE = "LOOPX_LADDER_FULL" # Rows whose assertions the in-repo CLI E2E suite already pins through the same @@ -64,7 +65,11 @@ "pinned by tests/control_plane/test_local_authority_shadow_cli_e2e.py; " "run examples/shared-goal-authority-e2e/ladder.py or set LOOPX_LADDER_FULL=1" ) -REQUIRED_PENDING_ROW_IDS = ( +# The Stage 2C parity half: every row below is executable through the public +# CLI and the shadow management interfaces. They carry the ``stage2c_e2e`` +# marker so CI runs them in the stage2c correctness job with the other real +# CLI, process-death and recovery suites instead of the pytest shards. +STAGE_2C2_ROW_IDS = ( "s2c2.outbox_prepared_then_committed_entries", "s2c2.drain_idempotent", "s2c2.sigkill_between_primary_write_and_drain", @@ -72,9 +77,15 @@ "s2c2.rollback_with_pending_entries", "s2c2.parity_equal", "s2c2.parity_divergent_detects_foreign_edit", + "s2c2.event_only_todo_source_holds", "s2c2.migration_seeds_and_drains", "s2c2.growth_measurement_gate", ) +STAGE_2C2_POSIX_ONLY_ROW_IDS = ( + "s2c2.sigkill_between_primary_write_and_drain", + "s2c2.sigkill_mid_drain", + "s2c2.rollback_with_pending_entries", +) def _row_parameters() -> Iterator[object]: @@ -87,6 +98,8 @@ def _row_parameters() -> Iterator[object]: ) if row.id in CLI_E2E_COVERED_ROW_IDS and not full_ladder: marks.append(pytest.mark.skip(reason=CLI_E2E_COVERAGE_REASON)) + if row.stage == "2c2": + marks.append(pytest.mark.stage2c_e2e) yield pytest.param(row, id=row.id, marks=marks) @@ -117,10 +130,15 @@ def test_registry_vocabulary_and_pending_rows_are_declared_not_claimed() -> None pending_ids = [row.id for row in ladder.PENDING_ROWS] assert len(set(row_ids)) == len(row_ids) assert set(row_ids).isdisjoint(pending_ids) - assert set(REQUIRED_PENDING_ROW_IDS) <= set(pending_ids) - assert {row.stage for row in ladder.LADDER_ROWS} == {"0", "1", "2a", "2b", "2c1"} + assert [row.id for row in ladder.LADDER_ROWS if row.stage == "2c2"] == list(STAGE_2C2_ROW_IDS) + assert pending_ids == list(PENDING_ROW_IDS) + assert {row.stage for row in ladder.LADDER_ROWS} == {"0", "1", "2a", "2b", "2c1", "2c2"} assert {row.stage for row in ladder.PENDING_ROWS} == {"2c2"} assert all("#3819" not in row.pending_until for row in ladder.PENDING_ROWS) + for row_id in STAGE_2C2_ROW_IDS: + row = ladder.row_by_id(row_id) + assert row.product_path == "real_cli" and row.gate == "deterministic" + assert row.posix_only is (row_id in STAGE_2C2_POSIX_ONLY_ROW_IDS) assert all(row.gate in ladder.GATES for row in ladder.LADDER_ROWS) assert all(row.product_path in ladder.PRODUCT_PATHS for row in ladder.LADDER_ROWS) with pytest.raises(ValueError): @@ -261,8 +279,11 @@ def test_pending_rows_never_exit_green_without_allow_pending( assert allowed["summary"]["pending"] == 1 capsys.readouterr() - # A whole pending stage behaves the same way. - assert ladder.main(["--stage", "2c2", "--report-json", str(report_path)]) == 1 + # A stage whose only remaining declaration is pending behaves the same way; + # the executable 2c2 rows are removed from the registry for this check only. + with pytest.MonkeyPatch.context() as registry: + registry.setattr(ladder, "LADDER_ROWS", tuple(row for row in ladder.LADDER_ROWS if row.stage != "2c2")) + assert ladder.main(["--stage", "2c2", "--report-json", str(report_path)]) == 1 capsys.readouterr() # Mixed selection: one executable pass does not excuse a pending obligation. @@ -378,5 +399,6 @@ def test_list_prints_rows_and_pending_declarations( assert ladder.main(["--list", "--stage", "2c2"]) == 0 stage_listing = json.loads(capsys.readouterr().out) - assert stage_listing["rows"] == [] + assert [row["id"] for row in stage_listing["rows"]] == list(STAGE_2C2_ROW_IDS) + assert [row["id"] for row in stage_listing["pending"]] == list(PENDING_ROW_IDS) assert {row["stage"] for row in stage_listing["pending"]} == {"2c2"}