Recover an outbox after a crash

# Outbox recovery lab

Two SQLite stores, separate Python dispatcher/downstream processes, real SIGKILL injection. Synthetic tickets only; no network, credentials or third-party dependencies.

Requires Python 3.10+ with SQLite on Linux/macOS. Run from this directory:

```sh
python3 lab.py --output results.json
python3 -m unittest -v test_lab.py
python3 lab.py --help
```

Five scenarios: before-dispatch crash; crash after downstream commit but before local completion; duplicate negative control with downstream deduplication off; crash inside caller acceptance transaction; changed downstream payload rejected.

Default databases are temporary. To retain both stores per scenario:

```sh
python3 lab.py --directory ./run-001 --output results-001.json
```

`--directory` must not exist. `--output` overwrites its named file. Choose new names to retain earlier evidence. Each child has a 15-second deadline. Worker-only modes (`accept`, `dispatch`, `service`) exist so that the experiment crosses real process boundaries; see `--help`. Do not use them on unrelated database files.

`results.json` is an observed run. `output.txt` contains its console output; `test-output.txt` records unit/integration verification. Process IDs vary. Scenario state, return codes and resource counts are the assertions. Checks in the runner remain active with Python optimisation enabled.

Limitations: no real network, broker, remote vendor, power failure, multi-worker claiming, ordering, retention expiry or authentication implementation. Downstream deduplication is a synthetic service contract. The outbox alone is shown to permit duplicates. No universal exactly-once guarantee.

# Source provenance

Reviewed 8 September 2026. Original code, synthetic payload and original failure-injection results. No source text reproduced verbatim in the article.

1. AWS Prescriptive Guidance, [Transactional outbox pattern](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html). Opened official page and reviewed intent, issues/considerations and relational outbox implementation. Supports atomically storing application state and outgoing intent, separate dispatch and need for an idempotent consumer because duplicates remain possible. This lab is an original local demonstration, not AWS sample code or an AWS service benchmark.
2. SQLite, [Transaction](https://www.sqlite.org/lang_transaction.html). Official transaction reference, previously reviewed in the local retries lab and rechecked for this extension. Supports local transaction and rollback semantics, not atomicity across independent services.
3. Python, [subprocess.Popen.returncode](https://docs.python.org/3/library/subprocess.html#subprocess.Popen.returncode). Official return-code contract: a negative N denotes termination by signal N on POSIX. The observed -9 is tested as SIGKILL in the runner.

# Execution provenance

Fedora Python 3.14.3, SQLite 3.51.2. `python3 lab.py --output results.json` runs actual subprocesses and sends SIGKILL to the deliberate faulting child itself. The dispatcher invokes a separate downstream child and never holds a caller transaction over that invocation. Two different on-disk SQLite files contain independently committed state. Results include process IDs, snapshots and return codes; no credentials or real customer data.

The post-commit fault occurs after the downstream subprocess exits successfully but before caller completion. It demonstrates a process crash in the acknowledgement gap, not a network timeout or power-loss durability test. The negative control deliberately disables downstream deduplication and expects two tickets. Claims in the article are limited to this fixture.

# Recover an agent tool call after the worker crashes

A tool worker has two records to write: the action in another service, and the completion marker in its own database. If it crashes between those writes, its replacement cannot infer what happened from a pending status alone.

This lab makes that gap visible. It uses two SQLite files and separate Python processes. We kill the dispatcher with `SIGKILL` after the downstream service commits a synthetic ticket, then start a new dispatcher. With downstream deduplication, recovery returns the original ticket. Disable deduplication and the same recovery creates a duplicate.

The [first tool-retries lab](https://swarmsignal.net/engineering-lab-tool-retries/) tested a ticket and its replay record inside one database transaction. Here, the caller and downstream service have independent transactions. Everything remains offline: no model calls, credentials or external tickets.

## Run the crash experiment

Download the ZIP below and extract it. In the extracted `engineering-labs/outbox-recovery` folder, run:

```sh
python3 lab.py --output results.json
python3 -m unittest -v test_lab.py
```

You need Python 3.10 or newer with SQLite support on Linux or macOS. Windows is not supported because the fixture explicitly uses POSIX `SIGKILL`. There are no third-party packages.

The default run creates temporary databases and removes them afterwards. To keep the stores for inspection, choose a directory that does not already exist:

```sh
python3 lab.py --directory ./run-001 --output results-001.json
```

The evidence option overwrites the named JSON file. The directory option refuses an existing directory, so it cannot silently reuse an earlier experiment's state. Each child has a 15-second test deadline. The runner only kills child processes it created; it does not search for or stop other workers.

## What actually happened

Observed on Fedora with Python 3.14.3 and SQLite 3.51.2:

| Failure injected | Tickets before recovery | Tickets after recovery | Recovered result |
| --- | ---: | ---: | --- |
| Dispatcher killed before dispatch | 0 | 1 | Ticket 1 |
| Dispatcher killed after downstream commit | 1 | 1 | Ticket 1 |
| Same crash, downstream deduplication disabled | 1 | 2 | Ticket 2 |

The acceptance crash test left **zero operations and zero outbox entries** before retrying. After resubmission and dispatch, it produced one ticket. A separate request with the original operation ID and a changed priority was rejected; the downstream store still contained one ticket.

The normal runner ends with:

```text
PASS: 5 scenarios; real SIGKILL, separate caller and downstream stores
```

`results.json` preserves the full before-and-after rows, child process IDs, return codes and recovered responses. A killed child returns `-9` to Python on this platform. The negative control is a passing test because it successfully demonstrates the expected duplicate, not because duplication is acceptable behaviour.

## Follow the two transactions

The caller accepts `operation-001` by inserting its payload into `operations` and a pending entry into `outbox` inside one transaction. The outbox is a durable record of work still to dispatch. It is not an in-memory retry queue.

A separate dispatcher reads that committed entry, closes the caller database connection and starts the downstream process. The downstream service writes a ticket and a replay receipt in its own transaction. That receipt maps the original operation ID and payload to the ticket result.

After receiving the downstream result, the dispatcher updates the local operation and outbox entry to `complete` together. There is no transaction spanning both files. That is the point of the experiment: recovery must work across the gap between independent commits.

AWS's transactional outbox guidance describes recording application state and the outgoing event together, then delivering committed entries separately. It also warns that duplicate deliveries remain possible and recommends idempotent consumers. The pattern preserves delivery intent; the consumer's contract determines whether redelivery repeats the effect. [AWS Prescriptive Guidance](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html)

## Put the crash after the irreversible step

In the most useful scenario, the downstream child commits, prints its result and exits. Only then does the dispatcher kill itself, before marking local completion. The caller store says `pending` with no result, while the downstream store already contains ticket 1.

This placement deliberately makes recovery uncertain from the caller's perspective. A crash before the downstream write would not exercise that uncertainty. A Python exception handled inside the dispatcher would also be weaker evidence than the abrupt process termination used here.

The replacement dispatcher obtains the same operation ID from the persisted outbox. The downstream child recognises that ID and replays ticket 1. The caller stores that result and marks completion. A further dispatcher invocation finds no pending work and changes neither store.

The checks compare the returned ticket ID with the downstream receipt and caller result. A successful exit status alone would not detect a wrong result or a second ticket.

## Why the duplicate control matters

Run the identical post-commit crash with deduplication disabled. The replacement dispatcher still reads the original outbox entry, still uses the same operation ID and still completes successfully. But the downstream service creates ticket 2.

An outbox alone does not stop duplicate side effects. A stable caller identifier is useful only when the receiving system honours it. Adding a local dictionary to the agent would not repair this boundary after a restart.

The lab also rejects a changed payload for an existing downstream operation ID. Reusing an ID must not quietly reinterpret an earlier intention. Its comparison uses canonical JSON; it does not decide whether two differently written instructions mean the same thing.

## Adapt the acceptance checks to your adapter

For a real sandbox integration, preserve the crash point between downstream success and local acknowledgement. Restart with the original durable operation ID, then inspect the downstream resource count and the result stored by the caller. Do not generate a fresh key simply because recovery started a new process.

Keep the pre-dispatch crash test too. It checks that accepted work remains discoverable. The acceptance-transaction crash verifies another boundary: an operation must not become durable without its corresponding outbox entry. SQLite's documented transaction behaviour underpins that local recovery, but this run is not a physical power-loss test. [SQLite transactions](https://www.sqlite.org/lang_transaction.html)

Before using the pattern in production, define replay retention, caller identity, retry deadlines and handling for permanently rejected work. Multi-worker dispatch needs a claiming or lease strategy; this lab runs one dispatcher at a time. It does not test ordering across multiple operations, broker delivery, lease expiry, network partitions or a real vendor's API.

The result is deliberately narrow: separate restarted processes recovered the same stored operation across an injected commit gap, provided the downstream receipt remained available. If your service cannot replay or reconcile that operation, this experiment identifies a requirement your adapter still has to meet.

## Sources

- [AWS: Transactional outbox pattern](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html)
- [SQLite: Transaction](https://www.sqlite.org/lang_transaction.html)
- [Python: subprocess return codes](https://docs.python.org/3/library/subprocess.html#subprocess.Popen.returncode)

Return to the [Engineering Lab](https://swarmsignal.net/engineering-lab/) for the local retry experiment and other runnable fixtures.

{"project":"demo-agent","title":"Investigate the synthetic queue backlog","priority":"normal"}

usage: lab.py [-h] [--directory DIRECTORY] [--output OUTPUT]
              [--operation-id OPERATION_ID] [--payload PAYLOAD]
              [--crash {during_accept,before_dispatch,after_downstream_commit}]
              [--no-deduplication]
              [{run,accept,dispatch,service}]

Offline outbox recovery with two SQLite files and real child-process crashes.

positional arguments:
  {run,accept,dispatch,service}

options:
  -h, --help            show this help message and exit
  --directory DIRECTORY
                        Fresh result directory for run; existing stores for
                        worker modes.
  --output OUTPUT       Write evidence JSON; overwrites the named file.
  --operation-id OPERATION_ID
  --payload PAYLOAD
  --crash {during_accept,before_dispatch,after_downstream_commit}
  --no-deduplication

#!/usr/bin/env python3
"""Offline outbox recovery with two SQLite files and real child-process crashes."""
import argparse
from contextlib import closing
import json
import os
from pathlib import Path
import platform
import signal
import sqlite3
import subprocess
import sys
import tempfile

HERE = Path(__file__).resolve().parent


def connect(path):
    db = sqlite3.connect(path, timeout=5, isolation_level=None)
    db.execute('PRAGMA synchronous=FULL')
    return db


def initialise(folder):
    folder.mkdir(parents=True, exist_ok=False)
    with closing(connect(folder / 'caller.sqlite')) as db:
        db.executescript('''
            CREATE TABLE operations (
                id TEXT PRIMARY KEY, payload TEXT NOT NULL,
                state TEXT NOT NULL, result TEXT
            );
            CREATE TABLE outbox (
                operation_id TEXT PRIMARY KEY REFERENCES operations(id),
                state TEXT NOT NULL
            );
        ''')
    with closing(connect(folder / 'downstream.sqlite')) as db:
        db.executescript('''
            CREATE TABLE tickets (
                id INTEGER PRIMARY KEY, operation_id TEXT NOT NULL, payload TEXT NOT NULL
            );
            CREATE TABLE receipts (
                operation_id TEXT PRIMARY KEY, payload TEXT NOT NULL, result TEXT NOT NULL
            );
        ''')


def encode(value):
    return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False)


def kill_here():
    os.kill(os.getpid(), signal.SIGKILL)


def accept(folder, operation_id, payload, crash=None):
    encoded = encode(payload)
    with closing(connect(folder / 'caller.sqlite')) as db:
        db.execute('BEGIN IMMEDIATE')
        db.execute('INSERT INTO operations VALUES (?, ?, ?, NULL)',
                   (operation_id, encoded, 'pending'))
        if crash == 'during_accept':
            kill_here()
        db.execute('INSERT INTO outbox VALUES (?, ?)', (operation_id, 'pending'))
        db.execute('COMMIT')
    return {'accepted': operation_id}


def service(folder, operation_id, payload, deduplicate=True):
    encoded = encode(payload)
    with closing(connect(folder / 'downstream.sqlite')) as db:
        try:
            db.execute('BEGIN IMMEDIATE')
            receipt = db.execute('SELECT payload, result FROM receipts WHERE operation_id=?',
                                 (operation_id,)).fetchone() if deduplicate else None
            if receipt:
                if receipt[0] != encoded:
                    db.execute('ROLLBACK')
                    return {'error': 'payload_conflict'}
                result = json.loads(receipt[1])
            else:
                row = db.execute('INSERT INTO tickets(operation_id, payload) VALUES (?, ?)',
                                 (operation_id, encoded))
                result = {'ticket_id': row.lastrowid}
                if deduplicate:
                    db.execute('INSERT INTO receipts VALUES (?, ?, ?)',
                               (operation_id, encoded, encode(result)))
            db.execute('COMMIT')
        except Exception:
            if db.in_transaction:
                db.execute('ROLLBACK')
            raise
    return {'result': result, 'service_pid': os.getpid()}


def child(mode, folder, operation_id='operation-001', payload=None, crash=None, deduplicate=True):
    command = [sys.executable, str(HERE / 'lab.py'), mode, '--directory', str(folder),
               '--operation-id', operation_id]
    if payload is not None:
        command += ['--payload', encode(payload)]
    if crash:
        command += ['--crash', crash]
    if not deduplicate:
        command += ['--no-deduplication']
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    try:
        stdout, stderr = process.communicate(timeout=15)
    except subprocess.TimeoutExpired:
        process.kill()
        process.communicate()
        raise RuntimeError('child exceeded the 15-second test deadline')
    return {'pid': process.pid, 'exit_code': process.returncode,
            'response': json.loads(stdout) if stdout.strip() else None,
            'stderr': stderr}


def dispatch(folder, crash=None, deduplicate=True):
    with closing(connect(folder / 'caller.sqlite')) as db:
        row = db.execute('''SELECT o.id, o.payload FROM operations o
            JOIN outbox b ON b.operation_id=o.id
            WHERE b.state='pending' ORDER BY o.id LIMIT 1''').fetchone()
    if row is None:
        return {'idle': True}
    operation_id, encoded = row
    if crash == 'before_dispatch':
        kill_here()
    remote = child('service', folder, operation_id, json.loads(encoded), deduplicate=deduplicate)
    if remote['exit_code'] != 0:
        return {'error': 'downstream_failure', 'downstream': remote}
    result = remote['response']['result']
    if crash == 'after_downstream_commit':
        kill_here()
    with closing(connect(folder / 'caller.sqlite')) as db:
        try:
            db.execute('BEGIN IMMEDIATE')
            db.execute('UPDATE operations SET state=?, result=? WHERE id=?',
                       ('complete', encode(result), operation_id))
            db.execute('UPDATE outbox SET state=? WHERE operation_id=?', ('complete', operation_id))
            db.execute('COMMIT')
        except Exception:
            if db.in_transaction:
                db.execute('ROLLBACK')
            raise
    return {'operation_id': operation_id, 'result': result, 'downstream': remote}


def snapshot(folder):
    with closing(connect(folder / 'caller.sqlite')) as db:
        operations = [dict(zip(('id', 'payload', 'state', 'result'), row))
                      for row in db.execute('SELECT * FROM operations ORDER BY id')]
        outbox = [dict(zip(('operation_id', 'state'), row))
                  for row in db.execute('SELECT * FROM outbox ORDER BY operation_id')]
    with closing(connect(folder / 'downstream.sqlite')) as db:
        tickets = [dict(zip(('id', 'operation_id', 'payload'), row))
                   for row in db.execute('SELECT * FROM tickets ORDER BY id')]
        receipts = [dict(zip(('operation_id', 'payload', 'result'), row))
                    for row in db.execute('SELECT * FROM receipts ORDER BY operation_id')]
    return {'operations': operations, 'outbox': outbox, 'tickets': tickets, 'receipts': receipts}


def require(condition, message):
    if not condition:
        raise AssertionError(message)


def run_case(root, name, crash, deduplicate=True):
    folder = root / name
    initialise(folder)
    payload = json.loads((HERE / 'fixture.json').read_text())
    acceptance = child('accept', folder, payload=payload)
    require(acceptance['exit_code'] == 0, 'acceptance failed')
    failed = child('dispatch', folder, crash=crash, deduplicate=deduplicate)
    require(failed['exit_code'] == -signal.SIGKILL, 'worker did not die from SIGKILL')
    before = snapshot(folder)
    expected_initial_tickets = 0 if crash == 'before_dispatch' else 1
    require(len(before['tickets']) == expected_initial_tickets, 'unexpected pre-recovery ticket count')
    require(before['operations'][0]['state'] == before['outbox'][0]['state'] == 'pending',
            'caller completed before crash')
    require(before['operations'][0]['result'] is None, 'caller has premature result')
    recovered = child('dispatch', folder, deduplicate=deduplicate)
    require(recovered['exit_code'] == 0, 'recovery failed')
    after = snapshot(folder)
    expected_tickets = 1 if deduplicate or crash == 'before_dispatch' else 2
    require(len(after['tickets']) == expected_tickets, 'unexpected recovered ticket count')
    require(after['operations'][0]['state'] == after['outbox'][0]['state'] == 'complete',
            'completion did not persist')
    result = recovered['response']['result']
    require(json.loads(after['operations'][0]['result']) == result, 'caller result mismatch')
    if deduplicate:
        require(result == {'ticket_id': 1}, 'original result not recovered')
        require(json.loads(after['receipts'][0]['result']) == result, 'receipt mismatch')
    idle = child('dispatch', folder, deduplicate=deduplicate)
    require(idle['response'] == {'idle': True}, 'completed outbox dispatched again')
    require(snapshot(folder) == after, 'idle dispatch changed state')
    return {'passed': True, 'crash': failed, 'acceptance': acceptance,
            'recovery': recovered, 'before_recovery': before, 'after_recovery': after,
            'summary': {'crash_exit': failed['exit_code'],
                        'tickets_before': expected_initial_tickets, 'tickets_after': len(after['tickets']),
                        'caller_state': after['operations'][0]['state'], 'ticket_id': result['ticket_id']}}


def run_suite(root):
    scenarios = {}
    scenarios['crash_before_dispatch'] = run_case(root, 'before', 'before_dispatch')
    scenarios['crash_after_downstream_commit'] = run_case(root, 'after', 'after_downstream_commit')
    scenarios['deduplication_disabled'] = run_case(root, 'negative', 'after_downstream_commit', False)
    folder = root / 'acceptance_rollback'
    initialise(folder)
    payload = json.loads((HERE / 'fixture.json').read_text())
    failed = child('accept', folder, payload=payload, crash='during_accept')
    require(failed['exit_code'] == -signal.SIGKILL, 'acceptance worker survived crash')
    before = snapshot(folder)
    require(before == {'operations': [], 'outbox': [], 'tickets': [], 'receipts': []},
            'partial accepted operation survived rollback')
    accepted = child('accept', folder, payload=payload)
    recovered = child('dispatch', folder)
    after = snapshot(folder)
    require(accepted['exit_code'] == recovered['exit_code'] == 0, 'acceptance recovery failed')
    require(len(after['tickets']) == 1 and after['outbox'][0]['state'] == 'complete',
            'accepted operation not recovered')
    scenarios['crash_during_acceptance'] = {'passed': True, 'crash': failed,
        'before_recovery': before, 'after_recovery': after,
        'summary': {'crash_exit': failed['exit_code'], 'partial_operations': len(before['operations']),
                    'partial_outbox': len(before['outbox']), 'tickets_after': len(after['tickets'])}}
    original = snapshot(root / 'after')
    conflict = child('service', root / 'after', payload={**payload, 'priority': 'urgent'})
    require(conflict['exit_code'] == 3 and conflict['response'] == {'error': 'payload_conflict'},
            'changed payload not rejected')
    require(snapshot(root / 'after') == original, 'conflict mutated persistent state')
    scenarios['changed_payload_rejected'] = {'passed': True, 'process': conflict,
        'summary': {'exit_code': conflict['exit_code'], 'tickets_after': len(original['tickets'])}}
    return {'experiment': 'two-store-outbox-recovery-v1',
            'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version,
            'platform': platform.system(), 'signal': 'SIGKILL', 'passed': True, 'scenarios': scenarios}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('mode', nargs='?', default='run', choices=['run', 'accept', 'dispatch', 'service'])
    parser.add_argument('--directory', type=Path, help='Fresh result directory for run; existing stores for worker modes.')
    parser.add_argument('--output', type=Path, help='Write evidence JSON; overwrites the named file.')
    parser.add_argument('--operation-id', default='operation-001')
    parser.add_argument('--payload', type=json.loads)
    parser.add_argument('--crash', choices=['during_accept', 'before_dispatch', 'after_downstream_commit'])
    parser.add_argument('--no-deduplication', action='store_true')
    args = parser.parse_args()
    if not hasattr(signal, 'SIGKILL'):
        parser.error('requires a POSIX system with SIGKILL (Linux or macOS)')
    if args.mode == 'run':
        if args.directory:
            try:
                args.directory.mkdir(parents=True, exist_ok=False)
            except FileExistsError:
                parser.error('--directory must not already exist; choose a fresh path')
            results = run_suite(args.directory)
        else:
            with tempfile.TemporaryDirectory(prefix='outbox-recovery-') as folder:
                results = run_suite(Path(folder))
        if args.output:
            args.output.write_text(json.dumps(results, indent=2) + '\n')
        for name, scenario in results['scenarios'].items():
            print(name + ': ' + json.dumps(scenario['summary'], sort_keys=True))
        print('PASS: 5 scenarios; real SIGKILL, separate caller and downstream stores')
        return
    if not args.directory:
        parser.error('worker modes require --directory')
    if args.mode in ('accept', 'service') and args.payload is None:
        parser.error('accept and service require --payload')
    if args.mode == 'accept':
        result = accept(args.directory, args.operation_id, args.payload, args.crash)
    elif args.mode == 'service':
        result = service(args.directory, args.operation_id, args.payload, not args.no_deduplication)
    else:
        result = dispatch(args.directory, args.crash, not args.no_deduplication)
    print(json.dumps(result))
    if 'error' in result:
        sys.exit(3)


if __name__ == '__main__':
    main()

crash_before_dispatch: {"caller_state": "complete", "crash_exit": -9, "ticket_id": 1, "tickets_after": 1, "tickets_before": 0}
crash_after_downstream_commit: {"caller_state": "complete", "crash_exit": -9, "ticket_id": 1, "tickets_after": 1, "tickets_before": 1}
deduplication_disabled: {"caller_state": "complete", "crash_exit": -9, "ticket_id": 2, "tickets_after": 2, "tickets_before": 1}
crash_during_acceptance: {"crash_exit": -9, "partial_operations": 0, "partial_outbox": 0, "tickets_after": 1}
changed_payload_rejected: {"exit_code": 3, "tickets_after": 1}
PASS: 5 scenarios; real SIGKILL, separate caller and downstream stores

{
  "experiment": "two-store-outbox-recovery-v1",
  "python": "3.14.3",
  "sqlite": "3.51.2",
  "platform": "Linux",
  "signal": "SIGKILL",
  "passed": true,
  "scenarios": {
    "crash_before_dispatch": {
      "passed": true,
      "crash": {
        "pid": 3674792,
        "exit_code": -9,
        "response": null,
        "stderr": ""
      },
      "acceptance": {
        "pid": 3674772,
        "exit_code": 0,
        "response": {
          "accepted": "operation-001"
        },
        "stderr": ""
      },
      "recovery": {
        "pid": 3674794,
        "exit_code": 0,
        "response": {
          "operation_id": "operation-001",
          "result": {
            "ticket_id": 1
          },
          "downstream": {
            "pid": 3674796,
            "exit_code": 0,
            "response": {
              "result": {
                "ticket_id": 1
              },
              "service_pid": 3674796
            },
            "stderr": ""
          }
        },
        "stderr": ""
      },
      "before_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "pending",
            "result": null
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "pending"
          }
        ],
        "tickets": [],
        "receipts": []
      },
      "after_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "complete",
            "result": "{\"ticket_id\":1}"
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "complete"
          }
        ],
        "tickets": [
          {
            "id": 1,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          }
        ],
        "receipts": [
          {
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "result": "{\"ticket_id\":1}"
          }
        ]
      },
      "summary": {
        "crash_exit": -9,
        "tickets_before": 0,
        "tickets_after": 1,
        "caller_state": "complete",
        "ticket_id": 1
      }
    },
    "crash_after_downstream_commit": {
      "passed": true,
      "crash": {
        "pid": 3674799,
        "exit_code": -9,
        "response": null,
        "stderr": ""
      },
      "acceptance": {
        "pid": 3674798,
        "exit_code": 0,
        "response": {
          "accepted": "operation-001"
        },
        "stderr": ""
      },
      "recovery": {
        "pid": 3674801,
        "exit_code": 0,
        "response": {
          "operation_id": "operation-001",
          "result": {
            "ticket_id": 1
          },
          "downstream": {
            "pid": 3674802,
            "exit_code": 0,
            "response": {
              "result": {
                "ticket_id": 1
              },
              "service_pid": 3674802
            },
            "stderr": ""
          }
        },
        "stderr": ""
      },
      "before_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "pending",
            "result": null
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "pending"
          }
        ],
        "tickets": [
          {
            "id": 1,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          }
        ],
        "receipts": [
          {
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "result": "{\"ticket_id\":1}"
          }
        ]
      },
      "after_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "complete",
            "result": "{\"ticket_id\":1}"
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "complete"
          }
        ],
        "tickets": [
          {
            "id": 1,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          }
        ],
        "receipts": [
          {
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "result": "{\"ticket_id\":1}"
          }
        ]
      },
      "summary": {
        "crash_exit": -9,
        "tickets_before": 1,
        "tickets_after": 1,
        "caller_state": "complete",
        "ticket_id": 1
      }
    },
    "deduplication_disabled": {
      "passed": true,
      "crash": {
        "pid": 3674805,
        "exit_code": -9,
        "response": null,
        "stderr": ""
      },
      "acceptance": {
        "pid": 3674804,
        "exit_code": 0,
        "response": {
          "accepted": "operation-001"
        },
        "stderr": ""
      },
      "recovery": {
        "pid": 3674807,
        "exit_code": 0,
        "response": {
          "operation_id": "operation-001",
          "result": {
            "ticket_id": 2
          },
          "downstream": {
            "pid": 3674808,
            "exit_code": 0,
            "response": {
              "result": {
                "ticket_id": 2
              },
              "service_pid": 3674808
            },
            "stderr": ""
          }
        },
        "stderr": ""
      },
      "before_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "pending",
            "result": null
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "pending"
          }
        ],
        "tickets": [
          {
            "id": 1,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          }
        ],
        "receipts": []
      },
      "after_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "complete",
            "result": "{\"ticket_id\":2}"
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "complete"
          }
        ],
        "tickets": [
          {
            "id": 1,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          },
          {
            "id": 2,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          }
        ],
        "receipts": []
      },
      "summary": {
        "crash_exit": -9,
        "tickets_before": 1,
        "tickets_after": 2,
        "caller_state": "complete",
        "ticket_id": 2
      }
    },
    "crash_during_acceptance": {
      "passed": true,
      "crash": {
        "pid": 3674810,
        "exit_code": -9,
        "response": null,
        "stderr": ""
      },
      "before_recovery": {
        "operations": [],
        "outbox": [],
        "tickets": [],
        "receipts": []
      },
      "after_recovery": {
        "operations": [
          {
            "id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "state": "complete",
            "result": "{\"ticket_id\":1}"
          }
        ],
        "outbox": [
          {
            "operation_id": "operation-001",
            "state": "complete"
          }
        ],
        "tickets": [
          {
            "id": 1,
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}"
          }
        ],
        "receipts": [
          {
            "operation_id": "operation-001",
            "payload": "{\"priority\":\"normal\",\"project\":\"demo-agent\",\"title\":\"Investigate the synthetic queue backlog\"}",
            "result": "{\"ticket_id\":1}"
          }
        ]
      },
      "summary": {
        "crash_exit": -9,
        "partial_operations": 0,
        "partial_outbox": 0,
        "tickets_after": 1
      }
    },
    "changed_payload_rejected": {
      "passed": true,
      "process": {
        "pid": 3674814,
        "exit_code": 3,
        "response": {
          "error": "payload_conflict"
        },
        "stderr": ""
      },
      "summary": {
        "exit_code": 3,
        "tickets_after": 1
      }
    }
  }
}

