From 7bee60434471df9e8306189021888278d414173a Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 18:58:36 -0400 Subject: [PATCH 1/2] fix: migrate an empty legacy pending-delivery table without data loss A run directory written by the pre-v2 durable authority carries a remote_delivery_pending table without the authority_origin column. An empty legacy table now gains the column in place; a populated one still refuses with DurableAuthorityBusy so no live pending delivery is ever rewritten by a schema migration. --- openadapt_flow/runtime/durable/authority.py | 23 +++++- tests/test_durable_authority_v13.py | 87 +++++++++++++++++++++ 2 files changed, 108 insertions(+), 2 deletions(-) diff --git a/openadapt_flow/runtime/durable/authority.py b/openadapt_flow/runtime/durable/authority.py index b7beb9d8..44d1e935 100644 --- a/openadapt_flow/runtime/durable/authority.py +++ b/openadapt_flow/runtime/durable/authority.py @@ -816,7 +816,7 @@ def _connect(self) -> sqlite3.Connection: "PRAGMA table_info(remote_delivery_pending)" ).fetchall() } - if pending_columns != { + expected_pending_columns = { "path_key", "authority_id", "authority_origin", @@ -831,7 +831,26 @@ def _connect(self) -> sqlite3.Connection: "runtime_delivery_sequence", "permit_artifact_sha256", "permit_artifact_bytes", - }: + } + legacy_pending_columns = expected_pending_columns - { + "authority_origin" + } + if pending_columns == legacy_pending_columns: + pending_count = int( + connection.execute( + "SELECT COUNT(*) FROM remote_delivery_pending" + ).fetchone()[0] + ) + if pending_count: + raise DurableAuthorityBusy( + "the legacy pending remote delivery cannot be migrated" + ) + connection.execute( + "ALTER TABLE remote_delivery_pending ADD COLUMN " + "authority_origin TEXT NOT NULL DEFAULT ''" + ) + pending_columns = expected_pending_columns + if pending_columns != expected_pending_columns: raise DurableAuthorityBusy( "the pending remote delivery schema is incompatible" ) diff --git a/tests/test_durable_authority_v13.py b/tests/test_durable_authority_v13.py index 62b3976f..f61884e9 100644 --- a/tests/test_durable_authority_v13.py +++ b/tests/test_durable_authority_v13.py @@ -134,6 +134,57 @@ def _fresh_run( return run_dir, store, manifest, authority +def _replace_pending_table_with_legacy_schema( + authority: DurableAuthority, + *, + populated: bool, +) -> None: + with sqlite3.connect(authority.db_path) as connection: + connection.execute("DROP TABLE remote_delivery_pending") + connection.execute( + """ + CREATE TABLE remote_delivery_pending ( + path_key TEXT PRIMARY KEY, + authority_id TEXT NOT NULL, + authority_signer_sha256 TEXT NOT NULL, + permit_id TEXT NOT NULL, + dispatch_session_id TEXT NOT NULL, + one_use_claim_id TEXT NOT NULL, + next_sequence INTEGER NOT NULL, + cursor_secret TEXT NOT NULL, + input_edge_sequence INTEGER NOT NULL, + authority_sequence INTEGER NOT NULL, + runtime_delivery_sequence INTEGER NOT NULL, + permit_artifact_sha256 TEXT NOT NULL, + permit_artifact_bytes BLOB NOT NULL + ) + """ + ) + if populated: + connection.execute( + """ + INSERT INTO remote_delivery_pending VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ) + """, + ( + authority.path_key, + "00000000-0000-4000-8000-000000000008", + "a" * 64, + "20000000-0000-4000-8000-000000000001", + "30000000-0000-4000-8000-000000000001", + "40000000-0000-4000-8000-000000000001", + 1, + "b" * 64, + 1, + 0, + 0, + "c" * 64, + b"{}", + ), + ) + + def _pause( store: CheckpointStore, manifest: RunManifest, @@ -372,6 +423,42 @@ def test_customer_controlled_standard_delivery_stays_local_without_cloud( authority.before_delivery(manifest, attempt_id="local", owner_nonce_sha256=owner) +def test_empty_legacy_pending_delivery_table_migrates_without_data_loss( + tmp_path: Path, +) -> None: + _run_dir, _store, manifest, authority = _fresh_run(tmp_path) + _replace_pending_table_with_legacy_schema(authority, populated=False) + + assert authority.validate(manifest).phase == "active" + with sqlite3.connect(authority.db_path) as connection: + columns = { + str(row[1]) + for row in connection.execute( + "PRAGMA table_info(remote_delivery_pending)" + ).fetchall() + } + pending_count = int( + connection.execute( + "SELECT COUNT(*) FROM remote_delivery_pending" + ).fetchone()[0] + ) + assert "authority_origin" in columns + assert pending_count == 0 + + +def test_populated_legacy_pending_delivery_table_fails_closed( + tmp_path: Path, +) -> None: + _run_dir, _store, manifest, authority = _fresh_run(tmp_path) + _replace_pending_table_with_legacy_schema(authority, populated=True) + + with pytest.raises( + DurableAuthorityBusy, + match="legacy pending remote delivery cannot be migrated", + ): + authority.validate(manifest) + + def test_production_delivery_requires_and_validates_remote_permit( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From f7c060e12895817edf46f192e68ab44a2a3d39f2 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Fri, 21 Aug 2026 19:12:30 -0400 Subject: [PATCH 2/2] style: ruff-format the legacy pending-table migration --- openadapt_flow/runtime/durable/authority.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/openadapt_flow/runtime/durable/authority.py b/openadapt_flow/runtime/durable/authority.py index 44d1e935..591853c8 100644 --- a/openadapt_flow/runtime/durable/authority.py +++ b/openadapt_flow/runtime/durable/authority.py @@ -832,9 +832,7 @@ def _connect(self) -> sqlite3.Connection: "permit_artifact_sha256", "permit_artifact_bytes", } - legacy_pending_columns = expected_pending_columns - { - "authority_origin" - } + legacy_pending_columns = expected_pending_columns - {"authority_origin"} if pending_columns == legacy_pending_columns: pending_count = int( connection.execute(