File size: 4,429 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 | 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]}..."
|