| from __future__ import annotations |
|
|
| import subprocess |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Protocol |
|
|
| from .models import Job, Phase |
|
|
|
|
| @dataclass(frozen=True) |
| class PhaseDecision: |
| title: str |
| context: str |
| options: list[str] |
|
|
|
|
| @dataclass(frozen=True) |
| class PhaseResult: |
| summary: str |
| blocked: bool |
| decisions: list[str] |
| next_steps: list[str] |
|
|
|
|
| @dataclass(frozen=True) |
| class PhaseOutcome: |
| status: str |
| summary: str |
| decision: PhaseDecision | None = None |
| result: PhaseResult | None = None |
|
|
|
|
| class AgentBackend(Protocol): |
| def run_phase(self, job: Job, phase: Phase, context: str) -> PhaseOutcome: |
| ... |
|
|
|
|
| class FakeAgentBackend: |
| """Deterministic backend that preserves the current V1 skeleton behavior.""" |
|
|
| _SUMMARIES = { |
| "research": "Relevante Annahmen und offene Risiken wurden für V1 verdichtet.", |
| "code": "Skeleton-Implementierung wurde als geplanter Arbeitsschritt abgeschlossen.", |
| "test": "Automatische V1-Prüfungen wurden als Schritt eingeplant und ausgewertet.", |
| "verify": "Ergebnis wurde gegen Intent, Human-Gates und V1-Scope geprüft.", |
| "result": "Verdichtetes Ergebnispaket wurde erstellt.", |
| } |
|
|
| def run_phase(self, job: Job, phase: Phase, context: str) -> PhaseOutcome: |
| summary = self._SUMMARIES.get(phase.name, f"{phase.name} abgeschlossen.") |
| return PhaseOutcome(status="completed", summary=summary) |
|
|
|
|
| class HermesAgentBackend: |
| """Small subprocess adapter for future Hermes-backed phases. |
| |
| This is intentionally not the default backend. It asks Hermes for a concise |
| phase summary and explicitly forbids mutating work. |
| """ |
|
|
| def __init__( |
| self, |
| executable: str = "hermes", |
| timeout_s: int = 120, |
| cwd: str | Path | None = None, |
| source: str = "clfgui", |
| max_turns: int = 1, |
| ) -> None: |
| self.executable = executable |
| self.timeout_s = timeout_s |
| self.cwd = Path(cwd) if cwd is not None else None |
| self.source = source |
| self.max_turns = max_turns |
|
|
| def run_phase(self, job: Job, phase: Phase, context: str) -> PhaseOutcome: |
| command = self.build_command(job, phase, context) |
| try: |
| completed = subprocess.run( |
| command, |
| cwd=str(self.cwd) if self.cwd is not None else None, |
| capture_output=True, |
| text=True, |
| timeout=self.timeout_s, |
| check=False, |
| ) |
| except subprocess.TimeoutExpired: |
| return PhaseOutcome( |
| status="failed", |
| summary=f"Hermes phase timed out after {self.timeout_s} seconds.", |
| ) |
|
|
| if completed.returncode == 0: |
| summary = completed.stdout.strip() or "Hermes returned no output." |
| return PhaseOutcome(status="completed", summary=summary) |
|
|
| message = self._short_output(completed.stderr) or self._short_output(completed.stdout) or "no output" |
| return PhaseOutcome( |
| status="failed", |
| summary=f"Hermes phase failed (exit {completed.returncode}): {message}", |
| ) |
|
|
| def build_command(self, job: Job, phase: Phase, context: str) -> list[str]: |
| return [ |
| self.executable, |
| "chat", |
| "-q", |
| self._build_prompt(job, phase, context), |
| "-Q", |
| "--source", |
| self.source, |
| "--max-turns", |
| str(self.max_turns), |
| ] |
|
|
| @staticmethod |
| def _build_prompt(job: Job, phase: Phase, context: str) -> str: |
| return "\n".join( |
| [ |
| "You are running one CLFGUI phase as a local Hermes backend.", |
| "Do not edit files.", |
| "Do not run tools or commands that mutate files.", |
| "Return only a concise phase summary for the owner cockpit.", |
| "", |
| f"Job #{job.id}: {job.intent}", |
| f"Phase: {phase.name}", |
| f"Current phase summary: {phase.summary}", |
| "Context:", |
| context, |
| ] |
| ) |
|
|
| @staticmethod |
| def _short_output(output: str, limit: int = 600) -> str: |
| collapsed = " ".join(output.split()) |
| if len(collapsed) <= limit: |
| return collapsed |
| return f"{collapsed[: limit - 3]}..." |
|
|
