51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
"""Smoke checks for durable loop-harness primitives."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from pipeline.harness_loop import CostBudget, append_event, read_events, run_one_cycle
|
|
|
|
|
|
def main() -> int:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tasks = os.path.join(tmp, "tasks.json")
|
|
events = os.path.join(tmp, "events.jsonl")
|
|
with open(tasks, "w", encoding="utf-8") as f:
|
|
json.dump([{"id": "1", "title": "first"}, {"id": "2", "title": "second"}], f)
|
|
|
|
receipt = run_one_cycle(tasks, events, "run-1", CostBudget(max_tokens=100), estimate_tokens=25)
|
|
assert receipt["status"] == "claimed"
|
|
assert receipt["task"]["id"] == "1"
|
|
assert receipt["budget"]["used_tokens"] == 25
|
|
assert read_events(events)[0].type == "claimed"
|
|
|
|
with open(tasks, "r", encoding="utf-8") as f:
|
|
saved = json.load(f)
|
|
assert saved[0]["status"] == "claimed"
|
|
assert saved[1]["status"] == "todo"
|
|
|
|
blocked = run_one_cycle(tasks, events, "run-1", CostBudget(max_tokens=10), estimate_tokens=25)
|
|
assert blocked == {"status": "blocked", "reason": "token_budget"}
|
|
assert read_events(events)[-1].type == "blocked"
|
|
|
|
saved[1]["status"] = "claimed"
|
|
with open(tasks, "w", encoding="utf-8") as f:
|
|
json.dump(saved, f)
|
|
done = run_one_cycle(tasks, events, "run-1", CostBudget(max_tokens=100), estimate_tokens=0)
|
|
assert done == {"status": "done"}
|
|
assert read_events(events)[-1].type == "done"
|
|
|
|
print("✓ durable loop harness")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|