From 2c6afa9e183dd6a98d8906a41ca44ef7bb6dcea1 Mon Sep 17 00:00:00 2001 From: fy59 Date: Tue, 10 Nov 2026 18:31:11 +0100 Subject: [PATCH] Fix generation capacity recovery with causal sibling proof and immutable retry --- CHANGELOG.md | 11 + docs/current_state.md | 22 ++ .../sync_generation_capacity_recovery_v1.md | 188 +++++++++++++++ docs/roadmap.md | 6 + docs/sync_exchange.md | 18 +- docs/tests.md | 17 ++ tests/test_sync_generation_exchange.py | 225 ++++++++++++++++++ tools/sync_generation_exchange.py | 84 +++++-- tools/sync_peer_worker.py | 29 ++- 9 files changed, 567 insertions(+), 33 deletions(-) create mode 100644 docs/reviews/sync_generation_capacity_recovery_v1.md diff --git a/CHANGELOG.md b/CHANGELOG.md index a978184..54ca740 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ ## Unreleased — 0.1.7 +- Fix desktop full-generation capacity exhaustion caused by retained pending + branches after interrupted conversations. Exclude a branch from active + admission only with exact durable acknowledged-sibling evidence, retain all + immutable files and causal records, and resume the unique unresolved + generation with its original manifest/ACK correlation. Keep the limit at 16 + and fail closed on divergent unresolved branches. No schema, Android data, + frozen exchange format or fabricated acknowledgement is involved. The local + helper rollout completed historical recovery and distinct current-state and + idempotence exchanges with authentic Android ACKs; both new cable exercises + returned from Android and existing business records remained intact. + - Present Sleep PDF nights and completed-sport days newest first while preserving chronological timed content within each night and sport day. Add a restrained diff --git a/docs/current_state.md b/docs/current_state.md index f00128a..1fc8728 100644 --- a/docs/current_state.md +++ b/docs/current_state.md @@ -1,5 +1,27 @@ # Current implementation state +## Generation capacity recovery + +Desktop full-generation admission now excludes retained pending branches only +when an exact durable consumed ACK proves a distinct sibling for the same +producer, consumer and parent. The limit remains 16. The worker replays the +unique unresolved immutable generation across conversation runs and accepts +its original-run ACK; it rejects ambiguous unresolved branches. This changes +no database schema, domain artifact, Android data or frozen V1 format. + +The 2026-11-10 incident was reproduced on a consistent private database copy: +13 of 14 pending generations had this sibling proof; the remaining generation +was recoverable by exact replay. All 14 immutable generations verified, and a +complete current capture succeeded on the copy without changing business +values. See the [recovery record](reviews/sync_generation_capacity_recovery_v1.md) +for deployment, controlled-run evidence and validation limits. The isolated +helper package is active for the Web and sync services. Historical replay, +current-state exchange and a distinct idempotence run completed with authentic +ACKs. Android returned all 46 exercises, 31 sessions and 7 body observations, +including the two newly added cable exercises. The 13 superseded historical +branches remain preserved; the active window is three with no unresolved retry. +The mission is PASS; the review records the unrelated baseline TUI-test failure. + ## Audited development line and private installation This 0.1.7 correction joins the audited `f4ddcfd` development lineage, diff --git a/docs/reviews/sync_generation_capacity_recovery_v1.md b/docs/reviews/sync_generation_capacity_recovery_v1.md new file mode 100644 index 0000000..7526739 --- /dev/null +++ b/docs/reviews/sync_generation_capacity_recovery_v1.md @@ -0,0 +1,188 @@ +# Generation capacity recovery V1 + +## Cause and preserved boundary + +The 2026-11-10 private desktop incident reached its unchanged limit of 16: +14 `waiting_acknowledgement` generations plus two acknowledged recovery +members. Recovery retrieved 32 authentic Android ACKs, none for those 14. +The desktop worker always captured a new generation after inbound consumption, +even after a previous conversation had interrupted publication or ACK handling. +Later successful siblings advanced from the same acknowledged parent while +old pending branches continued consuming admission. The desktop admission SQL +also omitted the acknowledged-sibling exclusion already documented and used +by Android. Archiving acknowledged generations alone cannot repair this state. + +The incident consumed an Android generation before desktop capture failed. +That inbound commit remains durable; the run is failed, never successful. +The retained Android `peer_capacity_exhausted` error has a different run ID +from this incident and is historical evidence, not proof of current saturation. + +## Recovery invariant + +A pending branch may leave active admission only if a distinct sibling has the +same producer, consumer and parent (including a null root), is acknowledged, +and owns a consumed `sqlite-commit-full` ACK matching its exact original run, +peer identities and manifest SHA-256. Archived siblings remain valid proof. +Neither elapsed time nor ACK absence proves non-consumption. All generation +rows, manifests, artifacts, hashes, causal relationships and memberships remain +unchanged. No sibling ACK advances the pending branch's domain publication +state. Its own authentic late ACK remains accepted and replay-protected. + +The worker resumes the unique unresolved branch using verified immutable +bytes. Its mutable reference uses the current conversation run while the +manifest and ACK use their original run. ACK selection additionally correlates +the generation ID and manifest digest; the production validator still checks +all ACK fields. Multiple unresolved branches fail closed. The limit remains +16 and no schema or frozen exchange format changes. + +## Private copy validation + +The existing backup and live database both passed SQLite `quick_check`. +A fresh SQLite backup, complete owned/transport directory copies, table-count +inventory and reproduction evidence are retained privately under: + +`~/.local/share/trainlog/backups/capacity-recovery-20261110T181125/` + +The original source reproduced `generation retention capacity exhausted` on a +separate backup. The corrected predicate identifies 13 superseded branches, +leaving three active members: two acknowledged and one unresolved. All 14 +pending immutable generations passed manifest and artifact verification. The +one unresolved generation was successfully republished byte-for-byte on the +copy. A current complete production capture then succeeded, including Programs +V2 and Sleep. Fingerprints covered 56 non-sync tables: the only capture change +was the expected Sleep generation-membership insertion. Session, Program, +body, Sleep facts, heart-rate and Cardio values remained unchanged. + +No personal payload or database is committed as a fixture. + +## Automated validation + +- C17 strict-warning desktop build, including the embedded Web frontend: pass. +- `test_sync_generation_exchange.py`: 21 tests pass, including new capacity, + exact-evidence rejection, late ACK, immutable replay, corruption rejection, + divergent pending branches, committed inbound data surviving outbound + saturation/restart, and archival regressions. +- Full Meson suite with Web enabled: 113/114 pass. The remaining failure is + `tui_workflows`, `tui/tests/test_tui_workflows.c:1680`, asserting three + Dashboard performance points in a dated fixture. Its C sources and binary + inputs are unchanged by this Python-only correction; this is an existing + baseline limitation, not claimed as a passing test. +- JSON catalog/schema validation and import-contract validation: pass. +- Changed C/H format check: pass (no changed C/H files). +- No C implementation changed, so a new C sanitizer checkpoint is not required. + +The first full-suite configuration lacked installed Web dependencies and +failed the deployment-package test. After `npm ci` and Web-enabled rebuild, +that test passed. No production warning policy was weakened. + +## Local deployment and rollback + +The deployed helper directory is separate from development and previous +rollouts: + +`~/.local/share/trainlog/rollouts/sync-capacity-recovery-v1-292bcc8/tools/` + +It copies the daily Web rollout's tools and reference catalog with only the two +corrected helpers replaced; a SHA-256 inventory is retained beside it. Both user services use +`90-capacity-recovery.conf` drop-ins. Existing desktop binaries, Web assets and +transport adapters remain in their previous rollouts. Services were stopped +for a fresh coherent predeployment backup and restarted after configuration +publication. A private `rollback-services.sh` removes only these drop-ins and +restarts the previous services; it does not roll back business data or delete +any backup or generation evidence. + +The initial deployment attempt showed that the daily Web shell launcher +unconditionally overwrites `TRAINLOG_SYNC_TOOLS_DIR`. The resulting first +controlled attempt still used the old helpers and failed after inbound commit. +Its failure evidence is retained. The corrected drop-ins bypass that wrapper +using its same underlying binary and explicit adapter paths. The sync daemon +uses the isolated tools and an explicit orchestrator path. Actual `/proc` +environment and command lines verified both paths after restart. No previous +rollout files were edited. + +A second attempt received a current Android generation but exposed a missing +reference catalog in the isolated deployment, before its import transaction. +The original catalog was copied into the isolated package. The deployed +consumer then accepted that exact Android generation on a fresh database copy, +and all 14 production capture participants exported successfully using that +same deployed package on the copy. Its private test ACK was never published to +the phone. The failed attempt remains recorded as failed. + +## Physical validation + +The Android-triggered recovery run +`sy_f9f0f9eb-e804-4149-a438-755ce0f3fdac` completed over Bluetooth on +2026-11-10. It consumed Android generation +`gen_5f1d5c2c-d87c-4f35-9ca1-2ea3254c65c1` and replayed the exact pending +desktop generation `gen_cf95dd3b-6fff-4791-9101-e0eceb24ee7d`. Android returned +its authentic original-run consumed ACK; the desktop ledger corroborated both +halves. The active count is three and the 13 superseded pending branches +remain retained. This proves historical recovery, not publication of current +PC edits. + +After the initial deployment faults were repaired, a Web waiter and the +phone's new explicit conversation had incompatible freshness baselines. The +stale waiting orchestrator was sent SIGTERM, recorded `interrupted`, released +its owned worker and lock, and preserved evidence. The Android-triggered daemon +then completed the recovery. A concurrent Web attempt correctly failed the +common lock instead of starting a second engine. A later Web waiter was also +interrupted before import so the remaining controlled validations could be +conducted through the phone's explicit trigger and daemon ledger alone. + +Before current-snapshot convergence, the returned Android snapshot contained +31 sessions, 7 body observations and 44 exercises. Desktop had 46 exercises, +including the two requested new ones; their actual Android return remains a +separate convergence check. All 5,817 pre-existing owned immutable files were +verified unchanged. The live non-sync table comparison found only Sleep +publication validation/acknowledgement timestamp updates, with no change to +business values or revision identities. +No Android APK or app data has been changed. ADB exposes no connected device; +the paired Bluetooth device is connected. The transport advertisement includes +manifest/ACK/archive/recovery V1, history V4 and Programs V1/V2 capabilities. +An advertisement is protocol evidence, not proof of the installed APK version. + +The current-snapshot run `sy_d18e8160-2917-458f-a1ce-3b3c58927a1b` +completed in both directions at 17:26:45 UTC. Its outbound +`gen_27f564c0-155b-41ba-8906-768894235499` has an authentic Android consumed +ACK. The user also confirmed completion without error on the phone. Comparing +all 56 non-sync tables against the predeployment backup found no business-value +or revision changes or deleted keys: only validation/ACK timestamps and 11 new +Sleep generation-membership rows changed. Existing sessions, Programs and +planning, body data, Sleep facts, heart-rate and Cardio tables were unchanged. + +The distinct idempotence run `sy_97e963d7-189f-4c44-b49d-bf339b14347d` +also completed in both directions, with outgoing +`gen_283f98e0-13fe-4def-ae9d-ad5b10f3063f`. Both current outgoing manifests +contain Programs V2 and Sleep and have exact authentic consumed ACKs; each +inbound generation has a committed consumed ledger row. The user confirmed +that this attempt entered progress and completed, and that both requested +exercises are visible in Android. A received Android history artifact also +contains both names, 46 exercises, 31 sessions and 7 body observations; all +three stable-ID sets equal the canonical desktop sets. + +The final 56-table comparison against the snapshot taken before idempotence +found no change to business values or revisions. Expected synchronization +metadata changes were limited to Sleep validation/ACK timestamps, 11 additional +Sleep generation-membership rows, and two body-zone synchronization markers +for the two newly propagated exercises. Existing body-zone rows were retained. +No session, set, Program, body, Sleep fact, heart-rate sample or Cardio row was +deleted or duplicated. Desktop retains 517 sets, 6 Programs, 98 Program +sessions, 23 execution links, 17 Sleep entries, 19 heart-rate captures and +93,903 samples plus 93,903 RR intervals. SQLite `quick_check` remains `ok`. +All 13 historical pending generations still validate from their immutable +files, and no unresolved unsuperseded desktop generation remains. The active +window is three after capture (the normal next archive pass returns it to two); +the unchanged admission limit is 16. + +Private proof owners include `final-convergence.json`, `android-roundtrip.json`, +the three completed-run reports and their authentic ACK files, coherent +predeployment/before-idempotence/final SQLite backups, copied immutable trees, +validation logs, installed-file digests and rollback instructions. The Web's +separate last waiter remains truthfully `interrupted`; the Android-triggered +service ledger owns the successful controlled runs. No success status was +fabricated or copied into a different conversation's history. + +`TRAINLOG_SYNC_GENERATION_CAPACITY_RECOVERY_V1=PASS` applies to this correction, +its deployment and the controlled bidirectional convergence/idempotence checks. +It does not claim the unrelated baseline TUI test passes. No Android APK, +identity, app data, database schema, or remote Git repository was changed. diff --git a/docs/roadmap.md b/docs/roadmap.md index 2a45b36..a06c7ba 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -22,6 +22,12 @@ This document owns future work. The implemented baseline is summarized in [current_state.md](current_state.md); completed narratives and evidence belong in [reviews](reviews/) and [CHANGELOG.md](../CHANGELOG.md). +The desktop generation-capacity correction preserves pending evidence outside +the active window only with exact acknowledged-sibling proof and resumes the +unique unresolved generation. It does not expand the default V3 rollout. +Current recovery validation and any remaining physical-device work are owned by +[current_state.md](current_state.md) and the linked recovery record. + ## Product boundary ```text diff --git a/docs/sync_exchange.md b/docs/sync_exchange.md index 294ba65..56ef784 100644 --- a/docs/sync_exchange.md +++ b/docs/sync_exchange.md @@ -1010,7 +1010,8 @@ transaction while calling the existing domain exporters. Both validate domain and outer limits before recording a captured generation. Publication copies verified immutable bytes into `generations//`, resumes only byte-identical partial publication, and writes `manifest.json` last. Capacity -admits at most eight active outgoing generations per consumer. Acknowledged +admits at most sixteen desktop and eight Android active outgoing generations +per consumer. Acknowledged generations older than the two newest lineage members leave active admission only after their exact consumed ACK, manifest, artifact sizes and digests have been reconciled and a complete atomic archive copy has been verified and @@ -1020,12 +1021,19 @@ active. Archive directories are outside bounded MTP polling and are never uploaded recursively. Failure or interruption leaves admission closed rather than fabricating an ACK or discarding evidence. -An unacknowledged generation whose parent also has a distinct, ledger-backed -acknowledged child is a retained superseded branch: it stays available for +An unacknowledged generation whose parent (including the null root) also has a +distinct acknowledged child for the same producer and consumer, with an exact +consumed `sqlite-commit-full` ACK matching its original run, peers and manifest +digest, is a retained superseded branch: it stays available for audit and is never deleted or assigned a fabricated ACK, but it no longer consumes active admission because it cannot become the causal tip. After a -transport reset, the next run republishes the newest unsuperseded immutable -generation for that peer before attempting a new capture. The mutable +transport reset, the next run republishes the unique unsuperseded immutable +generation for that peer before attempting a new capture. Multiple unresolved +branches fail closed; a timestamp never selects a winner. Desktop admission and +retry selection share this same proof predicate, including archived siblings. +All pending rows, payloads and publication memberships remain retained and +can still receive their own authentic late ACK. No sibling ACK acknowledges +the superseded generation or its domain publication state. The mutable generation reference uses the new conversation run; generation ID, manifest, artifact bytes and eventual ACK correlation remain unchanged. Because the immutable manifest and its durable ACK retain the original run identity, a diff --git a/docs/tests.md b/docs/tests.md index 1b8c429..bebd5d5 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -2132,3 +2132,20 @@ and the complete Web inventory **191/191** with typecheck and production build. JSON validators, changed-file formatting, and `git diff --check` also pass. This is software evidence only: it records no device rollout and authorizes no stable release. + +## Generation capacity recovery regressions + +`tests/test_sync_generation_exchange.py` covers the unchanged desktop limit +(15 pending generations admit one more; 16 reject), the incident-shaped 14 +pending branches with and without exact acknowledged-sibling evidence, wrong +peer/run/digest/result/durability evidence, null-parent branches, interrupted +publication and restart, byte-identical retry, authentic late ACK replay, +ambiguous unresolved branches, committed inbound facts surviving outgoing +capacity failure and restart, and archive/ACK evidence preservation. Fixtures +are fictitious and inspect actual ledger rows and immutable bytes. + +The recovery incident additionally validates the production codecs on a private +SQLite backup, with immutable-file digests and before/after business-table +fingerprints. Personal database files and payloads are never test fixtures or +Git artifacts. Physical-device convergence is recorded separately from these +automated checks in the recovery review record. diff --git a/tests/test_sync_generation_exchange.py b/tests/test_sync_generation_exchange.py index 3691565..1fbdc57 100644 --- a/tests/test_sync_generation_exchange.py +++ b/tests/test_sync_generation_exchange.py @@ -479,4 +479,229 @@ class GenerationTest(unittest.TestCase): self.assertEqual(0,db.execute("SELECT COUNT(*) FROM facts WHERE id='partial'").fetchone()[0]) self.assertEqual(1,db.execute("SELECT COUNT(*) FROM sync_consumed_generations WHERE result='rejected'").fetchone()[0]) + +class CapacityRecoveryTest(unittest.TestCase): + """Fictitious immutable generations exercise production admission/recovery.""" + + def setUp(self): + GenerationTest.setUp(self) + with closing(sqlite3.connect(self.database)) as db: + db.execute("INSERT OR REPLACE INTO sync_peer_identity VALUES(1,?,'desktop')", (PEER_A,)) + db.commit() + + tearDown = GenerationTest.tearDown + artifacts = staticmethod(GenerationTest.artifacts) + + def new_capture(self): + import uuid + + def exporter(_tool, _extra, output, snapshot): + with closing(sqlite3.connect(snapshot)) as db: + facts = dict(db.execute("SELECT id,value FROM facts")) + output.write_text(json.dumps({ + "format": "trainlog-mobile-export", "version": 4, "facts": facts, + })) + + with mock.patch.object(generation, "ARTIFACTS", self.artifacts()), \ + mock.patch.object(generation, "SUPPORTED", {("trainlog-mobile-export", 4)}), \ + mock.patch.object(generation, "run_export", exporter): + return generation.capture_desktop( + self.database, self.root / "owned", PEER_B, + "sy_" + str(uuid.uuid4()), "gen_" + str(uuid.uuid4()), + ) + + def pending_rows(self, count, parent=None): + import uuid + + identities = [] + with closing(sqlite3.connect(self.database)) as db: + for _ in range(count): + identity = "gen_" + str(uuid.uuid4()) + identities.append(identity) + db.execute( + "INSERT INTO sync_generations VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", + (identity, RUN, PEER_A, PEER_B, "desktop", "2026-01-01", + parent, "a" * 64, "waiting_acknowledgement", "{}", "unused", None), + ) + db.commit() + return identities + + def consumed_ack(self, captured): + _, manifest, checksum = captured + destination = self.root / "consumer.sqlite" + with closing(sqlite3.connect(destination)) as db: + if not db.execute("SELECT name FROM sqlite_master WHERE name='facts'").fetchone(): + db.executescript(SCHEMA) + db.execute("INSERT INTO sync_peer_identity VALUES(1,?,'desktop')", (PEER_B,)) + ack = generation.record_consumed(db, manifest, checksum) + db.commit() + return generation.canonical(ack) + + def publish_captured(self, captured): + with mock.patch.object(generation, "ARTIFACTS", self.artifacts()), \ + mock.patch.object(generation, "SUPPORTED", {("trainlog-mobile-export", 4)}): + return generation.publish( + self.database, captured[1]["generation_id"], self.root / "objects", + ) + + def test_fifteen_admits_sixteenth_and_sixteen_rejects(self): + self.pending_rows(15) + captured = self.new_capture() + self.assertTrue((captured[0] / "manifest.json").is_file()) + with self.assertRaisesRegex(generation.GenerationError, "capacity exhausted"): + self.new_capture() + with closing(sqlite3.connect(self.database)) as db: + self.assertEqual(16, generation.active_generation_count(db, PEER_B)) + self.assertEqual(16, db.execute("SELECT COUNT(*) FROM sync_generations").fetchone()[0]) + + def test_fourteen_orphans_require_correlated_sibling_and_preserve_evidence(self): + pending = self.pending_rows(14) + sibling = self.new_capture() + self.publish_captured(sibling) + raw = self.consumed_ack(sibling) + with closing(sqlite3.connect(self.database)) as db: + before = db.execute( + "SELECT * FROM sync_generations WHERE generation_id<>? ORDER BY generation_id", + (sibling[1]["generation_id"],), + ).fetchall() + self.assertEqual(15, generation.active_generation_count(db, PEER_B)) + self.assertEqual("acknowledged", generation.accept_ack_bytes(self.database, raw)) + with closing(sqlite3.connect(self.database)) as db: + self.assertEqual(1, generation.active_generation_count(db, PEER_B)) + self.assertEqual(before, db.execute( + "SELECT * FROM sync_generations WHERE generation_id<>? ORDER BY generation_id", + (sibling[1]["generation_id"],), + ).fetchall()) + # A wrong peer, run, digest or rejected result cannot prove a sibling. + for column, invalid in ( + ("consumer_peer_id", PEER_A), ("producer_peer_id", PEER_B), + ("run_id", RUN), ("manifest_sha256", "0" * 64), + ("result", "rejected"), ("durability", "none"), + ): + db.execute("SAVEPOINT invalid_evidence") + db.execute("UPDATE sync_acknowledgements SET " + column + "=?", (invalid,)) + self.assertEqual(15, generation.active_generation_count(db, PEER_B)) + db.execute("ROLLBACK TO invalid_evidence") + db.execute("RELEASE invalid_evidence") + self.assertEqual(14, len(pending)) + self.assertEqual("unchanged", generation.accept_ack_bytes(self.database, raw)) + + def test_interrupted_publication_restart_and_late_ack(self): + captured = self.new_capture() + published = self.publish_captured(captured) + original = {p.name: p.read_bytes() for p in published.iterdir()} + with mock.patch.object(generation, "ARTIFACTS", self.artifacts()), \ + mock.patch.object(generation, "SUPPORTED", {("trainlog-mobile-export", 4)}): + for _ in range(2): + recovered = generation.resumable_desktop(self.database, PEER_B) + self.assertEqual(captured, recovered) + self.publish_captured(recovered) + self.assertEqual(original, {p.name: p.read_bytes() for p in published.iterdir()}) + raw = self.consumed_ack(captured) + invalid = json.loads(raw) + invalid["consumer_peer_id"] = PEER_A + with self.assertRaises(generation.GenerationError): + generation.accept_ack_bytes(self.database, generation.canonical(invalid)) + self.assertEqual("acknowledged", generation.accept_ack_bytes(self.database, raw)) + self.assertEqual("unchanged", generation.accept_ack_bytes(self.database, raw)) + self.assertIsNone(generation.resumable_desktop(self.database, PEER_B)) + with closing(sqlite3.connect(self.database)) as db: + self.assertEqual([("state", "A")], db.execute("SELECT * FROM facts").fetchall()) + self.assertEqual(1, db.execute("SELECT COUNT(*) FROM sync_acknowledgements").fetchone()[0]) + + def test_late_ack_for_superseded_branch_keeps_its_own_identity(self): + older = self.new_capture() + self.publish_captured(older) + newer = self.new_capture() + self.publish_captured(newer) + newer_ack = self.consumed_ack(newer) + generation.accept_ack_bytes(self.database, newer_ack) + with closing(sqlite3.connect(self.database)) as db: + self.assertEqual(1, generation.active_generation_count(db, PEER_B)) + self.assertEqual("waiting_acknowledgement", db.execute( + "SELECT status FROM sync_generations WHERE generation_id=?", + (older[1]["generation_id"],), + ).fetchone()[0]) + late_ack = self.consumed_ack(older) + self.assertEqual("acknowledged", generation.accept_ack_bytes(self.database, late_ack)) + self.assertEqual("unchanged", generation.accept_ack_bytes(self.database, late_ack)) + # Authentic historical evidence must be retained even when it exposes + # a fork. It never authorizes choosing a new producer tip by timestamp. + with self.assertRaisesRegex(generation.GenerationError, "multiple tips"): + self.new_capture() + with closing(sqlite3.connect(self.database)) as db: + self.assertEqual(2, db.execute("SELECT COUNT(*) FROM sync_acknowledgements").fetchone()[0]) + self.assertEqual([("state", "A")], db.execute("SELECT * FROM facts").fetchall()) + + def test_recovery_rejects_bytes_that_disagree_with_ledger(self): + captured = self.new_capture() + (captured[0] / "history.json").write_text("{}") + with mock.patch.object(generation, "ARTIFACTS", self.artifacts()), \ + mock.patch.object(generation, "SUPPORTED", {("trainlog-mobile-export", 4)}): + with self.assertRaises(generation.GenerationError): + generation.resumable_desktop(self.database, PEER_B) + + def test_inbound_commit_survives_outbound_capacity_failure_and_restart(self): + self.pending_rows(16) + inbound = { + "generation_id": GEN, + "run_id": RUN, + "producer": {"peer_id": PEER_B}, + "consumer_peer_id": PEER_A, + "parent_generation_id": None, + } + # Model the worker's transaction boundary: business facts and the + # inbound receipt commit before outgoing admission is attempted. + with closing(sqlite3.connect(self.database)) as db: + db.execute("BEGIN IMMEDIATE") + db.execute("INSERT INTO facts VALUES('inbound','preserved')") + ack = generation.record_consumed(db, inbound, "c" * 64) + db.commit() + with self.assertRaisesRegex(generation.GenerationError, "capacity exhausted"): + self.new_capture() + with closing(sqlite3.connect(self.database)) as reopened: + self.assertEqual(ack, generation.record_consumed(reopened, inbound, "c" * 64)) + self.assertEqual([("inbound", "preserved")], reopened.execute( + "SELECT * FROM facts WHERE id='inbound'", + ).fetchall()) + self.assertEqual(1, reopened.execute( + "SELECT COUNT(*) FROM sync_consumed_generations", + ).fetchone()[0]) + self.assertEqual(16, generation.active_generation_count(reopened, PEER_B)) + + def test_divergent_pending_branches_fail_closed(self): + self.pending_rows(2) + with self.assertRaisesRegex(generation.GenerationError, "multiple unresolved"): + generation.resumable_desktop(self.database, PEER_B) + + def test_archival_retains_exact_bytes_and_ack_replay(self): + # Archiving the acknowledged sibling must retain its proof for all + # pending branches; moving it out of admission never resurrects them. + self.pending_rows(14) + captures = [] + for _ in range(3): + captured = self.new_capture() + self.publish_captured(captured) + raw = self.consumed_ack(captured) + generation.accept_ack_bytes(self.database, raw) + captures.append((captured, raw)) + with mock.patch.object(generation, "ARTIFACTS", self.artifacts()), \ + mock.patch.object(generation, "SUPPORTED", {("trainlog-mobile-export", 4)}): + self.assertEqual(1, generation.archive_acknowledged( + self.database, self.root / "owned", PEER_B, + )) + self.assertEqual(0, generation.archive_acknowledged( + self.database, self.root / "owned", PEER_B, + )) + captured, raw = captures[0] + archived = self.root / "owned/archives/generations" / captured[1]["generation_id"] + self.assertEqual( + {p.name: p.read_bytes() for p in captured[0].iterdir()}, + {p.name: p.read_bytes() for p in archived.iterdir()}, + ) + self.assertEqual("unchanged", generation.accept_ack_bytes(self.database, raw)) + with closing(sqlite3.connect(self.database)) as db: + self.assertEqual(2, generation.active_generation_count(db, PEER_B)) + self.assertEqual(3, db.execute("SELECT COUNT(*) FROM sync_acknowledgements").fetchone()[0]) + if __name__=="__main__":unittest.main() diff --git a/tools/sync_generation_exchange.py b/tools/sync_generation_exchange.py index 86056d3..cd36417 100644 --- a/tools/sync_generation_exchange.py +++ b/tools/sync_generation_exchange.py @@ -54,7 +54,8 @@ MAX_DIAGNOSTIC = 1024 # WHY: interrupted MTP publications can leave immutable generations awaiting # an ACK alongside the two acknowledged lineage members kept for recovery. # CONTRACT: admission remains bounded and never erases unresolved generations. -# INVARIANT: captured, published and waiting generations still consume a slot. +# INVARIANT: unresolved branches consume a slot unless an exact consumed ACK +# proves that a distinct sibling has advanced the same peer lineage. MAX_RETAINED_PER_PEER = 16 ID = re.compile(r"^(?:gen|peer|sy)_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$") HEX = re.compile(r"^[0-9a-f]{64}$") @@ -371,6 +372,68 @@ def run_export(tool: str, extra: tuple[str, ...], output: Path, snapshot: Path) raise GenerationError(f"export {tool} failed: {result.stdout}{result.stderr}") +# WHY: a failed conversation may leave a branch behind while a later sibling +# advances the consumer. CONTRACT: only an exact, durable consumed ACK for the +# same producer and consumer proves supersession; age or a missing ACK does not. +# INVARIANT: this predicate never updates status, membership, bytes or ACKs. +SUPERSEDED_BRANCH = """ +EXISTS(SELECT 1 FROM sync_generations sibling +JOIN sync_acknowledgements a ON a.generation_id=sibling.generation_id +WHERE sibling.parent_generation_id IS g.parent_generation_id +AND sibling.generation_id<>g.generation_id +AND sibling.producer_peer_id=g.producer_peer_id +AND sibling.consumer_peer_id=g.consumer_peer_id +AND sibling.status='acknowledged' +AND a.run_id=sibling.run_id +AND a.producer_peer_id=sibling.producer_peer_id +AND a.consumer_peer_id=sibling.consumer_peer_id +AND a.manifest_sha256=sibling.manifest_sha256 +AND a.result='consumed' AND a.durability='sqlite-commit-full') +""" + + +def active_generation_count(db: sqlite3.Connection, consumer: str) -> int: + """Count the bounded window without discarding retained causal evidence.""" + return db.execute( + "SELECT COUNT(*) FROM sync_generations g WHERE consumer_peer_id=? " + "AND ((g.status IN('captured','published','waiting_acknowledgement') " + "AND NOT " + SUPERSEDED_BRANCH + ") OR " + "(g.status='acknowledged' AND EXISTS(SELECT 1 FROM sync_acknowledgements k " + "WHERE k.generation_id=g.generation_id))) " + "AND NOT EXISTS(SELECT 1 FROM sync_generation_archives a " + "WHERE a.generation_id=g.generation_id)", + (consumer,), + ).fetchone()[0] + + +def resumable_desktop(database: Path, consumer: str): + """Recover the one unresolved branch; divergent branches require evidence. + + WHY: recapturing after transport interruption leaks admission slots and can + fork a lineage. CONTRACT: replay exact immutable bytes, including the old + manifest run, under the new conversation reference. INVARIANT: ambiguity + fails closed and neither missing ACKs nor restarts imply non-consumption. + """ + with closing(connect_database(database)) as db: + require_schema(db) + rows = db.execute( + "SELECT g.staging_path,g.manifest_json,g.manifest_sha256 " + "FROM sync_generations g WHERE g.consumer_peer_id=? " + "AND g.status IN('captured','published','waiting_acknowledgement') " + "AND NOT " + SUPERSEDED_BRANCH + " LIMIT 2", + (consumer,), + ).fetchall() + if len(rows) > 1: + raise GenerationError("outgoing generation lineage has multiple unresolved branches") + if not rows: + return None + staging, raw, checksum = rows[0] + manifest, actual = validate_published(Path(staging), consumer) + if canonical(manifest) != raw.encode() or actual != checksum: + raise GenerationError("resumable generation differs from immutable ledger") + return Path(staging), manifest, checksum + + def capture_desktop(database: Path, owned_root: Path, consumer: str, run_id: str | None = None, generation_id: str | None = None, peer_capabilities: set[str] | None = None, @@ -384,22 +447,9 @@ def capture_desktop(database: Path, owned_root: Path, consumer: str, archive_acknowledged(database, owned_root, consumer) with closing(connect_database(database)) as admission: require_schema(admission) - # WHY: early private generation builds can contain terminal - # acknowledged rows from before durable ACK-ledger insertion was - # enforced. They cannot be archived without fabricating missing ACK - # evidence, but they are terminal and must not permanently consume the - # active admission window. Current production acknowledgements always - # have a durable sync_acknowledgements row and remain counted until - # archive_acknowledged() safely moves them out of the active window. - retained = admission.execute( - "SELECT COUNT(*) FROM sync_generations g WHERE consumer_peer_id=? " - "AND (g.status IN('captured','published','waiting_acknowledgement') OR " - "(g.status='acknowledged' AND EXISTS(SELECT 1 FROM sync_acknowledgements k " - "WHERE k.generation_id=g.generation_id))) " - "AND NOT EXISTS(SELECT 1 FROM sync_generation_archives a " - "WHERE a.generation_id=g.generation_id)", - (consumer,), - ).fetchone()[0] + # Legacy terminal rows without ACK evidence remain preserved but cannot + # be archived. Pending branches leave admission only with sibling proof. + retained = active_generation_count(admission, consumer) if retained >= MAX_RETAINED_PER_PEER: raise GenerationError("generation retention capacity exhausted") stage = owned_root / "staging" / generation diff --git a/tools/sync_peer_worker.py b/tools/sync_peer_worker.py index 09ade4b..9b1d8ba 100644 --- a/tools/sync_peer_worker.py +++ b/tools/sync_peer_worker.py @@ -825,15 +825,21 @@ def main() -> int: inbound_generation_id=ack["generation_id"], manifest_sha256=ack["manifest_sha256"], ) - outgoing_id = "gen_" + str(uuid.uuid4()) - # Export codecs retain human diagnostics on stdout when invoked as CLIs. - # The worker stdout contract is typed NDJSON only, so route those bounded - # diagnostics to the separately bounded stderr channel. - with redirect_stdout(sys.stderr): - stage, manifest, digest = generation.capture_desktop( - args.database, args.owned_root, peer["peer_id"], args.run_id, outgoing_id, - set(peer["capabilities"]), True, - ) + # WHY: interrupted publication must resume its immutable generation rather + # than allocate another slot. The reference belongs to this conversation; + # the manifest and ACK retain their original run correlation. + recovered = generation.resumable_desktop(args.database, peer["peer_id"]) + if recovered is None: + outgoing_id = "gen_" + str(uuid.uuid4()) + # Exporter diagnostics must not corrupt the typed NDJSON stdout stream. + with redirect_stdout(sys.stderr): + stage, manifest, digest = generation.capture_desktop( + args.database, args.owned_root, peer["peer_id"], args.run_id, outgoing_id, + set(peer["capabilities"]), True, + ) + else: + stage, manifest, digest = recovered + outgoing_id = manifest["generation_id"] program_domain_incompatible = not any( artifact["logical_name"] in ("programs-v1", "programs-v2") for artifact in manifest["artifacts"] @@ -904,8 +910,9 @@ def main() -> int: android_ack_value = wait_json( android_ack, deadline, - lambda value: value.get("run_id") == args.run_id - and value.get("generation_id") == outgoing_id, + lambda value: value.get("run_id") == manifest["run_id"] + and value.get("generation_id") == outgoing_id + and value.get("manifest_sha256") == digest, pump, ) result = generation.accept_ack(args.database, android_ack)