286 lines
14 KiB
Python
286 lines
14 KiB
Python
"""Event-Tracking (events-Tabelle) + Agenten-Labels."""
|
|
|
|
import asyncio
|
|
|
|
import agents
|
|
from pipeline import GenContext
|
|
|
|
TOPIC = "t"
|
|
|
|
|
|
async def _events(db, kind=None):
|
|
conn = await db.get_db()
|
|
q = "SELECT topic, kind, key, label, status, dur_ms, wait_ms FROM events WHERE topic = ?"
|
|
args = [TOPIC]
|
|
if kind:
|
|
q += " AND kind = ?"
|
|
args.append(kind)
|
|
cur = await conn.execute(q, args)
|
|
return [dict(zip(("topic", "kind", "key", "label", "status", "dur_ms", "wait_ms"), r))
|
|
for r in await cur.fetchall()]
|
|
|
|
|
|
async def test_advance_many_writes_stage_events(testdb):
|
|
db = testdb
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "a", "block", "s1")
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "b", "block", "s1")
|
|
await db.kanban_advance_many(TOPIC, "inventory", [("a", "s2"), ("b", "s2")])
|
|
evs = await _events(db, "stage")
|
|
assert {(e["key"], e["status"]) for e in evs} == {("inventory:a", "s2"), ("inventory:b", "s2")}
|
|
|
|
|
|
async def test_fail_card_events_retry_then_dead(testdb):
|
|
db = testdb
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "a", "block", "s1")
|
|
assert await db.kanban_fail_card(TOPIC, "inventory", "a", "boom", max_retries=2) is False
|
|
assert await db.kanban_fail_card(TOPIC, "inventory", "a", "boom", max_retries=2) is True
|
|
evs = await _events(db, "fail")
|
|
assert [e["status"] for e in evs] == ["retry1", "dead"]
|
|
|
|
|
|
async def test_guide_stage_event(testdb):
|
|
db = testdb
|
|
await db.upsert_guide_card(TOPIC, "Guide", "alpha", "Alpha")
|
|
await db.set_guide_card(TOPIC, "Guide", "alpha", stage="writer")
|
|
evs = await _events(db, "stage")
|
|
assert evs and evs[-1]["key"] == "guide:Guide:alpha" and evs[-1]["status"] == "writer"
|
|
|
|
|
|
async def test_run_agent_emits_event_and_survives_broken_sink(testdb, monkeypatch):
|
|
recorded = []
|
|
|
|
async def sink(**kw):
|
|
recorded.append(kw)
|
|
|
|
async def fake_cli(agent_key, prompt, timeout, model, capabilities, label=""):
|
|
return 0, "out", ""
|
|
|
|
monkeypatch.setattr(agents, "on_event", sink)
|
|
monkeypatch.setattr(agents, "_run_claude_cli", fake_cli)
|
|
monkeypatch.setattr(agents.shutil, "which", lambda c: "/bin/true")
|
|
monkeypatch.setattr(agents, "resolve_role", lambda p, r: ("claude", "test-model"))
|
|
rc, out, err = await agents.run_agent("blocks-t-x", "p", 5, provider="claude",
|
|
role="judge", scope=TOPIC, label="Alpha · Judge")
|
|
assert rc == 0
|
|
assert recorded and recorded[0]["kind"] == "agent"
|
|
assert recorded[0]["label"] == "Alpha · Judge" and recorded[0]["status"] == "ok"
|
|
assert isinstance(recorded[0]["wait_ms"], int) and isinstance(recorded[0]["dur_ms"], int)
|
|
|
|
# broken sink never breaks the call; interactive/scope-less calls don't log
|
|
async def broken(**kw):
|
|
raise RuntimeError("sink down")
|
|
monkeypatch.setattr(agents, "on_event", broken)
|
|
rc, _, _ = await agents.run_agent("blocks-t-y", "p", 5, provider="claude", scope=TOPIC)
|
|
assert rc == 0
|
|
monkeypatch.setattr(agents, "on_event", sink)
|
|
recorded.clear()
|
|
await agents.run_agent("chat-1", "p", 5, provider="claude", lane="interactive")
|
|
assert recorded == []
|
|
|
|
|
|
async def test_active_agents_carry_labels():
|
|
async def run(key, label):
|
|
return await agents._communicate(key, ["sleep", "0.4"], None, 5, label=label)
|
|
|
|
t1 = asyncio.create_task(run("blocks-t-x", "Alpha · Facts 1"))
|
|
t2 = asyncio.create_task(run("blocks-t-x", "Alpha · Facts 2")) # key collision → ~2
|
|
await asyncio.sleep(0.15)
|
|
agents_now = agents.active_agents("blocks-t-")
|
|
assert sorted(a["label"] for a in agents_now) == ["Alpha · Facts 1", "Alpha · Facts 2"]
|
|
assert {a["key"] for a in agents_now} == {"blocks-t-x", "blocks-t-x~2"}
|
|
await asyncio.gather(t1, t2)
|
|
assert agents.active_agents("blocks-t-") == []
|
|
|
|
async def test_pull_prefers_bigger_blocks(testdb):
|
|
"""LPT: Karten mit größerem subs_n werden zuerst gezogen; ohne Feld bleibt FIFO."""
|
|
db = testdb
|
|
await db.kanban_upsert_card(TOPIC, "artefacts", "klein", "ablock", "facts", {"subs_n": 5})
|
|
await db.kanban_upsert_card(TOPIC, "artefacts", "gross", "ablock", "facts", {"subs_n": 40})
|
|
await db.kanban_upsert_card(TOPIC, "artefacts", "mittel", "ablock", "facts", {"subs_n": 15})
|
|
pulled = await db.kanban_pull(TOPIC, "artefacts", "facts", 10)
|
|
assert [c["card_id"] for c in pulled] == ["gross", "mittel", "klein"]
|
|
# ohne subs_n: FIFO nach updated_at
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "a", "block", "s1")
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "b", "block", "s1")
|
|
pulled = await db.kanban_pull(TOPIC, "inventory", "s1", 10)
|
|
assert [c["card_id"] for c in pulled] == ["a", "b"]
|
|
# n_size (Board 1) ist der Fallback-Schätzer; subs_n behält Vorrang
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "n-klein", "block", "s2", {"n_size": 2})
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "n-gross", "block", "s2", {"n_size": 9})
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "n-ohne", "block", "s2")
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "n-subs", "block", "s2", {"subs_n": 3, "n_size": 1})
|
|
pulled = await db.kanban_pull(TOPIC, "inventory", "s2", 10)
|
|
assert [c["card_id"] for c in pulled] == ["n-gross", "n-subs", "n-klein", "n-ohne"]
|
|
|
|
|
|
async def test_learnstate_smoke(testdb):
|
|
"""Regression: P5-Ausbau hatte die _LEVEL_CASE-Konstante mitgerissen —
|
|
load_learnstate (Guide-Start-Pfad) muss ohne NameError laufen."""
|
|
from rules import load_learnstate
|
|
guides, progress, levels = await load_learnstate()
|
|
assert isinstance(levels, dict)
|
|
|
|
|
|
async def test_guide_error_event(testdb):
|
|
db = testdb
|
|
await db.upsert_guide_card(TOPIC, "Guide", "alpha", "Alpha")
|
|
await db.set_guide_card(TOPIC, "Guide", "alpha", status="error", gate_info="Writer ohne Ergebnis")
|
|
evs = await _events(db, "fail")
|
|
assert evs and evs[-1]["key"] == "guide:Guide:alpha" and "Writer" in evs[-1]["status"]
|
|
|
|
|
|
def test_timeout_calibration_smoke():
|
|
from pipeline import _timeout
|
|
assert _timeout("subblock", 10) == 400 + 150
|
|
assert _timeout("content", 10) == 450 + 300
|
|
|
|
|
|
def test_env_file_wins(tmp_path, monkeypatch):
|
|
"""Regression: geerbte (veraltete) Env-Werte dürfen die .env nicht mehr überstimmen."""
|
|
import config
|
|
monkeypatch.setenv("X_CREATOR_TESTKEY", "alt")
|
|
p = tmp_path / ".env"
|
|
p.write_text("X_CREATOR_TESTKEY=neu\n", encoding="utf-8")
|
|
config._load_env(p)
|
|
import os
|
|
assert os.environ["X_CREATOR_TESTKEY"] == "neu"
|
|
|
|
|
|
async def test_restart_artefact_card_wipes_only_that_block(testdb):
|
|
import board_inventory as bi
|
|
db = testdb
|
|
for norm in ("alpha", "beta"):
|
|
await db.kanban_upsert_card(TOPIC, "artefacts", norm, "ablock", "done_artefact",
|
|
{"title": norm.title(), "raw": {norm: ["S"]}, "facts": {}})
|
|
await db.upsert_subblock(TOPIC, norm, "s1", norm.title(), "Sub Eins")
|
|
await db.upsert_question_pattern(TOPIC, norm, "s1", norm.title(), "Sub Eins", "Frage?")
|
|
await db.put_sub_artifact(TOPIC, norm, "s1", "flashcard", norm.title(), "Sub Eins", "{}")
|
|
assert await bi.restart_artefact_card(TOPIC, "alpha") is True
|
|
assert (await db.kanban_get_card(TOPIC, "artefacts", "alpha"))["stage"] == "subblocks"
|
|
assert await db.list_subblocks(TOPIC, "alpha") == []
|
|
assert len(await db.list_subblocks(TOPIC, "beta")) == 1 # untouched
|
|
assert await bi.restart_artefact_card(TOPIC, "gibtsnicht") is False
|
|
|
|
|
|
async def test_guide_reset_card_single(testdb):
|
|
import guide_board as gb
|
|
db = testdb
|
|
for n in ("alpha", "beta"):
|
|
await db.upsert_guide_card(TOPIC, "Guide", n, n.title())
|
|
await db.set_guide_card(TOPIC, "Guide", n, stage="done", status="ok",
|
|
writer_rounds=2, md="# SECTION Text", gate_info="x")
|
|
await db.put_lernziel(TOPIC, n, "z1", "Ziel eins")
|
|
assert await gb.reset_card(TOPIC, "Guide", "alpha", 0) is True
|
|
cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, "Guide")}
|
|
assert cards["alpha"]["stage"] == "lernziele" and cards["alpha"]["md"] == "" and cards["alpha"]["writer_rounds"] == 0
|
|
assert cards["beta"]["stage"] == "done" and cards["beta"]["md"] # untouched
|
|
assert await db.list_lernziele(TOPIC) and all(z["block_norm"] != "alpha" for z in await db.list_lernziele(TOPIC))
|
|
# ab_stage 3 (fakten_gate) behält md
|
|
assert await gb.reset_card(TOPIC, "Guide", "beta", 3) is True
|
|
cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, "Guide")}
|
|
assert cards["beta"]["stage"] == "fakten_gate" and cards["beta"]["md"]
|
|
|
|
|
|
async def test_completeness_route(testdb, tmp_path, monkeypatch):
|
|
import routes, paths
|
|
db = testdb
|
|
monkeypatch.setattr(paths, "arbeit_dir", lambda t: tmp_path)
|
|
await db.upsert_block(TOPIC, "alpha", "Alpha", "d", "[]")
|
|
await db.set_block_status(TOPIC, "alpha", "consensus")
|
|
await db.upsert_subblock(TOPIC, "alpha", "s1", "Alpha", "Sub Eins")
|
|
await db.set_subblock_fields(TOPIC, "alpha", "s1", status="consensus")
|
|
await db.upsert_question_pattern(TOPIC, "alpha", "s1", "Alpha", "Sub Eins", "Frage?")
|
|
await db.put_sub_artifact(TOPIC, "alpha", "s1", "flashcard", "Alpha", "Sub Eins", "{}")
|
|
await db.put_lernziel(TOPIC, "alpha", "z1", "Ziel")
|
|
await db.set_ziel_covered(TOPIC, "alpha", "z1", True)
|
|
(tmp_path / "inventar-filter-x.json").write_text(
|
|
'{"degradiert": 3, "ueberstimmt": ["A"], "floor_veto": []}', encoding="utf-8")
|
|
res = await routes.blocks_completeness(TOPIC)
|
|
assert res["bloecke"] == 1 and res["subs"] == 1
|
|
assert res["frage_bloecke"] == 1 and res["lernkarten"] == 1
|
|
assert res["ziele_total"] == 1 and res["ziele_covered"] == 1
|
|
assert res["degradiert_geprueft"] == 3 and res["panel_gerettet"] == 1
|
|
assert res["dead"] == 0
|
|
|
|
|
|
async def test_blocks_ready_from_db(testdb, monkeypatch):
|
|
"""Regression: gesynctes Topic ohne blocks.md muss trotzdem ready sein (DB zählt)."""
|
|
import blocks as blx
|
|
db = testdb
|
|
await db.kanban_upsert_card(TOPIC, "inventory", "b-1", "block", "done_block", {"title": "Alpha"})
|
|
st = await blx.blocks_status(TOPIC)
|
|
assert st["ready"] is True and st["partial"] is False
|
|
|
|
|
|
async def test_remove_guide_format_clears_everything(testdb, monkeypatch):
|
|
"""Board-Remove räumt ALLE Läufe eines Formats + Karten (8 error-Zeilen stapelten sich)."""
|
|
import routes
|
|
from models import GuideFormatRequest
|
|
db = testdb
|
|
for i in range(3):
|
|
await db.create_guide({"id": f"g{i}", "topic": TOPIC, "format": "Guide",
|
|
"instructions": "", "status": "error", "progress": None,
|
|
"created_at": "2026-01-01", "updated_at": "2026-01-01"})
|
|
await db.upsert_guide_card(TOPIC, "Guide", "alpha", "Alpha")
|
|
res = await routes.remove_guide_format(GuideFormatRequest(topic=TOPIC, format="Guide"))
|
|
assert res["removed"] == 3
|
|
assert await db.list_guides() == [] or all(g["topic"] != TOPIC for g in await db.list_guides())
|
|
assert await db.list_guide_cards(TOPIC, "Guide") == []
|
|
|
|
|
|
# ── Token-Logging pro Agent (OpenCode-Session → Event-Meta) ──────────────────────────
|
|
|
|
def test_session_tokens_reads_newest(tmp_path, monkeypatch):
|
|
"""--title = Agent-Key: neueste Session gewinnt (Retry); fehlende DB/Zeile → None."""
|
|
import sqlite3
|
|
dbf = tmp_path / "oc.db"
|
|
con = sqlite3.connect(dbf)
|
|
con.execute("CREATE TABLE session (title TEXT, time_created INT, tokens_input INT,"
|
|
" tokens_output INT, tokens_reasoning INT, tokens_cache_read INT, tokens_cache_write INT)")
|
|
con.execute("INSERT INTO session VALUES ('k', 1, 1, 1, 0, 10, 0)")
|
|
con.execute("INSERT INTO session VALUES ('k', 2, 7, 3, 0, 99, 5)")
|
|
con.commit()
|
|
con.close()
|
|
monkeypatch.setattr(agents, "_OPENCODE_DB", dbf)
|
|
assert agents._session_tokens("k") == {"input": 7, "output": 3, "reasoning": 0,
|
|
"cache_read": 99, "cache_write": 5}
|
|
assert agents._session_tokens("fehlt") is None
|
|
monkeypatch.setattr(agents, "_OPENCODE_DB", tmp_path / "nope.db")
|
|
assert agents._session_tokens("k") is None
|
|
|
|
|
|
async def test_opencode_cmd_sets_title(monkeypatch):
|
|
"""Der Agent-Key wird Session-Titel — der Join-Schlüssel fürs Token-Logging."""
|
|
seen = {}
|
|
|
|
async def fake_comm(agent_key, cmd, stdin, timeout, stagger=False, on_line=None, label="", env=None):
|
|
seen["cmd"] = cmd
|
|
return 0, "", ""
|
|
|
|
monkeypatch.setattr(agents, "_communicate", fake_comm)
|
|
await agents._run_opencode("blocks-t-z", "p", 5, "minimax", "m", "none")
|
|
i = seen["cmd"].index("--title")
|
|
assert seen["cmd"][i + 1] == "blocks-t-z"
|
|
|
|
|
|
async def test_run_agent_logs_opencode_tokens(testdb, monkeypatch):
|
|
"""OpenCode-Lauf: Token-Zähler der Session landen im Event-Meta."""
|
|
recorded = []
|
|
|
|
async def sink(**kw):
|
|
recorded.append(kw)
|
|
|
|
async def fake_oc(agent_key, prompt, timeout, provider, model, capabilities, on_line=None, label=""):
|
|
return 0, "out", ""
|
|
|
|
monkeypatch.setattr(agents, "on_event", sink)
|
|
monkeypatch.setattr(agents, "_run_opencode", fake_oc)
|
|
monkeypatch.setattr(agents.shutil, "which", lambda c: "/bin/true")
|
|
monkeypatch.setattr(agents, "resolve_role", lambda p, r: ("minimax", "test-model"))
|
|
monkeypatch.setattr(agents, "_session_tokens",
|
|
lambda k: {"input": 5, "output": 2, "reasoning": 0,
|
|
"cache_read": 100, "cache_write": 0})
|
|
rc, *_ = await agents.run_agent("blocks-t-tok", "p", 5, provider="minimax", scope=TOPIC)
|
|
assert rc == 0
|
|
assert recorded and recorded[0]["meta"]["tokens"]["cache_read"] == 100
|