342 lines
14 KiB
Python
342 lines
14 KiB
Python
"""Regress exact-run replay, provenance, and reclaim safety."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from unittest import mock
|
|
|
|
from hermes_cli import kanban_db
|
|
from hermes_execution_regression_support import ExecutionSafetyTestCase
|
|
|
|
|
|
class ExactRunSafetyTests(ExecutionSafetyTestCase):
|
|
"""Verify exact-run transactions reject every stale concurrent actor."""
|
|
|
|
def test_latest_ended_run_can_replay_a_durable_completion(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="journaled result")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
run_id = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertTrue(
|
|
kanban_db.block_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="legacy post-journal failure",
|
|
kind="capability",
|
|
expected_run_id=run_id,
|
|
)
|
|
)
|
|
|
|
with mock.patch.object(
|
|
kanban_db, "_fire_kanban_lifecycle_hook"
|
|
) as lifecycle:
|
|
self.assertTrue(
|
|
kanban_db.complete_task(
|
|
self.connection,
|
|
task_id,
|
|
result="durable terminal result",
|
|
summary="durable terminal result",
|
|
replay_ended_run_id=run_id,
|
|
)
|
|
)
|
|
lifecycle.assert_called_once()
|
|
self.assertEqual(lifecycle.call_args.args[:2], ("kanban_task_completed", task_id))
|
|
self.assertEqual(lifecycle.call_args.kwargs["run_id"], run_id)
|
|
task = kanban_db.get_task(self.connection, task_id)
|
|
self.assertEqual(task.status, "done")
|
|
self.assertEqual(task.completed_run_id, run_id)
|
|
runs = kanban_db.list_runs(self.connection, task_id)
|
|
self.assertEqual(len(runs), 1)
|
|
self.assertEqual(runs[0].id, run_id)
|
|
self.assertEqual(runs[0].status, "done")
|
|
self.assertEqual(runs[0].outcome, "completed")
|
|
self.assertEqual(runs[0].summary, "durable terminal result")
|
|
completed_events = self.connection.execute(
|
|
"SELECT run_id FROM task_events "
|
|
"WHERE task_id = ? AND kind = 'completed' ORDER BY id",
|
|
(task_id,),
|
|
).fetchall()
|
|
self.assertEqual([row["run_id"] for row in completed_events], [run_id])
|
|
|
|
def test_unchanged_triage_can_replay_its_exact_latest_ended_run(self) -> None:
|
|
task_id = self._route_executed_task_to_triage()
|
|
runs = kanban_db.list_runs(self.connection, task_id)
|
|
run_id = runs[-1].id
|
|
|
|
self.assertTrue(
|
|
kanban_db.complete_task(
|
|
self.connection,
|
|
task_id,
|
|
result="accepted before the loop breaker routed triage",
|
|
summary="accepted before the loop breaker routed triage",
|
|
replay_ended_run_id=run_id,
|
|
)
|
|
)
|
|
|
|
task = kanban_db.get_task(self.connection, task_id)
|
|
self.assertEqual(task.status, "done")
|
|
self.assertEqual(task.completed_run_id, run_id)
|
|
updated_runs = kanban_db.list_runs(self.connection, task_id)
|
|
self.assertEqual(len(updated_runs), kanban_db.BLOCK_RECURRENCE_LIMIT)
|
|
self.assertEqual(updated_runs[-1].id, run_id)
|
|
self.assertEqual(updated_runs[-1].outcome, "completed")
|
|
|
|
def test_triage_status_transition_prevents_stale_exact_run_replay(self) -> None:
|
|
task_id = self._route_executed_task_to_triage()
|
|
run_id = kanban_db.list_runs(self.connection, task_id)[-1].id
|
|
self.assertTrue(
|
|
kanban_db.specify_triage_task(
|
|
self.connection,
|
|
task_id,
|
|
title="operator changed the objective",
|
|
author="operator",
|
|
)
|
|
)
|
|
|
|
self.assertFalse(
|
|
kanban_db.complete_task(
|
|
self.connection,
|
|
task_id,
|
|
result="stale accepted result",
|
|
replay_ended_run_id=run_id,
|
|
)
|
|
)
|
|
self.assertNotEqual(self._status(task_id), "done")
|
|
|
|
def test_decomposition_lineage_prevents_replay_even_if_status_returns(self) -> None:
|
|
task_id = self._route_executed_task_to_triage()
|
|
run_id = kanban_db.list_runs(self.connection, task_id)[-1].id
|
|
children = kanban_db.decompose_triage_task(
|
|
self.connection,
|
|
task_id,
|
|
root_assignee="default",
|
|
children=[{"title": "new child objective"}],
|
|
author="operator",
|
|
auto_promote=False,
|
|
)
|
|
self.assertEqual(len(children or []), 1)
|
|
with kanban_db.write_txn(self.connection):
|
|
self.connection.execute(
|
|
"UPDATE tasks SET status = 'ready' WHERE id = ?",
|
|
(task_id,),
|
|
)
|
|
|
|
self.assertFalse(
|
|
kanban_db.complete_task(
|
|
self.connection,
|
|
task_id,
|
|
result="pre-decomposition accepted result",
|
|
replay_ended_run_id=run_id,
|
|
)
|
|
)
|
|
self.assertEqual(self._status(task_id), "ready")
|
|
|
|
def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="replacement guard")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
old_run = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertTrue(
|
|
kanban_db.block_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="first run",
|
|
kind="capability",
|
|
expected_run_id=old_run,
|
|
)
|
|
)
|
|
self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
replacement_run = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertNotEqual(old_run, replacement_run)
|
|
self.assertTrue(
|
|
kanban_db.block_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="replacement run",
|
|
kind="capability",
|
|
expected_run_id=replacement_run,
|
|
)
|
|
)
|
|
|
|
self.assertFalse(
|
|
kanban_db.complete_task(
|
|
self.connection,
|
|
task_id,
|
|
result="stale result",
|
|
replay_ended_run_id=old_run,
|
|
)
|
|
)
|
|
self.assertIn(self._status(task_id), {"blocked", "triage"})
|
|
|
|
def test_equal_result_bytes_remain_bound_to_the_completing_run(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="same result retries")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
first_run = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertTrue(
|
|
kanban_db.block_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="first attempt ended",
|
|
expected_run_id=first_run,
|
|
)
|
|
)
|
|
self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
second_run = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertTrue(
|
|
kanban_db.complete_task(
|
|
self.connection,
|
|
task_id,
|
|
result="identical result bytes",
|
|
summary="second run wins",
|
|
expected_run_id=second_run,
|
|
)
|
|
)
|
|
|
|
task = kanban_db.get_task(self.connection, task_id)
|
|
self.assertEqual(task.completed_run_id, second_run)
|
|
self.assertNotEqual(task.completed_run_id, first_run)
|
|
runs = {run.id: run for run in kanban_db.list_runs(self.connection, task_id)}
|
|
self.assertEqual(runs[first_run].outcome, "blocked")
|
|
self.assertEqual(runs[second_run].outcome, "completed")
|
|
|
|
def test_exact_run_reclaim_succeeds_for_the_authoritative_run(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="recover exact run")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
run_id = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
|
|
self.assertTrue(
|
|
kanban_db.reclaim_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="invalid exact journal",
|
|
expected_run_id=run_id,
|
|
)
|
|
)
|
|
task = kanban_db.get_task(self.connection, task_id)
|
|
self.assertEqual(task.status, "ready")
|
|
self.assertIsNone(task.current_run_id)
|
|
|
|
def test_stale_reclaim_before_transaction_preserves_replacement_run(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="replacement before txn")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
old_run = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertTrue(
|
|
kanban_db.block_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="old recovery run",
|
|
expected_run_id=old_run,
|
|
)
|
|
)
|
|
self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
replacement = kanban_db.get_task(self.connection, task_id)
|
|
replacement_run = replacement.current_run_id
|
|
self.assertNotEqual(old_run, replacement_run)
|
|
signals = []
|
|
|
|
self.assertFalse(
|
|
kanban_db.reclaim_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="stale journal",
|
|
expected_run_id=old_run,
|
|
signal_fn=lambda *args: signals.append(args),
|
|
)
|
|
)
|
|
latest = kanban_db.get_task(self.connection, task_id)
|
|
self.assertEqual(latest.status, "running")
|
|
self.assertEqual(latest.current_run_id, replacement_run)
|
|
self.assertEqual(latest.claim_lock, replacement.claim_lock)
|
|
self.assertEqual(signals, [])
|
|
|
|
def test_reclaim_update_guard_preserves_run_changed_inside_transaction(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="replacement in txn")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
old_run = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
replacement = {}
|
|
|
|
def install_replacement(*_args, **_kwargs):
|
|
cursor = self.connection.execute(
|
|
"INSERT INTO task_runs (task_id, status, claim_lock, started_at) "
|
|
"VALUES (?, 'running', ?, strftime('%s','now'))",
|
|
(task_id, "replacement-lock"),
|
|
)
|
|
replacement["run_id"] = int(cursor.lastrowid)
|
|
self.connection.execute(
|
|
"UPDATE tasks SET current_run_id = ?, claim_lock = ? WHERE id = ?",
|
|
(replacement["run_id"], "replacement-lock", task_id),
|
|
)
|
|
return {}
|
|
|
|
with mock.patch.object(
|
|
kanban_db,
|
|
"_terminate_reclaimed_worker",
|
|
side_effect=install_replacement,
|
|
):
|
|
self.assertFalse(
|
|
kanban_db.reclaim_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="journal for old run",
|
|
expected_run_id=old_run,
|
|
)
|
|
)
|
|
|
|
latest = kanban_db.get_task(self.connection, task_id)
|
|
self.assertEqual(latest.status, "running")
|
|
self.assertEqual(latest.current_run_id, replacement["run_id"])
|
|
self.assertEqual(latest.claim_lock, "replacement-lock")
|
|
|
|
def test_concurrent_exact_reclaim_and_completion_have_one_winner(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="concurrent finalizer")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
run_id = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
barrier = threading.Barrier(2)
|
|
outcomes = {}
|
|
errors = []
|
|
|
|
def reclaim() -> None:
|
|
try:
|
|
with kanban_db.connect_closing() as connection:
|
|
barrier.wait()
|
|
outcomes["reclaim"] = kanban_db.reclaim_task(
|
|
connection,
|
|
task_id,
|
|
reason="concurrent invalid journal",
|
|
expected_run_id=run_id,
|
|
)
|
|
except BaseException as error: # pragma: no cover - assertion relay
|
|
errors.append(error)
|
|
|
|
def finalize() -> None:
|
|
try:
|
|
with kanban_db.connect_closing() as connection:
|
|
barrier.wait()
|
|
outcomes["complete"] = kanban_db.complete_task(
|
|
connection,
|
|
task_id,
|
|
result="durable winner",
|
|
summary="durable winner",
|
|
expected_run_id=run_id,
|
|
)
|
|
except BaseException as error: # pragma: no cover - assertion relay
|
|
errors.append(error)
|
|
|
|
threads = [threading.Thread(target=reclaim), threading.Thread(target=finalize)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(timeout=10)
|
|
self.assertFalse(thread.is_alive())
|
|
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(set(outcomes), {"reclaim", "complete"})
|
|
self.assertEqual(sum(bool(value) for value in outcomes.values()), 1)
|
|
latest = kanban_db.get_task(self.connection, task_id)
|
|
self.assertIn(latest.status, {"done", "ready"})
|
|
self.assertIsNone(latest.current_run_id)
|
|
runs = kanban_db.list_runs(self.connection, task_id)
|
|
self.assertEqual(len(runs), 1)
|
|
self.assertIsNotNone(runs[0].ended_at)
|