diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 52ede77..54aacc4 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -65,6 +65,6 @@ | task | EMAIL-WP-0008-T04 | done | — | workplans/EMAIL-WP-0008-scoped-transactional-callers.md | | task | EMAIL-WP-0009-T01 | done | — | workplans/EMAIL-WP-0009-wave2-resource-right-sizing.md | | task | EMAIL-WP-0010-T01 | done | — | workplans/EMAIL-WP-0010-platform-sender-transition.md | -| task | EMAIL-WP-0010-T02 | todo | — | workplans/EMAIL-WP-0010-platform-sender-transition.md | +| task | EMAIL-WP-0010-T02 | progress | — | workplans/EMAIL-WP-0010-platform-sender-transition.md | | task | EMAIL-WP-0010-T03 | wait | — | workplans/EMAIL-WP-0010-platform-sender-transition.md | | task | EMAIL-WP-0010-T04 | todo | — | workplans/EMAIL-WP-0010-platform-sender-transition.md | diff --git a/docs/sender-transition-protocol-2026-10-31.json b/docs/sender-transition-protocol-2026-10-31.json new file mode 100644 index 0000000..b20a498 --- /dev/null +++ b/docs/sender-transition-protocol-2026-10-31.json @@ -0,0 +1,39 @@ +{ + "schema": "email-connect.sender-transition-protocol-proof.v1", + "native_adapter_complete": false, + "production_entrypoint_available": false, + "production_changed": false, + "mail_sent": false, + "native_fencing_verified": false, + "tests": { + "command": "PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=src python -m pytest -p no:cacheprovider -q", + "passed": 219, + "skipped": 21, + "new_protocol_cases": 27, + "real_child_exit_cases": 10, + "cluster_effects": "synthetic_file_backed_model", + "ledger": "actual_sqlite_and_owner_retry_source", + "seconds": 16.9, + "skipped_scope": "existing_optional_container_mail_harness", + "log": "/tmp/email-sender-transition-protocol-check-20261031.log" + }, + "sources": [ + { + "path": "scripts/sender_transition.py", + "sha256": "fa80b428b6231b74578baf23ff5977b6ce871a84321718adb2ba15d314f7abe1" + }, + { + "path": "tests/test_sender_transition.py", + "sha256": "572d3892c95231b5d04b85d47fa395c7c0c59bb323ec78bb43a08c635ac50534" + }, + { + "path": "scripts/inspect-sender-ledger.py", + "sha256": "262d6d52d5c10fb8583e0828ed34945b647f1d53bf9de5286374fef85cf76a8d" + }, + { + "path": "src/email_connect/transactional.py", + "sha256": "36fdfd8cb58f10148cd491a7e34d7aea798e4289fce760af70a48776bede204d" + } + ], + "technical_decision": "72341ab6-dc25-4974-a3d0-80186acb47f9" +} diff --git a/docs/sender-transition-protocol-2026-10-31.md b/docs/sender-transition-protocol-2026-10-31.md new file mode 100644 index 0000000..8b2f6e9 --- /dev/null +++ b/docs/sender-transition-protocol-2026-10-31.md @@ -0,0 +1,90 @@ +# Sender transition protocol preparation + +EMAIL-WP-0010-T02 is in progress. `scripts/sender_transition.py` is a tested +protocol core, with no production CLI or Kubernetes credentials. It does not +establish native fencing, credential admission or an execution-ready release. + +## Concrete behavior + +The proposed delta is derived from the observed configuration rather than +accepting arbitrary patches: retain the exact receiver image and every unrelated +field, replace the old runtime envFrom Secret with explicit platform SMTP username +and password projection references, and change only the configured sender. A +different image, existing environment shape or sender refuses preparation. + +Before every effect the protocol fsyncs an append-only private journal. The first +entry binds the complete before/after proposal. The directory must be operator +owned and 0700; the journal is 0600, exclusively locked, and refuses symlinks and +hardlinks. A truncated final line is retained for inspection, never silently +discarded. A completed or interrupted forward attempt cannot be blindly replayed. + +The forward sequence is admission, pause, fence, fresh capture, outcome review, +configuration change, deployment environment change, resume, native verification, +postflight capture, continuity comparison and helper cleanup. The protocol checks +that the receiver is still paused before configuration and environment changes. +Captured baseline bytes are hash-bound and must remain unchanged during the attempt. +Unknown outcomes before a switch stop the sequence without clearing reservations. + +Failures leave an inspection record; no finally block guesses a rollback or +resumes an uncertain receiver. Explicit recovery observes actual configuration, +recognizes only the reviewed old/new/mixed states, pauses again, captures the +**current** ledger and restores only the old configuration. It retains all earlier +journal entries and snapshots. It never copies an old ledger over current data, +clears reservations, sends mail or re-enables retired legacy caller authentication. +Rollback restores the previous SMTP environment references while retaining the +scoped callers-file configuration and its precedence over the legacy environment. +An additional interruption during recovery can be handled by observing the same +attempt again. Unknown outcomes remain unknown and preserve non-retryability. + +## Data comparison and limits + +The comparator reads all six tables and requires unchanged schema and preservation +of every pre-existing primary key. Accepted delivery receipts and suppression +decisions cannot be rewritten. Verification identity/handle/binding fields remain +exact; terminal consumption/cancellation may only advance from unset and cannot +be reversed or both set. Caller quota counters may advance without losing their +identity or reversing time. Attempt transitions retain their idempotency binding; +only a definitive retryable failure permits a new verification request handle. +New rows are reported, allowing legitimate traffic after receiver resume. + +This is structural continuity evidence, not independent proof that every new +application request was authorized or delivered. The unchanged image and caller +policy, actual runtime checks and productive mail evidence remain separate gates. +An unexpected deletion/change fails comparison; it never triggers a data restore. + +## Test evidence + +Twenty-seven new synthetic cases pass. Ten use real abrupt process exits around +pause, capture, configuration/deployment changes, resume, postflight, cleanup, +completion and lost configuration/scale replies. Cluster API effects are modeled +as durable file-backed state; the database and SQLite reopen/retry behavior are +real. Subsequent accepted and uncertain sends survive explicit recovery byte for +byte, and the old configuration is restored. Other cases cover a second recovery +interruption, foreign drift, absent admission, journal truncation/private modes, +writer resumption during capture, a broken snapshot before configuration change, +schema/row loss, ambiguous nullable SQLite primary keys and reversed consumption. + +Owner default suite: **219 passed, 21 skipped in 16.90s**. The 21 existing skipped tests +require the optional container mail harness; none of the new cases is skipped. +Ruff passes for both new Python files. Log: +`/tmp/email-sender-transition-protocol-check-20261031.log`. + +## Required native completion + +The adapter must implement complete UID/spec compare-and-swap operations, verify +the RPF-WP-0068 custody return and current projection, hold the existing capture +lock, prove the absence of other PVC writers, and bind each helper creation and +deletion to durable identity evidence. Snapshots must use the existing read-only +SQLite backup path and remain private. Runtime verification must inspect actual +sender/credential references, health and scoped denial behavior without sending. + +The final capsule must bind all executable dependencies, proposal and preimages; +the attended entry point must require the exact reviewed digest. Native server +dry-runs, source review and independent return checks must precede execution. +No production execution instruction is supplied by this protocol-only document. +T02 remains open until these obligations are implemented and verified; T03/T04 +remain attended execution and independent native acceptance. + +Technical decision72341ab6-dc25-4974-a3d0-80186acb47f9 records the configuration-only +recovery design. It is not credential admission or production execution approval. +Source hashes and validation scope: `sender-transition-protocol-2026-10-31.json`. diff --git a/scripts/sender_transition.py b/scripts/sender_transition.py new file mode 100644 index 0000000..d49c0d4 --- /dev/null +++ b/scripts/sender_transition.py @@ -0,0 +1,330 @@ +"""Durable sender-change protocol; adapters must supply native fencing and reads. + +This module has no CLI or cluster credentials. The native adapter must bind a +reviewed proposal, hold the existing capture lock and verify owner admission. +No protocol method writes or restores the delivery ledger. +""" + +from __future__ import annotations + +import copy +import fcntl +import hashlib +import json +import math +import os +import sqlite3 +import stat +from contextlib import closing +from datetime import datetime +from pathlib import Path + +IMAGE = ("forgejo.coulomb.social/coulomb/email-connect@sha256:" + "5d79fd1fec15d3a279383820f170d263de3a55f5308df01a0a34732ebaea6d3d") +TABLES = {"deliveries", "verifications", "suppressions", "send_attempts", + "verification_attempts", "caller_buckets"} +SCHEMA = "email-connect.sender-transition.v1" + + +class Refused(RuntimeError): + pass + + +def require(condition, reason): + if not condition: + raise Refused(reason) + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def desired(before): + """The only admitted configuration delta; never accepts a free-form patch.""" + after = copy.deepcopy(before) + spec = after["deployment_spec"] + containers = spec["template"]["spec"]["containers"] + require(spec["replicas"] == 1 and len(containers) == 1, "unexpected_receiver_shape") + container = containers[0] + require(container["name"] == "email-connect" and container["image"] == IMAGE, + "unexpected_receiver_image") + require(container["envFrom"] == [ + {"configMapRef": {"name": "email-connect-config"}}, + {"secretRef": {"name": "email-connect-runtime", "optional": False}}, + ], "unexpected_old_environment") + require(container["env"] == [ + {"name": "EMAIL_CONNECT_CALLERS_FILE", "value": "/etc/email-connect-callers/callers.json"}, + ], "unexpected_environment_override") + require(after["config_data"]["EMAIL_CONNECT_SENDER"] == "forgejo@coulomb.social", + "unexpected_old_sender") + container["envFrom"] = [{"configMapRef": {"name": "email-connect-config"}}] + container["env"] += [ + {"name": "EMAIL_CONNECT_SMTP_USERNAME", "value": "platform@coulomb.social"}, + {"name": "EMAIL_CONNECT_SMTP_PASSWORD", "valueFrom": {"secretKeyRef": { + "name": "email-connect-platform-smtp", "key": "password", "optional": False}}}, + ] + after["config_data"]["EMAIL_CONNECT_SENDER"] = "platform@coulomb.social" + return after + + +def with_replicas(spec, replicas): + result = copy.deepcopy(spec) + result["replicas"] = replicas + return result + + +def known(observed, before, after): + """Recognize lost-response/mixed phases, but never unrelated live drift.""" + for key in ("identity", "protected"): + require(observed[key] == before[key] == after[key], "protected_runtime_changed") + require(observed["config_data"] in (before["config_data"], after["config_data"]), + "configuration_changed_outside_proposal") + require(observed["deployment_spec"] in [with_replicas(item["deployment_spec"], replicas) + for item in (before, after) for replicas in (0, 1)], + "deployment_changed_outside_proposal") + + +def ledger(path): + path = Path(path).absolute() + require(not any(p.is_symlink() for p in [path, *path.parents]), "symlink_snapshot") + info = path.stat() + require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() + and not info.st_mode & 0o077 and 0 < info.st_size <= 128 * 1024**2, + "private_bounded_snapshot_required") + with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)) as db: + db.execute("PRAGMA trusted_schema=OFF") + db.execute("PRAGMA query_only=ON") + db.execute("BEGIN") + require(db.execute("PRAGMA integrity_check").fetchall() == [("ok",)], + "snapshot_integrity_failed") + schema = db.execute( + "SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name").fetchall() + require({r[1] for r in schema if r[0] == "table" and not r[1].startswith("sqlite_")} + == TABLES, "unexpected_snapshot_schema") + rows = {} + for table in sorted(TABLES): + columns = db.execute('PRAGMA table_info("' + table + '")').fetchall() + keys = [r[1] for r in sorted(columns, key=lambda r: r[5]) if r[5]] + require(len(keys) == 1, "unexpected_snapshot_key") + names = [r[1] for r in columns] + values = [dict(zip(names, r, strict=True)) + for r in db.execute('SELECT * FROM "' + table + '"')] + rows[table] = {r[keys[0]]: r for r in values} + require(len(rows[table]) == len(values) and None not in rows[table], + "ambiguous_snapshot_row_identity") + return schema, rows + + +def same_except(old, new, fields): + return {k: v for k, v in old.items() if k not in fields} == { + k: v for k, v in new.items() if k not in fields} + + +def row_preserved(table, old, new): + if old == new: + return True + if table == "verifications": + if not same_except(old, new, {"consumed_at", "canceled_at"}): + return False + if new["consumed_at"] is not None and new["canceled_at"] is not None: + return False + for key in ("consumed_at", "canceled_at"): + if old[key] == new[key]: + continue + if old[key] is not None or not isinstance(new[key], str): + return False + try: + if datetime.fromisoformat(new[key]).tzinfo is None: + return False + except ValueError: + return False + return True + if table == "caller_buckets": + return (same_except(old, new, {"tokens", "updated_at"}) + and all(isinstance(new[k], (int, float)) and math.isfinite(new[k]) + for k in ("tokens", "updated_at")) + and new["tokens"] >= 0 and new["updated_at"] >= old["updated_at"]) + if table in {"send_attempts", "verification_attempts"}: + mutable = {"state", "error", "retryable"} + if table == "verification_attempts": + mutable.add("provider_ref") + # Only a definitive retryable failure permits a fresh verification handle. + retry = old["state"] == "failed" and old["retryable"] == 1 + if retry and table == "verification_attempts": + mutable.add("request_id") + states = {"in_flight", "failed"} + if table == "verification_attempts": + states.add("accepted") + return (same_except(old, new, mutable) + and (retry or old["state"] == "in_flight") and new["state"] in states + and (new["state"] != "accepted" or bool(new.get("provider_ref")))) + # Accepted deliveries and suppression decisions have no authorized mutation + # in this configuration-only operation. New rows remain possible at runtime. + return False + + +def continuity(before, after): + old_schema, old = ledger(before) + new_schema, new = ledger(after) + require(old_schema == new_schema, "ledger_schema_changed") + changes = {} + for table in sorted(TABLES): + require(old[table].keys() <= new[table].keys(), "ledger_rows_removed") + require(all(row_preserved(table, value, new[table][key]) + for key, value in old[table].items()), "ledger_history_changed") + changes[table] = { + "preserved_rows": len(old[table]), + "new_rows": len(new[table]) - len(old[table]), + "permitted_updates": sum(value != new[table][key] + for key, value in old[table].items()), + } + return {"all_six_tables_checked": True, "schema_unchanged": True, "tables": changes, + "ledger_restored": False, "snapshot_bytes_equal": + Path(before).read_bytes() == Path(after).read_bytes()} + + +class Journal: + """Private append-only intents; incomplete lines refuse automatic recovery.""" + + def __init__(self, path, *, proposal=None): + self.path = Path(path).absolute() + require(not any(p.is_symlink() for p in [self.path, *self.path.parents]), + "real_journal_path_required") + parent = self.path.parent.stat() + require(stat.S_ISDIR(parent.st_mode) and parent.st_uid == os.getuid() + and not parent.st_mode & 0o077, "private_journal_parent_required") + flags = os.O_RDWR | os.O_APPEND | os.O_NOFOLLOW + if proposal is not None: + flags |= os.O_CREAT | os.O_EXCL + self.fd = os.open(self.path, flags, 0o600) + try: + info = os.fstat(self.fd) + require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() + and not info.st_mode & 0o077 and info.st_nlink == 1, + "private_owned_journal_required") + fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + self.entries = [] + if proposal is not None: + self.record("prepared", schema=SCHEMA, proposal=proposal, + proposal_digest=digest(proposal)) + directory = os.open(self.path.parent, os.O_DIRECTORY | os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + else: + require(info.st_size <= 16 * 1024**2, "journal_too_large") + data = os.read(self.fd, info.st_size) + require(data.endswith(b"\n"), "incomplete_journal_requires_inspection") + self.entries = [json.loads(line) for line in data.splitlines()] + first = self.entries[0] + require(first["schema"] == SCHEMA + and first["proposal_digest"] == digest(first["proposal"]), + "invalid_proposal_binding") + require(all(e["sequence"] == i for i, e in enumerate(self.entries)), + "invalid_journal_sequence") + except BaseException: + os.close(self.fd) + raise + + def record(self, phase, **data): + record = dict(data, phase=phase, sequence=len(self.entries)) + body = json.dumps(record, sort_keys=True).encode() + b"\n" + # Write all bytes before returning; short writes never authorize effects. + offset = 0 + while offset < len(body): + written = os.write(self.fd, body[offset:]) + require(written > 0, "journal_write_failed") + offset += written + os.fsync(self.fd) + self.entries.append(record) + + def close(self): + os.close(self.fd) + + +def run(adapter, journal, *, recover=False): + """One forward attempt, or explicit recovery to the original configuration. + + Adapter contract: observe/patch use complete UID/spec CAS; fence proves no + other PVC writers; snapshot creates a fresh private consistent copy and binds + helper identity through journal intents; cleanup removes only owned helpers; + verify proves actual process configuration, health and no-send denials. The + native adapter, not this protocol, proves credential admission and fencing. + """ + first = journal.entries[0] + proposal = first["proposal"] + before = proposal["before"] + after = desired(before) + require(proposal["after"] == after, "unreviewed_configuration_delta") + require(recover or len(journal.entries) == 1, "forward_replay_refused") + if recover: + require(journal.entries[-1]["phase"] != "recovered", "recovery_already_completed") + target = before if recover else after + mode = "recovery" if recover else "forward" + + def observe(): + current = adapter.observe() + known(current, before, after) + return current + + try: + live = observe() + if not recover: + require(live == before, "preflight_changed") + adapter.admission(proposal) + journal.record(mode + "_pause_intent") + if live["deployment_spec"]["replicas"] != 0: + adapter.patch_deployment(live["deployment_spec"], + with_replicas(live["deployment_spec"], 0)) + adapter.fence(journal) + live = observe() + require(live["deployment_spec"]["replicas"] == 0, "receiver_not_paused") + journal.record(mode + "_capture_intent") + baseline, outcomes = adapter.snapshot(journal, quiesced=True) + require(Path(baseline).resolve().is_relative_to(journal.path.parent), + "snapshot_outside_private_attempt") + ledger(baseline) # Refuse a broken or non-private baseline before switching. + baseline_sha256 = hashlib.sha256(Path(baseline).read_bytes()).hexdigest() + # Recovery may need to preserve an unknown outcome caused by an earlier + # interruption. It does not repair or clear that reservation. + if not recover: + require(not outcomes["needs_outcome_reconciliation"], "unknown_outcome_before_switch") + journal.record(mode + "_captured", outcomes=outcomes, + snapshot_name=Path(baseline).relative_to(journal.path.parent).as_posix(), + snapshot_sha256=baseline_sha256) + live = observe() + require(live["deployment_spec"]["replicas"] == 0, "receiver_resumed_before_config") + journal.record(mode + "_config_intent") + adapter.patch_config(live["config_data"], target["config_data"]) + live = observe() + require(live["deployment_spec"]["replicas"] == 0, "receiver_resumed_before_deployment") + journal.record(mode + "_deployment_intent") + adapter.patch_deployment(live["deployment_spec"], + with_replicas(target["deployment_spec"], 0)) + live = observe() + require(live["config_data"] == target["config_data"], "sender_config_not_verified") + require(live["deployment_spec"] == with_replicas(target["deployment_spec"], 0), + "paused_deployment_not_verified") + journal.record(mode + "_resume_intent") + adapter.patch_deployment(live["deployment_spec"], target["deployment_spec"]) + native = adapter.verify(target, journal) + require(observe() == target, "runtime_not_verified") + journal.record(mode + "_postflight_intent", native=native) + returned, _ = adapter.snapshot(journal, quiesced=False) + require(Path(returned).resolve().is_relative_to(journal.path.parent), + "snapshot_outside_private_attempt") + require(hashlib.sha256(Path(baseline).read_bytes()).hexdigest() == baseline_sha256, + "baseline_snapshot_changed") + result = continuity(baseline, returned) + journal.record(mode + "_cleanup_intent", continuity=result) + adapter.cleanup(journal) + journal.record("recovered" if recover else "completed", continuity=result, + ledger_restored=False, helper_cleanup_verified=True, + sender=target["config_data"]["EMAIL_CONNECT_SENDER"]) + return result + except BaseException: + # No guessed rollback or resume in finally. Explicit recovery observes + # actual state, captures the CURRENT ledger and retains prior attempts. + journal.record("requires_inspection", operation=mode) + raise diff --git a/tests/test_sender_transition.py b/tests/test_sender_transition.py new file mode 100644 index 0000000..cc1cb6c --- /dev/null +++ b/tests/test_sender_transition.py @@ -0,0 +1,379 @@ +"""Protocol fault injection uses synthetic Kubernetes state and real SQLite only.""" + +import copy +import importlib.util +import json +import os +import sqlite3 +import subprocess +import sys +from pathlib import Path + +import pytest + +from email_connect.transactional import ProviderError, SQLiteDeliveryStore + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" +sys.path.insert(0, str(SCRIPTS)) +import sender_transition as transition + +spec = importlib.util.spec_from_file_location("sender_observer_fixture", SCRIPTS / "inspect-sender-ledger.py") +observer = importlib.util.module_from_spec(spec) +spec.loader.exec_module(observer) + + +def write(path, value): + with path.open("w") as stream: + json.dump(value, stream) + stream.flush() + os.fsync(stream.fileno()) + + +def initial(): + return { + "identity": {"deployment": "fixture-deployment", "config": "fixture-config"}, + "protected": {"policy": "fixture-policy", "pvc": "fixture-pvc"}, + "config_data": {"EMAIL_CONNECT_SENDER": "forgejo@coulomb.social", "other": "preserve"}, + "deployment_spec": { + "replicas": 1, "strategy": {"type": "RollingUpdate"}, + "template": {"spec": {"containers": [{ + "name": "email-connect", "image": transition.IMAGE, + "envFrom": [{"configMapRef": {"name": "email-connect-config"}}, + {"secretRef": {"name": "email-connect-runtime", "optional": False}}], + "env": [{"name": "EMAIL_CONNECT_CALLERS_FILE", + "value": "/etc/email-connect-callers/callers.json"}], + "resources": {"requests": {"cpu": "5m", "memory": "64Mi"}}, + }]}}, + }, + } + + +class FixtureAdapter: + """File-backed API effects persist across real child-process termination.""" + + def __init__(self, root, lost_response=None): + self.root = Path(root) + self.lost_response = lost_response + + def observe(self): + return json.loads((self.root / "cluster.json").read_text()) + + def admission(self, proposal): + if (self.root / "deny-admission").exists(): + raise transition.Refused("fixture_admission_absent") + + def patch(self, key, expected, desired): + value = self.observe() + assert value[key] == expected + value[key] = desired + write(self.root / "cluster.json", value) + if self.lost_response == key: + os._exit(73) + + def patch_deployment(self, expected, desired): + self.patch("deployment_spec", expected, desired) + + def patch_config(self, expected, desired): + self.patch("config_data", expected, desired) + + def fence(self, journal): + assert self.observe()["deployment_spec"]["replicas"] == 0 + + def snapshot(self, journal, *, quiesced): + if quiesced: + assert self.observe()["deployment_spec"]["replicas"] == 0 + path = self.root / f"snapshot-{len(journal.entries)}.sqlite3" + assert not path.exists() + source = sqlite3.connect((self.root / "ledger.sqlite3").as_uri() + "?mode=ro", uri=True) + target = sqlite3.connect(path) + source.backup(target) + source.close() + target.close() + path.chmod(0o600) + return path, observer.inspect(path) + + def verify(self, expected, journal): + assert self.observe() == expected + return {"fixture_only": True} + + def cleanup(self, journal): + pass + + +@pytest.fixture +def attempt(tmp_path): + root = tmp_path / "private" + root.mkdir(mode=0o700) + before = initial() + write(root / "cluster.json", before) + store = SQLiteDeliveryStore(str(root / "ledger.sqlite3")) + store.db.close() + proposal = {"before": before, "after": transition.desired(before)} + write(root / "proposal.json", proposal) + return root, proposal + + +def execute(root, proposal): + journal = transition.Journal(root / "journal.jsonl", proposal=proposal) + try: + result = transition.run(FixtureAdapter(root), journal) + assert journal.entries[-1]["phase"] == "completed" + return result + finally: + journal.close() + + +def recover(root): + journal = transition.Journal(root / "journal.jsonl") + try: + result = transition.run(FixtureAdapter(root), journal, recover=True) + assert journal.entries[-1]["phase"] == "recovered" + return result + finally: + journal.close() + + +def test_only_intended_fields_change_and_replay_refuses(attempt): + root, proposal = attempt + result = execute(root, proposal) + assert result["all_six_tables_checked"] and not result["ledger_restored"] + assert FixtureAdapter(root).observe() == proposal["after"] + with pytest.raises(FileExistsError): + execute(root, proposal) + journal = transition.Journal(root / "journal.jsonl") + try: + with pytest.raises(transition.Refused, match="forward_replay"): + transition.run(FixtureAdapter(root), journal) + finally: + journal.close() + + +@pytest.mark.parametrize("phase", [ + "forward_pause_intent", "forward_capture_intent", "forward_config_intent", + "forward_deployment_intent", "forward_resume_intent", "forward_postflight_intent", + "forward_cleanup_intent", "completed", "lost_config_response", "lost_scale_response", +]) +def test_real_process_exit_can_restore_configuration_without_restoring_ledger(attempt, phase): + root, proposal = attempt + child = ''' +import json,os,sys +from pathlib import Path +from test_sender_transition import transition,FixtureAdapter +root=Path(sys.argv[1]);phase=sys.argv[2] +proposal=json.loads((root/'proposal.json').read_text()) +original=transition.Journal.record +def checkpoint(self, name, **data): + original(self,name,**data) + if name==phase:os._exit(73) +transition.Journal.record=checkpoint +journal=transition.Journal(root/'journal.jsonl',proposal=proposal) +lost={'lost_config_response':'config_data','lost_scale_response':'deployment_spec'}.get(phase) +transition.run(FixtureAdapter(root,lost),journal) +''' + result = subprocess.run([sys.executable, "-c", child, str(root), phase], + env={**os.environ, "PYTHONPATH": os.pathsep.join([ + str(SCRIPTS.parent / "src"), str(Path(__file__).parent)])}, + capture_output=True, timeout=30, check=False) + assert result.returncode == 73, result.stderr.decode() + # Writes which occurred after the original attempt must survive recovery. + store = SQLiteDeliveryStore(str(root / "ledger.sqlite3")) + store.reserve_send("later-accepted", "fp") + store.record("later-accepted", "provider-receipt") + store.reserve_send("later-unknown", "fp2") + store.db.close() + baseline = (root / "ledger.sqlite3").read_bytes() + old_journal = (root / "journal.jsonl").read_bytes() + result = recover(root) + assert FixtureAdapter(root).observe() == proposal["before"] + assert (root / "ledger.sqlite3").read_bytes() == baseline + assert (root / "journal.jsonl").read_bytes().startswith(old_journal) + assert not result["ledger_restored"] + store = SQLiteDeliveryStore(str(root / "ledger.sqlite3")) + with pytest.raises(ProviderError, match="delivery_outcome_unknown"): + store.reserve_send("later-unknown", "fp2") + assert store.reserve_send("later-accepted", "fp") is False + store.db.close() + + +@pytest.mark.parametrize("phase", ["recovery_config_intent", "recovery_resume_intent"]) +def test_second_recovery_interruption_is_restartable(attempt, monkeypatch, phase): + root, proposal = attempt + execute(root, proposal) + original = transition.Journal.record + + def stop(self, name, **data): + original(self, name, **data) + if name == phase: + raise RuntimeError("second interruption") + + monkeypatch.setattr(transition.Journal, "record", stop) + with pytest.raises(RuntimeError): + recover(root) + monkeypatch.setattr(transition.Journal, "record", original) + recover(root) + assert FixtureAdapter(root).observe() == proposal["before"] + + +def test_unknown_before_switch_holds_old_receiver_and_recovery_preserves_unknown(attempt): + root, proposal = attempt + store = SQLiteDeliveryStore(str(root / "ledger.sqlite3")) + store.reserve_send("unknown", "fp") + store.db.close() + with pytest.raises(transition.Refused, match="unknown_outcome"): + execute(root, proposal) + value = FixtureAdapter(root).observe() + assert value["config_data"] == proposal["before"]["config_data"] + assert value["deployment_spec"]["replicas"] == 0 + recover(root) + assert observer.inspect(root / "ledger.sqlite3")["needs_outcome_reconciliation"] + + +@pytest.mark.parametrize("drift", ["image", "protected", "sender"]) +def test_foreign_drift_refuses_recovery_without_overwrite(attempt, drift): + root, proposal = attempt + execute(root, proposal) + value = FixtureAdapter(root).observe() + if drift == "image": + value["deployment_spec"]["template"]["spec"]["containers"][0]["image"] = "foreign" + elif drift == "protected": + value["protected"]["pvc"] = "replaced-pvc" + else: + value["config_data"]["EMAIL_CONNECT_SENDER"] = "someone@example.test" + write(root / "cluster.json", value) + with pytest.raises(transition.Refused): + recover(root) + assert FixtureAdapter(root).observe() == value + + +def test_absent_admission_does_not_pause_or_patch(attempt): + root, proposal = attempt + (root / "deny-admission").touch() + with pytest.raises(transition.Refused, match="admission_absent"): + execute(root, proposal) + assert FixtureAdapter(root).observe() == proposal["before"] + + +def test_partial_journal_is_held_and_private_modes_enforced(attempt): + root, proposal = attempt + execute(root, proposal) + path = root / "journal.jsonl" + assert path.stat().st_mode & 0o777 == 0o600 + with path.open("ab") as stream: + stream.write(b'{"partial":') + with pytest.raises(transition.Refused, match="incomplete_journal"): + transition.Journal(path) + root.chmod(0o755) + with pytest.raises(transition.Refused, match="private_journal_parent"): + transition.Journal(path) + + +def test_continuity_permits_terminal_consumption_and_new_writes_but_not_rewrites(tmp_path): + path = tmp_path / "before.sqlite3" + store = SQLiteDeliveryStore(str(path)) + payload = {"tenant": "fixture", "client_id": "client", "registration_id": "registration", + "normalized_email": "one@example.test", "preferred_username": "one"} + request, _ = store.reserve_verification(payload, "handle", "2030-01-01T00:00:00+00:00") + store.verification_accepted(request, "receipt") + store.request_budget("caller", rate=60, burst=10, now=100) + store.db.close() + path.chmod(0o600) + after = tmp_path / "after.sqlite3" + after.write_bytes(path.read_bytes()) + after.chmod(0o600) + store = SQLiteDeliveryStore(str(after)) + store.consume_verification("handle") + store.request_budget("caller", rate=60, burst=10, now=102) + store.reserve_send("new", "fp") + store.record("new", "new-receipt") + result = transition.continuity(path, after) + assert result["tables"]["deliveries"]["new_rows"] == 1 + assert result["tables"]["verifications"]["permitted_updates"] == 1 + with store.db: + store.db.execute("UPDATE verifications SET email='different@example.test'") + with pytest.raises(transition.Refused, match="history_changed"): + transition.continuity(path, after) + store.db.close() + + +@pytest.mark.parametrize("change", ["delete", "schema", "unconsume"]) +def test_continuity_refuses_lost_rows_schema_drift_or_reversed_consumption(tmp_path, change): + before = tmp_path / "before.sqlite3" + store = SQLiteDeliveryStore(str(before)) + store.record("event", "provider") + payload = {"tenant": "fixture", "client_id": "client", "registration_id": "registration", + "normalized_email": "one@example.test", "preferred_username": "one"} + store.create_verification(payload, "handle", "2030-01-01T00:00:00+00:00") + store.consume_verification("handle") + store.db.close() + before.chmod(0o600) + after = tmp_path / "after.sqlite3" + after.write_bytes(before.read_bytes()) + after.chmod(0o600) + with sqlite3.connect(after) as db: + db.execute({"delete": "DELETE FROM deliveries", + "schema": "CREATE TABLE foreign_table (id TEXT)", + "unconsume": "UPDATE verifications SET consumed_at=NULL"}[change]) + with pytest.raises(transition.Refused): + transition.continuity(before, after) + + +def test_proposal_cannot_widen_the_configuration_change(attempt): + root, proposal = attempt + proposal = copy.deepcopy(proposal) + proposal["after"]["deployment_spec"]["template"]["spec"]["containers"][0]["image"] = "other" + with pytest.raises(transition.Refused, match="unreviewed_configuration_delta"): + execute(root, proposal) + assert FixtureAdapter(root).observe() == proposal["before"] + + +def test_writer_resumed_during_capture_refuses_configuration_change(attempt): + root, proposal = attempt + + class Interference(FixtureAdapter): + def snapshot(self, journal, *, quiesced): + result = super().snapshot(journal, quiesced=quiesced) + value = self.observe() + value["deployment_spec"]["replicas"] = 1 + write(root / "cluster.json", value) + return result + + journal = transition.Journal(root / "journal.jsonl", proposal=proposal) + try: + with pytest.raises(transition.Refused, match="resumed_before_config"): + transition.run(Interference(root), journal) + finally: + journal.close() + assert FixtureAdapter(root).observe() == proposal["before"] + + +def test_broken_capture_stops_before_switching_configuration(attempt): + root, proposal = attempt + + class BrokenCapture(FixtureAdapter): + def snapshot(self, journal, *, quiesced): + path, outcomes = super().snapshot(journal, quiesced=quiesced) + path.write_bytes(b"not a SQLite snapshot") + return path, outcomes + + journal = transition.Journal(root / "journal.jsonl", proposal=proposal) + try: + with pytest.raises(sqlite3.DatabaseError): + transition.run(BrokenCapture(root), journal) + assert journal.entries[-1]["phase"] == "requires_inspection" + finally: + journal.close() + expected = copy.deepcopy(proposal["before"]) + expected["deployment_spec"]["replicas"] = 0 + assert FixtureAdapter(root).observe() == expected + + +def test_nullable_sqlite_primary_keys_cannot_hide_rows_during_comparison(tmp_path): + path = tmp_path / "ambiguous.sqlite3" + store = SQLiteDeliveryStore(str(path)) + with store.db: + store.db.execute("INSERT INTO deliveries VALUES (NULL, 'first')") + store.db.execute("INSERT INTO deliveries VALUES (NULL, 'second')") + store.db.close() + path.chmod(0o600) + with pytest.raises(transition.Refused, match="ambiguous_snapshot_row_identity"): + transition.continuity(path, path) diff --git a/workplans/EMAIL-WP-0010-platform-sender-transition.md b/workplans/EMAIL-WP-0010-platform-sender-transition.md index 24c7d27..54d50f4 100644 --- a/workplans/EMAIL-WP-0010-platform-sender-transition.md +++ b/workplans/EMAIL-WP-0010-platform-sender-transition.md @@ -51,7 +51,7 @@ archive-fingerprint compatibility and private-output refusal. Owner suite: ```task id: EMAIL-WP-0010-T02 -status: todo +status: progress priority: high depends_on: [EMAIL-WP-0010-T01] state_hub_task_id: "acb6f995-9177-5a3c-909a-feb152d6bfed" @@ -76,6 +76,24 @@ Prepare a bounded recovery procedure that refuses unrelated drift. Deliver the exact review packet and passing synthetic/native dry-run evidence before asking for execution; a read-only ledger snapshot alone does not close this task. +2026-10-31 protocol implementation: `scripts/sender_transition.py` implements +private durable intent records, the exact sender/environment delta, interrupted +configuration recovery and six-table continuity comparison. Ten real child-process +exits cover phase boundaries and lost API replies against a file-backed cluster +fixture; recovery retains subsequent accepted/unknown sends and existing retry +refusal. Other cases cover repeated recovery interruption, unchanged protected +state, source/configuration drift, absent admission, journal privacy/truncation, +unexpected writer resume and preservation of consumed links, quotas and new rows. +All 27 new cases pass; owner default suite 219 passed / 21 existing optional harness skips. + +This is the protocol core, not a production executable or completed review packet. +The native adapter, identity/projection/admission verification, shared capture lock, +PVC fencing, source-bound helper capture/cleanup, runtime checks, capsule/digest, +attended entry point and native server dry-runs remain in T02. Existing capture +and projection tools are the integration surfaces; no credential/deployment/source +packet or native configuration was changed. Review: +`docs/sender-transition-protocol-2026-10-31.md`. + ## T03 — Execute the admitted sender switch ```task