paulpas CLI Agent Workflows
Implements command-line AI agent architectures — building terminal-resident agents with stdin/stdout streaming, subcommand routing, file system interaction, code generation from design specifications, and MCP server bridging. When loaded, this skill makes the model design production-grade CLI agents modeled after Gemini CLI, Claude Code, and similar developer tooling that operates entirely within the terminal environment.
TL;DR Checklist
- Parse all inputs at boundary before processing (Law 2: parse don't validate)
- Handle edge cases with early returns at function top (Law 1: early exit)
- Fail immediately with descriptive errors on invalid states (Law 4: fail fast)
- Return new data structures, never mutate inputs (Law 3: atomic predictability)
- Use explicit
subprocess.runwith list args — nevershell=Truewith user input - Implement tool sandboxing with path validation and command allowlists
- Design subcommand routing with click.Group or argparse subparsers for extensibility
- Reference
code-philosophy(5 Laws of Elegant Defense) in constraint design
When to Use
Use this skill when:
- Building a terminal-resident AI assistant that reads user prompts from stdin, processes them through an LLM, and writes structured output back to the terminal
- Designing code generation workflows where markdown architecture docs or design specifications are consumed as input and translated into executable source files
- Implementing prompt-based code review inside the CLI — accepting natural language review requests, running automated linting/style checks, and surfacing suggestions inline
- Creating developer tooling that bridges a local CLI agent to MCP servers for extended capabilities (database queries, API calls, external service automation)
- Building multi-turn interactive sessions in the terminal with context retention across commands (like Gemini CLI's persistent conversation mode)
- Automating repetitive developer workflows: scaffold new projects, run linters, generate boilerplate, execute tests — all orchestrated from a single CLI entry point
When NOT to Use
Avoid this skill for:
- GUI desktop applications or web-based interfaces — use
coding-agent-frameworks(LangChain, CrewAI) with their frontend integrations instead - Simple script automation without agent-level reasoning — direct bash scripting or
os-scriptingis lighter weight and avoids LLM overhead - High-throughput batch processing where sub-second latency matters — the LLM round-trip adds hundreds of milliseconds per request
- End-user productivity tools targeting non-technical audiences — CLI agents assume terminal familiarity; use a GUI agent (
gui-agent-interaction) instead
Core Workflow
Parse Input Stream — Read user input from stdin or command-line arguments. Classify the intent: code generation, file operation, terminal execution, review request, or tool invocation. Apply a lightweight rule-based classifier before delegating to the LLM for semantic routing. Checkpoint: Intent category is determined and all required parameters are extracted; if parameters are missing, return a structured error requesting them before proceeding.
Route to Tool Executor — Dispatch the classified request to the appropriate handler: code generator (reads design specs, writes source files), file engine (safe read/write with path validation), terminal executor (subprocess invocation with output capture), or MCP bridge (proxies to registered MCP servers). Each handler receives validated input and returns a structured result object. Checkpoint: The selected handler acknowledges receipt, confirms parameter validity, and begins execution; if no matching handler exists, fall back to the generic LLM completion path.
Execute with Safety Guards — Run the handler's logic within constrained boundaries: validate file paths against an allowlist directory, sandbox subprocess commands with timeout limits, rate-limit tool invocations, and capture all output (stdout, stderr, exit code) in a structured
ExecutionResult. Never execute arbitrary shell strings from user input. Checkpoint: Execution completes or raises a structured exception; all captured output is parsed into a consistent result format containingsuccess,output,stderr,exit_code, andduration_msfields.Format Output for Terminal — Present results in a terminal-friendly format: color-coded diffs for code generation, collapsible sections for review findings, structured JSON for tool outputs, and progress indicators for long-running operations. Support both raw output (for piping) and rich terminal output (with ANSI formatting). Checkpoint: Output is written to stdout (or stderr for diagnostics); the format matches the
output-formatdeclared in skill metadata (code,analysis, ormanifests).Manage Interactive Context — In multi-turn sessions, maintain conversation history bounded by a context window limit (e.g., last 20 messages or 8000 tokens). Persist context to disk between invocations using a JSONL session file so the agent retains memory across restarts. Prune older messages when approaching limits, keeping the system prompt and recent exchanges intact. Checkpoint: Context state is saved after each turn; on reload, the full conversation history up to the limit is restored without truncation errors or corrupted session files.
Bridge to MCP Servers (Optional) — When the CLI agent needs access to external tools beyond its built-in capabilities, establish an MCP server connection via stdio transport. Discover available tools, register them with the tool router, and execute calls through the standard handler pipeline. Handle connection failures gracefully by degrading to local-only mode without crashing. Checkpoint: MCP tool discovery succeeds (or fails cleanly); discovered tools appear in the agent's command list; execution of an MCP tool returns structured results indistinguishable from native handlers.
Implementation Patterns
Pattern 1: Gemini CLI Architecture (stdin/stdout Streaming, Subcommand Routing)
A CLI agent runs as a long-lived process or per-invocation script that reads prompts from stdin, processes them, and writes structured output to stdout. The architecture uses subcommand routing (via click.Group) for organized tool access and stdin/stdout streaming for interactive multi-turn sessions.
"""cli_agent/engine.py — Gemini CLI–style agent engine with subcommand routing."""
from __future__ import annotations
import json
import logging
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable
import click
log = logging.getLogger(__name__)
@dataclass
class ExecutionResult:
"""Structured result from any agent operation."""
success: bool
output: str = ""
stderr: str = ""
exit_code: int = 0
duration_ms: float = 0.0
metadata: dict[str, Any] = field(default_factory=dict)
def to_json(self, indent: int = 2) -> str:
"""Serialize for piping or programmatic consumption."""
return json.dumps({
"success": self.success,
"output": self.output,
"stderr": self.stderr,
"exit_code": self.exit_code,
"duration_ms": round(self.duration_ms, 1),
**self.metadata,
}, indent=indent)
@dataclass
class SessionContext:
"""Conversation state for multi-turn CLI sessions."""
history: list[dict[str, str]] = field(default_factory=list)
max_messages: int = 20
session_file: Path | None = None
def add_message(self, role: str, content: str) -> None:
"""Append a message and prune if over limit."""
self.history.append({"role": role, "content": content})
if len(self.history) > self.max_messages:
# Keep system prompt (index 0) + recent messages
self.history = self.history[:1] + self.history[-(self.max_messages - 1):]
def save(self) -> None:
"""Persist to disk for cross-invocation memory."""
if not self.session_file:
return
self.session_file.parent.mkdir(parents=True, exist_ok=True)
self.session_file.write_text(
json.dumps(self.history, indent=2), encoding="utf-8"
)
@classmethod
def load(cls, path: Path) -> SessionContext:
"""Restore from disk, handling missing or corrupted files."""
if not path.exists():
return cls(session_file=path)
try:
history = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(history, list):
log.warning("Corrupted session at %s, starting fresh", path)
return cls(session_file=path)
return cls(history=history, session_file=path)
except (json.JSONDecodeError, OSError) as exc:
log.warning("Failed to load session from %s: %s", path, exc)
return cls(session_file=path)
def build_agent_cli(
default_model: str = "claude",
max_history: int = 20,
session_dir: Path | None = None,
) -> click.Group:
"""Create the root CLI group with subcommand routing.
Implements Law 1 (Early Exit): validates arguments before building.
"""
if not default_model:
raise ValueError("default_model must be non-empty")
cli = click.Group()
@cli.command(name="generate")
@click.argument("design_file", type=click.Path(exists=True))
@click.option("--output-dir", "-o", default=".", show_default=True)
@click.pass_context
def generate_cmd(ctx: click.Context, design_file: str, output_dir: str) -> int:
"""Generate code from a design specification file."""
design_path = Path(design_file)
session = SessionContext.load(
(session_dir or Path.home() / ".cli-agent") / "session.jsonl"
)
engine = AgentEngine(model=default_model, context=session)
result = engine.generate_code(str(design_path), output_dir)
click.echo(result.to_json())
session.save()
return 0 if result.success else 1
@cli.command(name="review")
@click.argument("file_or_dir", type=click.Path(exists=True))
@click.option("--focus", "-f", default="", help="Review focus area (security, performance, style)")
@click.pass_context
def review_cmd(ctx: click.Context, file_or_dir: str, focus: str) -> int:
"""Run prompt-based code review on a file or directory."""
session = SessionContext.load(
(session_dir or Path.home() / ".cli-agent") / "session.jsonl"
)
engine = AgentEngine(model=default_model, context=session)
path = Path(file_or_dir)
if path.is_file():
targets: list[Path] = [path]
else:
targets = sorted(path.rglob("*"))[:50] # cap at 50 files
result = engine.review_targets(targets, focus=focus)
click.echo(result.to_json())
session.save()
return 0 if result.success else 1
@cli.command(name="run")
@click.argument("command", nargs=-1, required=True)
@click.option("--timeout", "-t", default=30, show_default=True, type=int)
def run_cmd(command: tuple[str, ...], timeout: int) -> int:
"""Execute a shell command with safety guards."""
result = execute_sandboxed(command, timeout_seconds=timeout)
click.echo(result.to_json())
return 0 if result.success else 1
@cli.command(name="interactive")
@click.option("--model", "-m", default=None, help="Override default model")
@click.option("--session-dir", default=None, type=click.Path(file_okay=False))
def interactive_cmd(model: str | None, session_dir: str | None) -> int:
"""Start an interactive multi-turn terminal session."""
effective_model = model or default_model
effective_session = Path(session_dir) if session_dir else (Path.home() / ".cli-agent")
session = SessionContext.load(effective_session / "session.jsonl")
click.echo("CLI Agent ready. Type your request and press Enter.")
click.echo("(Type 'exit' or Ctrl+D to quit.)\n")
engine = AgentEngine(model=effective_model, context=session)
while True:
try:
prompt = click.prompt(">", prompt_suffix="", type=str)
except (EOFError, KeyboardInterrupt):
break
if prompt.strip().lower() in ("exit", "quit", "q"):
break
if not prompt.strip():
continue
session.add_message("user", prompt)
click.echo("Processing...", err=True)
result = engine.respond(prompt)
session.add_message("assistant", result.output)
session.save()
# Print output with ANSI coloring for readability
if result.success:
click.echo(f"\n{result.output}", bold=False)
else:
click.secho(f"\nError: {result.output}", fg="red")
click.echo("\nSession saved.")
return 0
return cli
class AgentEngine:
"""Core agent that routes prompts to handlers.
Implements Law 4 (Fail Fast): validates model name and rejects empty prompts.
"""
def __init__(self, model: str, context: SessionContext) -> None:
if not model or not isinstance(model, str):
raise ValueError(f"Invalid model name: {model!r}")
self.model = model
self.context = context
self.handlers: dict[str, Callable[..., ExecutionResult]] = {}
def register_handler(self, intent: str, handler: Callable[..., ExecutionResult]) -> None:
"""Register an intent handler. Order matters — first registration wins."""
if intent in self.handlers:
log.warning("Handler for '%s' already registered, skipping", intent)
return
self.handlers[intent] = handler
def classify_intent(self, prompt: str) -> str:
"""Lightweight rule-based intent classifier.
Checks keyword patterns before delegating to LLM routing.
Returns one of: 'generate', 'review', 'execute', 'mcp', or 'general'.
"""
lower = prompt.lower().strip()
if any(term in lower for term in ("generate", "create code from", "build from design")):
return "generate"
if any(term in lower for term in ("review", "check", "audit", "lint")):
return "review"
if any(term in lower for term in ("run", "execute", "shell command", "run ")) or lower.startswith("$"):
return "execute"
if any(term in lower for term in ("query", "mcp call", "database", "api call")):
return "mcp"
return "general"
def respond(self, prompt: str) -> ExecutionResult:
"""Process a user prompt through the full pipeline.
Applies Law 1 (Early Exit): rejects empty prompts immediately.
Applies Law 4 (Fail Fast): raises on invalid state, never swallows errors.
"""
if not prompt or not prompt.strip():
return ExecutionResult(
success=False,
output="Empty prompt — nothing to process.",
metadata={"error": "empty_input"},
)
start = time.monotonic()
intent = self.classify_intent(prompt)
handler = self.handlers.get(intent) or self._default_handler
try:
result = handler(prompt)
elapsed = (time.monotonic() - start) * 1000
result.duration_ms = round(elapsed, 1)
return result
except Exception as exc:
elapsed = (time.monotonic() - start) * 1000
return ExecutionResult(
success=False,
output=f"Agent error: {exc}",
metadata={"error": type(exc).__name__, "intent": intent},
duration_ms=round(elapsed, 1),
)
def generate_code(self, design_path: str, output_dir: str) -> ExecutionResult:
"""Generate code from a design specification file."""
design = Path(design_path)
if not design.is_file():
return ExecutionResult(
success=False,
output=f"Design file not found: {design_path}",
metadata={"error": "file_not_found"},
)
try:
spec_text = design.read_text(encoding="utf-8")
except OSError as exc:
return ExecutionResult(
success=False,
output=f"Cannot read design file: {exc}",
metadata={"error": str(exc)},
)
# In production, this would call the LLM with the spec
generated_files = [f"{design.stem}.py", f"{design.stem}_test.py"]
out_path = Path(output_dir) / design.stem
out_path.mkdir(parents=True, exist_ok=True)
code_content = f'"""Generated from {design.name}."""\n\n# TODO: LLM-generated code would go here\n'
for fname in generated_files:
(out_path / fname).write_text(code_content, encoding="utf-8")
return ExecutionResult(
success=True,
output=f"Generated {len(generated_files)} files in {out_path}",
metadata={"files": [str(out_path / f) for f in generated_files], "design": str(design)},
)
def review_targets(self, targets: list[Path], focus: str = "") -> ExecutionResult:
"""Run code review on a list of files or directories."""
if not targets:
return ExecutionResult(
success=False,
output="No targets to review.",
metadata={"error": "no_targets"},
)
findings = []
for target in targets[:5]: # cap at 5 files per invocation
if target.is_file() and target.suffix == ".py":
try:
content = target.read_text(encoding="utf-8")
# Simple heuristic checks (production would use linters + LLM)
issues = self._simple_review(content, target.name)
findings.extend(issues)
except OSError:
continue
summary = f"Found {len(findings)} potential issues across {min(len(targets), 5)} files." if findings else "No issues detected."
return ExecutionResult(
success=True,
output=f"{summary}\n{''.join(findings[:10])}",
metadata={"focus": focus, "findings_count": len(findings)},
)
@staticmethod
def _simple_review(content: str, filename: str) -> list[str]:
"""Lightweight pre-LLM review heuristics."""
issues = []
if "eval(" in content or "exec(" in content:
issues.append(f" ⚠ {filename}: Contains eval()/exec() — potential code injection\n")
if '"""' in content and '"""' in content[content.index('"""') + 3:] == False and 'doc' not in content[:200].lower():
pass # would flag missing docstrings in production
return issues
def _default_handler(self, prompt: str) -> ExecutionResult:
"""Fallback handler for unclassified prompts."""
return ExecutionResult(
success=True,
output=f"[Default] Processed prompt via LLM ({self.model}). Intent was not explicitly classified.\n\nPrompt received: {prompt[:200]}{'...' if len(prompt) > 200 else ''}",
metadata={"model": self.model, "mode": "fallback"},
)
def execute_sandboxed(command: tuple[str, ...], timeout_seconds: int = 30) -> ExecutionResult:
"""Execute a command in a sandbox with safety guarantees.
Implements Law 1 (Early Exit): rejects empty commands and blocked binaries.
Implements Law 2 (Immutable State): never mutates the input command tuple.
Uses subprocess.run with list args — NEVER shell=True with user data.
Args:
command: Tuple of command arguments (e.g., ("git", "status")).
timeout_seconds: Maximum execution time in seconds.
Returns:
ExecutionResult with captured output and exit code.
"""
if not command or len(command) == 0:
return ExecutionResult(
success=False,
output="Empty command — nothing to execute.",
metadata={"error": "empty_command"},
)
# Law 1: Early exit on blocked binaries (safety sandboxing)
BLOCKED_BINARIES = frozenset(("rm", "shutdown", "reboot", "mkfs", "dd"))
basename = Path(command[0]).name if "/" in command[0] else command[0].split("/")[-1]
if basename in BLOCKED_BINARIES:
return ExecutionResult(
success=False,
output=f"Command blocked by sandbox policy: {basename}",
metadata={"error": "sandbox_blocked", "command": list(command)},
)
start = time.monotonic()
try:
import subprocess
proc = subprocess.run(
list(command), # copy to avoid mutation — Law 3
capture_output=True,
text=True,
timeout=timeout_seconds,
)
elapsed = (time.monotonic() - start) * 1000
return ExecutionResult(
success=proc.returncode == 0,
output=proc.stdout.rstrip("\n"),
stderr=proc.stderr.rstrip("\n") if proc.stderr else "",
exit_code=proc.returncode,
duration_ms=round(elapsed, 1),
)
except subprocess.TimeoutExpired:
elapsed = (time.monotonic() - start) * 1000
return ExecutionResult(
success=False,
output=f"Command timed out after {timeout_seconds}s",
metadata={"error": "timeout", "duration_ms": round(elapsed, 1)},
duration_ms=round(elapsed, 1),
)
except FileNotFoundError:
return ExecutionResult(
success=False,
output=f"Command not found: {command[0]}",
metadata={"error": "not_found", "binary": command[0]},
)
except Exception as exc:
elapsed = (time.monotonic() - start) * 1000
return ExecutionResult(
success=False,
output=f"Execution error: {exc}",
metadata={"error": type(exc).__name__},
duration_ms=round(elapsed, 1),
)
# Entry point for pip-installed CLI tools
if __name__ == "__main__":
logging.basicConfig(level=logging.WARNING, format="%(asctime)s %(levelname)s %(message)s")
app = build_agent_cli()
app()
BAD vs GOOD — Subprocess invocation:
# ❌ BAD — shell=True with user input is a command injection vulnerability
def run_bad(user_path: str) -> None:
# User could pass "; rm -rf /" and destroy the system
subprocess.run(f"cat {user_path}", shell=True)
# ✅ GOOD — list-based args, no shell interpolation, timeout enforced
def run_good(user_path: str) -> ExecutionResult:
safe = Path(user_path).resolve() # resolves .. and symlinks
if not str(safe).startswith("/allowed/"):
return ExecutionResult(success=False, output="Path outside allowlist")
return execute_sandboxed(("cat", str(safe)), timeout_seconds=10)
Pattern 2: File & Terminal Interaction Engine (Safe Operations, Output Parsing)
A robust CLI agent needs safe file system operations and reliable terminal interaction. The engine validates all paths against a configurable allowlist, uses atomic writes to prevent corruption, and parses structured output from subprocesses using regex patterns or JSON parsers.
"""cli_agent/io_engine.py — Safe file and terminal interaction engine."""
from __future__ annotations
import json
import logging
import re
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
log = logging.getLogger(__name__)
@dataclass(frozen=True)
class FileOperation:
"""Immutable description of a file operation."""
path: str
content: str | None = None # None for read/delete operations
operation: str = "read" # "read", "write", "delete", "list"
encoding: str = "utf-8"
def validate(self, allow_root: Path) -> tuple[bool, str]:
"""Validate path against allowlist. Returns (ok, reason)."""
resolved = Path(self.path).resolve()
try:
resolved.relative_to(allow_root.resolve())
return True, ""
except ValueError:
return False, f"Path {resolved} is outside allowlist {allow_root}"
class FileEngine:
"""Safely manages file read/write/delete within an allowlisted directory."""
def __init__(self, allow_root: str | Path = ".") -> None:
self.allow_root = Path(allow_root).resolve()
if not self.allow_root.is_dir():
raise FileNotFoundError(f"Allowlist root does not exist: {allow_root}")
def read_file(self, path: str) -> ExecutionResult: # type: ignore[name-defined]
"""Read a file within the allowlist. Returns content or structured error."""
operation = FileOperation(path=path, operation="read")
ok, reason = operation.validate(self.allow_root)
if not ok:
return ExecutionResult( # type: ignore[name-defined]
success=False, output=reason, metadata={"error": "path_outside_allowlist"}
)
resolved = Path(path).resolve()
try:
content = resolved.read_text(encoding="utf-8")
return ExecutionResult( # type: ignore[name-defined]
success=True,
output=content,
metadata={"path": str(resolved), "size_bytes": len(content.encode("utf-8"))},
)
except FileNotFoundError:
return ExecutionResult(
success=False,
output=f"File not found: {resolved}",
metadata={"error": "file_not_found"},
)
except PermissionError:
return ExecutionResult(
success=False,
output=f"Permission denied: {resolved}",
metadata={"error": "permission_denied"},
)
except UnicodeDecodeError as exc:
return ExecutionResult(
success=False,
output=f"File is not valid UTF-8: {exc}",
metadata={"error": "encoding_error"},
)
def write_file(self, path: str, content: str) -> ExecutionResult: # type: ignore[name-defined]
"""Atomically write a file within the allowlist using a temp file + rename."""
operation = FileOperation(path=path, content=content, operation="write")
ok, reason = operation.validate(self.allow_root)
if not ok:
return ExecutionResult(
success=False, output=reason, metadata={"error": "path_outside_allowlist"}
)
resolved = Path(path).resolve()
# Safety: limit file size to prevent disk exhaustion (10 MB cap)
if len(content.encode("utf-8")) > 10 * 1024 * 1024:
return ExecutionResult(
success=False,
output=f"Content exceeds 10 MB limit ({len(content)} chars)",
metadata={"error": "content_too_large"},
)
resolved.parent.mkdir(parents=True, exist_ok=True)
# Atomic write: temp file → fsync → os.replace (prevents partial writes)
try:
fd, tmp_path = tempfile.mkstemp(
dir=str(resolved.parent), prefix=".tmp_", suffix=resolved.suffix
)
try:
import os
with os.fdopen(fd, "w", encoding="utf-8") as tmp_file:
tmp_file.write(content)
tmp_file.flush()
os.fsync(tmp_file.fileno())
os.replace(tmp_path, str(resolved)) # atomic on POSIX and Windows
except Exception:
Path(tmp_path).unlink(missing_ok=True)
raise
return ExecutionResult(
success=True,
output=f"Written {len(content)} chars to {resolved}",
metadata={"path": str(resolved), "bytes_written": len(content.encode("utf-8"))},
)
except OSError as exc:
return ExecutionResult(
success=False,
output=f"Write failed: {exc}",
metadata={"error": str(exc)},
)
def list_directory(self, path: str = ".") -> ExecutionResult: # type: ignore[name-defined]
"""List files in a directory within the allowlist."""
operation = FileOperation(path=path, operation="list")
ok, reason = operation.validate(self.allow_root)
if not ok:
return ExecutionResult(
success=False, output=reason, metadata={"error": "path_outside_allowlist"}
)
resolved = Path(path).resolve()
if not resolved.is_dir():
return ExecutionResult(success=False, output=f"Not a directory: {resolved}")
entries = []
for item in sorted(resolved.iterdir()):
tag = "" if item.is_file() else "/"
entries.append(f"{item.name}{tag}")
return ExecutionResult(
success=True,
output="\n".join(entries[:100]), # cap output at 100 entries
metadata={"path": str(resolved), "entry_count": len(entries)},
)
def parse_terminal_output(output: str) -> dict[str, str | list[str]]:
"""Parse structured terminal output into categories.
Handles common patterns: file listings, command outputs with headers,
JSON/stderr separation, and progress indicators.
"""
if not output:
return {"type": "empty", "data": ""}
# Try parsing as JSON first (cleanest structured output)
stripped = output.strip()
try:
parsed = json.loads(stripped)
return {"type": "json", "data": parsed}
except (json.JSONDecodeError, ValueError):
pass
# Detect ANSI escape sequences and strip them for analysis
ansi_pattern = re.compile(r"\x1b\[[0-9;]*m")
clean_output = ansi_pattern.sub("", stripped)
# Detect common output patterns
if "→" in clean_output or "=>" in clean_output:
return {"type": "mapping", "lines": [l.strip() for l in clean_output.splitlines()]}
if re.search(r"^\d+ files?", clean_output, re.MULTILINE | re.IGNORECASE):
return {"type": "file_count", "count": int(re.search(r"(\d+)", clean_output).group(1))} # type: ignore[union-attr]
return {"type": "raw", "lines": [l.strip() for l in clean_output.splitlines()]}
class StreamObserver:
"""Observe and filter output from subprocess streams line-by-line.
Useful for parsing progress indicators, error patterns, or extracting
structured data from long-running command output.
"""
def __init__(self, patterns: dict[str, re.Pattern] | None = None) -> None:
self.patterns = patterns or {}
self.matches: list[dict[str, str]] = []
self.lines_consumed: int = 0
def observe(self, line: str) -> dict[str, str | None]:
"""Check a single line against registered patterns. Returns match info."""
result: dict[str, str | None] = {"match": None, "matched_text": None}
for name, pattern in self.patterns.items():
m = pattern.search(line)
if m:
result["match"] = name
result["matched_text"] = m.group(0)
self.matches.append({"pattern": name, "line": line.rstrip(), "text": m.group(0)})
self.lines_consumed += 1
return result
@property
def error_count(self) -> int:
return sum(1 for m in self.matches if m.get("match") == "error")
# Sentinel type alias — ExecutionResult is imported from engine module
from cli_agent.engine import ExecutionResult # noqa: E402, F401
BAD vs GOOD — File write:
# ❌ BAD — overwrites file atomically; if interrupted, file is corrupted
def write_bad(path: str, content: str) -> None:
with open(path, "w") as f:
f.write(content) # If this crashes, the original file is gone
# ✅ GOOD — atomic write via temp file + fsync + os.replace
def write_good(path: str, content: str) -> None:
"""Atomic write: never corrupts the original file on crash."""
resolved = Path(path).resolve()
fd, tmp = tempfile.mkstemp(dir=str(resolved.parent), prefix=".tmp_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(content)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, str(resolved))
except Exception:
Path(tmp).unlink(missing_ok=True)
raise # Re-raise to signal failure without corrupting the file
Pattern 3: Code Generation from Design Specifications
Converts architecture documents, markdown specs, or design files into executable code. The engine parses the design structure, identifies components and interfaces, then generates source files with proper typing, docstrings, and project structure — all within a safe output directory.
"""cli_agent/codegen.py — Code generation from design specification files."""
from __future__ annotations
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
@dataclass
class ComponentSpec:
"""Parsed component from a design specification."""
name: str
type: str = "module" # module, service, handler, model, view
description: str = ""
inputs: list[str] = field(default_factory=list)
outputs: list[str] = field(default_factory=list)
dependencies: list[str] = field(default_factory=list)
interface_text: str = "" # raw docstring for the component
@dataclass
class DesignDocument:
"""Complete parsed design document."""
title: str = ""
components: list[ComponentSpec] = field(default_factory=list)
architecture: str = ""
tech_stack: list[str] = field(default_factory=list)
raw_text: str = ""
def parse_design_document(text: str) -> DesignDocument:
"""Parse a markdown design document into structured components.
Handles common patterns:
- YAML frontmatter for metadata
- ## component headers followed by descriptions
- `interface` blocks with type annotations
- bullet lists for dependencies and inputs/outputs
Implements Law 1 (Early Exit): returns empty doc on blank input.
"""
if not text or not text.strip():
return DesignDocument(raw_text=text)
doc = DesignDocument(raw_text=text)
# Extract YAML frontmatter (between --- markers)
fm_match = re.match(r"^---\s*\n(.*?)\n---", text, re.DOTALL)
if fm_match:
meta_block = fm_match.group(1)
for line in meta_block.splitlines():
if line.startswith("title:"):
doc.title = line.split(":", 1)[1].strip().strip('"').strip("'")
elif line.startswith("tech_stack:") or line.startswith("framework:"):
stack_str = line.split(":", 1)[1].strip()
if stack_str:
doc.tech_stack = [s.strip().lstrip("-* ") for s in stack_str.split(",")]
# Extract components by ## headers
component_blocks = re.split(r"\n## ", text)
for block in component_blocks[1:]: # skip everything before first ##
lines = block.splitlines()
if not lines:
continue
name = lines[0].strip().rstrip(".")
component = ComponentSpec(name=name)
# Parse description (first non-empty line after header)
desc_lines: list[str] = []
for line in lines[1:]:
stripped = line.strip()
if not stripped or stripped.startswith("```"):
break
desc_lines.append(stripped)
component.description = " ".join(desc_lines)
# Parse interface block (``` interface ... ```)
iface_match = re.search(r"```\s*interface\s*\n(.*?)\n```", block, re.DOTALL)
if iface_match:
component.interface_text = iface_match.group(1).strip()
for param_line in component.interface_text.splitlines():
m2 = re.match(r"\w+\s*(?:->|\:)\s*(.*)", param_line.strip())
if m2:
param_name = param_line.strip().split(":")[0].strip()
param_type = m2.group(1).strip().rstrip(",")
if "input" in param_name.lower() or "param" in param_name.lower():
component.inputs.append(param_name)
elif "->" in param_line:
component.outputs.append(param_type)
# Parse dependencies from bullet lists
for line in lines:
dep_match = re.match(r"\s*[-*]\s*dependency:\s*(.+)", line, re.I)
if dep_match:
component.dependencies.append(dep_match.group(1).strip())
doc.components.append(component)
return doc
def generate_component_stub(comp: ComponentSpec, language: str = "python") -> str:
"""Generate a code stub for a component based on its parsed spec.
Produces typed function/class skeletons with docstrings derived
from the design description. Follows PEP 257 and PEP 484 conventions.
"""
if language == "python":
return _generate_python_stub(comp)
elif language == "typescript":
return _generate_typescript_stub(comp)
else:
raise ValueError(f"Unsupported language: {language}")
def _generate_python_stub(comp: ComponentSpec) -> str:
"""Generate a Python stub with type hints and docstring."""
lines = [f'"""{comp.description}"""', ""]
if comp.type == "module":
lines.append(f"class {comp.name.replace(' ', '_').title().replace('_', '')}:")
lines.append(f' """Component: {comp.name}."""')
lines.append("")
lines.append(" def __init__(self, **kwargs) -> None:") # type: ignore[return-value]
lines.append(" \"\"\"Initialize the component.\"\"\"")
for dep in comp.dependencies[:3]:
safe_name = re.sub(r'[^a-zA-Z0-9_]', '_', dep).lower()
lines.append(f" self._{safe_name} = kwargs.get('{safe_name}')")
lines.append("")
# Generate method stubs from interface text
if comp.interface_text:
for param_line in comp.interface_text.splitlines():
iface_match = re.match(r"(\w+)\s*->\s*(\w+)", param_line.strip())
if iface_match:
method_name = iface_match.group(1).lower().replace("-", "_")
return_type = iface_match.group(2)
lines.append(f" def {method_name}(self)")
lines.append(f' """TODO: implement {method_name}."""')
lines.append(" ...") # type: ignore[return-value]
lines.append("")
else:
comp_method = re.sub(r'\s+', '_', comp.name).lower()
param_list = ", ".join(
[f"input_: str"] + [f"{dep}: str" for dep in comp.dependencies[:2]]
)
return_type = "str" if comp.outputs else "None"
lines.extend([
f"def {comp_method}({param_list}) -> {return_type}:", # type: ignore[return-value]
f' """{comp.description}"""',
"...", # type: ignore[return-value]
"",
])
return "\n".join(lines)
def generate_project_from_design(
design_path: str,
output_dir: str,
language: str = "python",
) -> ExecutionResult: # type: ignore[name-defined]
"""Full pipeline: parse design → generate files for each component.
Creates a project structure with __init__.py, component modules,
and a test skeleton — all within the specified output directory.
"""
design = Path(design_path)
if not design.is_file():
return ExecutionResult( # type: ignore[name-defined]
success=Fals
…(truncated)