coulomb/email-connect
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prepare durable sender transition and current-ledger recovery protocol
Some checks failed
Transactional mail acceptance / mail (push) Failing after 1m11s
Details
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
This commit is contained in:
tegwick 2026-10-31 16:41:30 +01:00
parent c2ce58f10d
commit 422590934b
6 changed files with 858 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

2
WORK-RECORDS.md
View file

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

39
docs/sender-transition-protocol-2026-10-31.json Normal file
View file

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

90
docs/sender-transition-protocol-2026-10-31.md Normal file
View file

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

330
scripts/sender_transition.py Normal file
View file

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

379
tests/test_sender_transition.py Normal file
View file

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

20
workplans/EMAIL-WP-0010-platform-sender-transition.md
View file

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