diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md
index 4291ba016..1fcd327a8 100644
--- a/ARCHITECTURE.md
+++ b/ARCHITECTURE.md
@@ -474,9 +474,10 @@ revision and configuration digest prefixes.
`POST /api/analysis-runs` records a Pending run on a new authorized
cutoff capture (ADR 0017): snapshot, counts, frozen membership, run,
scope, and the first status in one transaction.
-`POST /api/analysis-runs/{id}/start` then runs ThreadWeave on that
-frozen bag and persists run-scoped edges (ADR 0021), or submits TEPP
-through `tepp_client` (ADR 0022). It does not invent a TEPP score.
+`POST /api/analysis-runs/{id}/start` then commits Running plus a
+durable outbox row, wakes Valkey, and delivers ThreadWeave on that
+frozen bag (ADR 0021 / ADR 0023) or submits TEPP through
+`tepp_client` (ADR 0022). It does not invent a TEPP score.
Request a lineage reconstruction from the home list, open the Pending
row, then start reconstruction. A Pending TEPP row starts a
measurement; a missing transport stays Failed /
@@ -506,7 +507,8 @@ payload is lookup labels plus non-negative aggregate counts -- never
source SQL, a DSN, a raw record, or a provider body. After `make seed`,
Demo Analyst and Demo Admin see "Lineage reconstruction · Succeeded ·
Demo Corp" with "3 documents" and Pending / Running / Succeeded times,
-and "TEPP measurement · Failed · Demo Corp" whose detail history ends
+the designed A-100 fork as clickable reconstructed edges, and
+"TEPP measurement · Failed · Demo Corp" whose detail history ends
in Failed / `tepp_not_available`.
A run-bearing registry is emptied only after an unrevoked
`analysis_run_retention_grant` and `GRANT analysis_run_retention_admin`,
diff --git a/CHANGELOG.d/0.93.0-seed-analysis-run-reconstruction.md b/CHANGELOG.d/0.93.0-seed-analysis-run-reconstruction.md
new file mode 100644
index 000000000..9bd198d8d
--- /dev/null
+++ b/CHANGELOG.d/0.93.0-seed-analysis-run-reconstruction.md
@@ -0,0 +1,4 @@
+# 0.93.0 Seed analysis-run reconstruction
+
+make seed persists the designed A-100 fork on the Demo Corp Succeeded
+lineage run. Open that run and click the revised-quote child.
diff --git a/CHANGELOG.d/0.94.0-analysis-run-outbox.md b/CHANGELOG.d/0.94.0-analysis-run-outbox.md
new file mode 100644
index 000000000..491436568
--- /dev/null
+++ b/CHANGELOG.d/0.94.0-analysis-run-outbox.md
@@ -0,0 +1,5 @@
+# 0.94.0 Analysis-run start outbox
+
+Start commits Running plus a durable outbox row, wakes Valkey, then
+delivers ThreadWeave or tepp_client. A crash no longer loses the work
+item. No TEPP theta is invented.
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b874d67de..a687e410c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,25 @@ All notable changes to this project are documented here. Format follows
[Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows
[Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## [0.94.0] - 2026-08-17
+
+### Added
+
+- `POST /api/analysis-runs/{id}/start` now commits Running plus one
+ durable outbox row, wakes Valkey (`analysis-run-outbox`), then
+ delivers ThreadWeave or `tepp_client` (ADR 0023). A crash after
+ Start leaves the work item; refresh finishes it. Period-report
+ stays 422. No TEPP theta is invented.
+
+## [0.93.0] - 2026-08-17
+
+### Added
+
+- `make seed` now persists the designed A-100 fork on the Demo Corp
+ Succeeded lineage run. Open that run: the revised quote and delivery
+ question follow the pricing follow-up and are buttons. Start is
+ unchanged. No TEPP theta is invented.
+
## [0.92.0] - 2026-08-17
### Added
diff --git a/CLAUDE.md b/CLAUDE.md
index d6d386499..3f347a2fe 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -34,8 +34,9 @@ after cutoff were rewritten after the run; compare those bodies
before treating them as reconstructed evidence (ADR 0016).
`POST /api/analysis-runs` records Pending on an authorized
cutoff capture (ADR 0017). `POST /api/analysis-runs/{id}/start`
-reconstructs that frozen cutoff bag (ADR 0021) or submits TEPP
-through `tepp_client` (ADR 0022). A missing transport or unused
-accepted envelope is Failed. Failed TEPP is terminal — request a
-new run, then start. Do not invent a theta. Hover the Result
-prefix to read the parent-choice digest.
+commits Running plus a durable outbox row, then reconstructs that
+frozen cutoff bag (ADR 0021 / ADR 0023) or submits TEPP through
+`tepp_client` (ADR 0022). A missing transport or unused accepted
+envelope is Failed. Failed TEPP is terminal — request a new run,
+then start. Do not invent a theta. Hover the Result prefix to read
+the parent-choice digest.
diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py
index fef422a59..2405b7b89 100644
--- a/backend/app/analysis_run_ingestion.py
+++ b/backend/app/analysis_run_ingestion.py
@@ -8,9 +8,9 @@
``create_pending_analysis_run`` (ADR 0017) writes snapshot, counts, frozen
membership, run, scope, and the first Pending event atomically.
-``start_pending_analysis_run`` later reconstructs lineage (ADR 0021)
-or submits TEPP through ``tepp_client`` (ADR 0022). Neither path
-invents a TEPP score.
+``enqueue_pending_analysis_run`` then ``deliver_queued_analysis_run``
+later reconstruct lineage (ADR 0021 / ADR 0023) or submit TEPP through
+``tepp_client`` (ADR 0022). Neither path invents a TEPP score.
"""
from __future__ import annotations
diff --git a/backend/app/analysis_run_outbox.py b/backend/app/analysis_run_outbox.py
new file mode 100644
index 000000000..7abeb2b31
--- /dev/null
+++ b/backend/app/analysis_run_outbox.py
@@ -0,0 +1,95 @@
+"""Durable start-work outbox. PostgreSQL is truth; Valkey is the wake-up.
+
+ADR 0023. Start writes Running plus one immutable outbox row, then a
+worker claims that row and runs ThreadWeave or ``tepp_client``. A crash
+after enqueue leaves the work item; it does not invent a theta.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from typing import Any
+
+import redis.asyncio as redis
+
+OUTBOX_STREAM_KEY = "analysis-run-outbox"
+_CLAIMED = "analysis_outbox_claimed"
+_DELIVERED = "analysis_outbox_delivered"
+
+
+def outbox_request_digest(
+ *,
+ analysis_run_id: str,
+ work_kind_code: str,
+ snapshot_sha256: str,
+ knowledge_cutoff: datetime,
+) -> str:
+ """SHA-256 of the frozen start request. Never hashes a post body."""
+ cutoff = knowledge_cutoff
+ if cutoff.tzinfo is None:
+ cutoff = cutoff.replace(tzinfo=timezone.utc)
+ material = json.dumps(
+ {
+ "analysis_run_id": str(analysis_run_id),
+ "knowledge_cutoff": cutoff.astimezone(timezone.utc).strftime(
+ "%Y-%m-%dT%H:%M:%SZ"
+ ),
+ "snapshot_sha256": snapshot_sha256,
+ "work_kind_code": work_kind_code,
+ },
+ separators=(",", ":"),
+ sort_keys=True,
+ )
+ return hashlib.sha256(material.encode()).hexdigest()
+
+
+def outbox_stream_fields(
+ *,
+ analysis_run_id: str,
+ work_kind_code: str,
+ request_sha256: str,
+) -> dict[str, str]:
+ """Valkey XADD fields for one start-work wake-up. No body, no theta."""
+ return {
+ "analysis_run_id": str(analysis_run_id),
+ "request_sha256": request_sha256,
+ "work_kind_code": work_kind_code,
+ }
+
+
+async def publish_outbox_event(
+ client: redis.Redis | None,
+ *,
+ analysis_run_id: str,
+ work_kind_code: str,
+ request_sha256: str,
+) -> str | None:
+ """``XADD`` the wake-up. A missing Valkey leaves PostgreSQL durable."""
+ if client is None:
+ return None
+ try:
+ entry_id = await client.xadd(
+ OUTBOX_STREAM_KEY,
+ outbox_stream_fields(
+ analysis_run_id=analysis_run_id,
+ work_kind_code=work_kind_code,
+ request_sha256=request_sha256,
+ ),
+ maxlen=1000,
+ approximate=True,
+ )
+ except redis.RedisError:
+ return None
+ return str(entry_id)
+
+
+def latest_outbox_delivery_is_delivered(status_code: str | None) -> bool:
+ """True when the newest delivery event already finished the work."""
+ return status_code == _DELIVERED
+
+
+def latest_outbox_delivery_is_claimed(status_code: str | None) -> bool:
+ """True when a worker already claimed the row and may retry."""
+ return status_code == _CLAIMED
diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py
index 474bcd2ef..37058d11b 100644
--- a/backend/app/analysis_run_start.py
+++ b/backend/app/analysis_run_start.py
@@ -1,8 +1,10 @@
"""Start a Pending lineage reconstruction or TEPP measurement.
ADR 0021 reconstructs lineage. ADR 0022 starts TEPP through
-``tepp_client`` only. Period-report stays another path. Neither start
-invents a theta or a calibrated report score.
+``tepp_client`` only. ADR 0023 enqueues that work on a durable outbox
+so a crash after Running does not lose the item. Period-report stays
+another path. Neither start invents a theta or a calibrated report
+score.
"""
from __future__ import annotations
@@ -19,6 +21,11 @@
AnalysisRunCreateError,
fetch_visible_analysis_run,
)
+from backend.app.analysis_run_outbox import (
+ latest_outbox_delivery_is_claimed,
+ latest_outbox_delivery_is_delivered,
+ outbox_request_digest,
+)
from backend.app.lineage_ingestion import records_from_source_posts
from lineageweave.http_client import HttpClientError, post_json
from lineageweave.lineage_persistence import lineage_edge_specs
@@ -253,23 +260,138 @@ async def _next_status_ordinal(
return int(current_max) + 1
-async def start_pending_analysis_run(
+async def _latest_outbox_delivery(
+ conn: asyncpg.Connection,
+ analysis_run_id: str,
+) -> str | None:
+ """Newest outbox delivery status, or None when the row was never claimed."""
+ return await conn.fetchval(
+ """
+ select delivery_status_code
+ from analysis_run_outbox_delivery
+ where analysis_run_id = $1
+ order by delivery_ordinal desc
+ limit 1
+ """,
+ analysis_run_id,
+ )
+
+
+async def _next_outbox_delivery_ordinal(
+ conn: asyncpg.Connection,
+ analysis_run_id: str,
+) -> int:
+ """Return the next contiguous outbox delivery ordinal for this run."""
+ current_max = await conn.fetchval(
+ """
+ select coalesce(max(delivery_ordinal), 0)
+ from analysis_run_outbox_delivery
+ where analysis_run_id = $1
+ """,
+ analysis_run_id,
+ )
+ return int(current_max) + 1
+
+
+async def _append_outbox_delivery(
+ conn: asyncpg.Connection,
+ analysis_run_id: str,
+ delivery_ordinal: int,
+ delivery_status_code: str,
+ occurred_at: datetime,
+ valkey_stream_entry_id: str | None = None,
+) -> None:
+ """Append one claim or delivery event. Stream id is optional."""
+ await conn.execute(
+ """
+ insert into analysis_run_outbox_delivery
+ (analysis_run_id, delivery_ordinal, delivery_status_code,
+ occurred_at, valkey_stream_entry_id)
+ values ($1, $2, $3, $4, $5)
+ """,
+ analysis_run_id,
+ delivery_ordinal,
+ delivery_status_code,
+ occurred_at,
+ valkey_stream_entry_id,
+ )
+
+
+async def _visible_or_404(
+ conn: asyncpg.Connection,
+ analysis_run_id: str,
+ account_id: str,
+ affiliated_entity_ids: list[str],
+) -> dict[str, Any]:
+ """Reload the authorized projection or hide the run."""
+ started = await fetch_visible_analysis_run(
+ conn,
+ analysis_run_id,
+ account_id,
+ affiliated_entity_ids,
+ )
+ if started is None:
+ raise AnalysisRunStartError(404, "This analysis run is not visible.")
+ return started
+
+
+async def _attach_outbox_digest(
+ conn: asyncpg.Connection,
+ started: dict[str, Any],
+) -> dict[str, Any]:
+ """Expose the wake-up digest to the start API, never to the client body."""
+ digest = await conn.fetchval(
+ """
+ select request_sha256
+ from analysis_run_outbox
+ where analysis_run_id = $1
+ """,
+ started["analysis_run_id"],
+ )
+ if not digest:
+ return started
+ attached = dict(started)
+ attached["outbox_request_sha256"] = str(digest)
+ return attached
+
+
+async def _lock_start_run(
+ conn: asyncpg.Connection,
+ analysis_run_id: str,
+) -> asyncpg.Record:
+ """Lock the run row used by enqueue and delivery."""
+ locked = await conn.fetchrow(
+ """
+ select run.analysis_run_id, run.knowledge_cutoff, run.run_kind_code,
+ run.idempotency_key, run.analysis_source_snapshot_id,
+ snapshot.snapshot_sha256, scope.corporate_entity_id
+ from analysis_run run
+ join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id
+ join analysis_source_snapshot snapshot
+ on snapshot.analysis_source_snapshot_id = run.analysis_source_snapshot_id
+ where run.analysis_run_id = $1
+ for update of run
+ """,
+ analysis_run_id,
+ )
+ if locked is None:
+ raise AnalysisRunStartError(404, "This analysis run is not visible.")
+ return locked
+
+
+async def enqueue_pending_analysis_run(
conn: asyncpg.Connection,
*,
analysis_run_id: str,
account_id: str,
affiliated_entity_ids: list[str],
- tepp_client: TeppClient | None = None,
) -> dict[str, Any]:
- """Run ThreadWeave or submit TEPP on a visible Pending row.
+ """Append Running and one outbox row, or resume an undelivered item.
Period-report is rejected so this path cannot invent a calibrated
- score. TEPP goes through ``tepp_client`` and stays Failed when the
- transport is missing or the envelope is not persistable. A Succeeded
- retry returns the stored reconstruction (documented no-op replay).
- A Running or concurrent write is 409. Hidden runs 404. The run row
- is locked before Running so a double-click is 409 or a replay,
- never a 500.
+ score. A Succeeded retry returns the stored reconstruction. A
+ Running row with an undelivered outbox is a crash resume. A Running
+ row without pending work is 409. Hidden runs 404.
"""
try:
UUID(analysis_run_id)
@@ -289,27 +411,8 @@ async def start_pending_analysis_run(
raise kind_error
if current["status_code"] == _SUCCEEDED:
return current
- if current["status_code"] != _PENDING:
- raise AnalysisRunStartError(
- 409,
- "Open this run. Start is only for a Pending lineage reconstruction "
- "or TEPP measurement.",
- )
- locked = await conn.fetchrow(
- """
- select run.analysis_run_id, run.knowledge_cutoff, run.run_kind_code,
- run.idempotency_key, run.analysis_source_snapshot_id,
- snapshot.snapshot_sha256, scope.corporate_entity_id
- from analysis_run run
- join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id
- join analysis_source_snapshot snapshot
- on snapshot.analysis_source_snapshot_id = run.analysis_source_snapshot_id
- where run.analysis_run_id = $1
- for update of run
- """,
- analysis_run_id,
- )
+ locked = await _lock_start_run(conn, analysis_run_id)
locked_status = await conn.fetchval(
"""
select status_code
@@ -319,15 +422,35 @@ async def start_pending_analysis_run(
analysis_run_id,
)
if locked_status == _SUCCEEDED:
- replayed = await fetch_visible_analysis_run(
- conn,
+ return await _visible_or_404(
+ conn, analysis_run_id, account_id, affiliated_entity_ids
+ )
+ if locked_status == _RUNNING:
+ latest = await _latest_outbox_delivery(conn, analysis_run_id)
+ if latest_outbox_delivery_is_delivered(latest):
+ raise AnalysisRunStartError(
+ 409,
+ "Open this run. Start is only for a Pending lineage reconstruction "
+ "or TEPP measurement.",
+ )
+ has_outbox = await conn.fetchval(
+ """
+ select 1 from analysis_run_outbox where analysis_run_id = $1
+ """,
analysis_run_id,
- account_id,
- affiliated_entity_ids,
)
- if replayed is None:
- raise AnalysisRunStartError(404, "This analysis run is not visible.")
- return replayed
+ if has_outbox is None:
+ raise AnalysisRunStartError(
+ 409,
+ "Open this run. Start is only for a Pending lineage reconstruction "
+ "or TEPP measurement.",
+ )
+ return await _attach_outbox_digest(
+ conn,
+ await _visible_or_404(
+ conn, analysis_run_id, account_id, affiliated_entity_ids
+ ),
+ )
if locked_status != _PENDING:
raise AnalysisRunStartError(
409,
@@ -335,123 +458,262 @@ async def start_pending_analysis_run(
"or TEPP measurement.",
)
- if locked["run_kind_code"] == _TEPP_KIND:
- return await _start_tepp_measurement(
- conn,
- analysis_run_id=analysis_run_id,
- account_id=account_id,
- affiliated_entity_ids=affiliated_entity_ids,
- locked=locked,
- tepp_client=tepp_client or TeppClient(),
- )
-
now = datetime.now(timezone.utc)
- running_ordinal = await _next_status_ordinal(conn, analysis_run_id)
+ digest = outbox_request_digest(
+ analysis_run_id=str(locked["analysis_run_id"]),
+ work_kind_code=str(locked["run_kind_code"]),
+ snapshot_sha256=str(locked["snapshot_sha256"]),
+ knowledge_cutoff=locked["knowledge_cutoff"],
+ )
try:
- await _append_status(conn, analysis_run_id, running_ordinal, _RUNNING, now)
- member_rows = await _snapshot_member_posts(
+ await _append_status(
conn,
- locked["analysis_source_snapshot_id"],
+ analysis_run_id,
+ await _next_status_ordinal(conn, analysis_run_id),
+ _RUNNING,
+ now,
)
- if member_rows:
- rows = member_rows
- else:
- rows = await _cutoff_source_posts(
- conn,
- corporate_entity_id=locked["corporate_entity_id"],
- knowledge_cutoff=locked["knowledge_cutoff"],
- affiliated_entity_ids=affiliated_entity_ids,
- )
- edges = lineage_edge_specs(records_from_source_posts(rows))
- digest = reconstruction_result_digest(edges)
- finished = datetime.now(timezone.utc)
- if finished < now:
- finished = now
await conn.execute(
"""
- insert into analysis_run_reconstruction
- (analysis_run_id, result_sha256, edge_count, reconstructed_at)
+ insert into analysis_run_outbox
+ (analysis_run_id, work_kind_code, request_sha256, enqueued_at)
values ($1, $2, $3, $4)
""",
analysis_run_id,
+ locked["run_kind_code"],
digest,
- len(edges),
- finished,
- )
- for edge in edges:
- await conn.execute(
- """
- insert into analysis_run_lineage_edge
- (analysis_run_id, child_post_id, parent_post_id,
- fused_score, reconstructed_at)
- values ($1, $2, $3, $4, $5)
- """,
- analysis_run_id,
- edge.child_id,
- edge.parent_id,
- edge.fused_score,
- finished,
- )
- await _append_status(
- conn,
- analysis_run_id,
- running_ordinal + 1,
- _SUCCEEDED,
- finished,
+ now,
)
except asyncpg.UniqueViolationError as exc:
raise start_write_conflict_error() from exc
- started = await fetch_visible_analysis_run(
+ return await _attach_outbox_digest(
conn,
- analysis_run_id,
- account_id,
- affiliated_entity_ids,
+ await _visible_or_404(
+ conn, analysis_run_id, account_id, affiliated_entity_ids
+ ),
)
- if started is None:
- raise AnalysisRunStartError(404, "This analysis run is not visible.")
- return started
-async def _start_tepp_measurement(
+async def deliver_queued_analysis_run(
conn: asyncpg.Connection,
*,
analysis_run_id: str,
account_id: str,
affiliated_entity_ids: list[str],
- locked: asyncpg.Record,
- tepp_client: TeppClient,
+ tepp_client: TeppClient | None = None,
+ valkey_stream_entry_id: str | None = None,
) -> dict[str, Any]:
- """Submit the frozen snapshot through ``tepp_client``. Never persist a theta."""
- now = datetime.now(timezone.utc)
- running_ordinal = await _next_status_ordinal(conn, analysis_run_id)
+ """Claim the outbox row and finish ThreadWeave or TEPP.
+
+ A delivered row replays the stored result. Missing work is 409.
+ TEPP stays Failed when the transport is missing or the envelope is
+ not persistable. No theta is invented.
+ """
try:
- await _append_status(conn, analysis_run_id, running_ordinal, _RUNNING, now)
- request = tepp_run_request(
- idempotency_key=str(locked["idempotency_key"]),
- snapshot_sha256=str(locked["snapshot_sha256"]),
- knowledge_cutoff=locked["knowledge_cutoff"],
- corporate_entity_id=str(locked["corporate_entity_id"]),
+ UUID(analysis_run_id)
+ except ValueError as exc:
+ raise AnalysisRunStartError(404, "This analysis run is not visible.") from exc
+
+ current = await fetch_visible_analysis_run(
+ conn,
+ analysis_run_id,
+ account_id,
+ affiliated_entity_ids,
+ )
+ if current is None:
+ raise AnalysisRunStartError(404, "This analysis run is not visible.")
+ if current["status_code"] == _SUCCEEDED:
+ return current
+
+ outbox = await conn.fetchrow(
+ """
+ select outbox.analysis_run_id, outbox.work_kind_code,
+ run.knowledge_cutoff, run.idempotency_key,
+ run.analysis_source_snapshot_id, snapshot.snapshot_sha256,
+ scope.corporate_entity_id
+ from analysis_run_outbox outbox
+ join analysis_run run on run.analysis_run_id = outbox.analysis_run_id
+ join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id
+ join analysis_source_snapshot snapshot
+ on snapshot.analysis_source_snapshot_id = run.analysis_source_snapshot_id
+ where outbox.analysis_run_id = $1
+ for update of outbox
+ """,
+ analysis_run_id,
+ )
+ if outbox is None:
+ raise AnalysisRunStartError(
+ 409,
+ "Open this run. Start is only for a Pending lineage reconstruction "
+ "or TEPP measurement.",
+ )
+ latest = await _latest_outbox_delivery(conn, analysis_run_id)
+ if latest_outbox_delivery_is_delivered(latest):
+ return await _visible_or_404(
+ conn, analysis_run_id, account_id, affiliated_entity_ids
)
- status_code, failure_code = tepp_submit_outcome(tepp_client, request)
+ now = datetime.now(timezone.utc)
+ try:
+ if not latest_outbox_delivery_is_claimed(latest):
+ await _append_outbox_delivery(
+ conn,
+ analysis_run_id,
+ await _next_outbox_delivery_ordinal(conn, analysis_run_id),
+ "analysis_outbox_claimed",
+ now,
+ valkey_stream_entry_id,
+ )
+ if outbox["work_kind_code"] == _TEPP_KIND:
+ await _deliver_tepp_measurement(
+ conn,
+ analysis_run_id=analysis_run_id,
+ locked=outbox,
+ tepp_client=tepp_client or TeppClient(),
+ )
+ else:
+ await _deliver_lineage_reconstruction(
+ conn,
+ analysis_run_id=analysis_run_id,
+ locked=outbox,
+ affiliated_entity_ids=affiliated_entity_ids,
+ )
finished = datetime.now(timezone.utc)
if finished < now:
finished = now
- await _append_status(
+ await _append_outbox_delivery(
conn,
analysis_run_id,
- running_ordinal + 1,
- status_code,
+ await _next_outbox_delivery_ordinal(conn, analysis_run_id),
+ "analysis_outbox_delivered",
finished,
- failure_code,
+ valkey_stream_entry_id,
)
except asyncpg.UniqueViolationError as exc:
raise start_write_conflict_error() from exc
- started = await fetch_visible_analysis_run(
+ return await _visible_or_404(
+ conn, analysis_run_id, account_id, affiliated_entity_ids
+ )
+
+
+async def start_pending_analysis_run(
+ conn: asyncpg.Connection,
+ *,
+ analysis_run_id: str,
+ account_id: str,
+ affiliated_entity_ids: list[str],
+ tepp_client: TeppClient | None = None,
+ valkey_stream_entry_id: str | None = None,
+) -> dict[str, Any]:
+ """Enqueue then deliver on one connection.
+
+ The HTTP start path commits the outbox before this delivery so a
+ crash leaves Running plus a durable work item. Callers that wrap
+ both steps in one transaction keep the older all-or-nothing
+ behavior.
+ """
+ queued = await enqueue_pending_analysis_run(
+ conn,
+ analysis_run_id=analysis_run_id,
+ account_id=account_id,
+ affiliated_entity_ids=affiliated_entity_ids,
+ )
+ if queued["status_code"] == _SUCCEEDED:
+ return queued
+ return await deliver_queued_analysis_run(
+ conn,
+ analysis_run_id=analysis_run_id,
+ account_id=account_id,
+ affiliated_entity_ids=affiliated_entity_ids,
+ tepp_client=tepp_client,
+ valkey_stream_entry_id=valkey_stream_entry_id,
+ )
+
+
+async def _deliver_lineage_reconstruction(
+ conn: asyncpg.Connection,
+ *,
+ analysis_run_id: str,
+ locked: asyncpg.Record,
+ affiliated_entity_ids: list[str],
+) -> None:
+ """Persist ThreadWeave parent choices for the frozen bag."""
+ now = datetime.now(timezone.utc)
+ member_rows = await _snapshot_member_posts(
+ conn,
+ locked["analysis_source_snapshot_id"],
+ )
+ if member_rows:
+ rows = member_rows
+ else:
+ rows = await _cutoff_source_posts(
+ conn,
+ corporate_entity_id=locked["corporate_entity_id"],
+ knowledge_cutoff=locked["knowledge_cutoff"],
+ affiliated_entity_ids=affiliated_entity_ids,
+ )
+ edges = lineage_edge_specs(records_from_source_posts(rows))
+ digest = reconstruction_result_digest(edges)
+ finished = datetime.now(timezone.utc)
+ if finished < now:
+ finished = now
+ await conn.execute(
+ """
+ insert into analysis_run_reconstruction
+ (analysis_run_id, result_sha256, edge_count, reconstructed_at)
+ values ($1, $2, $3, $4)
+ """,
+ analysis_run_id,
+ digest,
+ len(edges),
+ finished,
+ )
+ for edge in edges:
+ await conn.execute(
+ """
+ insert into analysis_run_lineage_edge
+ (analysis_run_id, child_post_id, parent_post_id,
+ fused_score, reconstructed_at)
+ values ($1, $2, $3, $4, $5)
+ """,
+ analysis_run_id,
+ edge.child_id,
+ edge.parent_id,
+ edge.fused_score,
+ finished,
+ )
+ await _append_status(
conn,
analysis_run_id,
- account_id,
- affiliated_entity_ids,
+ await _next_status_ordinal(conn, analysis_run_id),
+ _SUCCEEDED,
+ finished,
+ )
+
+
+async def _deliver_tepp_measurement(
+ conn: asyncpg.Connection,
+ *,
+ analysis_run_id: str,
+ locked: asyncpg.Record,
+ tepp_client: TeppClient,
+) -> None:
+ """Submit the frozen snapshot through ``tepp_client``. Never persist a theta."""
+ now = datetime.now(timezone.utc)
+ request = tepp_run_request(
+ idempotency_key=str(locked["idempotency_key"]),
+ snapshot_sha256=str(locked["snapshot_sha256"]),
+ knowledge_cutoff=locked["knowledge_cutoff"],
+ corporate_entity_id=str(locked["corporate_entity_id"]),
+ )
+ status_code, failure_code = tepp_submit_outcome(tepp_client, request)
+ finished = datetime.now(timezone.utc)
+ if finished < now:
+ finished = now
+ await _append_status(
+ conn,
+ analysis_run_id,
+ await _next_status_ordinal(conn, analysis_run_id),
+ status_code,
+ finished,
+ failure_code,
)
- if started is None:
- raise AnalysisRunStartError(404, "This analysis run is not visible.")
- return started
diff --git a/backend/app/main.py b/backend/app/main.py
index da7067f43..55576ca19 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -72,10 +72,12 @@
fetch_visible_analysis_run,
fetch_visible_analysis_runs,
)
+from backend.app.analysis_run_outbox import publish_outbox_event
from backend.app.analysis_run_start import (
AnalysisRunStartError,
configured_tepp_client,
- start_pending_analysis_run,
+ deliver_queued_analysis_run,
+ enqueue_pending_analysis_run,
)
from backend.app.activity_stream import (
create_valkey_client,
@@ -1263,26 +1265,53 @@ async def start_analysis_run(
analysis_run_id: str,
account: CurrentAccount = Depends(get_current_account),
pool: asyncpg.Pool = Depends(get_pool),
+ valkey: redis.Redis = Depends(get_valkey),
) -> dict[str, Any]:
- """Start ThreadWeave or submit TEPP on a visible Pending run.
+ """Enqueue start work, then deliver ThreadWeave or TEPP.
post_read is enough. Hidden runs 404. Period-report is 422 so this
path cannot invent a calibrated score. TEPP goes through
``tepp_client`` and stays Failed when the transport is missing or
the envelope is not persistable. A Succeeded lineage retry returns
- the stored tree. A Running restart is 409.
+ the stored tree. A Running restart with an undelivered outbox
+ finishes that work. A Running restart without pending work is 409.
+ The outbox commits before reconstruct/TEPP so a crash leaves a
+ durable work item (ADR 0023).
"""
_require_post_read(account)
settings = load_settings()
async with pool.acquire() as conn:
async with conn.transaction():
try:
- started = await start_pending_analysis_run(
+ queued = await enqueue_pending_analysis_run(
+ conn,
+ analysis_run_id=analysis_run_id,
+ account_id=account.user_account_id,
+ affiliated_entity_ids=list(account.corporate_entity_ids),
+ )
+ except AnalysisRunStartError as exc:
+ raise HTTPException(exc.status_code, exc.detail) from exc
+ if queued.get("status_code") == "analysis_status_succeeded":
+ return queued
+ request_digest = queued.pop("outbox_request_sha256", None)
+ stream_id = None
+ if request_digest:
+ stream_id = await publish_outbox_event(
+ valkey,
+ analysis_run_id=analysis_run_id,
+ work_kind_code=str(queued.get("run_kind_code") or ""),
+ request_sha256=request_digest,
+ )
+ async with pool.acquire() as conn:
+ async with conn.transaction():
+ try:
+ started = await deliver_queued_analysis_run(
conn,
analysis_run_id=analysis_run_id,
account_id=account.user_account_id,
affiliated_entity_ids=list(account.corporate_entity_ids),
tepp_client=configured_tepp_client(settings.tepp_transport_url),
+ valkey_stream_entry_id=stream_id,
)
except AnalysisRunStartError as exc:
raise HTTPException(exc.status_code, exc.detail) from exc
diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py
index 9d5995bfd..3d08b8650 100644
--- a/backend/tests/test_api.py
+++ b/backend/tests/test_api.py
@@ -39,6 +39,9 @@
_SNAPSHOT_MEMBER_MIGRATION = (
Path(__file__).resolve().parents[2] / "migrations" / "0022_analysis_source_snapshot_member.sql"
)
+_OUTBOX_MIGRATION = (
+ Path(__file__).resolve().parents[2] / "migrations" / "0023_analysis_run_outbox.sql"
+)
def _postgres_available() -> bool:
@@ -125,6 +128,7 @@ def seeded_db(demo_analyst_token):
cur.execute(_RETENTION_MIGRATION.read_text())
cur.execute(_RECONSTRUCTION_MIGRATION.read_text())
cur.execute(_SNAPSHOT_MEMBER_MIGRATION.read_text())
+ cur.execute(_OUTBOX_MIGRATION.read_text())
cur.execute(
"insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values "
"('corporate_entity_level', 'group', 'Group'), "
@@ -669,6 +673,39 @@ def test_start_analysis_run_recovers_the_a100_fork(
assert "Pricing renegotiation: revised quote sent" in children
assert "Delivery schedule question raised" in children
assert "theta" not in str(body).lower()
+ assert "outbox_request_sha256" not in body
+
+ admin_conn = psycopg2.connect(seeded_db["dsn"])
+ admin_conn.autocommit = True
+ try:
+ with admin_conn.cursor() as cur:
+ cur.execute(
+ """
+ select outbox.work_kind_code, delivery.delivery_status_code
+ from analysis_run_outbox outbox
+ join analysis_run_outbox_delivery delivery
+ on delivery.analysis_run_id = outbox.analysis_run_id
+ where outbox.analysis_run_id = %s
+ order by delivery.delivery_ordinal desc
+ limit 1
+ """,
+ (run_id,),
+ )
+ outbox_row = cur.fetchone()
+ assert outbox_row == ("analysis_run_lineage", "analysis_outbox_delivered")
+ finally:
+ admin_conn.close()
+ valkey = redis.from_url(_VALKEY_URL, decode_responses=True)
+ try:
+ entries = valkey.xrevrange("analysis-run-outbox", count=50)
+ assert any(
+ fields.get("analysis_run_id") == run_id
+ and fields.get("work_kind_code") == "analysis_run_lineage"
+ and "theta" not in str(fields).casefold()
+ for _entry_id, fields in entries
+ )
+ finally:
+ valkey.close()
replay = client.post(
f"/api/analysis-runs/{run_id}/start",
@@ -823,6 +860,83 @@ def test_start_analysis_run_recovers_the_a100_fork(
)
assert hidden.status_code == 404
+ admin_conn = psycopg2.connect(seeded_db["dsn"])
+ admin_conn.autocommit = True
+ try:
+ with admin_conn.cursor() as cur:
+ cur.execute(
+ """
+ insert into analysis_source_snapshot
+ (snapshot_sha256, source_contract_version,
+ maximum_available_time, captured_at)
+ values (%s, 'source-contract-v1',
+ '2026-01-12T00:00:00Z', '2026-01-12T00:05:00Z')
+ returning analysis_source_snapshot_id
+ """,
+ ("3" * 64,),
+ )
+ crash_snapshot_id = cur.fetchone()[0]
+ cur.execute(
+ """
+ insert into analysis_run
+ (analysis_source_snapshot_id, run_kind_code, idempotency_key,
+ requested_by_account_id, knowledge_cutoff,
+ configuration_schema_version, configuration_sha256,
+ code_revision_sha, requested_at)
+ values (%s, 'analysis_run_lineage', 'buyer-start-outbox-resume',
+ %s, '2026-02-15T00:00:00Z', 'lineage-run-v1', %s, %s,
+ '2026-02-15T12:30:00Z')
+ returning analysis_run_id
+ """,
+ (crash_snapshot_id, requester_id, "2" * 64, "1" * 40),
+ )
+ crash_run_id = str(cur.fetchone()[0])
+ cur.execute(
+ """
+ insert into analysis_run_scope
+ (analysis_run_id, scope_kind_code, corporate_entity_id)
+ values (%s, 'analysis_scope_corporate_entity', %s)
+ """,
+ (crash_run_id, seeded_db["own_corp_id"]),
+ )
+ for ordinal, status, occurred in (
+ (1, "analysis_status_pending", "2026-02-15T12:31:00Z"),
+ (2, "analysis_status_running", "2026-02-15T12:32:00Z"),
+ ):
+ cur.execute(
+ """
+ insert into analysis_run_status_event
+ (analysis_run_id, status_ordinal, status_code, occurred_at)
+ values (%s, %s, %s, %s)
+ """,
+ (crash_run_id, ordinal, status, occurred),
+ )
+ cur.execute(
+ """
+ insert into analysis_run_outbox
+ (analysis_run_id, work_kind_code, request_sha256, enqueued_at)
+ values (%s, 'analysis_run_lineage', %s, '2026-02-15T12:32:00Z')
+ """,
+ (crash_run_id, "a" * 64),
+ )
+ finally:
+ admin_conn.close()
+
+ resumed = client.post(
+ f"/api/analysis-runs/{crash_run_id}/start",
+ headers={"Authorization": f"Bearer {demo_analyst_token}"},
+ )
+ assert resumed.status_code == 200, resumed.text
+ resumed_body = resumed.json()
+ assert resumed_body["status_label"] == "Succeeded"
+ assert "theta" not in str(resumed_body).lower()
+ children = {
+ edge["child_post_title"]
+ for edge in resumed_body["reconstructed_edges"]
+ if edge["parent_post_title"] == "Pricing renegotiation follow-up"
+ }
+ assert "Pricing renegotiation: revised quote sent" in children
+
def test_me_reflects_the_authenticated_account(client, demo_analyst_token) -> None:
response = client.get("/api/me", headers={"Authorization": f"Bearer {demo_analyst_token}"})
@@ -836,7 +950,12 @@ def test_post_list_includes_public_and_own_corp_but_excludes_other_corp(client,
response = client.get("/api/posts", headers={"Authorization": f"Bearer {demo_analyst_token}"})
assert response.status_code == 200
titles = {post["post_title"] for post in response.json()}
- assert titles == {"Public post", "Own-corp private post", "Late own-corp private post"}
+ assert titles == {
+ "Public post",
+ "Own-corp private post",
+ "Late own-corp private post",
+ "Edited own-corp private post",
+ }
public = next(post for post in response.json() if post["post_title"] == "Public post")
assert public["voc_type_label"] == "Voice of Customer"
assert public["visibility_label"] == "Public"
@@ -1934,14 +2053,19 @@ def test_thread_group_run_list_honors_knowledge_cutoff(
""",
(run_id,),
)
- cur.execute(
- """
- insert into analysis_run_status_event
- (analysis_run_id, status_ordinal, status_code, occurred_at)
- values (%s, 1, 'analysis_status_succeeded', '2026-01-12T12:33:00Z')
- """,
- (run_id,),
- )
+ for ordinal, status, occurred in (
+ (1, "analysis_status_pending", "2026-01-12T12:31:00Z"),
+ (2, "analysis_status_running", "2026-01-12T12:32:00Z"),
+ (3, "analysis_status_succeeded", "2026-01-12T12:33:00Z"),
+ ):
+ cur.execute(
+ """
+ insert into analysis_run_status_event
+ (analysis_run_id, status_ordinal, status_code, occurred_at)
+ values (%s, %s, %s, %s)
+ """,
+ (run_id, ordinal, status, occurred),
+ )
finally:
admin_conn.close()
diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile
index 71e9fc733..2e016a607 100644
--- a/docker/postgres-init/Dockerfile
+++ b/docker/postgres-init/Dockerfile
@@ -28,6 +28,7 @@ COPY migrations/0019_role_catalog_identity.sql /docker-entrypoint-initdb.d/20-ro
COPY migrations/0020_analysis_run_retention_purge.sql /docker-entrypoint-initdb.d/21-analysis-run-retention-purge.sql
COPY migrations/0021_analysis_run_reconstruction.sql /docker-entrypoint-initdb.d/22-analysis-run-reconstruction.sql
COPY migrations/0022_analysis_source_snapshot_member.sql /docker-entrypoint-initdb.d/23-analysis-source-snapshot-member.sql
+COPY migrations/0023_analysis_run_outbox.sql /docker-entrypoint-initdb.d/24-analysis-run-outbox.sql
# Official image already drops to this account at runtime; declare it so
# the Dockerfile itself satisfies DS-0002 (explicit non-root USER).
USER postgres
diff --git a/docs/adr/0013-normalized-analysis-run-registry.md b/docs/adr/0013-normalized-analysis-run-registry.md
index 8184c9ffd..96e1d8917 100644
--- a/docs/adr/0013-normalized-analysis-run-registry.md
+++ b/docs/adr/0013-normalized-analysis-run-registry.md
@@ -160,7 +160,7 @@ runs. This migration does not claim that an API or UI exists.
reviewed API.
- **fast-mlsirm** owns Rust psychometric arithmetic and calibration interfaces.
- **Valkey** remains the event queue. Durable registry truth remains in
- PostgreSQL; a later outbox slice bridges the two.
+ PostgreSQL; the start outbox (ADR 0023) bridges the two.
No component reads another service's private application tables.
@@ -242,11 +242,13 @@ Acceptance requires:
`POST /api/analysis-runs` now records that Pending write (ADR 0017).
`POST /api/analysis-runs/{id}/start` now reconstructs a Pending
lineage cutoff bag in-process from frozen snapshot membership
- (ADR 0021). A durable outbox / Valkey worker remains a later slice.
- Live TEPP start now submits through `tepp_client` (ADR 0022).
+ (ADR 0021). Start now commits a durable outbox row and wakes Valkey
+ before reconstruct / TEPP (ADR 0023). Live TEPP start submits
+ through `tepp_client` (ADR 0022).
2. Add RBAC/ABAC-protected run list/detail endpoints and the DB-grounded
read-only administrator surface.
-3. Add a normalized PostgreSQL outbox and Valkey delivery worker.
+3. Add a normalized PostgreSQL outbox and Valkey delivery worker
+ (ADR 0023).
4. Add TEPP and contextual-orchestrator adapters only after their versioned
contracts are present on reviewed main branches. Seed and
`POST /api/analysis-runs/{id}/start` now record Failed TEPP through
diff --git a/docs/adr/0017-authorized-analysis-run-create.md b/docs/adr/0017-authorized-analysis-run-create.md
index ef6446a81..da2d5661e 100644
--- a/docs/adr/0017-authorized-analysis-run-create.md
+++ b/docs/adr/0017-authorized-analysis-run-create.md
@@ -39,7 +39,7 @@ The home panel's **Request a lineage reconstruction** button records a
Pending row the operator can open immediately. `POST
/api/analysis-runs/{id}/start` then reconstructs that frozen bag
(ADR 0021). TEPP start now goes through `tepp_client` (ADR 0022). The
-outbox worker remains a later slice.
+outbox worker is ADR 0023.
Do not stamp Succeeded or invent a theta from this write.
## References — APA 7th
diff --git a/docs/adr/0021-authorized-analysis-run-start.md b/docs/adr/0021-authorized-analysis-run-start.md
index 572bd884a..5e0921a0b 100644
--- a/docs/adr/0021-authorized-analysis-run-start.md
+++ b/docs/adr/0021-authorized-analysis-run-start.md
@@ -5,7 +5,7 @@
**Depends on:** ADR 0013 registry; ADR 0014 authorized read; ADR 0016 cutoff
posts; ADR 0017 authorized create
**Refs:** Issue #79 (Milestone 2 parent); ADR 0013 follow-up 3 (in-process
-start; durable outbox remains later)
+start; durable outbox is ADR 0023)
## Context
@@ -15,10 +15,8 @@ Seed still owned the only Succeeded Demo Corp tree. A buyer cannot
treat a request they cannot start as a product.
ADR 0013 follow-up 3 asked for a PostgreSQL outbox and Valkey worker.
-That durable delivery path is still later. This slice starts
-reconstruction in the authorized request so the operator can see the
-cutoff tree immediately. A crash after Running and before Succeeded
-rolls the transaction back to Pending.
+That durable delivery path is ADR 0023. This slice starts
+reconstruction so the operator can see the cutoff tree immediately.
Landed #145 occupies ADR 0020 / package 0.87.0 for granted retention
purge. ADR 0019 binds R&R catalog identity. This decision is the next
@@ -89,10 +87,12 @@ period-report rows do not show the button.
## Consequences
Demo Analyst can request a run, start it, and confirm the designed A-100
-fork (revised quote and delivery question under the pricing follow-up)
-without a seed-only Succeeded row. The durable outbox / Valkey worker
-and live TEPP transport remain later slices. Do not stamp Succeeded
-from a missing reconstruct library, and do not invent a theta.
+fork (revised quote and delivery question under the pricing follow-up).
+`make seed` also persists that fork on the Demo Corp Succeeded row so
+open-after-seed is not empty. The durable outbox / Valkey worker is
+ADR 0023. Live TEPP transport remains a later persistable-result
+slice. Do not stamp Succeeded from a missing reconstruct library, and
+do not invent a theta.
## References — APA 7th
diff --git a/docs/adr/0022-authorized-tepp-start.md b/docs/adr/0022-authorized-tepp-start.md
index 6949209fa..bf6e54d5e 100644
--- a/docs/adr/0022-authorized-tepp-start.md
+++ b/docs/adr/0022-authorized-tepp-start.md
@@ -70,9 +70,9 @@ sequenceDiagram
end
```
-A durable outbox / Valkey worker remains a later slice. Start still
-holds the request through the TEPP call so a crash rolls back to
-Pending.
+A durable outbox / Valkey worker is ADR 0023. Start commits Running
+plus the outbox row before `tepp_client` so a crash leaves the work
+item instead of rolling back to Pending.
## Consequences
diff --git a/docs/adr/0023-analysis-run-outbox.md b/docs/adr/0023-analysis-run-outbox.md
new file mode 100644
index 000000000..460b83a0a
--- /dev/null
+++ b/docs/adr/0023-analysis-run-outbox.md
@@ -0,0 +1,90 @@
+# ADR 0023 — Durable start outbox and Valkey wake-up
+
+**Decision status:** Accepted on this active PR; not protected-main truth until merge
+**Date:** 2026-08-17
+**Depends on:** ADR 0013 registry; ADR 0017 authorized create; ADR 0021
+authorized lineage start; ADR 0022 authorized TEPP start
+**Refs:** Issue #79 (Milestone 2 parent); ADR 0013 follow-up 3
+
+## Context
+
+ADR 0021 and ADR 0022 start a Pending lineage or TEPP run in the same
+request transaction as reconstruct / `tepp_client`. That is honest: a
+crash rolls back to Pending. It is not durable. A buyer who clicked
+Start and then lost the process cannot tell whether work began. TEPP's
+HTTP call also sits inside the registry write.
+
+ADR 0013 follow-up 3 asked for a normalized PostgreSQL outbox and a
+Valkey delivery worker. The activity stream already uses Valkey as an
+event queue, not a second database. The missing slice is one immutable
+start-work row committed with Running, then a worker that claims that
+row and finishes ThreadWeave or `tepp_client`.
+
+## Decision
+
+`POST /api/analysis-runs/{id}/start` splits into two authorized
+transactions:
+
+1. lock the visible Pending run, append Running, insert one
+ `analysis_run_outbox` row whose digest hashes run id, kind, snapshot
+ digest, and cutoff — never a post body or a theta — then commit;
+2. `XADD` the wake-up onto the Valkey stream `analysis-run-outbox`
+ (a missing Valkey does not roll back the outbox);
+3. lock that outbox row, append `analysis_outbox_claimed`, run the same
+ reconstruct or `tepp_client` path as ADR 0021 / ADR 0022, append the
+ terminal status, then append `analysis_outbox_delivered`.
+
+A Succeeded retry still replays the stored digest. A Running restart
+with an undelivered outbox is delivery, not a second start. A Running
+row without pending work stays 409. Period-report stays 422. Failed
+TEPP remains `tepp_not_available` / `tepp_result_not_persisted`. The
+HTTP response still waits for delivery so the operator sees the A-100
+fork or the Failed TEPP row without polling.
+
+```mermaid
+sequenceDiagram
+ participant Operator
+ participant API
+ participant Registry
+ participant Valkey
+ participant Worker
+ Operator->>API: POST /api/analysis-runs/{id}/start
+ API->>Registry: Running + outbox
+ Registry-->>API: committed work item
+ API->>Valkey: XADD analysis-run-outbox
+ API->>Worker: claim outbox
+ alt lineage
+ Worker->>Registry: reconstruction + Succeeded
+ else TEPP
+ Worker->>Registry: Failed tepp_not_available or tepp_result_not_persisted
+ end
+ API-->>Operator: 200 stored result
+```
+
+`make seed` writes a delivered outbox row on the Demo Corp lineage and
+TEPP runs so open-after-seed matches the start path. Retention purge
+deletes delivery, outbox, reconstruction, and snapshot membership
+before the registry rows.
+
+## Consequences
+
+Start survives a crash after Running. Refreshing a queued run finishes
+the same work item. Valkey is a wake-up, not a source of truth. Do not
+invent a theta, and do not stamp Succeeded from a missing reconstruct
+library.
+
+## References — APA 7th
+
+Hohpe, G., & Woolf, B. (2003). *Enterprise integration patterns:
+Designing, building, and deploying messaging solutions*.
+Addison-Wesley.
+
+International Organization for Standardization. (2019). *ISO 8601-1:2019:
+Date and time—Representations for information interchange—Part 1: Basic
+rules* (confirmed 2024; Amendment 1:2022).
+
+Moreau, L., & Missier, P. (Eds.). (2013). *PROV-DM: The PROV data model*.
+World Wide Web Consortium. https://www.w3.org/TR/prov-dm/
+
+World Wide Web Consortium. (2013). *PROV-O: The PROV ontology* (W3C
+Recommendation). https://www.w3.org/TR/prov-o/
diff --git a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md
index 951f72a17..aab8bb551 100644
--- a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md
+++ b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md
@@ -1,8 +1,8 @@
# Analysis-run registry standards and research traceability
**Status:** Active PR evidence; not protected-main truth until merge.
-**Scope:** Migrations 0018–0022, ADR 0013 / 0017 / 0020 / 0021, rollback, and
-real-PostgreSQL contract tests.
+**Scope:** Migrations 0018–0023, ADR 0013 / 0017 / 0020 / 0021 / 0022 /
+0023, rollback, and real-PostgreSQL contract tests.
## Standards mapped to implementation
@@ -16,7 +16,7 @@ real-PostgreSQL contract tests.
| NIST SP 800-92 | Treat audit records as bounded, protected operational evidence rather than unstructured application logging. | Append-only status events, machine failure codes, actor identity, occurrence/record clocks, fail-closed rollback, `invoking_session_role` on each retention event, and exclusion of raw source/provider payloads. |
| NIST SP 800-53 Rev. 5 AC-3 | Enforce least privilege on privileged procedures; a well-known procedure name is not an authorization secret. | `REVOKE ALL` on `purge_analysis_run_registry` from `PUBLIC`; `GRANT EXECUTE` only to `analysis_run_retention_admin`; unrevoked `analysis_run_retention_grant` required (ADR 0020). |
| OpenAPI 3.2.0 | Define explicit versioned API schemas rather than exposing database rows or implementation-specific payloads. | `GET` / `POST /api/analysis-runs` and `POST /api/analysis-runs/{id}/start` return labels, clocks, aggregates, and titled reconstruction edges — never source SQL or a provider body. |
-| ThreadWeave tree assembly | Persist the same parent choices the library reconstructs on the cutoff bag. | `start_pending_analysis_run` calls `lineage_edge_specs` on frozen `analysis_source_snapshot_member` rows (or the live cutoff query when membership is absent); tests require the designed A-100 fork through `records_from_source_posts` (revised quote + delivery question under the pricing follow-up). TEPP start uses `tepp_client` only. |
+| ThreadWeave tree assembly | Persist the same parent choices the library reconstructs on the cutoff bag. | Start enqueues `analysis_run_outbox` then `deliver_queued_analysis_run` calls `lineage_edge_specs` on frozen `analysis_source_snapshot_member` rows (or the live cutoff query when membership is absent); tests require the designed A-100 fork through `records_from_source_posts` (revised quote + delivery question under the pricing follow-up). TEPP start uses `tepp_client` only. Valkey `analysis-run-outbox` is the wake-up (ADR 0023). |
## Temporal reasoning
diff --git a/frontend/package.json b/frontend/package.json
index dc98dbd98..3b2e21acb 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -1,7 +1,7 @@
{
"name": "frontend",
"private": true,
- "version": "0.92.0",
+ "version": "0.94.0",
"type": "module",
"scripts": {
"dev": "vite",
diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx
index 0b8b5b824..de5a6367b 100644
--- a/frontend/src/App.test.tsx
+++ b/frontend/src/App.test.tsx
@@ -60,6 +60,7 @@ describe("App, authenticated", () => {
searchUnavailable?: boolean;
verificationEvidenceUrl?: string | null;
failedLineageRun?: boolean;
+ runningLineageRun?: boolean;
failedReportRun?: boolean;
succeededTeppRun?: boolean;
pendingTeppRun?: boolean;
@@ -320,8 +321,16 @@ describe("App, authenticated", () => {
scope_kind_code: "analysis_scope_corporate_entity",
scope_kind_label: "Corporate entity",
scope_entity_name: "Demo Corp",
- status_code: "analysis_status_succeeded",
- status_label: "Succeeded",
+ status_code: options?.runningLineageRun
+ ? "analysis_status_running"
+ : options?.failedLineageRun
+ ? "analysis_status_failed"
+ : "analysis_status_succeeded",
+ status_label: options?.runningLineageRun
+ ? "Running"
+ : options?.failedLineageRun
+ ? "Failed"
+ : "Succeeded",
knowledge_cutoff: "2026-01-12T12:00:00Z",
requested_at: "2026-01-12T12:30:00Z",
source_counts: [
@@ -345,6 +354,23 @@ describe("App, authenticated", () => {
live_after_cutoff: false,
},
],
+ reconstructed_edges: [
+ {
+ parent_post_id: "post-2",
+ parent_post_title: "Pricing renegotiation follow-up",
+ child_post_id: "post-1",
+ child_post_title: "Pricing renegotiation: revised quote sent",
+ fused_score: 0.72,
+ },
+ {
+ parent_post_id: "post-2",
+ parent_post_title: "Pricing renegotiation follow-up",
+ child_post_id: "post-delivery",
+ child_post_title: "Delivery schedule question raised",
+ fused_score: 0.68,
+ },
+ ],
+ reconstruction_result_sha256: "aa".repeat(32),
code_revision_sha: "abcdef0123456789deadbeefcafebabe",
configuration_sha256:
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
@@ -552,8 +578,14 @@ describe("App, authenticated", () => {
scope_entity_name: "Demo Corp",
status_code: options?.failedLineageRun
? "analysis_status_failed"
- : "analysis_status_succeeded",
- status_label: options?.failedLineageRun ? "Failed" : "Succeeded",
+ : options?.runningLineageRun
+ ? "analysis_status_running"
+ : "analysis_status_succeeded",
+ status_label: options?.failedLineageRun
+ ? "Failed"
+ : options?.runningLineageRun
+ ? "Running"
+ : "Succeeded",
knowledge_cutoff: "2026-01-12T12:00:00Z",
requested_at: "2026-01-12T12:30:00Z",
source_counts: [
@@ -1869,6 +1901,21 @@ describe("App, authenticated", () => {
expect(history).toHaveTextContent("Running 2026-01-12 12:32");
expect(history).toHaveTextContent("Succeeded 2026-01-12 12:33");
expect(screen.getByRole("list", { name: "Posts known at this run cutoff" })).toBeInTheDocument();
+ const seededFork = screen.getByRole("list", { name: "Reconstructed lineage edges" });
+ expect(seededFork).toHaveTextContent(
+ "Pricing renegotiation: revised quote sent follows Pricing renegotiation follow-up",
+ );
+ expect(seededFork).toHaveTextContent(
+ "Delivery schedule question raised follows Pricing renegotiation follow-up",
+ );
+ await userEvent.click(
+ screen.getByRole("button", {
+ name: "Open reconstructed child: Pricing renegotiation: revised quote sent",
+ }),
+ );
+ await waitFor(() => expect(screen.getByText("The full body text.")).toBeInTheDocument());
+ await userEvent.click(screen.getByRole("button", { name: "Close" }));
+ expect(screen.getByRole("list", { name: "Posts known at this run cutoff" })).toBeInTheDocument();
expect(
screen.getByText(
"Opening a title shows the live post. Titles marked updated after cutoff were rewritten after 2026-01-12. Compare those bodies with this run before you treat them as reconstructed evidence.",
@@ -1955,6 +2002,23 @@ describe("App, authenticated", () => {
expect(screen.queryByRole("status", { name: "Live body warning" })).not.toBeInTheDocument();
});
+ it("tells a running lineage run to refresh the durable outbox", async () => {
+ stubBackend({ runningLineageRun: true });
+ render();
+
+ const lineageButton = await screen.findByRole("button", {
+ name: "Open analysis run: Lineage reconstruction · Running · Demo Corp",
+ });
+ expect(lineageButton).toHaveTextContent(
+ "Refresh this run. Start already queued the work on the durable outbox.",
+ );
+ await userEvent.click(lineageButton);
+ expect(screen.getByRole("button", { name: "Start reconstruction" })).toBeInTheDocument();
+ expect(
+ screen.getAllByText("Refresh this run. Start already queued the work on the durable outbox."),
+ ).not.toHaveLength(0);
+ });
+
it("does not tell a failed lineage run to connect the measurement service", async () => {
stubBackend({ failedLineageRun: true });
render();
diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx
index 5b1a1eea5..67767b419 100644
--- a/frontend/src/App.tsx
+++ b/frontend/src/App.tsx
@@ -1491,6 +1491,7 @@ function analysisRunNextAction(run: AnalysisRun): string | null {
}
}
case "analysis_status_running":
+ return "Refresh this run. Start already queued the work on the durable outbox.";
case "analysis_status_succeeded":
case "analysis_status_cancelled":
case null:
@@ -1664,7 +1665,8 @@ function AnalysisRunReproducibilityDigests({
function analysisRunCanStart(run: AnalysisRun): boolean {
return (
(run.run_kind_code === "analysis_run_lineage" || run.run_kind_code === "analysis_run_tepp") &&
- run.status_code === "analysis_status_pending"
+ (run.status_code === "analysis_status_pending" ||
+ run.status_code === "analysis_status_running")
);
}
diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py
index cc28322f0..eebd45bcb 100644
--- a/lineageweave/__init__.py
+++ b/lineageweave/__init__.py
@@ -55,4 +55,4 @@
"sentence_excerpts",
]
-__version__ = "0.92.0"
+__version__ = "0.94.0"
diff --git a/migrations/0023_analysis_run_outbox.sql b/migrations/0023_analysis_run_outbox.sql
new file mode 100644
index 000000000..2aaef8c2c
--- /dev/null
+++ b/migrations/0023_analysis_run_outbox.sql
@@ -0,0 +1,241 @@
+-- Durable analysis-run start outbox (ADR 0023).
+--
+-- Start appends Running and one immutable outbox row in the same
+-- transaction. Reconstruct and TEPP then run from that row so a crash
+-- no longer rolls the work item back to Pending. Valkey carries the
+-- wake-up; PostgreSQL stays the source of truth. No post body or
+-- fabricated psychometric score is stored.
+
+insert into common_lookup_value (
+ lookup_category, lookup_code, lookup_label, display_order
+) values
+ ('analysis_outbox_delivery', 'analysis_outbox_claimed', 'Claimed', 0),
+ ('analysis_outbox_delivery', 'analysis_outbox_delivered', 'Delivered', 1)
+on conflict (lookup_code) do nothing;
+
+create table if not exists analysis_run_outbox (
+ analysis_run_id uuid primary key
+ references analysis_run (analysis_run_id),
+ work_kind_code text not null,
+ request_sha256 text not null,
+ enqueued_at timestamptz not null,
+ recorded_at timestamptz not null default clock_timestamp(),
+ constraint analysis_run_outbox_kind_check
+ check (work_kind_code in ('analysis_run_lineage', 'analysis_run_tepp')),
+ constraint analysis_run_outbox_digest_check
+ check (request_sha256 ~ '^[0-9a-f]{64}$'),
+ constraint analysis_run_outbox_time_check
+ check (enqueued_at <= recorded_at)
+);
+
+comment on table analysis_run_outbox is
+ 'One immutable start-work item per analysis run; never a post body '
+ 'or a fabricated psychometric score.';
+
+create table if not exists analysis_run_outbox_delivery (
+ analysis_run_id uuid not null
+ references analysis_run_outbox (analysis_run_id),
+ delivery_ordinal integer not null,
+ delivery_status_code text not null,
+ occurred_at timestamptz not null,
+ valkey_stream_entry_id text,
+ primary key (analysis_run_id, delivery_ordinal),
+ constraint analysis_run_outbox_delivery_ordinal_check
+ check (delivery_ordinal >= 1),
+ constraint analysis_run_outbox_delivery_status_check
+ check (
+ delivery_status_code in (
+ 'analysis_outbox_claimed',
+ 'analysis_outbox_delivered'
+ )
+ ),
+ constraint analysis_run_outbox_delivery_stream_check
+ check (
+ valkey_stream_entry_id is null
+ or char_length(valkey_stream_entry_id) between 1 and 64
+ )
+);
+
+comment on table analysis_run_outbox_delivery is
+ 'Append-only claim and delivery events for one start-work item.';
+
+create or replace function reject_analysis_run_outbox_mutation()
+returns trigger
+language plpgsql
+as $$
+begin
+ raise exception 'analysis_run_outbox_is_immutable';
+end
+$$;
+
+comment on function reject_analysis_run_outbox_mutation() is
+ 'Rejects mutation of an enqueued start-work item.';
+
+drop trigger if exists analysis_run_outbox_mutation_reject
+ on analysis_run_outbox;
+create trigger analysis_run_outbox_mutation_reject
+before update or delete on analysis_run_outbox
+for each row execute function reject_analysis_run_outbox_mutation();
+
+create or replace function reject_analysis_run_outbox_delivery_mutation()
+returns trigger
+language plpgsql
+as $$
+begin
+ raise exception 'analysis_run_outbox_delivery_is_append_only';
+end
+$$;
+
+comment on function reject_analysis_run_outbox_delivery_mutation() is
+ 'Rejects mutation of a start-work delivery event.';
+
+drop trigger if exists analysis_run_outbox_delivery_mutation_reject
+ on analysis_run_outbox_delivery;
+create trigger analysis_run_outbox_delivery_mutation_reject
+before update or delete on analysis_run_outbox_delivery
+for each row execute function reject_analysis_run_outbox_delivery_mutation();
+
+create or replace function purge_analysis_run_registry(approval_token text)
+returns void
+language plpgsql
+security definer
+set search_path = public
+as $$
+declare
+ run_count bigint;
+ snapshot_count bigint;
+begin
+ if not exists (
+ select 1
+ from analysis_run_retention_grant
+ where database_role_name = session_user
+ and revoked_at is null
+ ) then
+ raise exception 'analysis_run_retention_not_granted';
+ end if;
+
+ if not pg_has_role(session_user, 'analysis_run_retention_admin', 'member') then
+ raise exception 'analysis_run_retention_not_admin';
+ end if;
+
+ if approval_token is distinct from 'approved-retention-purge' then
+ raise exception 'analysis_run_retention_not_approved';
+ end if;
+
+ select count(*) into run_count from analysis_run;
+ select count(*) into snapshot_count from analysis_source_snapshot;
+
+ alter table analysis_run_status_event
+ disable trigger analysis_run_status_event_delete_reject;
+ alter table analysis_run_scope
+ disable trigger analysis_run_scope_mutation_reject;
+ alter table analysis_run
+ disable trigger analysis_run_mutation_reject;
+ if to_regclass('public.analysis_run_outbox') is not null then
+ alter table analysis_run_outbox
+ disable trigger analysis_run_outbox_mutation_reject;
+ alter table analysis_run_outbox_delivery
+ disable trigger analysis_run_outbox_delivery_mutation_reject;
+ end if;
+ if to_regclass('public.analysis_run_reconstruction') is not null then
+ alter table analysis_run_reconstruction
+ disable trigger analysis_run_reconstruction_update_reject;
+ alter table analysis_run_lineage_edge
+ disable trigger analysis_run_lineage_edge_update_reject;
+ end if;
+ if to_regclass('public.analysis_source_snapshot_member') is not null then
+ alter table analysis_source_snapshot_member
+ disable trigger analysis_source_snapshot_member_update_reject;
+ end if;
+
+ begin
+ if to_regclass('public.analysis_run_outbox_delivery') is not null then
+ delete from analysis_run_outbox_delivery;
+ delete from analysis_run_outbox;
+ end if;
+ if to_regclass('public.analysis_run_lineage_edge') is not null then
+ delete from analysis_run_lineage_edge;
+ delete from analysis_run_reconstruction;
+ end if;
+ if to_regclass('public.analysis_source_snapshot_member') is not null then
+ delete from analysis_source_snapshot_member;
+ end if;
+ delete from analysis_run_status_event;
+ delete from analysis_run_scope;
+ delete from analysis_run;
+ delete from analysis_source_count;
+ delete from analysis_source_snapshot;
+ exception
+ when others then
+ alter table analysis_run
+ enable trigger analysis_run_mutation_reject;
+ alter table analysis_run_scope
+ enable trigger analysis_run_scope_mutation_reject;
+ alter table analysis_run_status_event
+ enable trigger analysis_run_status_event_delete_reject;
+ if to_regclass('public.analysis_run_outbox') is not null then
+ alter table analysis_run_outbox
+ enable trigger analysis_run_outbox_mutation_reject;
+ alter table analysis_run_outbox_delivery
+ enable trigger analysis_run_outbox_delivery_mutation_reject;
+ end if;
+ if to_regclass('public.analysis_run_reconstruction') is not null then
+ alter table analysis_run_reconstruction
+ enable trigger analysis_run_reconstruction_update_reject;
+ alter table analysis_run_lineage_edge
+ enable trigger analysis_run_lineage_edge_update_reject;
+ end if;
+ if to_regclass('public.analysis_source_snapshot_member') is not null then
+ alter table analysis_source_snapshot_member
+ enable trigger analysis_source_snapshot_member_update_reject;
+ end if;
+ raise;
+ end;
+
+ alter table analysis_run
+ enable trigger analysis_run_mutation_reject;
+ alter table analysis_run_scope
+ enable trigger analysis_run_scope_mutation_reject;
+ alter table analysis_run_status_event
+ enable trigger analysis_run_status_event_delete_reject;
+ if to_regclass('public.analysis_run_outbox') is not null then
+ alter table analysis_run_outbox
+ enable trigger analysis_run_outbox_mutation_reject;
+ alter table analysis_run_outbox_delivery
+ enable trigger analysis_run_outbox_delivery_mutation_reject;
+ end if;
+ if to_regclass('public.analysis_run_reconstruction') is not null then
+ alter table analysis_run_reconstruction
+ enable trigger analysis_run_reconstruction_update_reject;
+ alter table analysis_run_lineage_edge
+ enable trigger analysis_run_lineage_edge_update_reject;
+ end if;
+ if to_regclass('public.analysis_source_snapshot_member') is not null then
+ alter table analysis_source_snapshot_member
+ enable trigger analysis_source_snapshot_member_update_reject;
+ end if;
+
+ insert into analysis_run_retention_event (
+ purged_run_count,
+ purged_snapshot_count,
+ approval_token_digest,
+ invoking_session_role,
+ invoking_current_role,
+ client_network_address
+ ) values (
+ run_count,
+ snapshot_count,
+ encode(sha256(convert_to(approval_token, 'UTF8')), 'hex'),
+ session_user,
+ current_user,
+ inet_client_addr()
+ );
+end
+$$;
+
+comment on function purge_analysis_run_registry(text) is
+ 'Empties immutable registry, reconstruction, membership, and outbox '
+ 'relations after an unrevoked role grant, analysis_run_retention_admin '
+ 'membership, and the documented approval token; records one '
+ 'analysis_run_retention_event. Next action: export that event, delete '
+ 'it, then roll back 0023, 0022, 0021, 0020, and 0018.';
diff --git a/migrations/rollback/0023_analysis_run_outbox.sql b/migrations/rollback/0023_analysis_run_outbox.sql
new file mode 100644
index 000000000..c8aa18248
--- /dev/null
+++ b/migrations/rollback/0023_analysis_run_outbox.sql
@@ -0,0 +1,38 @@
+-- Fail-closed rollback for migration 0023.
+--
+-- Outbox evidence must be exported or explicitly deleted under an
+-- approved retention procedure before these objects can be removed.
+-- The extended purge function stays; it already guards missing tables.
+
+begin;
+
+do $$
+declare
+ relation_name text;
+ relation_has_rows boolean;
+begin
+ foreach relation_name in array array[
+ 'analysis_run_outbox_delivery',
+ 'analysis_run_outbox'
+ ] loop
+ if to_regclass('public.' || relation_name) is not null then
+ execute format('select exists (select 1 from %I)', relation_name)
+ into relation_has_rows;
+ if relation_has_rows then
+ raise exception 'analysis_run_outbox_not_empty';
+ end if;
+ end if;
+ end loop;
+end
+$$;
+
+drop trigger if exists analysis_run_outbox_delivery_mutation_reject
+ on analysis_run_outbox_delivery;
+drop trigger if exists analysis_run_outbox_mutation_reject
+ on analysis_run_outbox;
+drop function if exists reject_analysis_run_outbox_delivery_mutation();
+drop function if exists reject_analysis_run_outbox_mutation();
+drop table if exists analysis_run_outbox_delivery;
+drop table if exists analysis_run_outbox;
+
+commit;
diff --git a/pyproject.toml b/pyproject.toml
index 2a84bfa96..d3682e502 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "lineageweave"
-version = "0.92.0"
+version = "0.94.0"
description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication."
readme = "README.md"
license = { text = "MIT" }
diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py
index 7ce3c31d6..e215e3e41 100644
--- a/scripts/seed_demo_data.py
+++ b/scripts/seed_demo_data.py
@@ -124,6 +124,7 @@ def seed(
cur.execute((migrations / "0020_analysis_run_retention_purge.sql").read_text())
cur.execute((migrations / "0021_analysis_run_reconstruction.sql").read_text())
cur.execute((migrations / "0022_analysis_source_snapshot_member.sql").read_text())
+ cur.execute((migrations / "0023_analysis_run_outbox.sql").read_text())
cur.execute(
"""
insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values
@@ -1386,6 +1387,73 @@ def _seed_demo_analysis_run(cur, requested_by_account_id, corporate_entity_id) -
""",
(run_id, ordinal, status, occurred),
)
+ _seed_demo_run_reconstruction(cur, run_id, corporate_entity_id)
+ _seed_demo_run_outbox(cur, run_id)
+
+
+def seed_reconstruction_edges(rows: list[dict]) -> tuple:
+ """ThreadWeave parent choices and digest for seed and start. Never a theta."""
+ from backend.app.analysis_run_start import reconstruction_result_digest
+ from backend.app.lineage_ingestion import records_from_source_posts
+ from lineageweave.lineage_persistence import lineage_edge_specs
+
+ edges = lineage_edge_specs(records_from_source_posts(rows))
+ return edges, reconstruction_result_digest(edges)
+
+
+def _seed_demo_run_reconstruction(cur, analysis_run_id, corporate_entity_id) -> None:
+ """Persist the designed A-100 fork on the seeded Succeeded lineage run.
+
+ Seed already stamps Succeeded. Without run-scoped edges the home
+ detail has cutoff titles and no fork. Reuses the same ThreadWeave
+ path start uses. Does not invent a TEPP score.
+ """
+ from datetime import datetime, timezone
+
+ cur.execute(
+ "select 1 from analysis_run_reconstruction where analysis_run_id = %s",
+ (analysis_run_id,),
+ )
+ if cur.fetchone() is not None:
+ return
+ cur.execute(
+ """
+ select post_id, post_title, created_at, visibility_code,
+ corporate_entity_id, process_unit_id,
+ thread_group_key, secondary_grouping_key
+ from source_post
+ where corporate_entity_id = %s
+ and created_at <= %s
+ order by created_at, post_title
+ """,
+ (corporate_entity_id, datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc)),
+ )
+ columns = [desc[0] for desc in cur.description]
+ rows = [dict(zip(columns, row)) for row in cur.fetchall()]
+ if not rows:
+ return
+ edges, digest = seed_reconstruction_edges(rows)
+ finished = datetime(2026, 1, 12, 12, 33, tzinfo=timezone.utc)
+ cur.execute(
+ """
+ insert into analysis_run_reconstruction
+ (analysis_run_id, result_sha256, edge_count, reconstructed_at)
+ values (%s, %s, %s, %s)
+ on conflict do nothing
+ """,
+ (analysis_run_id, digest, len(edges), finished),
+ )
+ for edge in edges:
+ cur.execute(
+ """
+ insert into analysis_run_lineage_edge
+ (analysis_run_id, child_post_id, parent_post_id,
+ fused_score, reconstructed_at)
+ values (%s, %s, %s, %s, %s)
+ on conflict do nothing
+ """,
+ (analysis_run_id, edge.child_id, edge.parent_id, edge.fused_score, finished),
+ )
def tepp_seed_request() -> AnalysisRunRequest:
@@ -1484,6 +1552,73 @@ def _seed_demo_tepp_run(cur, requested_by_account_id, corporate_entity_id) -> No
""",
(run_id, ordinal, status, occurred, fail),
)
+ _seed_demo_run_outbox(cur, run_id)
+
+
+def _seed_demo_run_outbox(cur, analysis_run_id) -> None:
+ """Record a delivered start-work item for the seeded run.
+
+ Seed already stamped the terminal status. The outbox row proves the
+ same durable path start uses. No theta is stored.
+ """
+ from datetime import datetime, timezone
+
+ from backend.app.analysis_run_outbox import outbox_request_digest
+
+ cur.execute(
+ "select 1 from analysis_run_outbox where analysis_run_id = %s",
+ (analysis_run_id,),
+ )
+ if cur.fetchone() is not None:
+ return
+ cur.execute(
+ """
+ select run.run_kind_code, run.knowledge_cutoff, snapshot.snapshot_sha256
+ from analysis_run run
+ join analysis_source_snapshot snapshot
+ on snapshot.analysis_source_snapshot_id = run.analysis_source_snapshot_id
+ where run.analysis_run_id = %s
+ """,
+ (analysis_run_id,),
+ )
+ row = cur.fetchone()
+ if row is None:
+ return
+ work_kind_code, knowledge_cutoff, snapshot_sha256 = row
+ digest = outbox_request_digest(
+ analysis_run_id=str(analysis_run_id),
+ work_kind_code=work_kind_code,
+ snapshot_sha256=snapshot_sha256,
+ knowledge_cutoff=knowledge_cutoff,
+ )
+ if work_kind_code == "analysis_run_tepp":
+ claimed = datetime(2026, 1, 12, 12, 36, tzinfo=timezone.utc)
+ delivered = datetime(2026, 1, 12, 12, 37, tzinfo=timezone.utc)
+ else:
+ claimed = datetime(2026, 1, 12, 12, 32, tzinfo=timezone.utc)
+ delivered = datetime(2026, 1, 12, 12, 33, tzinfo=timezone.utc)
+ cur.execute(
+ """
+ insert into analysis_run_outbox
+ (analysis_run_id, work_kind_code, request_sha256, enqueued_at)
+ values (%s, %s, %s, %s)
+ on conflict do nothing
+ """,
+ (analysis_run_id, work_kind_code, digest, claimed),
+ )
+ for ordinal, status, occurred in (
+ (1, "analysis_outbox_claimed", claimed),
+ (2, "analysis_outbox_delivered", delivered),
+ ):
+ cur.execute(
+ """
+ insert into analysis_run_outbox_delivery
+ (analysis_run_id, delivery_ordinal, delivery_status_code, occurred_at)
+ values (%s, %s, %s, %s)
+ on conflict do nothing
+ """,
+ (analysis_run_id, ordinal, status, occurred),
+ )
def main() -> None:
diff --git a/tests/test_analysis_run_outbox.py b/tests/test_analysis_run_outbox.py
new file mode 100644
index 000000000..240006ba1
--- /dev/null
+++ b/tests/test_analysis_run_outbox.py
@@ -0,0 +1,105 @@
+"""Static and unit contracts for the durable analysis-run start outbox."""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+
+from backend.app.analysis_run_outbox import (
+ OUTBOX_STREAM_KEY,
+ latest_outbox_delivery_is_claimed,
+ latest_outbox_delivery_is_delivered,
+ outbox_request_digest,
+ outbox_stream_fields,
+)
+from backend.app.analysis_run_start import start_kind_rejection
+
+_ROOT = Path(__file__).resolve().parents[1]
+_OUTBOX_MIGRATION = _ROOT / "migrations" / "0023_analysis_run_outbox.sql"
+_OUTBOX_ROLLBACK = _ROOT / "migrations" / "rollback" / "0023_analysis_run_outbox.sql"
+_POSTGRES_IMAGE = _ROOT / "docker" / "postgres-init" / "Dockerfile"
+
+
+def test_outbox_migration_is_normalized_and_wired() -> None:
+ """Static contract: 3NF names, no payload JSON, Dockerfile copy, rollback."""
+ migration = _OUTBOX_MIGRATION.read_text(encoding="utf-8")
+ rollback = _OUTBOX_ROLLBACK.read_text(encoding="utf-8")
+ dockerfile = _POSTGRES_IMAGE.read_text(encoding="utf-8")
+ created_tables = set(
+ re.findall(r"create table if not exists\s+([a-z0-9_]+)", migration, re.I)
+ )
+ assert {"analysis_run_outbox", "analysis_run_outbox_delivery"} <= created_tables
+ assert "jsonb" not in migration.casefold()
+ assert "metadata_payload" not in migration
+ assert "theta" not in migration.casefold()
+ assert "0023_analysis_run_outbox.sql" in dockerfile
+ assert "analysis_run_outbox_not_empty" in rollback
+ assert "reject_analysis_run_outbox_mutation" in migration
+ assert "reject_analysis_run_outbox_delivery_mutation" in migration
+ assert "analysis_run_lineage_edge" in migration
+ assert "analysis_source_snapshot_member" in migration
+ object_patterns = (
+ r"create table if not exists\s+([a-z0-9_]+)",
+ r"create or replace function\s+([a-z0-9_]+)",
+ r"create trigger\s+([a-z0-9_]+)",
+ )
+ for pattern in object_patterns:
+ for object_name in re.findall(pattern, migration, re.I):
+ assert len(object_name.split("_")) >= 2, object_name
+
+
+def test_outbox_request_digest_is_stable_and_ignores_bodies() -> None:
+ """The same frozen start hashes the same way and never includes a body."""
+ cutoff = datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc)
+ digest = outbox_request_digest(
+ analysis_run_id="11111111-1111-1111-1111-111111111111",
+ work_kind_code="analysis_run_lineage",
+ snapshot_sha256="ab" * 32,
+ knowledge_cutoff=cutoff,
+ )
+ again = outbox_request_digest(
+ analysis_run_id="11111111-1111-1111-1111-111111111111",
+ work_kind_code="analysis_run_lineage",
+ snapshot_sha256="ab" * 32,
+ knowledge_cutoff=datetime(2026, 1, 12, 12, 0),
+ )
+ other = outbox_request_digest(
+ analysis_run_id="11111111-1111-1111-1111-111111111111",
+ work_kind_code="analysis_run_tepp",
+ snapshot_sha256="ab" * 32,
+ knowledge_cutoff=cutoff,
+ )
+ assert digest == again
+ assert digest != other
+ assert "theta" not in digest
+ assert "Pricing renegotiation" not in digest
+
+
+def test_outbox_stream_fields_are_the_wake_up_only() -> None:
+ """Valkey carries the run id and digest, never a measurement."""
+ fields = outbox_stream_fields(
+ analysis_run_id="11111111-1111-1111-1111-111111111111",
+ work_kind_code="analysis_run_tepp",
+ request_sha256="cd" * 32,
+ )
+ assert fields["work_kind_code"] == "analysis_run_tepp"
+ assert fields["request_sha256"] == "cd" * 32
+ assert "theta" not in str(fields).casefold()
+ assert OUTBOX_STREAM_KEY == "analysis-run-outbox"
+
+
+def test_outbox_delivery_helpers_distinguish_claimed_from_done() -> None:
+ """A claimed row is retryable. A delivered row is finished."""
+ assert latest_outbox_delivery_is_claimed("analysis_outbox_claimed")
+ assert not latest_outbox_delivery_is_claimed("analysis_outbox_delivered")
+ assert latest_outbox_delivery_is_delivered("analysis_outbox_delivered")
+ assert not latest_outbox_delivery_is_delivered(None)
+
+
+def test_period_report_never_enters_the_start_outbox() -> None:
+ """The outbox is for lineage and TEPP start, not a fabricated report."""
+ report = start_kind_rejection("analysis_run_report")
+ assert report is not None
+ assert report.status_code == 422
+ assert "invent a measurement" in report.detail
diff --git a/tests/test_analysis_run_reconstruction_schema.py b/tests/test_analysis_run_reconstruction_schema.py
index 8ca265f37..9362edc80 100644
--- a/tests/test_analysis_run_reconstruction_schema.py
+++ b/tests/test_analysis_run_reconstruction_schema.py
@@ -47,6 +47,7 @@ def test_reconstruction_migration_is_normalized_and_wired() -> None:
assert "theta" not in migration.casefold()
assert "0021_analysis_run_reconstruction.sql" in dockerfile
assert "0022_analysis_source_snapshot_member.sql" in dockerfile
+ assert "0023_analysis_run_outbox.sql" in dockerfile
assert "analysis_run_reconstruction_not_empty" in rollback
assert "reject_analysis_run_reconstruction_update" in migration
assert "reject_analysis_run_lineage_edge_update" in migration
diff --git a/tests/test_analysis_run_registry_schema.py b/tests/test_analysis_run_registry_schema.py
index 6041d3090..18a1a91c4 100644
--- a/tests/test_analysis_run_registry_schema.py
+++ b/tests/test_analysis_run_registry_schema.py
@@ -278,6 +278,7 @@ def test_registry_contract_is_normalized_and_has_one_temporal_authority() -> Non
assert "0020_analysis_run_retention_purge.sql" in dockerfile
assert "0021_analysis_run_reconstruction.sql" in dockerfile
assert "0022_analysis_source_snapshot_member.sql" in dockerfile
+ assert "0023_analysis_run_outbox.sql" in dockerfile
seed = (_ROOT / "scripts" / "seed_demo_data.py").read_text(encoding="utf-8")
assert seed.index("0019_role_catalog_identity.sql") < seed.index(
"0020_analysis_run_retention_purge.sql"
@@ -288,6 +289,9 @@ def test_registry_contract_is_normalized_and_has_one_temporal_authority() -> Non
assert seed.index("0021_analysis_run_reconstruction.sql") < seed.index(
"0022_analysis_source_snapshot_member.sql"
)
+ assert seed.index("0022_analysis_source_snapshot_member.sql") < seed.index(
+ "0023_analysis_run_outbox.sql"
+ )
assert "analysis_run_registry_not_empty" in rollback
retention = _RETENTION_MIGRATION.read_text(encoding="utf-8")
retention_rollback = _RETENTION_ROLLBACK.read_text(encoding="utf-8")
diff --git a/tests/test_analysis_run_start.py b/tests/test_analysis_run_start.py
index 795586a7e..e46aa4a0c 100644
--- a/tests/test_analysis_run_start.py
+++ b/tests/test_analysis_run_start.py
@@ -161,7 +161,7 @@ def test_hidden_run_start_is_not_found() -> None:
def test_running_restart_conflicts_and_succeeded_replay_is_documented() -> None:
- """Running is 409. Succeeded replay is a documented no-op (200 in the API)."""
+ """Running without pending outbox is 409. Succeeded replay is a no-op."""
conflict = start_write_conflict_error()
assert conflict.status_code == 409
assert "Refresh to see the stored tree" in conflict.detail
diff --git a/tests/test_seed_analysis_run_reconstruction.py b/tests/test_seed_analysis_run_reconstruction.py
new file mode 100644
index 000000000..7f73800e7
--- /dev/null
+++ b/tests/test_seed_analysis_run_reconstruction.py
@@ -0,0 +1,38 @@
+"""Seeded Succeeded lineage runs persist the designed A-100 fork."""
+
+from datetime import datetime, timezone
+
+from lineageweave.fixtures import sample_records
+from scripts.seed_demo_data import seed_reconstruction_edges
+
+
+def _rows_from_fixtures() -> list[dict]:
+ cutoff = datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc)
+ rows: list[dict] = []
+ for record in sample_records():
+ occurred = record.occurred_at.replace(tzinfo=timezone.utc)
+ if occurred > cutoff:
+ continue
+ rows.append(
+ {
+ "post_id": record.record_id,
+ "post_title": record.label,
+ "created_at": occurred,
+ "visibility_code": "public",
+ "corporate_entity_id": "demo-corp",
+ "process_unit_id": "demo-pu",
+ "thread_group_key": record.group_key,
+ "secondary_grouping_key": record.secondary_key,
+ }
+ )
+ return rows
+
+
+def test_seed_reconstruction_recovers_the_a100_fork() -> None:
+ """Seed must persist the same parent choices start uses."""
+ edges, digest = seed_reconstruction_edges(_rows_from_fixtures())
+ children = {edge.child_id for edge in edges if edge.parent_id == "rec-002"}
+ assert children >= {"rec-003", "rec-004"}
+ assert "rec-006" not in {edge.child_id for edge in edges}
+ assert digest
+ assert "theta" not in digest
diff --git a/uv.lock b/uv.lock
index 313978937..499690dbf 100644
--- a/uv.lock
+++ b/uv.lock
@@ -454,7 +454,7 @@ wheels = [
[[package]]
name = "lineageweave"
-version = "0.92.0"
+version = "0.94.0"
source = { virtual = "." }
dependencies = [
{ name = "certifi" },