Adam Moussa
0eae5dbfc3
Reworks the P1 sim so the four §7.1 exit criteria are demonstrated against the ACTUAL mechanic, not a model (resolves the verifier's "sim models the ledger, not the LangGraph integration" finding). - New tests/sim/test_p1_graph_integration.py drives the real agent_team.graph StateGraph (interrupt/Command(resume)) + the real langgraph SqliteSaver checkpointer + the committed pending_questions compare-and-set, proving: (a) suspend survives a simulated restart (drop saver/conn, rebuild over the same checkpoint DB) and resumes; (b) duplicate answer loses the CAS and the graph never double-advances; (c) a post-deadline answer loses to expire and the task is not resumed; (d) two concurrent tasks resume to the correct thread, with a turn-guarded no-double-apply check. - graph.py: derive a STABLE question_id from uuid5(thread_id, turn). The clarifier node replays on resume, so the prior fresh-uuid id changed between the delivered/ledgered question and the qa_history entry — breaking the §3.3.1 identity contract. Now the delivered id == ledger key == history entry (unit-tested in test_graph.py). - harness._connect() now uses the committed schema.connect() (WAL + busy_timeout) instead of a raw sqlite3.connect, so concurrent responders genuinely serialize; the criterion-(d) concurrency test no longer swallows OperationalError (it asserts zero errors + exactly one CAS winner). - requirements.txt: pin langgraph-checkpoint-sqlite==3.1.0 (design D9 durable checkpointer), now exercised by the integration test. Full suite: 564 passed; ruff + format clean.
601 lines
22 KiB
Python
601 lines
22 KiB
Python
"""P1 exit-criteria simulation harness (design §7.1 P1, demonstrating §3.3.1).
|
|
|
|
Phase P1 of the Plane-2 pipeline must *demonstrate* the durable
|
|
human-in-the-loop suspend/resume contract before anything else is built. The
|
|
four exit criteria (§7.1 P1) are:
|
|
|
|
* (a) kill the box mid-wait and have the task resume after restart;
|
|
* (b) submit a duplicate answer and confirm it no-ops;
|
|
* (c) submit an answer after the deadline expired and confirm it is rejected
|
|
and the task parked;
|
|
* (d) two tasks suspended concurrently resume independently to the correct
|
|
thread.
|
|
|
|
This module is a *simulation* harness, not the production pipeline. There is no
|
|
LangGraph runtime on the box yet (that lands in the P1 build proper), so the
|
|
harness stands in a minimal, faithful model of the riskiest mechanic — the
|
|
``pending_questions`` ledger and the §3.3.1 first-answer-wins / deadline-race /
|
|
turn-guarded-resume compare-and-set — *built on the real committed foundation*:
|
|
|
|
* :mod:`agent_team.db.schema` — the real ``pending_questions`` ledger DDL and
|
|
the real ``answer_question`` / ``expire_question`` / ``supersede_question``
|
|
``BEGIN IMMEDIATE`` compare-and-set helpers. The harness never re-implements
|
|
the atomic statements; it drives the committed ones.
|
|
* :mod:`agent_team.state_store` — the real atomic, integrity-checked durable
|
|
state store. The "graph checkpoint" each task suspends on is written through
|
|
:func:`agent_team.state_store.write_checked` and read back through
|
|
:func:`agent_team.state_store.read_checked`, so a simulated "kill the box"
|
|
(drop the in-memory harness, reconstruct from disk) exercises real durable
|
|
recovery, not a Python dict.
|
|
* :mod:`agent_team.task_model` — the real :class:`TaskRecord` / :class:`Phase`
|
|
/ :class:`TaskStatus` model and its JSON serialization.
|
|
* :mod:`agent_team.transport.base` — the real :class:`Transport` ABC and
|
|
:class:`QuestionSet` payload; :class:`RecordingTransport` is a faithful
|
|
in-memory adapter subclassing the committed contract (no live Slack).
|
|
|
|
Nothing here provisions, schedules, or reaches live infrastructure. It is
|
|
pre-deployment scaffolding that proves the design's durable mechanic holds.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from agent_team.db.schema import (
|
|
answer_question,
|
|
connect,
|
|
expire_question,
|
|
init_db,
|
|
supersede_question,
|
|
)
|
|
from agent_team.state_store import IntegrityError, read_checked, write_checked
|
|
from agent_team.task_model import (
|
|
Phase,
|
|
TaskRecord,
|
|
TaskStatus,
|
|
new_thread_id,
|
|
task_from_json,
|
|
task_to_json,
|
|
)
|
|
from agent_team.transport.base import QuestionSet, Transport
|
|
|
|
__all__ = [
|
|
"CHECKPOINT_SCHEMA_VERSION",
|
|
"PostFailingTransport",
|
|
"RecordingTransport",
|
|
"ResumeOutcome",
|
|
"SimClock",
|
|
"SimPipeline",
|
|
"SuspendedTask",
|
|
]
|
|
|
|
# Schema version stamped on the harness's durable "graph checkpoint" sidecars.
|
|
# Distinct from the SQL ``SCHEMA_VERSION``; this versions the checkpoint blob
|
|
# format the harness round-trips through ``state_store``.
|
|
CHECKPOINT_SCHEMA_VERSION: int = 1
|
|
|
|
|
|
class SimClock:
|
|
"""A monotonically advanceable fake clock for deadline-race tests (§3.3.1).
|
|
|
|
Each open question carries a ``deadline_at``. Rather than sleep in tests,
|
|
the harness compares a question's deadline against this clock's "now", and
|
|
tests advance the clock past a deadline to drive the timer loop. Times are
|
|
plain integer ticks (seconds since an arbitrary epoch); the ledger stores
|
|
them as ISO-like sortable strings so the durable column is human-readable.
|
|
"""
|
|
|
|
def __init__(self, start: int = 0) -> None:
|
|
self._now = int(start)
|
|
|
|
def now(self) -> int:
|
|
"""Return the current tick."""
|
|
return self._now
|
|
|
|
def advance(self, ticks: int) -> int:
|
|
"""Advance the clock by ``ticks`` and return the new now."""
|
|
if ticks < 0:
|
|
raise ValueError("cannot advance the clock backwards")
|
|
self._now += int(ticks)
|
|
return self._now
|
|
|
|
def stamp(self, tick: int | None = None) -> str:
|
|
"""Render ``tick`` (default: now) as a sortable durable timestamp."""
|
|
value = self._now if tick is None else int(tick)
|
|
# Zero-padded so lexical order == numeric order in the ledger column.
|
|
return f"t{value:020d}"
|
|
|
|
|
|
@dataclass
|
|
class RecordingTransport(Transport):
|
|
"""In-memory :class:`Transport` adapter that records posts (no live Slack).
|
|
|
|
A faithful subclass of the committed :class:`agent_team.transport.base.
|
|
Transport` contract: :meth:`post_question` embeds the ``question_id`` in
|
|
the returned ``channel_ref`` (mirroring the real "the post MUST embed the
|
|
question_id" rule) and records the post so recovery/reconcile tests can
|
|
inspect delivery. :meth:`parse_answer` normalizes a ``(question_id,
|
|
answer, via)`` raw payload, mapping it back via the embedded id.
|
|
"""
|
|
|
|
name: str = "sim"
|
|
posts: list[dict[str, Any]] = field(default_factory=list)
|
|
|
|
def post_question(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
question_id: str,
|
|
turn: int,
|
|
question_set: QuestionSet,
|
|
deadline: str,
|
|
) -> str:
|
|
channel_ref = f"{self.name}:{question_id}"
|
|
self.posts.append(
|
|
{
|
|
"thread_id": thread_id,
|
|
"question_id": question_id,
|
|
"turn": turn,
|
|
"questions": list(question_set.questions),
|
|
"deadline": deadline,
|
|
"channel_ref": channel_ref,
|
|
}
|
|
)
|
|
return channel_ref
|
|
|
|
def parse_answer(self, raw: Any) -> tuple[str, Any, str]:
|
|
question_id = raw["question_id"]
|
|
answer = raw.get("answer")
|
|
via = raw.get("via", self.name)
|
|
return question_id, answer, via
|
|
|
|
|
|
@dataclass
|
|
class PostFailingTransport(RecordingTransport):
|
|
"""A transport whose first ``post_question`` raises (lost-post simulation).
|
|
|
|
Used to exercise the §3.3.1 "if the post fails, the row stays ``open`` with
|
|
no ref and a reconcile loop retries idempotently" path. The first post
|
|
raises; subsequent posts succeed and record normally.
|
|
"""
|
|
|
|
fail_times: int = 1
|
|
_attempts: int = 0
|
|
|
|
def post_question(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
question_id: str,
|
|
turn: int,
|
|
question_set: QuestionSet,
|
|
deadline: str,
|
|
) -> str:
|
|
self._attempts += 1
|
|
if self._attempts <= self.fail_times:
|
|
raise RuntimeError("simulated transport post failure")
|
|
return super().post_question(
|
|
thread_id=thread_id,
|
|
question_id=question_id,
|
|
turn=turn,
|
|
question_set=question_set,
|
|
deadline=deadline,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SuspendedTask:
|
|
"""Handle to a task suspended on a question (returned by :meth:`SimPipeline.submit`)."""
|
|
|
|
thread_id: str
|
|
question_id: str
|
|
turn: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResumeOutcome:
|
|
"""Result of attempting to resume a task on an answered question.
|
|
|
|
``resumed`` is True iff the turn guard passed and the graph advanced;
|
|
``superseded`` is True iff a stale/redelivered resume was skipped (§3.3.1
|
|
"a resume can never double-apply").
|
|
"""
|
|
|
|
thread_id: str
|
|
resumed: bool
|
|
superseded: bool
|
|
new_phase: Phase | None
|
|
|
|
|
|
class SimPipeline:
|
|
"""A minimal, durable simulation of the §3.3.1 suspend/resume mechanic.
|
|
|
|
The pipeline owns two real durable stores under ``root``:
|
|
|
|
* the SQLite ``pending_questions`` ledger (via the committed
|
|
:mod:`agent_team.db.schema`), the single source of truth for the
|
|
question lifecycle, and
|
|
* one integrity-checked "graph checkpoint" file per task (via the
|
|
committed :mod:`agent_team.state_store`), holding the durable
|
|
:class:`agent_team.task_model.TaskRecord`.
|
|
|
|
A task is submitted, suspends on a question (status ``WAITING_HUMAN``,
|
|
ledger row ``open``), and later resumes when a first valid answer wins the
|
|
compare-and-set. "Killing the box" is modelled by dropping the in-memory
|
|
pipeline and calling :meth:`reopen`, which reconstructs purely from the two
|
|
durable stores — proving recovery has no in-memory-only state.
|
|
"""
|
|
|
|
def __init__(self, root: Path, *, clock: SimClock, transport: Transport) -> None:
|
|
self._root = Path(root)
|
|
self._db_path = self._root / "agent_team.sqlite"
|
|
self._checkpoints = self._root / "checkpoints"
|
|
self._clock = clock
|
|
self._transport = transport
|
|
self._root.mkdir(parents=True, exist_ok=True)
|
|
self._checkpoints.mkdir(parents=True, exist_ok=True)
|
|
init_db(self._db_path)
|
|
|
|
# -- durable checkpoint helpers (real state_store) --------------------
|
|
|
|
def _checkpoint_path(self, thread_id: str) -> Path:
|
|
return self._checkpoints / f"{thread_id}.json"
|
|
|
|
def _write_checkpoint(self, record: TaskRecord) -> None:
|
|
"""Persist a task record through the real atomic state-store."""
|
|
write_checked(
|
|
self._checkpoint_path(record.thread_id),
|
|
task_to_json(record).encode("utf-8"),
|
|
schema_version=CHECKPOINT_SCHEMA_VERSION,
|
|
)
|
|
|
|
def load_record(self, thread_id: str) -> TaskRecord:
|
|
"""Read a task record back, integrity-checked (raises IntegrityError)."""
|
|
data = read_checked(
|
|
self._checkpoint_path(thread_id),
|
|
schema_version=CHECKPOINT_SCHEMA_VERSION,
|
|
)
|
|
return task_from_json(data.decode("utf-8"))
|
|
|
|
# -- ledger helpers (real db.schema connection) -----------------------
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
# Use the committed foundation connection helper (WAL + busy_timeout +
|
|
# the stashed db path the compare-and-set relies on) rather than a raw
|
|
# sqlite3.connect — so concurrent responders genuinely serialize on the
|
|
# write lock (§3.3.1) instead of racing without a busy timeout.
|
|
return connect(self._db_path)
|
|
|
|
def ledger_row(self, question_id: str) -> sqlite3.Row | None:
|
|
"""Return the durable ``pending_questions`` row for ``question_id``."""
|
|
conn = self._connect()
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
return conn.execute(
|
|
"SELECT * FROM pending_questions WHERE question_id = ?",
|
|
(question_id,),
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
# -- pipeline operations ----------------------------------------------
|
|
|
|
def submit(self, *, questions: list[str], deadline_in: int) -> SuspendedTask:
|
|
"""Submit a task; it advances to CLARIFY and suspends on a question.
|
|
|
|
Writes the ledger row ``open`` *first*, then posts to the transport and
|
|
stores the returned ``channel_ref`` (the §3.3.1 delivery order). If the
|
|
post fails the row stays ``open`` with no ref for the reconcile loop to
|
|
retry. The durable task record is checkpointed as ``WAITING_HUMAN``.
|
|
"""
|
|
thread_id = new_thread_id()
|
|
question_id = uuid.uuid4().hex
|
|
turn = 0
|
|
deadline_at = self._clock.stamp(self._clock.now() + int(deadline_in))
|
|
|
|
record = TaskRecord(
|
|
thread_id=thread_id,
|
|
status=TaskStatus.WAITING_HUMAN,
|
|
current_phase=Phase.CLARIFY,
|
|
transport=getattr(self._transport, "name", "sim"),
|
|
created_at=self._clock.stamp(),
|
|
updated_at=self._clock.stamp(),
|
|
)
|
|
self._write_checkpoint(record)
|
|
|
|
# Ledger row first (open, no channel_ref yet).
|
|
conn = self._connect()
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO pending_questions "
|
|
"(question_id, thread_id, turn, status, transport, posted_at, deadline_at) "
|
|
"VALUES (?, ?, ?, 'open', ?, ?, ?)",
|
|
(
|
|
question_id,
|
|
thread_id,
|
|
turn,
|
|
record.transport,
|
|
self._clock.stamp(),
|
|
deadline_at,
|
|
),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
# Then deliver; tolerate a lost post (row stays open, no ref).
|
|
self._deliver(
|
|
thread_id=thread_id,
|
|
question_id=question_id,
|
|
turn=turn,
|
|
questions=questions,
|
|
deadline_at=deadline_at,
|
|
)
|
|
return SuspendedTask(thread_id=thread_id, question_id=question_id, turn=turn)
|
|
|
|
def _deliver(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
question_id: str,
|
|
turn: int,
|
|
questions: list[str],
|
|
deadline_at: str,
|
|
) -> str | None:
|
|
"""Post the question and store the channel_ref; tolerate post failure."""
|
|
question_set = QuestionSet(
|
|
thread_id=thread_id,
|
|
question_id=question_id,
|
|
turn=turn,
|
|
questions=list(questions),
|
|
)
|
|
try:
|
|
channel_ref = self._transport.post_question(
|
|
thread_id=thread_id,
|
|
question_id=question_id,
|
|
turn=turn,
|
|
question_set=question_set,
|
|
deadline=deadline_at,
|
|
)
|
|
except Exception:
|
|
# Lost post: row stays open with no ref; reconcile retries later.
|
|
return None
|
|
conn = self._connect()
|
|
try:
|
|
conn.execute(
|
|
"UPDATE pending_questions SET channel_ref = ? WHERE question_id = ?",
|
|
(channel_ref, question_id),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
return channel_ref
|
|
|
|
def reconcile(self, *, questions_by_qid: dict[str, list[str]]) -> int:
|
|
"""Retry delivery for ``open`` rows lacking a ``channel_ref`` (§3.3.1).
|
|
|
|
Returns the number of rows for which a (re)delivery now succeeded.
|
|
``questions_by_qid`` supplies the question text per id (the harness
|
|
does not persist question text on the ledger, mirroring the design's
|
|
ledger schema which carries lifecycle, not prompt bodies).
|
|
"""
|
|
conn = self._connect()
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT question_id, thread_id, turn, deadline_at "
|
|
"FROM pending_questions "
|
|
"WHERE status = 'open' AND channel_ref IS NULL"
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
redelivered = 0
|
|
for row in rows:
|
|
ref = self._deliver(
|
|
thread_id=row["thread_id"],
|
|
question_id=row["question_id"],
|
|
turn=row["turn"],
|
|
questions=questions_by_qid.get(row["question_id"], []),
|
|
deadline_at=row["deadline_at"],
|
|
)
|
|
if ref is not None:
|
|
redelivered += 1
|
|
return redelivered
|
|
|
|
def submit_answer(self, raw: Any) -> bool:
|
|
"""Normalize ``raw`` via the transport and run the §3.3.1 compare-and-set.
|
|
|
|
Returns ``True`` when this answer won the race (ledger rowcount 1 — the
|
|
first valid answer, a resume is now eligible) and ``False`` when it
|
|
lost (rowcount 0 — duplicate, late, or for a closed question, ignored).
|
|
Drives the *committed* :func:`agent_team.db.schema.answer_question`
|
|
``BEGIN IMMEDIATE`` statement; the harness never re-implements it.
|
|
"""
|
|
question_id, answer, via = self._transport.parse_answer(raw)
|
|
conn = self._connect()
|
|
try:
|
|
return answer_question(
|
|
conn,
|
|
question_id=question_id,
|
|
answer_json=json.dumps(answer),
|
|
answered_via=via,
|
|
answered_at=self._clock.stamp(),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
def run_deadline_sweep(self) -> list[str]:
|
|
"""Expire every overdue ``open`` question and park its task (§3.3.1).
|
|
|
|
A timer loop flips overdue ``open`` rows to ``expired`` via the
|
|
committed compare-and-set (:func:`agent_team.db.schema.
|
|
expire_question`) and applies the park policy: the task record flips to
|
|
:attr:`TaskStatus.PARKED` / :attr:`Phase.PARKED`. Returns the list of
|
|
``question_id`` s expired by this sweep. An answer arriving for an
|
|
already-expired question will lose its own compare-and-set.
|
|
"""
|
|
now = self._clock.now()
|
|
conn = self._connect()
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT question_id, thread_id, deadline_at "
|
|
"FROM pending_questions WHERE status = 'open'"
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
expired: list[str] = []
|
|
for row in rows:
|
|
if not self._is_overdue(row["deadline_at"], now):
|
|
continue
|
|
conn2 = self._connect()
|
|
try:
|
|
won = expire_question(conn2, question_id=row["question_id"])
|
|
finally:
|
|
conn2.close()
|
|
if won:
|
|
expired.append(row["question_id"])
|
|
self._park(row["thread_id"])
|
|
return expired
|
|
|
|
@staticmethod
|
|
def _is_overdue(deadline_at: str | None, now: int) -> bool:
|
|
"""Decode a ``SimClock``-stamped deadline and test it against ``now``."""
|
|
if not deadline_at:
|
|
return False
|
|
try:
|
|
deadline_tick = int(deadline_at.lstrip("t"))
|
|
except ValueError:
|
|
return False
|
|
return now >= deadline_tick
|
|
|
|
def _park(self, thread_id: str) -> None:
|
|
"""Flip a task record to PARKED durably (idempotent)."""
|
|
record = self.load_record(thread_id)
|
|
record.status = TaskStatus.PARKED
|
|
record.current_phase = Phase.PARKED
|
|
record.updated_at = self._clock.stamp()
|
|
self._write_checkpoint(record)
|
|
|
|
def resume(self, thread_id: str, question_id: str) -> ResumeOutcome:
|
|
"""Turn-guarded resume of an ``answered`` question (§3.3.1, single-flight).
|
|
|
|
Mirrors the design's resume worker: before advancing the graph it
|
|
checks the live checkpoint is still interrupted on this turn. The
|
|
durable task record is the checkpoint here, so:
|
|
|
|
* if the record is still ``WAITING_HUMAN`` on the answered question, the
|
|
graph advances (CLARIFY -> PLAN), the record is checkpointed
|
|
``ACTIVE``, and ``resumed`` is True;
|
|
* if the record already advanced (a stale/redelivered resume), the
|
|
question is marked ``superseded`` via the committed compare-and-set
|
|
and the resume is skipped (``superseded`` True), so a resume can
|
|
never double-apply.
|
|
"""
|
|
row = self.ledger_row(question_id)
|
|
if row is None or row["status"] != "answered":
|
|
return ResumeOutcome(
|
|
thread_id=thread_id, resumed=False, superseded=False, new_phase=None
|
|
)
|
|
|
|
record = self.load_record(thread_id)
|
|
# Turn guard: only resume if still suspended on this turn/phase.
|
|
if (
|
|
record.status is not TaskStatus.WAITING_HUMAN
|
|
or record.current_phase is not Phase.CLARIFY
|
|
):
|
|
conn = self._connect()
|
|
try:
|
|
supersede_question(conn, question_id=question_id)
|
|
finally:
|
|
conn.close()
|
|
return ResumeOutcome(
|
|
thread_id=thread_id,
|
|
resumed=False,
|
|
superseded=True,
|
|
new_phase=record.current_phase,
|
|
)
|
|
|
|
# Apply the won answer into the durable Q&A history and advance a phase.
|
|
answer = json.loads(row["answer_json"]) if row["answer_json"] else None
|
|
record.qa_history.append(
|
|
{"question_id": question_id, "turn": row["turn"], "answer": answer}
|
|
)
|
|
record.status = TaskStatus.ACTIVE
|
|
record.current_phase = Phase.PLAN
|
|
record.updated_at = self._clock.stamp()
|
|
self._write_checkpoint(record)
|
|
return ResumeOutcome(
|
|
thread_id=thread_id,
|
|
resumed=True,
|
|
superseded=False,
|
|
new_phase=Phase.PLAN,
|
|
)
|
|
|
|
# -- restart recovery -------------------------------------------------
|
|
|
|
def reopen(self) -> SimPipeline:
|
|
"""Simulate "kill the box": return a fresh pipeline over the same disk.
|
|
|
|
The new pipeline shares the durable SQLite ledger and the
|
|
integrity-checked checkpoints but holds *no* in-memory state, so any
|
|
recovery must come entirely from disk (§3.3.1 "No in-memory-only
|
|
state."). The transport and clock are re-used (a real restart would
|
|
re-instantiate adapters; reusing them keeps recorded posts visible to
|
|
the assertions).
|
|
"""
|
|
return SimPipeline(self._root, clock=self._clock, transport=self._transport)
|
|
|
|
def startup_sweep(
|
|
self, *, questions_by_qid: dict[str, list[str]]
|
|
) -> dict[str, Any]:
|
|
"""Run the §3.3.1 startup convergence sweep after a restart.
|
|
|
|
Concretely: (1) retry delivery for ``open`` rows lacking a ref; (2)
|
|
re-enqueue a resume for ``answered`` rows whose task is still suspended
|
|
on that turn (idempotent via the turn guard); (3) apply the deadline
|
|
policy for overdue ``open`` rows. Returns a summary of what converged.
|
|
"""
|
|
redelivered = self.reconcile(questions_by_qid=questions_by_qid)
|
|
|
|
conn = self._connect()
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
answered = conn.execute(
|
|
"SELECT question_id, thread_id FROM pending_questions "
|
|
"WHERE status = 'answered'"
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
resumed: list[str] = []
|
|
for row in answered:
|
|
outcome = self.resume(row["thread_id"], row["question_id"])
|
|
if outcome.resumed:
|
|
resumed.append(row["thread_id"])
|
|
|
|
expired = self.run_deadline_sweep()
|
|
return {
|
|
"redelivered": redelivered,
|
|
"resumed": resumed,
|
|
"expired": expired,
|
|
}
|
|
|
|
|
|
def assert_no_integrity_error(pipeline: SimPipeline, thread_id: str) -> TaskRecord:
|
|
"""Load a record and surface :class:`IntegrityError` as an explicit failure.
|
|
|
|
Convenience for tests that want the durable read to be part of the
|
|
assertion (the foundation fails closed on corruption rather than returning
|
|
junk).
|
|
"""
|
|
try:
|
|
return pipeline.load_record(thread_id)
|
|
except IntegrityError as exc: # pragma: no cover - defensive
|
|
raise AssertionError(
|
|
f"durable checkpoint failed integrity check: {exc}"
|
|
) from exc
|