fegger/time_to_leave
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
55c77c757296cbd2b1768dc2c90cda094da87c6c
time_to_leave/agent_loop/agent_base_gemma4.py
T
fegger 5bcfafcbaf Add mobile services, web proxy, and Gemma4 agent loop 
- Mobile: add push notification and calendar sync services with full test suite (calendar, eventStore, notifications,
  screens)
- Mobile: add EAS build config and Jest setup
- Web: add API proxy, update useDestinationStation/useJourneys hooks, add middleware tests
- Web: update next.config and rebuild
- Agent: add Gemma4-based agent loop (agent_base, ttl_agent, ts_agent)
- Docs: add privacy policy, post-MVP plan, and aider rules
2026-05-11 18:32:54 +02:00

1479 lines
59 KiB
Python
Raw Blame History

from __future__ import annotations
import argparse
import json
import os
import re
import shlex
import shutil
import subprocess
import time
from contextvars import ContextVar
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
import requests
# Strips <think>...</think> blocks that Gemma 4 may emit inline in the content
# field rather than in the dedicated thinking field, depending on Ollama version.
_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
WRITER_SCHEMA = {
"type": "object",
"properties": {
"summary": {"type": "string"},
"files": {
"type": "array",
"items": {
"type": "object",
"properties": {"path": {"type": "string"}, "content": {"type": "string"}},
"required": ["path", "content"],
"additionalProperties": False,
},
},
"tests": {"type": "array", "items": {"type": "string"}},
"notes": {"type": "array", "items": {"type": "string"}},
},
"required": ["summary", "files", "tests", "notes"],
"additionalProperties": False,
}
# Unified review schema — used by both the inline reviewer and as the canonical
# definition for any external tooling that consumes agent state files.
REVIEW_SCHEMA = {
"type": "object",
"properties": {
"verdict": {"type": "string", "enum": ["approve", "needs_changes"]},
"summary": {"type": "string"},
"critical_issues": {"type": "array", "items": {"type": "string"}},
"important_improvements": {"type": "array", "items": {"type": "string"}},
"preserve": {"type": "array", "items": {"type": "string"}},
"rewrite_strategy": {"type": "array", "items": {"type": "string"}},
"missing_files": {"type": "array", "items": {"type": "string"}},
"test_gaps": {"type": "array", "items": {"type": "string"}},
"file_comments": {
"type": "array",
"items": {
"type": "object",
"properties": {"path": {"type": "string"}, "comment": {"type": "string"}},
"required": ["path", "comment"],
"additionalProperties": False,
},
},
},
"required": [
"verdict", "summary", "critical_issues", "important_improvements",
"preserve", "rewrite_strategy", "missing_files", "test_gaps", "file_comments",
],
"additionalProperties": False,
}
DESIGN_SCHEMA = {
"type": "object",
"properties": {
"summary": {"type": "string"},
"files": {
"type": "array",
"items": {
"type": "object",
"properties": {"path": {"type": "string"}, "purpose": {"type": "string"}},
"required": ["path", "purpose"],
"additionalProperties": False,
},
},
"responsibilities": {"type": "array", "items": {"type": "string"}},
"interfaces": {"type": "array", "items": {"type": "string"}},
"integration_points": {"type": "array", "items": {"type": "string"}},
"dependencies": {"type": "array", "items": {"type": "string"}},
"testing_plan": {"type": "array", "items": {"type": "string"}},
"risks": {"type": "array", "items": {"type": "string"}},
"assumptions": {"type": "array", "items": {"type": "string"}},
},
"required": [
"summary", "files", "responsibilities", "interfaces", "integration_points",
"dependencies", "testing_plan", "risks", "assumptions",
],
"additionalProperties": False,
}
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
@dataclass
class Config:
api_base: str
writer_model: str
reviewer_model: str
design_model: str
workspace: Path
outputs: Path
task: str
max_review_loops: int
max_context_chars: int
max_files: int
language: str
writer_system_prompt: str
reviewer_system_prompt: str
design_system_prompt: str
log_dir: Path
redesign_after: int # trigger design revision after this many consecutive failed loops (0 = disabled)
per_file_write: bool = True # write one file per pass instead of all files at once
# Write approved files directly into workspace instead of a separate output dir.
# Also marks the ✔️ column in CHECKLIST.md after a reviewer approval.
write_to_workspace: bool = False
# Keep running until all ✅ boxes in CHECKLIST.md are checked. Implies write_to_workspace.
run_until_done: bool = False
# Maximum seconds of silence before a stream is considered hung.
# Must be longer than the worst-case prefill time (prompt processing before
# the first token arrives). For a 32B model + large context this can be
# 5-10 minutes. Default: 600s. Between-token gaps in practice are <1s.
read_timeout: int = 600
# Context chars passed to each individual file write. Smaller than
# max_context_chars to keep per-file prompts short and reduce prefill time.
per_file_context_chars: int = 40_000
# Writer implementation. "ollama" keeps the historical JSON full-file
# writer; "aider" lets the Aider CLI edit the workspace and then captures
# changed files back into the existing review loop.
writer_backend: str = "ollama"
aider_command: str = "aider"
aider_model: str = ""
aider_extra_args: str = ""
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
# ContextVar instead of a plain module global so that concurrent runs in
# different threads or async tasks each maintain their own log file pointer.
_LOG_FILE: ContextVar[Optional[Path]] = ContextVar("_LOG_FILE", default=None)
def _set_log_file(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
_LOG_FILE.set(path)
def _emit(text: str) -> None:
print(text, flush=True)
lf = _LOG_FILE.get()
if lf is not None:
with lf.open("a", encoding="utf-8") as f:
f.write(text + "\n")
def _log_header(title: str) -> None:
_emit(f"\n=== {title} ===")
def _log_kv(key: str, value: str) -> None:
_emit(f"{key}: {value}")
# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------
def _now_ts() -> float:
return time.time()
def _fmt_elapsed(seconds: float) -> str:
if seconds < 1:
return f"{seconds:.2f}s"
if seconds < 60:
return f"{seconds:.1f}s"
minutes = int(seconds // 60)
rem = seconds % 60
return f"{minutes}m {rem:.1f}s"
def _truncate(text: str, limit: int = 4000) -> str:
text = (text or "").strip()
if len(text) <= limit:
return text
return text[:limit] + "\n...[truncated]..."
def _extract_json(text: str) -> str:
t = text.strip()
start = t.find("{")
if start == -1:
return t
depth = 0
for i, ch in enumerate(t[start:], start):
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
return t[start : i + 1]
return t[start:]
def _pretty_json_or_text(text: str) -> str:
text = (text or "").strip()
if not text:
return text
try:
obj = json.loads(_extract_json(text))
return json.dumps(obj, ensure_ascii=False, indent=2)
except Exception:
return text
def _log_model_output(
label: str, model: str, content: str, elapsed: float, iteration: int | None = None
) -> None:
_log_header(label)
if iteration is not None:
_log_kv("Iteration", str(iteration))
_log_kv("Model", model)
_log_kv("Elapsed", _fmt_elapsed(elapsed))
_emit("--- model output ---")
_emit(_truncate(_pretty_json_or_text(content), limit=4000))
_emit("--------------------")
# ---------------------------------------------------------------------------
# Workspace reader
# ---------------------------------------------------------------------------
_SKIP_SUFFIXES = frozenset({
".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".pdf",
".zip", ".tar", ".gz", ".db", ".sqlite", ".lock",
".map", ".pyc", ".pyo", ".tsbuildinfo",
})
# Directory names that are never useful context for a coding agent.
_SKIP_DIRS = frozenset({
"node_modules", ".next", ".git", "__pycache__",
".idea", ".vscode", "dist", "build", "out", "coverage",
"logs", "runs", "agent_loop",
})
# Specific filenames to exclude regardless of location.
_SKIP_FILENAMES = frozenset({
"package-lock.json", "yarn.lock", "pnpm-lock.yaml",
})
def _read_context(workspace: Path, max_context_chars: int, max_files: int) -> List[Dict[str, str]]:
files: List[Dict[str, str]] = []
total = 0
for path in sorted(workspace.rglob("*")):
if not path.is_file():
continue
if any(part in _SKIP_DIRS for part in path.parts):
continue
if path.suffix.lower() in _SKIP_SUFFIXES:
continue
if path.name in _SKIP_FILENAMES:
continue
try:
text = path.read_text(encoding="utf-8")
except Exception:
continue
remaining = max_context_chars - total
if remaining <= 0:
_emit(f"[warn] context truncated at {max_context_chars} chars after {len(files)} files")
break
if len(text) > remaining:
text = text[:remaining]
_emit(f"[warn] truncated {path.name} to fit context limit")
total += len(text)
files.append({"name": str(path.relative_to(workspace)), "content": text})
if len(files) >= max_files:
_emit(f"[warn] file limit reached ({max_files} files); remaining files excluded")
break
if total >= max_context_chars:
_emit(f"[warn] context truncated at {max_context_chars} chars after {len(files)} files")
break
return files
# ---------------------------------------------------------------------------
# Ollama chat
# ---------------------------------------------------------------------------
def _chat(
api_base: str,
model: str,
system_prompt: str,
user_prompt: str,
schema: Dict[str, Any],
temperature: float,
allow_empty: bool = False,
log_label: str = "Model call",
iteration: int | None = None,
read_timeout: int = 600,
) -> str:
"""Call the Ollama /api/chat endpoint using streaming.
Streaming is essential for long generations (large source files, reasoning
chains). With stream=False the HTTP connection sits idle waiting for the
entire response and hits the read-timeout long before the model finishes.
With streaming each token keeps the connection alive, so generation can
take arbitrarily long without timing out.
connect_timeout=30 — fail fast if Ollama is unreachable
read_timeout=600 (default) — maximum silence before the FIRST token
(covers model load + prefill time) and
between subsequent tokens. For a 32B model
with a large prompt, prefill alone can take
5-10 minutes. Between-token gaps in
practice are <1 second.
"""
start = _now_ts()
url = f"{api_base.rstrip('/')}/api/chat"
base_options: Dict[str, Any] = {
"temperature": temperature,
# Gemma 4 recommended sampling settings (harmless on other models).
"top_p": 0.95,
"top_k": 64,
# Unlimited output tokens. Without this Ollama defaults to ~2048
# tokens which is far too small for complete source files.
"num_predict": -1,
}
def _stream_chat(fmt: Any) -> str:
"""Stream a /api/chat request and return the concatenated content."""
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"format": fmt,
"options": base_options,
"stream": True,
}
content_chunks: list[str] = []
thinking_chunks: list[str] = []
# connect_timeout=30; read_timeout covers both prefill silence and
# between-token gaps — must be large enough for worst-case prefill.
with requests.post(url, json=payload, stream=True, timeout=(30, read_timeout)) as resp:
if not resp.ok:
raise RuntimeError(
f"Ollama API error {resp.status_code}: {resp.text[:2000]}"
)
for raw_line in resp.iter_lines():
if not raw_line:
continue
try:
chunk = json.loads(raw_line)
except json.JSONDecodeError:
continue
msg = chunk.get("message", {}) or {}
delta_content = msg.get("content", "") or ""
delta_thinking = msg.get("thinking", "") or ""
if delta_content:
content_chunks.append(delta_content)
if delta_thinking:
thinking_chunks.append(delta_thinking)
if chunk.get("done"):
break
content = "".join(content_chunks).strip()
thinking = "".join(thinking_chunks).strip()
# Gemma 4 may emit <think> blocks inline in content rather than in the
# dedicated thinking field depending on Ollama version — strip them so
# raw reasoning never leaks into JSON output or multi-turn history.
content = _THINK_RE.sub("", content).strip()
return content if content else thinking
# ── Primary attempt: structured schema ───────────────────────────────
try:
text = _stream_chat(schema)
except RuntimeError as exc:
# Non-2xx from Ollama — fall through to unstructured retry
_emit(f"[warn] Primary stream failed: {exc} — retrying without schema")
text = ""
elapsed = _now_ts() - start
if text.strip():
_log_model_output(log_label, model, text, elapsed, iteration)
return text.strip()
# ── Fallback: unstructured JSON mode ──────────────────────────────────
# Some models produce empty content when schema enforcement is strict.
# Retry without the schema object, relying on the system prompt instead.
_emit(
f"[warn] {log_label}: primary response empty after {_fmt_elapsed(elapsed)} "
f"— retrying in unstructured JSON mode"
)
try:
fallback_system = system_prompt + "\n\nReturn valid JSON only. No extra text."
payload_fb = {
"model": model,
"messages": [
{"role": "system", "content": fallback_system},
{"role": "user", "content": user_prompt},
],
"format": "json",
"options": base_options,
"stream": True,
}
fb_chunks: list[str] = []
with requests.post(url, json=payload_fb, stream=True, timeout=(30, read_timeout)) as resp:
resp.raise_for_status()
for raw_line in resp.iter_lines():
if not raw_line:
continue
try:
chunk = json.loads(raw_line)
except json.JSONDecodeError:
continue
msg = chunk.get("message", {}) or {}
fb_chunks.append(msg.get("content", "") or "")
if chunk.get("done"):
break
text = "".join(fb_chunks).strip()
except Exception as exc:
raise RuntimeError(
f"Both primary and fallback streams failed for model {model}. "
f"Last error: {exc}"
) from exc
elapsed = _now_ts() - start
if not text and not allow_empty:
raise RuntimeError(
f"Model {model} returned empty content on both attempts "
f"after {_fmt_elapsed(elapsed)}. "
f"Check that Ollama is reachable at {url} and the model is loaded."
)
_log_model_output(f"{log_label} (fallback)", model, text, elapsed, iteration)
return text.strip()
# ---------------------------------------------------------------------------
# Normalisation helpers
# ---------------------------------------------------------------------------
def _safe_json(raw: str, fallback: Dict[str, Any]) -> Dict[str, Any]:
try:
return json.loads(_extract_json(raw))
except Exception:
return fallback
def _fallback_review(reason: str, approve: bool = False) -> Dict[str, Any]:
# Safe default is needs_changes — never silently approve on error/timeout.
return {
"verdict": "approve" if approve else "needs_changes",
"summary": reason,
"critical_issues": [] if approve else [reason],
"important_improvements": [] if approve else ["Retry review manually if needed."],
"preserve": [],
"rewrite_strategy": [],
"missing_files": [],
"test_gaps": [],
"file_comments": [],
}
def _normalize_draft(data: Dict[str, Any]) -> Dict[str, Any]:
if not isinstance(data.get("files"), list):
data["files"] = []
if not isinstance(data.get("tests"), list):
data["tests"] = []
if not isinstance(data.get("notes"), list):
data["notes"] = []
if not isinstance(data.get("summary"), str):
data["summary"] = "No summary provided."
return data
def _normalize_design(data: Dict[str, Any]) -> Dict[str, Any]:
for key in (
"files", "responsibilities", "interfaces", "integration_points",
"dependencies", "testing_plan", "risks", "assumptions",
):
if not isinstance(data.get(key), list):
data[key] = []
if not isinstance(data.get("summary"), str):
data["summary"] = "No design summary provided."
return data
def _normalize_review(data: Dict[str, Any]) -> Dict[str, Any]:
if data.get("verdict") not in {"approve", "needs_changes"}:
data["verdict"] = "needs_changes"
for k in (
"critical_issues", "important_improvements", "preserve",
"rewrite_strategy", "missing_files", "test_gaps", "file_comments",
):
if not isinstance(data.get(k), list):
data[k] = []
if not isinstance(data.get("summary"), str):
data["summary"] = "Reviewer returned no summary."
return data
# ---------------------------------------------------------------------------
# I/O helpers
# ---------------------------------------------------------------------------
def _write_state(outputs: Path, stem: str, state: Dict[str, Any], suffix: str = "") -> Path:
state_dir = outputs / "coding-agent-state"
state_dir.mkdir(parents=True, exist_ok=True)
path = state_dir / f"{stem}{suffix}.json"
path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
return path
def _write_files(root: Path, draft: Dict[str, Any]) -> Path:
root.mkdir(parents=True, exist_ok=True)
root_resolved = root.resolve()
for item in draft.get("files", []):
raw = item.get("path", "").strip()
parts = [p for p in Path(raw).parts if p not in ("..", ".", "") and p != "/"]
if not parts:
continue
dest = root / Path(*parts)
try:
if not dest.resolve().is_relative_to(root_resolved):
_emit(f"[warn] skipping unsafe path: {raw!r}")
continue
except Exception:
continue
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(item["content"], encoding="utf-8")
return root
def _snapshot_workspace_files(workspace: Path) -> Dict[str, str]:
"""Capture current text-file contents so Aider edits can be diffed safely."""
snapshot: Dict[str, str] = {}
for path in sorted(workspace.rglob("*")):
if not path.is_file():
continue
if any(part in _SKIP_DIRS for part in path.relative_to(workspace).parts):
continue
if path.suffix.lower() in _SKIP_SUFFIXES:
continue
if path.name in _SKIP_FILENAMES:
continue
try:
snapshot[str(path.relative_to(workspace))] = path.read_text(encoding="utf-8")
except Exception:
continue
return snapshot
def _changed_files_from_snapshots(before: Dict[str, str], after: Dict[str, str]) -> List[Dict[str, str]]:
changed: List[Dict[str, str]] = []
for path in sorted(after):
if before.get(path) != after[path]:
changed.append({"path": path, "content": after[path]})
return changed
def _merge_draft_files(base: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]:
merged: Dict[str, Dict[str, str]] = {
f["path"]: f
for f in base.get("files", [])
if isinstance(f, dict) and isinstance(f.get("path"), str)
}
for f in updates.get("files", []):
if isinstance(f, dict) and isinstance(f.get("path"), str):
merged[f["path"]] = f
return _normalize_draft({
"summary": updates.get("summary") or base.get("summary", ""),
"files": list(merged.values()),
"tests": [*base.get("tests", []), *updates.get("tests", [])],
"notes": [*base.get("notes", []), *updates.get("notes", [])],
})
def _assert_git_workspace(workspace: Path) -> None:
try:
subprocess.run(
["git", "rev-parse", "--is-inside-work-tree"],
cwd=workspace,
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
)
except (FileNotFoundError, subprocess.CalledProcessError) as exc:
raise RuntimeError("Aider backend requires --workspace to be inside a git repository.") from exc
def _build_aider_command(config: Config, message: str, file_paths: List[str]) -> List[str]:
cmd = shlex.split(config.aider_command)
if not cmd:
raise RuntimeError("--aider-command cannot be empty.")
cmd.extend(["--yes-always", "--no-auto-commits"])
if config.aider_model:
cmd.extend(["--model", config.aider_model])
if config.aider_extra_args:
cmd.extend(shlex.split(config.aider_extra_args))
cmd.extend(["--message", message])
# Passing expected files keeps Aider focused while still allowing it to
# create extra files when the prompt calls for tests or supporting modules.
cmd.extend([p for p in file_paths if p and not Path(p).is_absolute()])
return cmd
def _run_aider_writer(
config: Config,
prompt: str,
file_paths: List[str],
label: str,
iteration: int,
) -> Dict[str, Any]:
if shutil.which(shlex.split(config.aider_command)[0]) is None:
raise RuntimeError(
f"Aider command not found: {config.aider_command!r}. "
"Install aider-chat or pass --aider-command."
)
_assert_git_workspace(config.workspace)
before = _snapshot_workspace_files(config.workspace)
cmd = _build_aider_command(config, prompt, file_paths)
_log_header(label)
_log_kv("Iteration", str(iteration))
_log_kv("Command", " ".join(shlex.quote(part) for part in cmd[:4]) + " ...")
start = _now_ts()
proc = subprocess.run(
cmd,
cwd=config.workspace,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
elapsed = _now_ts() - start
_log_kv("Elapsed", _fmt_elapsed(elapsed))
_emit("--- aider output ---")
_emit(_truncate(proc.stdout or "", limit=6000))
_emit("--------------------")
if proc.returncode != 0:
raise RuntimeError(f"Aider exited with status {proc.returncode}. See log for output.")
after = _snapshot_workspace_files(config.workspace)
files = _changed_files_from_snapshots(before, after)
return _normalize_draft({
"summary": f"Aider changed {len(files)} file(s).",
"files": files,
"tests": [],
"notes": [f"Aider backend: {label}"],
})
def _mark_checklist_reviewed(workspace: Path) -> bool:
"""After reviewer approves, mark the first step where ✅=[x] and ✔️=[ ] as reviewed."""
checklist = workspace / "CHECKLIST.md"
if not checklist.exists():
return False
text = checklist.read_text(encoding="utf-8")
# Match the first table row where the written column is [x] and reviewed column is [ ]
# Row format: | N | description | [x] | [ ] |
pattern = re.compile(r'(\|[^|]+\|[^|]+\| *)\[x\]( *\| *)\[ \]( *\|)')
match = pattern.search(text)
if not match:
return False
updated = (
text[: match.start()]
+ match.group(1) + "[x]"
+ match.group(2) + "[x]"
+ match.group(3)
+ text[match.end() :]
)
checklist.write_text(updated, encoding="utf-8")
_emit("[checklist] Marked ✔️ reviewed in CHECKLIST.md")
return True
def _checklist_has_pending(workspace: Path) -> bool:
"""Return True if CHECKLIST.md has any step where ✅ is still [ ]."""
checklist = workspace / "CHECKLIST.md"
if not checklist.exists():
return False
text = checklist.read_text(encoding="utf-8")
# Match any table row where the written (✅) column is [ ]
# Row format: | N | description | [ ] | ... |
return bool(re.search(r'\|[^|]+\|[^|]+\| *\[ \] *\|', text))
# ---------------------------------------------------------------------------
# Design stage (extracted so it can be called again for revisions)
# ---------------------------------------------------------------------------
def _run_design(config: Config, ctx: str, extra_context: str = "") -> Dict[str, Any]:
"""Run (or re-run) the design stage.
Pass *extra_context* to inject accumulated reviewer feedback when triggering
a mid-loop design revision.
"""
parts = [
f"Task:\n{config.task}",
f"\nContext files:\n{ctx}",
]
if extra_context:
parts.append(f"\n{extra_context}")
parts.append(f"\nProduce a concise implementation design for a {config.language} project.")
design_raw = _chat(
config.api_base, config.design_model, config.design_system_prompt,
"\n".join(parts), DESIGN_SCHEMA, 0.0,
log_label="Design stage",
iteration=0,
read_timeout=config.read_timeout,
)
return _normalize_design(_safe_json(design_raw, {
"summary": "Could not parse design output",
"files": [], "responsibilities": [], "interfaces": [],
"integration_points": [], "dependencies": [],
"testing_plan": [], "risks": [], "assumptions": [],
}))
# ---------------------------------------------------------------------------
# Per-file writing helpers
# ---------------------------------------------------------------------------
def _write_single_file(
config: "Config",
design: dict,
file_entry: dict,
already_written: list,
ctx: str,
iteration: int,
) -> dict:
"""Write a single file as specified by its design entry.
*already_written* contains files produced in previous passes so the writer
can maintain consistent imports, types, and naming conventions.
The workspace context (*ctx*) is capped at *per_file_context_chars* rather
than the full *max_context_chars*. Shorter prompts mean shorter prefill
times and fewer read-timeout risks when writing individual files.
"""
# Cap workspace context for per-file calls — a single file only needs
# enough context to understand the surrounding project, not the full repo.
trimmed_ctx = ctx[: config.per_file_context_chars]
if len(ctx) > config.per_file_context_chars:
trimmed_ctx += "\n... [workspace context trimmed — see full context in design] ..."
already_ctx = ""
if already_written:
already_ctx = (
"\n\nAlready-written files (read-only reference — use consistent "
"types, imports, and naming):\n\n"
+ "\n\n".join(
f"--- {f['path']} ---\n{f['content']}" for f in already_written
)
)
prompt = (
f"Task:\n{config.task}\n\n"
f"Full design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"YOUR CURRENT ASSIGNMENT — write this file only:\n"
f" path : {file_entry['path']}\n"
f" purpose: {file_entry.get('purpose', '(see design)')}\n\n"
f"Existing workspace files (read-only context):\n{trimmed_ctx}"
f"{already_ctx}\n\n"
f"Return only the file at '{file_entry['path']}' plus any directly "
f"associated test file. Do not write files outside your assignment."
)
raw = _chat(
config.api_base, config.writer_model, config.writer_system_prompt,
prompt, WRITER_SCHEMA, 0.1,
log_label=f"Writer [{file_entry['path']}]",
iteration=iteration,
read_timeout=config.read_timeout,
)
result = _normalize_draft(_safe_json(raw, {
"summary": f"Could not parse output for {file_entry['path']}",
"files": [], "tests": [], "notes": [raw[:1000]],
}))
if not result.get("files"):
_emit(f"[warn] Writer produced no files for {file_entry['path']} — skipping")
return result
def _files_needing_rewrite(review: dict, draft_files: list) -> set:
"""Return the set of file paths the reviewer flagged for changes."""
flagged = {
fc.get("path", "")
for fc in review.get("file_comments", [])
if fc.get("path")
}
draft_paths = {f["path"] for f in draft_files}
for issue in review.get("critical_issues", []):
for path in draft_paths:
if path in issue:
flagged.add(path)
if review.get("critical_issues") and not flagged:
return draft_paths
return flagged
def _rewrite_targeted_files(
config: "Config",
design: dict,
draft: dict,
review: dict,
ctx: str,
iteration: int,
temperature: float,
) -> dict:
"""Rewrite only the files flagged in the review, merging back into draft."""
all_files = draft.get("files", [])
flagged_paths = _files_needing_rewrite(review, all_files)
if not flagged_paths:
_emit("[per-file] No specific files flagged — falling back to full rewrite")
return draft
_emit(f"[per-file] Targeted rewrite for: {sorted(flagged_paths)}")
design_file_map = {f["path"]: f for f in design.get("files", [])}
current_file_map: Dict[str, Dict[str, str]] = {f["path"]: f for f in all_files}
for path in sorted(flagged_paths):
if path not in current_file_map:
_emit(f"[per-file] Skipping unknown path: {path}")
continue
file_entry = design_file_map.get(path, {"path": path, "purpose": "(not in design)"})
current_content = current_file_map[path].get("content", "")
other_files_ctx = "\n\n".join(
f"--- {f['path']} ---\n{f['content']}"
for f in all_files if f["path"] != path
)
file_comments = [
fc["comment"]
for fc in review.get("file_comments", [])
if fc.get("path") == path
]
prompt = (
f"Task:\n{config.task}\n\n"
f"Full design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"YOUR CURRENT ASSIGNMENT — fix this file only:\n"
f" path : {path}\n"
f" purpose: {file_entry.get('purpose', '(see design)')}\n\n"
f"Current content of '{path}':\n{current_content}\n\n"
f"Other project files (read-only reference):\n{other_files_ctx}\n\n"
f"Review issues specific to '{path}':\n"
+ ("\n".join(f"- {c}" for c in file_comments) if file_comments
else "(none — fix based on general review below)")
+ f"\n\nGeneral review summary: {review.get('summary', '')}\n"
f"Critical issues: {review.get('critical_issues', [])}\n"
f"Rewrite strategy: {review.get('rewrite_strategy', [])}\n\n"
f"Return only '{path}' in the files array. Do not rewrite other files."
)
raw = _chat(
config.api_base, config.writer_model, config.writer_system_prompt,
prompt, WRITER_SCHEMA, temperature,
log_label=f"Writer rewrite [{path}]",
iteration=iteration,
read_timeout=config.read_timeout,
)
result = _normalize_draft(_safe_json(raw, {
"summary": f"Could not parse rewrite for {path}",
"files": [], "tests": [], "notes": [raw[:1000]],
}))
for updated_file in result.get("files", []):
updated_path = updated_file.get("path", "")
current_file_map[updated_path] = updated_file
_emit(f"[per-file] Updated: {updated_path}")
return _normalize_draft({
"summary": draft.get("summary", ""),
"files": list(current_file_map.values()),
"tests": draft.get("tests", []),
"notes": draft.get("notes", []),
})
# ---------------------------------------------------------------------------
# Main agent loop
# ---------------------------------------------------------------------------
def run_agent(config: Config) -> Dict[str, Any]:
run_stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
_set_log_file(config.log_dir / f"{config.language.lower()}-agent-{run_stamp}.log")
_log_header("Agent run started")
_log_kv("Language", config.language)
_log_kv("Writer model", config.writer_model)
_log_kv("Reviewer model", config.reviewer_model)
_log_kv("Design model", config.design_model)
_log_kv("Writer backend", config.writer_backend)
if config.writer_backend == "aider" and not config.write_to_workspace:
raise RuntimeError(
"The Aider writer backend edits the workspace directly. "
"Run with --write-to-workspace or --run-until-done."
)
context_files = _read_context(config.workspace, config.max_context_chars, config.max_files)
ctx = (
"\n\n".join(f"--- FILE: {f['name']} ---\n{f['content']}" for f in context_files)
if context_files else "(no workspace files found)"
)
# --- Design ---
design = _run_design(config, ctx)
_log_header("Design parsed")
_emit(_truncate(json.dumps(design, ensure_ascii=False, indent=2)))
# --- Initial write ---
# Per-file mode: write one file at a time so each generation stays within
# a manageable output size and avoids streaming timeouts on large projects.
# Bulk mode: write everything in one pass (useful for very small tasks).
if config.writer_backend == "aider":
_log_header("Initial write — aider backend")
aider_prompt = (
f"Task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Implement only this task. Keep the change scoped to the design, "
f"update CHECKLIST.md as instructed by the project prompt, and add "
f"or update tests required for the current step."
)
draft = _run_aider_writer(
config,
aider_prompt,
[f.get("path", "") for f in design.get("files", [])],
label="Aider writer stage",
iteration=0,
)
elif config.per_file_write and design.get("files"):
_log_header(f"Initial write — per-file mode ({len(design['files'])} files)")
combined: Dict[str, Any] = {} # path → file dict (deduplicates by path)
combined_tests: List[str] = []
combined_notes: List[str] = []
already_written: List[Dict[str, str]] = []
for i, file_entry in enumerate(design["files"]):
_log_header(f"Writing file {i + 1}/{len(design['files'])}: {file_entry['path']}")
result = _write_single_file(config, design, file_entry, already_written, ctx, iteration=0)
for f in result.get("files", []):
combined[f["path"]] = f
already_written.append(f)
combined_tests.extend(result.get("tests", []))
combined_notes.extend(result.get("notes", []))
draft = _normalize_draft({
"summary": f"Per-file write: {len(combined)} files",
"files": list(combined.values()),
"tests": combined_tests,
"notes": combined_notes,
})
else:
_log_header("Initial write — bulk mode")
writer_prompt = (
f"Task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Context files:\n{ctx}\n\n"
f"Generate code that follows the design."
)
draft_raw = _chat(
config.api_base, config.writer_model, config.writer_system_prompt,
writer_prompt, WRITER_SCHEMA, 0.1,
log_label="Writer stage", iteration=0,
read_timeout=config.read_timeout,
)
draft = _normalize_draft(_safe_json(draft_raw, {
"summary": "Could not parse writer output",
"files": [], "tests": [], "notes": [draft_raw[:3000]],
}))
_log_header("Writer parsed")
_emit(_truncate(json.dumps(draft, ensure_ascii=False, indent=2)))
if not draft.get("files"):
# Diagnose and retry regardless of mode.
_log_header("Writer returned no files — diagnosing")
_emit(
"[diagnose] All files are empty. In per-file mode this means every "
"individual file call returned nothing. In bulk mode check the log for "
"truncation. Retrying with an explicit files-or-fail prompt."
)
if config.writer_backend == "aider":
retry_prompt = (
f"Task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"The previous Aider run produced no changed files. Make concrete "
f"workspace edits for the current task now. Do not stop at an "
f"explanation."
)
draft = _run_aider_writer(
config,
retry_prompt,
[f.get("path", "") for f in design.get("files", [])],
label="Aider writer retry stage",
iteration=0,
)
else:
retry_prompt = (
f"Task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Context files:\n{ctx}\n\n"
f'You MUST write all source files listed in the design. '
f'Return a JSON object: '
f'{{"summary":"...","files":[{{"path":"...","content":"..."}}],"tests":[],"notes":[]}}\n\n'
f"The files array MUST contain at least one entry."
)
retry_raw = _chat(
config.api_base, config.writer_model, config.writer_system_prompt,
retry_prompt, {"type": "object"}, temperature=0.4,
log_label="Writer retry stage", iteration=0,
read_timeout=config.read_timeout,
)
draft = _normalize_draft(_safe_json(retry_raw, {
"summary": "Could not parse writer retry output",
"files": [], "tests": [], "notes": [retry_raw[:3000]],
}))
_log_header("Writer retry parsed")
_emit(_truncate(json.dumps(draft, ensure_ascii=False, indent=2)))
if not draft.get("files"):
raise RuntimeError(
"Writer returned no files on initial attempt and retry. "
"Check the log for '[diagnose]' lines. Common causes: "
"streaming error, model not loaded, or design produced no file entries."
)
stem = run_stamp
# Partial state — written immediately so progress isn't lost on crash
partial = {
"language": config.language, "task": config.task, "design": design,
"writer_model": config.writer_model, "reviewer_model": config.reviewer_model,
"design_model": config.design_model, "draft": draft,
"review": _fallback_review("Review not completed yet."),
"loop_info": {
"loops_used": 0, "max_review_loops": config.max_review_loops,
"stop_reason": "review_not_started", "final_verdict": "needs_changes",
"designs_used": 1, "history": [],
},
"log_file": str(_LOG_FILE.get() or ""),
}
_write_state(config.outputs, f"{config.language.lower()}-{stem}", partial)
# --- Inline reviewer ---
def do_review(cur_draft: Dict[str, Any], iteration: int) -> Dict[str, Any]:
review_prompt = (
f"Original task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Draft JSON:\n{json.dumps(cur_draft, ensure_ascii=False, indent=2)}"
)
try:
raw = _chat(
config.api_base, config.reviewer_model, config.reviewer_system_prompt,
review_prompt, REVIEW_SCHEMA, 0.0,
allow_empty=True, log_label="Reviewer stage", iteration=iteration,
read_timeout=config.read_timeout,
)
if not raw.strip():
# Empty response — conservative: don't approve
return _fallback_review("Reviewer returned an empty response.")
data = _safe_json(raw, _fallback_review("Reviewer returned non-parseable output."))
rv = _normalize_review(data)
_log_header("Reviewer parsed")
_emit(_truncate(json.dumps(rv, ensure_ascii=False, indent=2)))
return rv
except Exception as e:
# Never auto-approve on error — safe default is needs_changes
return _fallback_review(f"Reviewer error: {type(e).__name__}: {e}", approve=False)
# --- Review / rewrite loop ---
review = do_review(draft, 0)
# full_history stores the complete draft summary + review per iteration so
# the -history file is genuinely useful for debugging convergence.
full_history: List[Dict[str, Any]] = [{
"iteration": 0,
"draft_summary": draft.get("summary", ""),
"file_paths": [f.get("path", "") for f in draft.get("files", [])],
"review": review,
}]
# Compact summary used inside loop_info for quick inspection
history = [{
"iteration": 0,
"review_verdict": review.get("verdict", "needs_changes"),
"review_summary": review.get("summary", ""),
"file_count": len(draft.get("files", [])),
}]
loops_used = 0
designs_used = 1
consecutive_needs_changes = 0 if review.get("verdict") == "approve" else 1
stop_reason = "approved" if review.get("verdict") == "approve" else "not_entered"
while review.get("verdict") == "needs_changes" and loops_used < config.max_review_loops:
loops_used += 1
_log_header("Review loop")
_log_kv("Iteration", str(loops_used))
_log_kv("Verdict", review.get("verdict", "needs_changes"))
_log_kv("Summary", review.get("summary", ""))
# --- Optional design revision when the writer is stuck ---
if (
config.redesign_after > 0
and consecutive_needs_changes > 0
and consecutive_needs_changes % config.redesign_after == 0
and designs_used < 3 # cap to prevent infinite redesign cycles
):
_log_header("Design revision triggered")
feedback_window = full_history[-config.redesign_after:]
accumulated = "\n".join(
f"Iteration {h['iteration']}: {h['review'].get('summary', '')}\n"
f" Critical: {h['review'].get('critical_issues', [])}\n"
f" Strategy: {h['review'].get('rewrite_strategy', [])}"
for h in feedback_window
)
extra = (
f"The previous design caused {consecutive_needs_changes} consecutive failed reviews.\n"
f"Accumulated reviewer feedback:\n{accumulated}\n\n"
f"Revise the design to address these issues directly. "
f"Do not repeat a structure that has already failed."
)
design = _run_design(config, ctx, extra_context=extra)
designs_used += 1
_log_header("Revised design parsed")
_emit(_truncate(json.dumps(design, ensure_ascii=False, indent=2)))
# Pass the last 3 reviews to the writer so it can see repeated failures
recent_history = full_history[-3:]
history_context = json.dumps(
[{"iteration": h["iteration"], "review": h["review"]} for h in recent_history],
ensure_ascii=False, indent=2,
)
# Gradually raise temperature so a stuck writer tries different approaches.
# Starts at 0.1 and climbs toward 0.6 over successive loops.
write_temperature = min(0.1 + (loops_used - 1) * 0.1, 0.6)
# Per-file rewrite: only touch files the reviewer flagged.
# Bulk rewrite: rewrite the whole draft (fallback / non-per-file mode).
if config.writer_backend == "aider":
flagged_paths = sorted(_files_needing_rewrite(review, draft.get("files", [])))
missing_paths = [
path for path in review.get("missing_files", [])
if isinstance(path, str) and path
]
aider_paths = sorted(set(flagged_paths + missing_paths))
if not aider_paths:
aider_paths = [f.get("path", "") for f in draft.get("files", [])]
aider_rewrite_prompt = (
f"Original task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Review history (last {len(recent_history)} iterations):\n{history_context}\n\n"
f"Current review feedback:\n{json.dumps(review, ensure_ascii=False, indent=2)}\n\n"
f"Patch the workspace to resolve the review feedback. Keep the "
f"change scoped to the current CHECKLIST step."
)
aider_updates = _run_aider_writer(
config,
aider_rewrite_prompt,
aider_paths,
label="Aider rewrite stage",
iteration=loops_used,
)
new_draft = _merge_draft_files(draft, aider_updates)
elif config.per_file_write:
new_draft = _rewrite_targeted_files(
config, design, draft, review, ctx,
iteration=loops_used, temperature=write_temperature,
)
else:
rewrite_prompt = (
f"Follow-up: Rewrite considering the review.\n\n"
f"Original task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Previous proposal:\n{json.dumps(draft, ensure_ascii=False, indent=2)}\n\n"
f"Review history (last {len(recent_history)} iterations):\n{history_context}\n\n"
f"Current review feedback:\n{json.dumps(review, ensure_ascii=False, indent=2)}"
)
rewrite_raw = _chat(
config.api_base, config.writer_model, config.writer_system_prompt,
rewrite_prompt, WRITER_SCHEMA, write_temperature,
log_label="Writer rewrite stage", iteration=loops_used,
read_timeout=config.read_timeout,
)
new_draft = _normalize_draft(_safe_json(rewrite_raw, draft))
_log_header("Writer rewrite parsed")
_emit(_truncate(json.dumps(new_draft, ensure_ascii=False, indent=2)))
if not new_draft.get("files"):
stop_reason = "rewrite_returned_no_files"
break
old_files = [(f.get("path", ""), f.get("content", "")) for f in draft.get("files", [])]
new_files_list = [(f.get("path", ""), f.get("content", "")) for f in new_draft.get("files", [])]
if old_files == new_files_list:
# Writer returned identical output — stall-break at high temperature.
# In per-file mode, rewrite every file unconditionally to break the deadlock.
_log_header("Stall detected — retrying at high temperature")
if config.writer_backend == "aider":
stall_prompt = (
f"Original task:\n{config.task}\n\n"
f"Design:\n{json.dumps(design, ensure_ascii=False, indent=2)}\n\n"
f"Current review feedback:\n{json.dumps(review, ensure_ascii=False, indent=2)}\n\n"
f"Your previous Aider pass did not change any tracked text files. "
f"Make concrete edits that address the review feedback now."
)
stall_updates = _run_aider_writer(
config,
stall_prompt,
[f.get("path", "") for f in draft.get("files", [])],
label="Aider stall-break attempt",
iteration=loops_used,
)
stall_draft = _merge_draft_files(draft, stall_updates)
elif config.per_file_write:
stall_draft = _rewrite_targeted_files(
config, design, draft, review, ctx,
iteration=loops_used, temperature=0.7,
)
else:
stall_prompt = (
f"{rewrite_prompt}\n\n"
f"IMPORTANT: Your last rewrite returned files identical to the previous attempt. "
f"You must make concrete changes. "
f"Try a different implementation strategy for the issues flagged in the review."
)
stall_raw = _chat(
config.api_base, config.writer_model, config.writer_system_prompt,
stall_prompt, WRITER_SCHEMA, 0.7,
log_label="Writer stall-break attempt", iteration=loops_used,
read_timeout=config.read_timeout,
)
stall_draft = _normalize_draft(_safe_json(stall_raw, new_draft))
stall_files = [(f.get("path", ""), f.get("content", "")) for f in stall_draft.get("files", [])]
if stall_files == old_files:
stop_reason = "no_file_changes"
break
new_draft = stall_draft
draft = new_draft
review = do_review(draft, loops_used)
history.append({
"iteration": loops_used,
"review_verdict": review.get("verdict", "needs_changes"),
"review_summary": review.get("summary", ""),
"file_count": len(draft.get("files", [])),
})
full_history.append({
"iteration": loops_used,
"draft_summary": draft.get("summary", ""),
"file_paths": [f.get("path", "") for f in draft.get("files", [])],
"review": review,
})
if review.get("verdict") == "approve":
consecutive_needs_changes = 0
stop_reason = "approved"
break
consecutive_needs_changes += 1
else:
stop_reason = "approved" if review.get("verdict") == "approve" else "max_review_loops_reached"
state = {
"language": config.language, "task": config.task, "design": design,
"writer_model": config.writer_model, "reviewer_model": config.reviewer_model,
"design_model": config.design_model, "draft": draft, "review": review,
"loop_info": {
"loops_used": loops_used,
"max_review_loops": config.max_review_loops,
"stop_reason": stop_reason,
"final_verdict": review.get("verdict", "needs_changes"),
"designs_used": designs_used,
"history": history,
},
"log_file": str(_LOG_FILE.get() or ""),
}
# Final state — overwrites the partial written at the start
_write_state(config.outputs, f"{config.language.lower()}-{stem}", state)
# Genuine iteration history: one entry per pass with draft summary + full review
_write_state(
config.outputs,
f"{config.language.lower()}-{stem}",
{"task": config.task, "full_history": full_history},
suffix="-history",
)
if config.writer_backend == "aider":
project_dir = config.workspace
if stop_reason == "approved":
_mark_checklist_reviewed(config.workspace)
elif config.write_to_workspace:
output_root = config.workspace
project_dir = _write_files(output_root, draft)
if stop_reason == "approved":
_mark_checklist_reviewed(config.workspace)
else:
output_root = config.outputs / f"{config.language.lower()}-agent-output-{stem}"
project_dir = _write_files(output_root, draft)
state["project_dir"] = str(project_dir)
return state
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(language: str, writer_prompt: str, reviewer_prompt: str, design_prompt: str) -> Config:
p = argparse.ArgumentParser()
_default_task = os.getenv("AGENT_TASK")
p.add_argument("--task", default=_default_task, required=_default_task is None)
p.add_argument("--workspace", default=os.getenv("AGENT_WORKSPACE", "/workspace"))
p.add_argument("--outputs", default=os.getenv("AGENT_OUTPUTS", "./runs"))
p.add_argument("--writer-model", default=os.getenv("WRITER_MODEL", "gemma4:31b"))
p.add_argument("--reviewer-model", default=os.getenv("REVIEWER_MODEL", "gemma4:31b"))
p.add_argument(
"--design-model",
default=os.getenv("DESIGN_MODEL", os.getenv("REVIEWER_MODEL", "gemma4:31b")),
)
p.add_argument("--api-base", default=os.getenv("OLLAMA_API_BASE", "http://ollama:11434"))
p.add_argument("--max-review-loops", type=int, default=int(os.getenv("AGENT_MAX_REVIEW_LOOPS", "6")))
p.add_argument("--max-context-chars", type=int, default=120_000)
p.add_argument("--max-files", type=int, default=80)
p.add_argument(
"--redesign-after", type=int, default=3,
help="Trigger a design revision after this many consecutive failed review loops (0 = disabled).",
)
p.add_argument("--log-dir", default=os.getenv("AGENT_LOG_DIR", "./logs"))
p.add_argument(
"--no-per-file", action="store_true",
help=(
"Disable per-file writing and write all files in a single pass. "
"Per-file mode is the default: it keeps each generation short, "
"avoids streaming timeouts on large projects, and enables targeted "
"rewrites in the review loop."
),
)
p.add_argument(
"--read-timeout", type=int, default=600,
help=(
"Seconds of silence before a stream is considered hung. "
"Must cover worst-case prefill time (model load + prompt processing "
"before the first token). For a 32B model with a large context this "
"can be 5-10 minutes. Default: 600."
),
)
p.add_argument(
"--per-file-context-chars", type=int, default=40_000,
help=(
"Maximum workspace context chars passed to each individual file write. "
"Smaller than --max-context-chars to keep per-file prompts short and "
"reduce prefill time. Default: 40000."
),
)
p.add_argument(
"--writer-backend",
choices=("ollama", "aider"),
default=os.getenv("AGENT_WRITER_BACKEND", "ollama"),
help=(
"Writer backend to use. 'ollama' asks the configured writer model for "
"JSON full-file contents. 'aider' runs the Aider CLI, lets it edit "
"the workspace directly, then captures changed files for review."
),
)
p.add_argument(
"--aider-command",
default=os.getenv("AIDER_COMMAND", "aider"),
help="Aider executable or command prefix. Default: aider.",
)
p.add_argument(
"--aider-model",
default=os.getenv("AIDER_MODEL", ""),
help="Optional model passed to Aider as --model.",
)
p.add_argument(
"--aider-extra-args",
default=os.getenv("AIDER_EXTRA_ARGS", ""),
help="Optional extra CLI args appended to Aider before --message.",
)
p.add_argument(
"--write-to-workspace", action="store_true",
help=(
"Write approved files directly into --workspace instead of a timestamped "
"subdirectory under --outputs. Also marks the ✔️ column in CHECKLIST.md "
"automatically after the reviewer approves."
),
)
p.add_argument(
"--run-until-done", action="store_true",
help=(
"Loop automatically through every pending CHECKLIST step without human input. "
"Stops when all ✅ boxes are checked or a step fails to get approved. "
"Implies --write-to-workspace."
),
)
a = p.parse_args()
return Config(
api_base=a.api_base,
writer_model=a.writer_model,
reviewer_model=a.reviewer_model,
design_model=a.design_model,
workspace=Path(a.workspace),
outputs=Path(a.outputs),
task=a.task,
max_review_loops=a.max_review_loops,
max_context_chars=a.max_context_chars,
max_files=a.max_files,
language=language,
writer_system_prompt=writer_prompt,
reviewer_system_prompt=reviewer_prompt,
design_system_prompt=design_prompt,
log_dir=Path(a.log_dir),
redesign_after=a.redesign_after,
per_file_write=not a.no_per_file,
write_to_workspace=a.write_to_workspace or a.run_until_done,
run_until_done=a.run_until_done,
read_timeout=a.read_timeout,
per_file_context_chars=a.per_file_context_chars,
writer_backend=a.writer_backend,
aider_command=a.aider_command,
aider_model=a.aider_model,
aider_extra_args=a.aider_extra_args,
)
def main(language: str, writer_prompt: str, reviewer_prompt: str, design_prompt: str) -> int:
cfg = parse_args(language, writer_prompt, reviewer_prompt, design_prompt)
if cfg.run_until_done:
step = 0
while _checklist_has_pending(cfg.workspace):
step += 1
_emit(f"\n{'=' * 60}\n[loop] Starting step {step}\n{'=' * 60}")
state = run_agent(cfg)
stop_reason = state["loop_info"]["stop_reason"]
verdict = state["review"]["verdict"]
_emit(f"[loop] Step {step} finished — stop_reason={stop_reason} verdict={verdict}")
if stop_reason != "approved":
_emit(f"[loop] Step {step} did not get approved. Halting.")
print(f"Stopped at step {step}: {stop_reason}")
return 1
_emit("[loop] All CHECKLIST steps complete.")
print(f"All {step} steps completed.")
return 0
state = run_agent(cfg)
print(f"{cfg.language} agent completed")
print(f"Iteration count: {state['loop_info']['loops_used']}/{state['loop_info']['max_review_loops']}")
print(f"Stop reason: {state['loop_info']['stop_reason']}")
print(f"Final verdict: {state['review']['verdict']}")
print(f"Designs used: {state['loop_info']['designs_used']}")
print(f"Project output: {state['project_dir']}")
print(f"Log file: {state.get('log_file', '')}")
return 0
Reference in New Issue View Git Blame Copy Permalink