File size: 6,165 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 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | from __future__ import annotations
from .backends import AgentBackend, FakeAgentBackend
from .models import Decision, JobSnapshot
from .store import SQLiteStatusStore
PHASES = ("plan", "research", "code", "test", "verify", "result")
class LocalAgentKernel:
"""Small in-process orchestration boundary for V1.
The public methods are intentionally daemon-shaped. The GUI should keep
calling this facade when the implementation later moves behind IPC.
"""
def __init__(self, store: SQLiteStatusStore | None = None, agent_backend: AgentBackend | None = None) -> None:
self.store = store or SQLiteStatusStore()
self.agent_backend = agent_backend or FakeAgentBackend()
def submit_intent(self, text: str) -> int:
intent = text.strip()
if not intent:
raise ValueError("Intent must not be empty.")
job = self.store.create_job(intent)
self.store.create_phases(job.id, PHASES)
self.store.update_phase(job.id, "plan", "running", "Intent wird in einen Arbeitsplan verdichtet.")
self.store.update_phase(
job.id,
"plan",
"completed",
f"Plan erstellt: {intent}",
)
self.store.update_phase(
job.id,
"research",
"waiting",
"Wartet auf menschliche Freigabe des verdichteten Plans.",
)
self.store.create_decision(
job_id=job.id,
phase_name="plan",
title="Plan freigeben",
context=(
"Der Kernel hat den Intent in einen kleinen V1-Plan übersetzt. "
"Die nächsten Phasen laufen automatisch, wenn der Plan freigegeben wird."
),
options=["Plan freigeben", "Plan anpassen", "Auftrag blockieren"],
)
self.store.update_job_status(job.id, "waiting_decision")
return job.id
def get_job_status(self, job_id: int) -> JobSnapshot:
return self.store.snapshot(job_id)
def list_jobs(self):
return self.store.list_jobs()
def list_open_decisions(self, job_id: int) -> list[Decision]:
return self.store.list_open_decisions(job_id)
def answer_decision(self, decision_id: int, selected_option: str, note: str = "") -> JobSnapshot:
answered = self.store.answer_decision(decision_id, selected_option, note)
if selected_option == "Auftrag blockieren":
self._block_job(answered.job_id, note)
return self.store.snapshot(answered.job_id)
if selected_option == "Plan anpassen":
self._request_plan_adjustment(answered.job_id, note)
return self.store.snapshot(answered.job_id)
self._run_remaining_phases(answered.job_id)
return self.store.snapshot(answered.job_id)
def approve_result(self, job_id: int, approved: bool) -> JobSnapshot:
self.store.set_result_approval(job_id, approved)
self.store.update_job_status(job_id, "approved" if approved else "rejected")
return self.store.snapshot(job_id)
def _request_plan_adjustment(self, job_id: int, note: str) -> None:
adjustment = note.strip() or "Bitte Plan vor Ausführung anpassen."
self.store.update_phase(job_id, "plan", "waiting", "Plananpassung angefordert.")
self.store.create_decision(
job_id=job_id,
phase_name="plan",
title="Angepassten Plan freigeben",
context=(
"Gewünschte Anpassung: "
f"{adjustment} "
"Der Skeleton-Kernel hält den Auftrag, bis diese Anpassung freigegeben wird."
),
options=["Plan freigeben", "Auftrag blockieren"],
)
self.store.update_job_status(job_id, "waiting_decision")
def _block_job(self, job_id: int, note: str) -> None:
reason = note.strip() or "Auftrag wurde an einem Human-Gate blockiert."
for phase in self.store.list_phases(job_id):
if phase.status in {"pending", "waiting", "running"}:
self.store.update_phase(job_id, phase.name, "blocked", reason)
self.store.save_result(
job_id=job_id,
summary="Der Auftrag wurde vor der automatischen Ausführung blockiert.",
blocked=True,
decisions=self._decision_summaries(job_id),
next_steps=["Blocker klären", "Auftrag mit geschärftem Ziel neu starten"],
)
self.store.update_job_status(job_id, "blocked")
def _run_remaining_phases(self, job_id: int) -> None:
job = self.store.get_job(job_id)
for phase_name in PHASES[1:]:
phase = self.store.update_phase(job_id, phase_name, "running", f"{phase_name} läuft.")
outcome = self.agent_backend.run_phase(
job,
phase,
context="Current CLFGUI V1 skeleton flow. Return a concise phase summary only.",
)
self.store.update_phase(job_id, phase_name, outcome.status, outcome.summary)
self.store.save_result(
job_id=job_id,
summary=(
"V1-Skeleton abgeschlossen: Intent aufgenommen, Plan freigegeben, "
"Phasen abgearbeitet und Ergebnispaket erzeugt."
),
blocked=False,
decisions=self._decision_summaries(job_id),
next_steps=[
"Echte Agent-Adapter für Research/Code/Test anbinden",
"Kernel bei Bedarf in separaten Headless-Daemon auslagern",
f"Folgeauftrag für '{job.intent}' starten",
],
)
self.store.update_job_status(job_id, "awaiting_approval")
def _decision_summaries(self, job_id: int) -> list[str]:
summaries: list[str] = []
for decision in self.store.list_decisions(job_id):
if decision.selected_option:
note = f" ({decision.note})" if decision.note else ""
summaries.append(f"{decision.title}: {decision.selected_option}{note}")
else:
summaries.append(f"{decision.title}: offen")
return summaries
|