diff --git a/openadapt_flow/runtime/durable/authority.py b/openadapt_flow/runtime/durable/authority.py index b7beb9d8..591853c8 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,24 @@ 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: