File size: 2,059 Bytes
a4514b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 | from clfgui.kernel import LocalAgentKernel, SQLiteStatusStore
def make_kernel(tmp_path):
return LocalAgentKernel(SQLiteStatusStore(tmp_path / "kernel.sqlite3"))
def test_kernel_runs_from_intent_to_approval(tmp_path):
kernel = make_kernel(tmp_path)
job_id = kernel.submit_intent("Build the CLFGUI V1")
snapshot = kernel.get_job_status(job_id)
assert snapshot.job.status == "waiting_decision"
assert snapshot.result is None
assert [decision.title for decision in kernel.list_open_decisions(job_id)] == ["Plan freigeben"]
decision = kernel.list_open_decisions(job_id)[0]
after_answer = kernel.answer_decision(decision.id, "Plan freigeben", "Go")
assert after_answer.job.status == "awaiting_approval"
assert after_answer.result is not None
assert after_answer.result.approved is None
assert all(phase.status == "completed" for phase in after_answer.phases)
approved = kernel.approve_result(job_id, True)
assert approved.job.status == "approved"
assert approved.result is not None
assert approved.result.approved is True
def test_kernel_blocks_job_at_human_gate(tmp_path):
kernel = make_kernel(tmp_path)
job_id = kernel.submit_intent("Risky task")
decision = kernel.list_open_decisions(job_id)[0]
snapshot = kernel.answer_decision(decision.id, "Auftrag blockieren", "Need owner input")
assert snapshot.job.status == "blocked"
assert snapshot.result is not None
assert snapshot.result.blocked is True
assert kernel.list_open_decisions(job_id) == []
def test_kernel_can_request_plan_adjustment(tmp_path):
kernel = make_kernel(tmp_path)
job_id = kernel.submit_intent("Do something broad")
decision = kernel.list_open_decisions(job_id)[0]
snapshot = kernel.answer_decision(decision.id, "Plan anpassen", "Make it smaller")
assert snapshot.job.status == "waiting_decision"
open_decisions = kernel.list_open_decisions(job_id)
assert len(open_decisions) == 1
assert open_decisions[0].title == "Angepassten Plan freigeben"
|