fy59/trainlog
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions

Fix generation capacity recovery with causal sibling proof and immutable retry

Browse source
This commit is contained in:
fy59 2026-11-10 18:31:11 +01:00
parent 292bcc8246
commit 2c6afa9e18
9 changed files with 567 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
CHANGELOG.md
View file

@ -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

22
docs/current_state.md
View file

@ -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,

188
docs/reviews/sync_generation_capacity_recovery_v1.md Normal file
View file

@ -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.

6
docs/roadmap.md
View file

@ -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

18
docs/sync_exchange.md
View file

@ -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/<generation_id>/`, 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

17
docs/tests.md
View file

@ -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.

225
tests/test_sync_generation_exchange.py
View file

@ -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()

84
tools/sync_generation_exchange.py
View file

@ -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

17
tools/sync_peer_worker.py
View file

@ -825,15 +825,21 @@ def main() -> int:
inbound_generation_id=ack["generation_id"],
manifest_sha256=ack["manifest_sha256"],
)
# 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())
# 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.
# 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)