클로드가 없는 루프와 루틴

Claude's recent /loop and routines move coding agents from one-shot CLI assistants toward scheduled, event-driven workers. It is also perfectly possible to replicate multiple worktrees, loops, and routines with a small agent-written runner. The public description says loops repeat prompts inside a session, while routines run in Claude Code's cloud on schedules, API calls, or GitHub events. I do not know Anthropic's internals, but from the description it sounds like a Temporal-shaped workflow layer: persisted intent, wakeups, retries, and resumable sessions. I have been using loops and specs in Foundry for a while, but with Restate instead of Temporal. The local scripts around the repository do the unglamorous work: snapshot the current state, find the next bounded obligation in the spec, generate a precise prompt, run the coding agent, collect output, and leave a trail that can survive interruption. That is the practical version of spec driven development. The spec is not only prose for humans. It is the source of durable work items. The loop does not ask the model to remember the whole project or invent priority from scratch; it gives the model one narrow contract, a live repo, and a verification expectation. If the pass fails, the failure becomes input for the next pass instead of disappearing into a terminal scrollback. This is different from prompt driven development. A prompt can be a useful instruction, but it is usually disposable. A spec is durable. It has scope, constraints, acceptance criteria, and a place in the product contract. When the loop reads from the spec, the agent is not just following a clever instruction; it is advancing a named obligation that can be audited later. That matters for systems work because correctness is rarely visible in one file. The useful artifact is the chain from requirement to patch to verification. The surrounding automation is deliberately plain. One part gathers context without changing durable progress. Another part chooses or prepares a small unit of work. Another runs the agent and captures logs, prompts, patches, status, and failure output. A proof pass can then exercise real workflows and attach evidence. None of that requires the model to be consistent over days. The model can be interrupted, replaced, upgraded, or wrong; the loop still has enough state to decide what to do next. It also changes review. Instead of asking whether an agent was impressive, I can ask whether a specific spec claim moved forward, whether the produced patch is narrow, whether verification actually covers the changed behavior, and whether the remaining gap is recorded. That makes automation useful even when the answer is "not done yet." The loop still produces a better next input. Restate fits this shape because the workflow owns retry, recovery, and wakeups while the agent owns reading code, editing files, and reporting verification. In other words, the agent is replaceable. The durable system is the combination of spec, workflow state, logs, patches, commits, and evidence. Instead of a heroic session, the unit of progress becomes an auditable repeatable cycle. That is what makes long-running agent work boring enough to trust. foundry_restate_codex_loop.py #!/usr/bin/env python3 """Standalone durable Codex loop for Foundry specs. This script has two modes: 1. Local mode, available with the repository's normal Python: it runs audit/implementation cycles, streams all child output, and writes a durable local state/log trail under .foundry-restate-loop/. 2. Restate service mode, available when running Python >= 3.11 with restate_sdk[serde] and hypercorn installed: it exposes the same cycle as a Restate Workflow so Restate can resume, schedule, and inspect long-running Codex/spec implementation work. It never edits .foundry-progress directly. Spec progress goes through scripts/foundry_spec_runner.py status, list, run, prompt, and logs. """ from __future__ import annotations import argparse import contextlib import datetime as dt import difflib import hashlib import html import json import os from pathlib import Path import shutil import shlex import signal import subprocess import sys import threading import time from typing import Any, Dict, Iterable, List, Optional, Tuple ROOT = Path(__file__).resolve().parents[1] STATE_DIR = ROOT / ".foundry-restate-loop" LOG_DIR = STATE_DIR / "logs" RESTATE_LOG_DIR = STATE_DIR / "restate-logs" PROMPT_DIR = STATE_DIR / "prompts" AUDIT_DIR = STATE_DIR / "audits" CHANGE_DIR = STATE_DIR / "changes" WORKTREE_DIR = STATE_DIR / "worktrees" STATE_FILE = STATE_DIR / "state.json" DECISIONS_FILE = STATE_DIR / "decisions.jsonl" EVENTS_FILE = STATE_DIR / "events.jsonl" INTERRUPTED_FILE = STATE_DIR / "last-interrupted.json" STOP_FILE = STATE_DIR / "stop-request.json" RESTATE_SERVER_FILE = STATE_DIR / "serve-restate.json" DEFAULT_RESTATE_DATA_DIR = ROOT / "restate-data" DEFAULT_CODEX_CMD = "codex exec -" DEFAULT_VERIFY_CMD = "python3 -m py_compile scripts/foundry_restate_codex_loop.py" DEFAULT_REMOTE_VERIFY_HOST = "selectel-day" STATE_OUTPUT_PREVIEW_CHARS = 4000 MAX_SNAPSHOT_TEXT_BYTES = 2_000_000 DEFAULT_DIFF_LIMIT = 60000 CHANGE_FORMATS = ("text", "json", "patch") CODEX_STREAM_MODES = ("auto", "raw", "assistant", "quiet") DEFAULT_ISOLATED_BASE = "HEAD" DEFAULT_COMMIT_BRANCH_PREFIX = "codex" SPINNER_FRAMES = ("|", "/", "-", "\\") ACTIVE: Dict[str, Any] = { "cycle": None, "step": None, "pid": None, "prompt_path": None, "log_path": None, "codex_running": False, } STOP_REQUESTED = False def utc_now() -> str: return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() def stamp() -> str: return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).strftime("%Y%m%dT%H%M%SZ") def slug(value: str, max_len: int = 56) -> str: chars: List[str] = [] prev_dash = False for ch in value.lower(): if ch.isalnum(): chars.append(ch) prev_dash = False elif not prev_dash: chars.append("-") prev_dash = True result = "".join(chars).strip("-") return result[:max_len].strip("-") or "item" def rel(path: Optional[Path]) -> Optional[str]: if path is None: return None try: return str(path.relative_to(ROOT)) except ValueError: return str(path) def ensure_dirs() -> None: STATE_DIR.mkdir(parents=True, exist_ok=True) LOG_DIR.mkdir(parents=True, exist_ok=True) RESTATE_LOG_DIR.mkdir(parents=True, exist_ok=True) PROMPT_DIR.mkdir(parents=True, exist_ok=True) AUDIT_DIR.mkdir(parents=True, exist_ok=True) CHANGE_DIR.mkdir(parents=True, exist_ok=True) WORKTREE_DIR.mkdir(parents=True, exist_ok=True) def atomic_write_json(path: Path, value: Dict[str, Any]) -> None: ensure_dirs() tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp") tmp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") tmp.replace(path) def append_jsonl(path: Path, value: Dict[str, Any]) -> None: ensure_dirs() with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(value, sort_keys=True) + "\n") fh.flush() with contextlib.suppress(OSError): os.fsync(fh.fileno()) def event_fields_text(fields: Dict[str, Any]) -> str: parts: List[str] = [] for key, value in fields.items(): if value is None: continue if isinstance(value, (dict, list, tuple)): text = json.dumps(value, sort_keys=True) else: text = str(value) parts.append(f"{key}={shlex.quote(text)}") return " ".join(parts) def record_loop_event(event: str, cycle: Optional[int] = None, **fields: Any) -> Dict[str, Any]: payload: Dict[str, Any] = {"time": utc_now(), "event": event} if cycle is not None: payload["cycle"] = cycle payload.update({key: value for key, value in fields.items() if value is not None}) append_jsonl(EVENTS_FILE, payload) return payload def log_loop_event(args: Optional[argparse.Namespace], event: str, message: str, cycle: Optional[int] = None, **fields: Any) -> Dict[str, Any]: payload = record_loop_event(event, cycle=cycle, **fields) if args is not None and not bool(getattr(args, "event_log", True)): return payload visible_fields = {key: value for key, value in payload.items() if key not in {"time", "event", "cycle"}} prefix = f"[foundry-loop] {event}" if cycle is not None: prefix += f" cycle={cycle}" suffix = event_fields_text(visible_fields) line = f"{prefix} {message}" if suffix: line += f" | {suffix}" print(line, file=sys.stderr, flush=True) return payload def tail_jsonl(path: Path, lines: int) -> List[str]: if not path.exists(): return [] try: return path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:] except OSError as exc: return [json.dumps({"time": utc_now(), "event": "read-error", "path": rel(path), "error": str(exc)}, sort_keys=True)] def read_json(path: Path) -> Dict[str, Any]: if not path.exists(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, UnicodeDecodeError) as exc: return {"error": str(exc)} return data if isinstance(data, dict) else {"error": "state root is not an object"} def trim_text(value: str, max_chars: int) -> str: if len(value) Optional[Dict[str, Any]]: if result is None: return None compact: Dict[str, Any] = {} for key in ( "step", "exitCode", "startedAt", "completedAt", "logPath", "cwd", "auditInput", "auditReport", "auditHtml", "promptPath", "skipped", ): if key in result: compact[key] = result[key] argv = result.get("argv") if isinstance(argv, list): compact["command"] = shell_join(str(part) for part in argv) output = result.get("outputTail") if output: compact["outputPreview"] = trim_text(str(output), STATE_OUTPUT_PREVIEW_CHARS) return compact def shell_join(argv: Iterable[str]) -> str: return " ".join(shlex.quote(part) for part in argv) def codex_argv(args: argparse.Namespace) -> List[str]: argv = shlex.split(args.codex_cmd) if not argv: return argv extra: List[str] = [] if getattr(args, "dangerously_bypass_approvals_and_sandbox", False): extra.append("--dangerously-bypass-approvals-and-sandbox") if getattr(args, "search", False): extra.append("--search") extra = [flag for flag in extra if flag not in argv] if not extra: return argv if Path(argv[0]).name == "codex": return [argv[0], *extra, *argv[1:]] return argv def effective_codex_cmd(args: argparse.Namespace) -> str: return shell_join(codex_argv(args)) def repo_path(value: str) -> Path: path = Path(value).expanduser() if not path.is_absolute(): path = ROOT / path return path def work_root(args: Optional[argparse.Namespace] = None) -> Path: if args is not None and getattr(args, "_work_root", None): return Path(str(args._work_root)) return ROOT def read_prompt_file(path_text: str) -> str: path = repo_path(path_text) if not path.exists(): raise SystemExit(f"prompt file does not exist: {path}") if not path.is_file(): raise SystemExit(f"prompt path is not a file: {path}") return path.read_text(encoding="utf-8") def operator_prompt(args: argparse.Namespace) -> str: parts: List[str] = [] prompt_file = getattr(args, "prompt_file", None) if prompt_file: parts.append(read_prompt_file(str(prompt_file)).strip()) prompt = getattr(args, "prompt", None) if prompt: parts.append(str(prompt).strip()) return "\n\n".join(part for part in parts if part).strip() def operator_prompt_source(args: argparse.Namespace) -> Optional[str]: sources: List[str] = [] prompt_file = getattr(args, "prompt_file", None) if prompt_file: sources.append(f"file:{rel(repo_path(str(prompt_file)))}") if getattr(args, "prompt", None): sources.append("inline") return ",".join(sources) if sources else None def verification_policy(args: argparse.Namespace) -> Dict[str, Any]: return { "remoteHost": str(getattr(args, "remote_verify_host", None) or DEFAULT_REMOTE_VERIFY_HOST), "allowLocalVerification": bool(getattr(args, "allow_local_verification", False)), "localVerificationReason": getattr(args, "local_verification_reason", None), } def verification_policy_prompt(args: argparse.Namespace) -> str: policy = verification_policy(args) remote_host = policy["remoteHost"] if policy["allowLocalVerification"]: reason = policy.get("localVerificationReason") or ( f"Remote SSH to `{remote_host}` is blocked or unavailable in this execution environment." ) return f"""Verification policy: - Prefer the repo-required remote verification path on `{remote_host}` when the check is relevant and SSH is available. - This run explicitly allows local verification fallback if SSH to `{remote_host}` is blocked by sandbox, permission, or network restrictions. - When falling back locally, run the narrowest relevant local checks that exercise the changed code path. - In the completion response, state the exact remote verification blocker and mark acceptance as local-only pending remote proof. - Local fallback reason: {reason} """ return f"""Verification policy: - Prefer the repo-required remote verification path on `{remote_host}` when the check is relevant. - Do not silently downgrade remote-required verification to local-only checks. - If SSH to `{remote_host}` is blocked, report the blocker and leave remote verification pending; local checks may still be run as supporting evidence. """ def normalize_changes_format(value: str) -> str: if value not in CHANGE_FORMATS: raise SystemExit(f"changes format must be one of: {', '.join(CHANGE_FORMATS)}") return value def normalize_codex_stream(value: str) -> str: if value not in CODEX_STREAM_MODES: raise SystemExit(f"codex stream must be one of: {', '.join(CODEX_STREAM_MODES)}") return value def normalize_run_args(args: argparse.Namespace) -> argparse.Namespace: if getattr(args, "plan_only", False) and getattr(args, "implement_only", False): raise SystemExit("--plan-only and --implement-only are mutually exclusive") if getattr(args, "plan_only", False) and getattr(args, "commit", False): raise SystemExit("--commit cannot be used with --plan-only") if getattr(args, "plan_only", False): args.skip_audit = False if getattr(args, "implement_only", False): args.skip_audit = True dangerous_codex = bool(getattr(args, "dangerously_bypass_approvals_and_sandbox", False)) if getattr(args, "isolated_worktree", None) is None: args.isolated_worktree = bool(getattr(args, "changes_only", False) or getattr(args, "commit", False) or dangerous_codex) if getattr(args, "commit", False) and not bool(getattr(args, "isolated_worktree", False)): raise SystemExit("--commit requires --isolated-worktree or --changes-only so unrelated current-checkout changes are not committed") if dangerous_codex and not bool(getattr(args, "isolated_worktree", False)): raise SystemExit("--dangerously-bypass-approvals-and-sandbox requires isolated worktree mode; remove --no-isolated-worktree") args.remote_verify_host = str(getattr(args, "remote_verify_host", None) or DEFAULT_REMOTE_VERIFY_HOST) args.changes_format = normalize_changes_format(str(getattr(args, "changes_format", "text"))) args.codex_stream = normalize_codex_stream(str(getattr(args, "codex_stream", "auto"))) return args def animation_enabled(args: Optional[argparse.Namespace] = None) -> bool: if args is not None and hasattr(args, "animation"): return bool(args.animation) ret