"""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: # Isolation law: the event-bus journal dir is a module-level # constant resolved from the ambient RUNE_HOME at import time # (memory_env.memory_dir() → hardcoded default). Without # pinning it here, the test reads whatever journal the host # resolves to — on a CI runner the default path is # unwritable/nonexistent, emit()'s deliberate degradation # swallows the write, and the scan below finds nothing. from rune import event_bus as _eb orig_journal = _eb.JOURNAL_DIR orig_event_log = _eb.EVENT_LOG journal_dir = Path(tmp) / "journal" _eb.JOURNAL_DIR = str(journal_dir) _eb.EVENT_LOG = str(journal_dir / "events.jsonl") self.addCleanup(setattr, _eb, "JOURNAL_DIR", orig_journal) self.addCleanup(setattr, _eb, "EVENT_LOG", orig_event_log) 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. 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()