Nick
4eefc15837
CI Pipeline / best-practices (push) Successful in 8s
Auto-version and Release / release (push) Failing after 13s
CI Pipeline / documentation (push) Successful in 5s
CI Pipeline / canary (push) Successful in 33s
CI Pipeline / lint (push) Failing after 41s
CI Pipeline / type-check (push) Successful in 27s
CI Pipeline / gitea-offline (push) Successful in 14s
CI Pipeline / unit-tests (push) Successful in 59s
CI Pipeline / required-checks (push) Successful in 1s
deploy / deploy (push) Successful in 37s
- scripts/gitea_auth.py: single credential ladder (env -> 0600 token file -> git credential fill); 6 consumers migrated off direct os.environ reads - rune_operator: consolidate 5 duplicated failure blocks into _terminalize_execution_failure; fix 3 fault-injection gaps: (1) dual-module PreparedTool identity split crashed approval-path saves, (2) fallback-model failure escaped send_task (created/pending orphan), (3) running-transition save uncaught (running-with-no-owner) - envelope task.status now agrees with canonical ledger (awaiting_approval) - orphan task_ac35f4e287c8 terminalized failed with durable evidence - tests: gitea auth ladder (6), lifecycle fault matrix (9 boundaries + restart/reload + helper contract); 1280 green in ci-venv; ratchet 352/80
261 lines
12 KiB
Python
261 lines
12 KiB
Python
"""Task lifecycle fault-injection tests (handoff 2026-09-11 Parts I/J/K).
|
|
|
|
Invariant: NO execution exception may leave a task permanently RUNNING.
|
|
Every exit path produces a legitimate durable state."""
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
PKG_ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(PKG_ROOT / "src"))
|
|
|
|
import importlib.util
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"rune_operator", PKG_ROOT / "src" / "rune" / "rune_operator.py")
|
|
rune_operator = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(rune_operator)
|
|
|
|
from rune.tool_protocol import OperatorResponse # noqa: E402
|
|
|
|
|
|
class _FailingOperator(rune_operator.RuneOperator):
|
|
"""Operator that raises at a specific injection point."""
|
|
|
|
failure_point = None
|
|
|
|
def _route_task_class(self, description, task_id=""):
|
|
if self.failure_point == "route":
|
|
raise RuntimeError("route execution failed")
|
|
return super()._route_task_class(description, task_id)
|
|
|
|
def _build_model_router(self):
|
|
if self.failure_point == "model_selection":
|
|
raise RuntimeError("model selection failed")
|
|
return super()._build_model_router()
|
|
|
|
def _execute_on_model(self, task, model, **kw):
|
|
if self.failure_point == "execute":
|
|
raise RuntimeError("model execution failed")
|
|
return super()._execute_on_model(task, model, **kw)
|
|
|
|
def _route_execution_context(self, task, description, task_route):
|
|
if self.failure_point == "context":
|
|
raise RuntimeError("context assembly failed")
|
|
return super()._route_execution_context(task, description, task_route)
|
|
|
|
def _fallback_model(self, description):
|
|
if self.failure_point == "fallback_model":
|
|
raise RuntimeError("fallback model resolution failed")
|
|
return super()._fallback_model(description)
|
|
|
|
def _route_instruction(self, task_route, route_exec):
|
|
if self.failure_point == "route_instruction":
|
|
raise RuntimeError("route instruction failed")
|
|
return super()._route_instruction(task_route, route_exec)
|
|
|
|
def _emit_task_outcome(self, task, channel="", verification=None, route=None):
|
|
if self.failure_point == "post_model_outcome_emit":
|
|
raise RuntimeError("outcome emission failed")
|
|
return super()._emit_task_outcome(task, channel, verification, route)
|
|
|
|
def _save(self):
|
|
if self.failure_point == "durable_save":
|
|
raise RuntimeError("durable state save failed")
|
|
return super()._save()
|
|
|
|
|
|
class TestTaskLifecycle(unittest.TestCase):
|
|
"""Fault injection across pre-model/model/post-model boundaries."""
|
|
|
|
def _make_op(self, failure_point=None, tmp=None):
|
|
op = _FailingOperator(storage_path=str(Path(tmp) / "operator.json"))
|
|
op.failure_point = failure_point
|
|
return op
|
|
|
|
def _assert_terminalized(self, result, task_state_path):
|
|
self.assertFalse(result.get("success"), "task must not succeed")
|
|
task = result.get("task") or {}
|
|
self.assertIn(task.get("status"), ("failed", "research_required"),
|
|
f"task must terminalize, got: {task.get('status')}")
|
|
self.assertTrue(result.get("error"), "failure reason must be populated")
|
|
# Verify checkpoint
|
|
state = json.loads(Path(task_state_path).read_text())
|
|
tasks = {t["task_id"]: t for t in state["tasks"]}
|
|
for t in tasks.values():
|
|
if t.get("status") == "running":
|
|
self.fail(f"orphan RUNNING task: {t['task_id']}")
|
|
# outcome event exists
|
|
events_path = Path(task_state_path).parent / "events.jsonl"
|
|
if events_path.exists():
|
|
outcomes = [json.loads(l) for l in events_path.read_text().splitlines()
|
|
if "task.outcome" in l]
|
|
self.assertTrue(len(outcomes) > 0, "task.outcome event must exist")
|
|
|
|
def test_exception_during_route(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = self._make_op("route", tmp)
|
|
result = op.send_task("Fix the failing test in tests/test_x.py")
|
|
self._assert_terminalized(result, str(Path(tmp) / "hot" / "task_state.json"))
|
|
|
|
def test_exception_during_model_selection(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = self._make_op("model_selection", tmp)
|
|
result = op.send_task("Fix the failing test in tests/test_x.py")
|
|
self._assert_terminalized(result, str(Path(tmp) / "hot" / "task_state.json"))
|
|
|
|
def test_exception_during_execution(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = self._make_op("execute", tmp)
|
|
result = op.send_task("Fix the failing test in tests/test_x.py")
|
|
self._assert_terminalized(result, str(Path(tmp) / "hot" / "task_state.json"))
|
|
|
|
def test_normal_success_not_broken(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = self._make_op(None, tmp)
|
|
with mock.patch.object(op, "_execute_on_model",
|
|
return_value=OperatorResponse(
|
|
success=True, user_text="Done.",
|
|
model="qwen3:8b")):
|
|
result = op.send_task("hello, what models are available?")
|
|
self.assertTrue(result.get("success"), "normal task must succeed")
|
|
|
|
|
|
class TestLifecycleExtended(unittest.TestCase):
|
|
"""Full fault-injection matrix (handoff 2026-09-11 Part D/E).
|
|
|
|
Boundaries: route decision, context assembly, route instruction
|
|
(pre-transport prep), model selection, fallback-model resolution,
|
|
_execute_on_model, post-model processing (outcome emission), durable
|
|
save. Invariant: once a task is active, every exceptional exit leaves
|
|
durable truthful state — never RUNNING without a live owner.
|
|
"""
|
|
|
|
def _make_op(self, failure_point, tmp):
|
|
op = _FailingOperator(storage_path=str(Path(tmp) / "operator.json"))
|
|
op.failure_point = failure_point
|
|
return op
|
|
|
|
def _assert_no_orphan_running(self, tmp):
|
|
state = json.loads((Path(tmp) / "hot" / "task_state.json").read_text())
|
|
for t in state["tasks"]:
|
|
self.assertNotEqual(
|
|
t.get("status"), "running",
|
|
f"orphan RUNNING task after fault injection: {t['task_id']}",
|
|
)
|
|
|
|
def _assert_failed_with_evidence(self, result, tmp, expected_ref=None):
|
|
self.assertFalse(result.get("success"))
|
|
task = result.get("task") or {}
|
|
self.assertEqual(task.get("status"), "failed")
|
|
self.assertTrue(task.get("result"), "failure text must be populated")
|
|
self.assertTrue(task.get("completed_at"), "terminal time must be recorded")
|
|
state = json.loads((Path(tmp) / "hot" / "task_state.json").read_text())
|
|
rec = next(t for t in state["tasks"] if t["task_id"] == task["task_id"])
|
|
self.assertEqual(rec["status"], "failed")
|
|
self.assertTrue(rec.get("outcome_ref"), "durable outcome_ref required")
|
|
if expected_ref:
|
|
self.assertEqual(rec["outcome_ref"], expected_ref)
|
|
self._assert_no_orphan_running(tmp)
|
|
|
|
def test_fault_route_instruction(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
result = self._make_op("route_instruction", tmp).send_task(
|
|
"Fix the failing test in tests/test_x.py")
|
|
self._assert_failed_with_evidence(result, tmp, "pre_model_error")
|
|
|
|
def test_fault_fallback_model(self):
|
|
"""Fallback-model resolution failure must terminalize (this fault
|
|
injection found a real gap: the raise escaped send_task leaving the
|
|
task created/pending with the pointer set)."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = _FailingOperator(storage_path=str(Path(tmp) / "operator.json"))
|
|
op.failure_point = "fallback_model"
|
|
# Router answers None → the else-branch fallback path is taken.
|
|
with mock.patch.object(type(op), "_build_model_router", lambda self: None):
|
|
result = op.send_task("Fix the failing test in tests/test_x.py")
|
|
self._assert_failed_with_evidence(result, tmp, "model_selection_error")
|
|
|
|
def test_fault_post_model_outcome_emit(self):
|
|
"""Outcome-emission failure must not crash send_task or corrupt
|
|
state (the guard's documented contract: journal failure never
|
|
masks the operator path). The task lands in a legitimate owned
|
|
state — awaiting_approval or terminal — never orphan RUNNING."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
result = self._make_op("post_model_outcome_emit", tmp).send_task(
|
|
"Fix the failing test in tests/test_x.py")
|
|
self.assertIn(
|
|
(result.get("task") or {}).get("status"),
|
|
("awaiting_approval", "failed", "completed", "research_required"),
|
|
"emit failure must leave a legitimate owned state",
|
|
)
|
|
self._assert_no_orphan_running(tmp)
|
|
|
|
def test_fault_durable_save_inside_helper(self):
|
|
"""Save failure during terminalization surfaces loudly (no silent
|
|
swallow) but the in-memory task state is still truthful."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = self._make_op(None, tmp)
|
|
calls = {"n": 0}
|
|
real_save = type(op)._save
|
|
def failing_save(self):
|
|
calls["n"] += 1
|
|
if calls["n"] > 1: # first save = task creation; later = terminal
|
|
raise RuntimeError("durable state save failed")
|
|
return real_save(self)
|
|
with mock.patch.object(type(op), "_save", failing_save):
|
|
with self.assertRaises(RuntimeError):
|
|
op.send_task("Fix the failing test in tests/test_x.py")
|
|
task = op.state.tasks[-1]
|
|
self.assertEqual(task.status, "failed", "in-memory state must be truthful even if save raised")
|
|
|
|
def test_restart_reload_of_terminalized_failure(self):
|
|
"""A failed task survives process restart: fresh operator loads
|
|
durable state, sees failed, and no task is RUNNING without owner."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op1 = self._make_op("execute", tmp)
|
|
result = op1.send_task("Fix the failing test in tests/test_x.py")
|
|
task_id = result["task"]["task_id"]
|
|
# Fresh instance = restart simulation
|
|
op2 = _FailingOperator(storage_path=str(Path(tmp) / "operator.json"))
|
|
state = json.loads((Path(tmp) / "hot" / "task_state.json").read_text())
|
|
rec = next(t for t in state["tasks"] if t["task_id"] == task_id)
|
|
self.assertEqual(rec["status"], "failed")
|
|
self._assert_no_orphan_running(tmp)
|
|
# The active-channel pointer must not point at the dead task
|
|
self.assertNotIn("task", op2.state.current_task or {})
|
|
|
|
def test_terminalization_helper_single_path(self):
|
|
"""The consolidated helper preserves: task state, durable
|
|
task_state, outcome event, current-task pointer, error text."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
op = self._make_op("execute", tmp)
|
|
result = op.send_task("Fix the failing test in tests/test_x.py")
|
|
task_id = result["task"]["task_id"]
|
|
self.assertIsNone(op.state.current_task, "current-task pointer cleared")
|
|
self.assertEqual(result["error"], "Execution error: model execution failed")
|
|
events_path = Path(tmp) / "events.jsonl"
|
|
self.assertTrue(events_path.exists())
|
|
# task.outcome rides the event_bus journal (memory_dir/journal),
|
|
# not the operator's task-transition events.jsonl.
|
|
from rune import event_bus as _eb
|
|
journal_dir = Path(_eb.JOURNAL_DIR)
|
|
outcomes = []
|
|
if journal_dir.exists():
|
|
for j in journal_dir.glob("*.jsonl"):
|
|
outcomes += [json.loads(l) for l in j.read_text().splitlines()
|
|
if "task.outcome" in l]
|
|
self.assertTrue(outcomes, "task.outcome emitted by helper")
|
|
state = json.loads((Path(tmp) / "hot" / "task_state.json").read_text())
|
|
rec = next(t for t in state["tasks"] if t["task_id"] == task_id)
|
|
self.assertEqual(rec["status"], "failed")
|
|
self.assertEqual(rec["outcome_ref"], "execution_error")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|