| 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" | |