test_existing_output_directory_is_rejected_without_changes (test_lab.OutboxRecoveryTests.test_existing_output_directory_is_rejected_without_changes) ... ok
test_process_recovery_boundaries (test_lab.OutboxRecoveryTests.test_process_recovery_boundaries) ... ok
test_service_requires_payload (test_lab.OutboxRecoveryTests.test_service_requires_payload) ... ok

----------------------------------------------------------------------
Ran 3 tests in 0.909s

OK

from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

import lab


class OutboxRecoveryTests(unittest.TestCase):
    def test_process_recovery_boundaries(self):
        with tempfile.TemporaryDirectory() as folder:
            results = lab.run_suite(Path(folder))
        self.assertTrue(results['passed'])
        self.assertEqual(len(results['scenarios']), 5)
        after = results['scenarios']['crash_after_downstream_commit']
        self.assertEqual(after['before_recovery']['operations'][0]['result'], None)
        self.assertEqual(after['after_recovery']['operations'][0]['result'], '{"ticket_id":1}')
        self.assertNotEqual(after['crash']['pid'], after['recovery']['pid'])
        self.assertNotEqual(after['recovery']['pid'], after['recovery']['response']['downstream']['pid'])
        self.assertEqual(after['recovery']['response']['downstream']['pid'],
                         after['recovery']['response']['downstream']['response']['service_pid'])

    def test_existing_output_directory_is_rejected_without_changes(self):
        with tempfile.TemporaryDirectory() as folder:
            sentinel = Path(folder) / 'keep.txt'
            sentinel.write_text('keep me')
            result = subprocess.run([sys.executable, str(lab.HERE / 'lab.py'), '--directory', folder],
                                    text=True, capture_output=True, timeout=15)
            self.assertEqual(result.returncode, 2)
            self.assertEqual(sentinel.read_text(), 'keep me')
            self.assertEqual(sorted(p.name for p in Path(folder).iterdir()), ['keep.txt'])

    def test_service_requires_payload(self):
        with tempfile.TemporaryDirectory() as folder:
            result = subprocess.run([sys.executable, str(lab.HERE / 'lab.py'), 'service',
                                     '--directory', folder], text=True, capture_output=True, timeout=15)
            self.assertEqual(result.returncode, 2)
            self.assertIn('require --payload', result.stderr)
            self.assertEqual(list(Path(folder).iterdir()), [])


if __name__ == '__main__':
    unittest.main()