mirror of
https://github.com/razzant/ouroboros.git
synced 2026-08-22 09:13:21 +00:00
24394dc0d1
This RC makes review evidence first-class across repo and skill review: skill findings are no longer silently truncated, review.json is readable cognitive evidence, and skill review status is computed live from findings plus the current enforcement mode. It also fixes the SKILL_NOT_FINALIZED transcript bug, adds LLM-first recent task context recovery, hardens owner-state protection for skill review history, removes MagicMock JSONL pollution, and updates the release/docs/UI surfaces for the new review flow.
108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
import json
|
|
import os
|
|
from types import SimpleNamespace
|
|
|
|
from ouroboros.tools.recent_tasks import _handle_recent_tasks
|
|
|
|
|
|
def _ctx(tmp_path):
|
|
return SimpleNamespace(drive_root=tmp_path)
|
|
|
|
|
|
def _write_task(root, name, *, result="done", ts="2026-01-01T00:00:00Z", **extra):
|
|
path = root / "task_results" / f"{name}.json"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = {
|
|
"task_id": name,
|
|
"ts": ts,
|
|
"status": "completed",
|
|
"description": f"task {name}",
|
|
"cost_usd": 1.25,
|
|
"total_rounds": 3,
|
|
"result": result,
|
|
**extra,
|
|
}
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_recent_tasks_empty_drive(tmp_path):
|
|
data = json.loads(_handle_recent_tasks(_ctx(tmp_path)))
|
|
assert data == {"running": [], "tasks": [], "unreadable_tasks": []}
|
|
|
|
|
|
def test_recent_tasks_returns_preview_by_default(tmp_path):
|
|
_write_task(tmp_path, "abc123", result="hello world")
|
|
|
|
data = json.loads(_handle_recent_tasks(_ctx(tmp_path)))
|
|
|
|
assert data["tasks"][0]["task_id"] == "abc123"
|
|
assert data["tasks"][0]["result_preview"] == "hello world"
|
|
assert "result" not in data["tasks"][0]
|
|
assert "trace_summary" not in data["tasks"][0]
|
|
|
|
|
|
def test_recent_tasks_include_results_and_traces(tmp_path):
|
|
_write_task(tmp_path, "abc123", result="full result", trace_summary="trace")
|
|
|
|
data = json.loads(_handle_recent_tasks(
|
|
_ctx(tmp_path),
|
|
include_results=True,
|
|
include_traces=True,
|
|
))
|
|
|
|
assert data["tasks"][0]["result"] == "full result"
|
|
assert data["tasks"][0]["trace_summary"] == "trace"
|
|
|
|
|
|
def test_recent_tasks_sorted_by_mtime(tmp_path):
|
|
older = _write_task(tmp_path, "older")
|
|
newer = _write_task(tmp_path, "newer")
|
|
os.utime(older, (100, 100))
|
|
os.utime(newer, (200, 200))
|
|
|
|
data = json.loads(_handle_recent_tasks(_ctx(tmp_path)))
|
|
|
|
assert [row["task_id"] for row in data["tasks"]] == ["newer", "older"]
|
|
|
|
|
|
def test_recent_tasks_limit_is_clamped(tmp_path):
|
|
for idx in range(25):
|
|
path = _write_task(tmp_path, f"task{idx}")
|
|
os.utime(path, (idx, idx))
|
|
|
|
data = json.loads(_handle_recent_tasks(_ctx(tmp_path), limit=99))
|
|
|
|
assert len(data["tasks"]) == 20
|
|
|
|
|
|
def test_recent_tasks_reports_malformed_json(tmp_path):
|
|
_write_task(tmp_path, "good")
|
|
bad = tmp_path / "task_results" / "bad.json"
|
|
bad.write_text("{not json", encoding="utf-8")
|
|
os.utime(bad, (999, 999))
|
|
|
|
data = json.loads(_handle_recent_tasks(_ctx(tmp_path), limit=10))
|
|
|
|
assert [row["task_id"] for row in data["tasks"]] == ["good"]
|
|
assert len(data["unreadable_tasks"]) == 1
|
|
assert data["unreadable_tasks"][0]["path"].endswith("bad.json")
|
|
assert "JSONDecodeError" in data["unreadable_tasks"][0]["error"]
|
|
|
|
|
|
def test_recent_tasks_reports_running_queue(tmp_path):
|
|
state = tmp_path / "state"
|
|
state.mkdir()
|
|
(state / "queue_snapshot.json").write_text(json.dumps({
|
|
"ts": "2026-01-01T00:00:00Z",
|
|
"running": [{"id": "run1", "text": "keep going"}],
|
|
}), encoding="utf-8")
|
|
|
|
data = json.loads(_handle_recent_tasks(_ctx(tmp_path)))
|
|
|
|
assert data["running"] == [{
|
|
"task_id": "run1",
|
|
"status": "running",
|
|
"description": "keep going",
|
|
"ts": "2026-01-01T00:00:00Z",
|
|
}]
|