Adam Moussa
1f8c7e1ee3
Remediates the Phase-0 adversarial BLOCKs: - Durable ci_pending_provider (_enumerate_ci_pending) walks the LangGraph SQLite checkpointer to enumerate threads suspended at VERIFY awaiting CI; re-derives across restart. Excludes human-clarify gates + advanced threads. - run-team serve wires ci_pending_provider + ci_poller + ci_timeout ONLY on a configured box; inert path unchanged. Closes the 'VERIFY suspended forever' defect: tick()->_ci_watch resumes on terminal CI or timeout-parks. - CI resume routes through the single-flight, turn-guarded ResumeWorker. - FIXes: run-locator skips cancelled/stale runs on rapid re-dispatch; inert-mode wording matches behavior; added node-level fail-closed + spurious-resume tests. - end-to-end async-resume proof (test_p3_async_resume.py, real checkpointer). Suite: 1270 passed, ruff clean. Branch only; not merged/deployed.
594 lines
20 KiB
Python
594 lines
20 KiB
Python
"""Unit tests for agent_team.resume_worker (§3.3.1 single-flight, turn-guarded).
|
|
|
|
These tests prove the design's three resume guarantees:
|
|
|
|
* turn-guarded: resume applies only while the graph is interrupted on the
|
|
answer's turn; a graph that already advanced is superseded and skipped;
|
|
* no double-apply: a redelivered job for an already-advanced thread no-ops;
|
|
* single-flight: resumes for one ``thread_id`` are serialized while different
|
|
threads run concurrently;
|
|
|
|
plus the restart-recovery sweep, and an end-to-end check against a real
|
|
compiled LangGraph app when ``langgraph`` is importable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from agent_team import resume_worker
|
|
from agent_team.db.schema import answer_question, init_db, connect
|
|
from agent_team.resume_worker import (
|
|
GraphLike,
|
|
ResumeOutcome,
|
|
ResumeResult,
|
|
ResumeWorker,
|
|
build_resume_command,
|
|
snapshot_interrupt_turns,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fakes
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@dataclass
|
|
class _FakeInterrupt:
|
|
"""Mimics a langgraph Interrupt: carries a ``.value`` payload."""
|
|
|
|
value: Any
|
|
|
|
|
|
@dataclass
|
|
class _FakeSnapshot:
|
|
"""Mimics a langgraph StateSnapshot's relevant surface."""
|
|
|
|
next: tuple[str, ...] = ()
|
|
interrupts: tuple[_FakeInterrupt, ...] = ()
|
|
|
|
|
|
class _FakeGraph:
|
|
"""A GraphLike test double over an explicit interrupt-turn.
|
|
|
|
``interrupted_turn`` is the turn the graph is currently suspended on, or
|
|
``None`` if it has advanced past every interrupt. ``invoke`` records every
|
|
resume payload so double-apply is directly observable, and advances the
|
|
graph (clears the interrupt) the way a real resume would.
|
|
"""
|
|
|
|
def __init__(self, interrupted_turn: int | None) -> None:
|
|
self.interrupted_turn = interrupted_turn
|
|
self.invocations: list[Any] = []
|
|
self.get_state_calls: list[dict[str, Any]] = []
|
|
self._invoke_hook: Any = None
|
|
|
|
def get_state(self, config: dict[str, Any]) -> _FakeSnapshot:
|
|
self.get_state_calls.append(config)
|
|
if self.interrupted_turn is None:
|
|
return _FakeSnapshot(next=(), interrupts=())
|
|
payload = {"turn": self.interrupted_turn, "question_id": "q"}
|
|
return _FakeSnapshot(
|
|
next=("clarify",),
|
|
interrupts=(_FakeInterrupt(value=payload),),
|
|
)
|
|
|
|
def invoke(self, command: Any, config: dict[str, Any]) -> Any:
|
|
if self._invoke_hook is not None:
|
|
self._invoke_hook()
|
|
self.invocations.append(command)
|
|
# A real resume clears the interrupt and advances the graph.
|
|
self.interrupted_turn = None
|
|
return {"resumed": True, "command": command}
|
|
|
|
|
|
def _insert_question(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
question_id: str,
|
|
thread_id: str,
|
|
turn: int,
|
|
status: str = "open",
|
|
) -> None:
|
|
conn.execute(
|
|
"INSERT INTO pending_questions "
|
|
"(question_id, thread_id, turn, status, transport) "
|
|
"VALUES (?, ?, ?, ?, 'slack')",
|
|
(question_id, thread_id, turn, status),
|
|
)
|
|
|
|
|
|
def _status(conn: sqlite3.Connection, question_id: str) -> str:
|
|
row = conn.execute(
|
|
"SELECT status FROM pending_questions WHERE question_id = ?",
|
|
(question_id,),
|
|
).fetchone()
|
|
return row["status"]
|
|
|
|
|
|
@pytest.fixture()
|
|
def conn(tmp_path: Path) -> sqlite3.Connection:
|
|
db = tmp_path / "agent_team.sqlite"
|
|
init_db(db)
|
|
connection = connect(db)
|
|
yield connection
|
|
connection.close()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# snapshot_interrupt_turns
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_snapshot_turns_from_mapping_payload() -> None:
|
|
snap = _FakeSnapshot(
|
|
next=("n",),
|
|
interrupts=(_FakeInterrupt(value={"turn": 7}),),
|
|
)
|
|
assert snapshot_interrupt_turns(snap) == {7}
|
|
|
|
|
|
def test_snapshot_turns_from_object_payload() -> None:
|
|
@dataclass
|
|
class _ObjPayload:
|
|
turn: int
|
|
|
|
snap = _FakeSnapshot(interrupts=(_FakeInterrupt(value=_ObjPayload(turn=2)),))
|
|
assert snapshot_interrupt_turns(snap) == {2}
|
|
|
|
|
|
def test_snapshot_turns_empty_when_not_interrupted() -> None:
|
|
assert snapshot_interrupt_turns(_FakeSnapshot()) == set()
|
|
|
|
|
|
def test_snapshot_turns_ignores_unreadable_turn() -> None:
|
|
snap = _FakeSnapshot(interrupts=(_FakeInterrupt(value={"no_turn": 1}),))
|
|
assert snapshot_interrupt_turns(snap) == set()
|
|
|
|
|
|
def test_snapshot_turns_ignores_bool_turn() -> None:
|
|
# bool is an int subclass; a True/False must not be read as a turn number.
|
|
snap = _FakeSnapshot(interrupts=(_FakeInterrupt(value={"turn": True}),))
|
|
assert snapshot_interrupt_turns(snap) == set()
|
|
|
|
|
|
def test_snapshot_turns_collects_multiple() -> None:
|
|
snap = _FakeSnapshot(
|
|
interrupts=(
|
|
_FakeInterrupt(value={"turn": 1}),
|
|
_FakeInterrupt(value={"turn": 4}),
|
|
)
|
|
)
|
|
assert snapshot_interrupt_turns(snap) == {1, 4}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# resume — turn guard
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_resume_applies_when_interrupted_on_turn(conn: sqlite3.Connection) -> None:
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=3, status="answered")
|
|
graph = _FakeGraph(interrupted_turn=3)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume(
|
|
thread_id="t1", question_id="q1", turn=3, answer="the answer"
|
|
)
|
|
|
|
assert result.outcome is ResumeOutcome.RESUMED
|
|
assert result.resumed is True
|
|
assert result.graph_result == {"resumed": True, "command": graph.invocations[0]}
|
|
assert len(graph.invocations) == 1
|
|
# Question is untouched by the worker on a successful resume (the responder
|
|
# already flipped it to answered).
|
|
assert _status(conn, "q1") == "answered"
|
|
|
|
|
|
def test_resume_superseded_when_graph_advanced(conn: sqlite3.Connection) -> None:
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=3, status="answered")
|
|
graph = _FakeGraph(interrupted_turn=None) # already advanced past turn 3
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume(thread_id="t1", question_id="q1", turn=3, answer="x")
|
|
|
|
assert result.outcome is ResumeOutcome.SUPERSEDED
|
|
assert result.resumed is False
|
|
assert graph.invocations == [] # never invoked -> never applied
|
|
assert _status(conn, "q1") == "superseded"
|
|
|
|
|
|
def test_resume_superseded_when_interrupted_on_different_turn(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
# Graph moved on to a *later* interrupt (turn 4); a turn-3 job is stale.
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=3, status="answered")
|
|
graph = _FakeGraph(interrupted_turn=4)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume(thread_id="t1", question_id="q1", turn=3, answer="x")
|
|
|
|
assert result.outcome is ResumeOutcome.SUPERSEDED
|
|
assert graph.invocations == []
|
|
assert _status(conn, "q1") == "superseded"
|
|
|
|
|
|
def test_resume_stale_when_nothing_to_supersede(conn: sqlite3.Connection) -> None:
|
|
# Question already expired; graph advanced. Nothing to supersede.
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=3, status="expired")
|
|
graph = _FakeGraph(interrupted_turn=None)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume(thread_id="t1", question_id="q1", turn=3, answer="x")
|
|
|
|
assert result.outcome is ResumeOutcome.STALE
|
|
assert graph.invocations == []
|
|
assert _status(conn, "q1") == "expired"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# no double-apply
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_redelivered_job_does_not_double_apply(conn: sqlite3.Connection) -> None:
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=3, status="answered")
|
|
graph = _FakeGraph(interrupted_turn=3)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
first = worker.resume(thread_id="t1", question_id="q1", turn=3, answer="a")
|
|
# A redelivered/duplicate resume job for the same turn arrives.
|
|
second = worker.resume(thread_id="t1", question_id="q1", turn=3, answer="a")
|
|
|
|
assert first.outcome is ResumeOutcome.RESUMED
|
|
assert second.outcome is ResumeOutcome.SUPERSEDED
|
|
# The graph was invoked exactly once across both jobs.
|
|
assert len(graph.invocations) == 1
|
|
assert _status(conn, "q1") == "superseded"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# single-flight serialization
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_same_thread_resumes_are_serialized(conn: sqlite3.Connection) -> None:
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=3, status="answered")
|
|
graph = _FakeGraph(interrupted_turn=3)
|
|
|
|
in_invoke = threading.Event()
|
|
overlap_detected: list[bool] = []
|
|
concurrency = {"current": 0, "max": 0}
|
|
lock = threading.Lock()
|
|
|
|
def _hook() -> None:
|
|
with lock:
|
|
concurrency["current"] += 1
|
|
concurrency["max"] = max(concurrency["max"], concurrency["current"])
|
|
in_invoke.set()
|
|
time.sleep(0.05)
|
|
with lock:
|
|
concurrency["current"] -= 1
|
|
|
|
graph._invoke_hook = _hook
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
results: list[ResumeResult] = []
|
|
results_lock = threading.Lock()
|
|
|
|
def _run(answer: str) -> None:
|
|
r = worker.resume(thread_id="t1", question_id="q1", turn=3, answer=answer)
|
|
with results_lock:
|
|
results.append(r)
|
|
|
|
threads = [threading.Thread(target=_run, args=(f"a{i}",)) for i in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Lock serialized them: invoke never overlapped.
|
|
assert concurrency["max"] == 1
|
|
assert overlap_detected == []
|
|
# Exactly one resumed; the rest were superseded (turn guard) -> no double.
|
|
resumed = [r for r in results if r.outcome is ResumeOutcome.RESUMED]
|
|
assert len(resumed) == 1
|
|
assert len(graph.invocations) == 1
|
|
|
|
|
|
def test_distinct_threads_use_distinct_locks(conn: sqlite3.Connection) -> None:
|
|
graph_a = _FakeGraph(interrupted_turn=1)
|
|
graph_b = _FakeGraph(interrupted_turn=1)
|
|
# One worker can only hold one graph; emulate isolation by giving each
|
|
# thread its own worker over its own graph, sharing the ledger.
|
|
_insert_question(conn, question_id="qa", thread_id="ta", turn=1, status="answered")
|
|
_insert_question(conn, question_id="qb", thread_id="tb", turn=1, status="answered")
|
|
|
|
worker_a = ResumeWorker(graph_a, conn)
|
|
worker_b = ResumeWorker(graph_b, conn)
|
|
|
|
# Distinct thread_ids must mint distinct locks within a single worker.
|
|
single = ResumeWorker(_FakeGraph(interrupted_turn=1), conn)
|
|
assert single._lock_for("ta") is not single._lock_for("tb")
|
|
assert single._lock_for("ta") is single._lock_for("ta")
|
|
|
|
ra = worker_a.resume(thread_id="ta", question_id="qa", turn=1, answer="x")
|
|
rb = worker_b.resume(thread_id="tb", question_id="qb", turn=1, answer="y")
|
|
assert ra.outcome is ResumeOutcome.RESUMED
|
|
assert rb.outcome is ResumeOutcome.RESUMED
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# restart recovery sweep
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_recover_resumes_answered_rows_still_interrupted(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=2, status="open")
|
|
answer_question(
|
|
conn, question_id="q1", answer_json=json.dumps("ans"), answered_via="slack"
|
|
)
|
|
graph = _FakeGraph(interrupted_turn=2)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
results = worker.recover_pending_resumes()
|
|
|
|
assert len(results) == 1
|
|
assert results[0].outcome is ResumeOutcome.RESUMED
|
|
assert results[0].thread_id == "t1"
|
|
# The decoded answer reached the graph as a Command(resume=...).
|
|
assert len(graph.invocations) == 1
|
|
|
|
|
|
def test_recover_is_idempotent_when_graph_already_advanced(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
_insert_question(conn, question_id="q1", thread_id="t1", turn=2, status="open")
|
|
answer_question(
|
|
conn, question_id="q1", answer_json=json.dumps("ans"), answered_via="slack"
|
|
)
|
|
# Graph already advanced (the resume applied before the crash).
|
|
graph = _FakeGraph(interrupted_turn=None)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
results = worker.recover_pending_resumes()
|
|
|
|
assert len(results) == 1
|
|
assert results[0].outcome is ResumeOutcome.SUPERSEDED
|
|
assert graph.invocations == [] # no double-apply across a restart
|
|
assert _status(conn, "q1") == "superseded"
|
|
|
|
|
|
def test_recover_skips_non_answered_rows(conn: sqlite3.Connection) -> None:
|
|
_insert_question(conn, question_id="open1", thread_id="t1", turn=0, status="open")
|
|
_insert_question(conn, question_id="exp1", thread_id="t2", turn=0, status="expired")
|
|
graph = _FakeGraph(interrupted_turn=0)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
results = worker.recover_pending_resumes()
|
|
|
|
assert results == []
|
|
assert graph.invocations == []
|
|
|
|
|
|
def test_recover_processes_answered_oldest_first(conn: sqlite3.Connection) -> None:
|
|
# Two answered rows on distinct threads; recovery must visit older first.
|
|
_insert_question(
|
|
conn, question_id="q_old", thread_id="t_old", turn=0, status="open"
|
|
)
|
|
answer_question(
|
|
conn,
|
|
question_id="q_old",
|
|
answer_json=json.dumps("old"),
|
|
answered_via="slack",
|
|
answered_at="2026-01-01T00:00:00+00:00",
|
|
)
|
|
_insert_question(
|
|
conn, question_id="q_new", thread_id="t_new", turn=0, status="open"
|
|
)
|
|
answer_question(
|
|
conn,
|
|
question_id="q_new",
|
|
answer_json=json.dumps("new"),
|
|
answered_via="slack",
|
|
answered_at="2026-06-01T00:00:00+00:00",
|
|
)
|
|
graph = _FakeGraph(interrupted_turn=0)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
results = worker.recover_pending_resumes()
|
|
|
|
assert [r.thread_id for r in results] == ["t_old", "t_new"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# answer decoding
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_decode_answer_json_roundtrip() -> None:
|
|
assert resume_worker._decode_answer(json.dumps({"k": 1})) == {"k": 1}
|
|
|
|
|
|
def test_decode_answer_none() -> None:
|
|
assert resume_worker._decode_answer(None) is None
|
|
|
|
|
|
def test_decode_answer_non_json_passthrough() -> None:
|
|
assert resume_worker._decode_answer("not-json{{") == "not-json{{"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# resume_ci (CI machine-gate: turn-guarded on the awaiting-CI marker)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class _CiGraph:
|
|
"""A GraphLike double suspended at VERIFY awaiting a given CI ``run_id``.
|
|
|
|
``awaiting_run`` is the run the thread is currently suspended on, or ``None``
|
|
if it has advanced past the CI gate (resumed / parked / done). ``invoke``
|
|
records every resume and advances the graph (clears the interrupt) as a real
|
|
resume would, so a double-resume is directly observable.
|
|
"""
|
|
|
|
def __init__(self, awaiting_run: str | None) -> None:
|
|
self.awaiting_run = awaiting_run
|
|
self.invocations: list[Any] = []
|
|
|
|
def get_state(self, config: dict[str, Any]) -> _FakeSnapshot:
|
|
if self.awaiting_run is None:
|
|
return _FakeSnapshot(next=(), interrupts=())
|
|
payload = {"awaiting_ci": True, "run_id": self.awaiting_run}
|
|
return _FakeSnapshot(
|
|
next=("verify",),
|
|
interrupts=(_FakeInterrupt(value=payload),),
|
|
)
|
|
|
|
def invoke(self, command: Any, config: dict[str, Any]) -> Any:
|
|
self.invocations.append(command)
|
|
self.awaiting_run = None
|
|
return {"resumed": True}
|
|
|
|
|
|
def test_resume_ci_applies_when_suspended_on_run(conn: sqlite3.Connection) -> None:
|
|
graph = _CiGraph(awaiting_run="999")
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume_ci(thread_id="t1", run_id="999", answer={"conclusion": "ok"})
|
|
|
|
assert result.outcome is ResumeOutcome.RESUMED
|
|
assert result.resumed is True
|
|
assert len(graph.invocations) == 1
|
|
|
|
|
|
def test_resume_ci_skips_when_thread_already_advanced(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
# Already resumed/parked/done: no awaiting-CI interrupt -> guard skips invoke.
|
|
graph = _CiGraph(awaiting_run=None)
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume_ci(thread_id="t1", run_id="999", answer={})
|
|
|
|
assert result.outcome is ResumeOutcome.STALE
|
|
assert graph.invocations == []
|
|
|
|
|
|
def test_resume_ci_skips_when_awaiting_a_different_run(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
# Suspended awaiting a DIFFERENT run (e.g. a re-dispatch): must not resume.
|
|
graph = _CiGraph(awaiting_run="other")
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
result = worker.resume_ci(thread_id="t1", run_id="999", answer={})
|
|
|
|
assert result.outcome is ResumeOutcome.STALE
|
|
assert graph.invocations == []
|
|
|
|
|
|
def test_resume_ci_double_resume_is_idempotent(conn: sqlite3.Connection) -> None:
|
|
# Two terminal observations of the same run on overlapping sweeps: the first
|
|
# applies; the second finds the thread advanced (guard) and skips. State is
|
|
# never double-applied.
|
|
graph = _CiGraph(awaiting_run="999")
|
|
worker = ResumeWorker(graph, conn)
|
|
|
|
first = worker.resume_ci(thread_id="t1", run_id="999", answer={})
|
|
second = worker.resume_ci(thread_id="t1", run_id="999", answer={})
|
|
|
|
assert first.outcome is ResumeOutcome.RESUMED
|
|
assert second.outcome is ResumeOutcome.STALE
|
|
assert len(graph.invocations) == 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# command builder
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_build_resume_command_wraps_answer() -> None:
|
|
pytest.importorskip("langgraph")
|
|
cmd = build_resume_command("hello")
|
|
assert getattr(cmd, "resume", None) == "hello"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# module contract
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_module_exports_public_contract() -> None:
|
|
for name in (
|
|
"GraphLike",
|
|
"ResumeOutcome",
|
|
"ResumeResult",
|
|
"ResumeWorker",
|
|
"build_resume_command",
|
|
"snapshot_interrupt_turns",
|
|
):
|
|
assert name in resume_worker.__all__
|
|
assert hasattr(resume_worker, name)
|
|
|
|
|
|
def test_graphlike_is_runtime_checkable() -> None:
|
|
assert isinstance(_FakeGraph(interrupted_turn=None), GraphLike)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# end-to-end against a real compiled LangGraph app
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_end_to_end_against_real_langgraph(conn: sqlite3.Connection) -> None:
|
|
pytest.importorskip("langgraph")
|
|
from langgraph.graph import StateGraph, START, END
|
|
from langgraph.checkpoint.memory import MemorySaver
|
|
from langgraph.types import interrupt
|
|
from typing import TypedDict
|
|
|
|
class S(TypedDict, total=False):
|
|
turn: int
|
|
answer: Any
|
|
|
|
def clarify(state: S) -> dict[str, Any]:
|
|
ans = interrupt({"turn": state.get("turn", 0), "question_id": "q1"})
|
|
return {"answer": ans}
|
|
|
|
g = StateGraph(S)
|
|
g.add_node("clarify", clarify)
|
|
g.add_edge(START, "clarify")
|
|
g.add_edge("clarify", END)
|
|
app = g.compile(checkpointer=MemorySaver())
|
|
|
|
cfg = {"configurable": {"thread_id": "real-1"}}
|
|
app.invoke({"turn": 5}, cfg) # suspends on interrupt at turn 5
|
|
|
|
_insert_question(
|
|
conn, question_id="q1", thread_id="real-1", turn=5, status="answered"
|
|
)
|
|
worker = ResumeWorker(app, conn)
|
|
|
|
first = worker.resume(
|
|
thread_id="real-1", question_id="q1", turn=5, answer="confirmed"
|
|
)
|
|
assert first.outcome is ResumeOutcome.RESUMED
|
|
assert first.graph_result.get("answer") == "confirmed"
|
|
|
|
# A redelivered job after the real graph advanced must not double-apply.
|
|
second = worker.resume(
|
|
thread_id="real-1", question_id="q1", turn=5, answer="confirmed"
|
|
)
|
|
assert second.outcome is ResumeOutcome.SUPERSEDED
|
|
assert _status(conn, "q1") == "superseded"
|