Nick
b81edb2029
test_terminalization_helper_single_path scanned the ambient event_bus.JOURNAL_DIR — an import-time-frozen constant resolved via memory_env.memory_dir(), whose RUNE_HOME defaults to the hardcoded /mnt/workdrive/rune. On this machine that path coincidentally IS the repo, so the test passed locally; on the CI runner the default path is unwritable, emit()'s deliberate degradation swallowed the write, and the scan found no task.outcome — the sole unit red of 2,219 on run 13155. Fix pins JOURNAL_DIR/EVENT_LOG to the test's TemporaryDirectory with addCleanup restore — same isolation-seam family as the hot-store law (ambient state must never be read by tests). Red-first repro'd with a hostile RUNE_HOME locally: exact CI signature ([] is not true) before the pin, green after.
275 lines
13 KiB
Python
275 lines
13 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:
|
|
# 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()
|