#!/usr/bin/env python3 """Run one governed goal through a local worker and a cloud reviewer. The job file is tenant-scoped and created by the Control Panel. Credentials are passed only through the child environment and are never copied into job, trace, audit, or metric records. """ import argparse import fcntl import hashlib import json import os import re import shlex import subprocess import tempfile import time import urllib.error import urllib.request from datetime import datetime, timezone def project_root() -> str: current = os.path.abspath(os.path.dirname(__file__)) while current != os.path.dirname(current): if os.path.isdir(os.path.join(current, ".specify")): return current current = os.path.dirname(current) raise SystemExit("GOAL_ROOT_NOT_FOUND") ROOT = project_root() BIN = os.path.join(ROOT, "packages", "casan-harness", "scripts", "bash") MODEL_ROUTER = os.environ.get("CASAN_GOAL_MODEL_ROUTER") or os.path.join(BIN, "model-router.sh") SECURITY = os.path.join(BIN, "security-check.sh") STATE_ROOT = os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify") PROJECT_REGISTRY = os.path.join(ROOT, "packages", "casan-harness", "level5", "project-registry.json") APPROVAL_INBOX = os.path.join(BIN, "approval-inbox.py") CONTEXT_EXTENSIONS = {".md", ".txt", ".json", ".yaml", ".yml", ".ts", ".tsx", ".js", ".mjs", ".py", ".sh", ".prisma", ".css", ".html"} CONTEXT_IGNORED = {"node_modules", ".git", "dist", "build", "coverage", ".vite", "tmp", "logs", "__pycache__"} SENSITIVE_NAMES = {".env", ".env.local", "credentials", "credentials.json", "secrets.json", "id_rsa", "id_ed25519"} def now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def sha(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def atomic_json(path: str, payload: dict) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=".goal-", dir=os.path.dirname(path), text=True) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) handle.write("\n") os.chmod(temporary, 0o600) os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def load_json(path: str) -> dict: with open(path, encoding="utf-8") as handle: payload = json.load(handle) if not isinstance(payload, dict): raise ValueError("GOAL_JOB_INVALID") return payload def append_jsonl(path: str, payload: dict) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "a", encoding="utf-8") as handle: fcntl.flock(handle.fileno(), fcntl.LOCK_EX) handle.write(json.dumps(payload, ensure_ascii=False) + "\n") handle.flush() fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def trace_path(goal_id: str) -> str: directory = os.environ.get("CASAN_TRACE_EVENTS_DIR") or os.path.join(STATE_ROOT, "logs", "trace-events") return os.path.join(directory, f"{goal_id}.jsonl") def emit(goal_id: str, gate_id: str, status: str, reason: str, evidence=None) -> None: append_jsonl(trace_path(goal_id), { "timestamp": now(), "trace_id": goal_id, "gate_id": gate_id, "status": status, "reason": reason, "evidence": evidence or {}, }) def update_job(path: str, **changes) -> dict: job = load_json(path) job.update(changes) job["updated_at"] = now() atomic_json(path, job) return job def stage(path: str, stage_id: str, status: str, detail: str, provider="", model="") -> None: job = load_json(path) stages = list(job.get("stages", [])) row = next((item for item in stages if item.get("id") == stage_id), None) if row is None: row = {"id": stage_id} stages.append(row) row.update({"status": status, "detail": detail, "provider": provider, "model": model, "updated_at": now()}) job["stages"] = stages job["updated_at"] = now() atomic_json(path, job) def registered_project(project_id: str) -> dict: registry = load_json(PROJECT_REGISTRY) entry = next((item for item in registry.get("projects", []) if item.get("project_id") == project_id and item.get("status") == "active"), None) if not entry: raise ValueError("goal_project_not_allowed") raw_roots = entry.get("context_roots") or [entry.get("domain_root")] root_real = os.path.realpath(ROOT) resolved = [] for relative in raw_roots: if not isinstance(relative, str) or not relative: raise ValueError("goal_context_root_invalid") absolute = os.path.realpath(os.path.join(ROOT, relative)) if absolute != root_real and not absolute.startswith(root_real + os.sep): raise ValueError("goal_context_root_denied") if not os.path.exists(absolute): raise ValueError("goal_context_root_missing") resolved.append((relative, absolute)) return {"project_id": project_id, "domain": entry.get("domain", project_id), "domain_root": entry.get("domain_root", ""), "roots": resolved} def redact_context(text: str) -> str: patterns = [ (r"(?im)^[^\n]*(?:api[_-]?key|secret|password|token)[^:=\n]*[:=]\s*[^\s\n]+", "[REDACTED SECRET]"), (r"(?i)bearer\s+[A-Za-z0-9._~+/-]{12,}", "Bearer [REDACTED]"), (r"-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----", "[REDACTED PRIVATE KEY]"), ] redacted = text for pattern, replacement in patterns: redacted = re.sub(pattern, replacement, redacted) return redacted def context_excerpt_is_sensitive(text: str) -> bool: """Exclude source excerpts that would turn the shared snapshot into unsafe input.""" normalized = " ".join(text.lower().split()) blocked_phrases = ( "ignore previous instruction", "ignore prior instruction", "drop table", "shutdown system", "export secrets", "dump database", # The security gate normalizes identifiers, so Prisma's `onDelete` # otherwise looks like an imperative delete request. "ondelete", ) if any(phrase in normalized for phrase in blocked_phrases): return True return bool(re.search( r"(?i)(?:api[_-]?key|access[_-]?token|refresh[_-]?token|password|jwt[_-]?secret|secret)" r"\s*[:=]\s*(?!\[REDACTED\])\S+", text, )) def context_candidates(project: dict, goal: str): terms = {term.lower() for term in re.findall(r"[A-Za-z0-9_-]{3,}", goal)} candidates = [] seen = set() for relative_root, absolute_root in project["roots"]: if os.path.isfile(absolute_root): paths = [absolute_root] else: paths = [] for base, directories, files in os.walk(absolute_root): directories[:] = [item for item in directories if item not in CONTEXT_IGNORED and not item.startswith('.')] paths.extend(os.path.join(base, name) for name in files) for path in paths: relative = os.path.relpath(path, ROOT) if relative in seen or os.path.basename(path).lower() in SENSITIVE_NAMES: continue seen.add(relative) extension = os.path.splitext(path)[1].lower() if extension not in CONTEXT_EXTENSIONS or os.path.getsize(path) > 256_000: continue try: with open(path, encoding="utf-8", errors="replace") as handle: raw = handle.read(16_000) except OSError: continue haystack = (relative + "\n" + raw[:4000]).lower() score = sum(4 if term in relative.lower() else 1 for term in terms if term in haystack) if relative.endswith(("README.md", "architecture.md", "technical_architecture.md", "package.json")): score += 3 candidates.append((score, relative, raw)) return sorted(candidates, key=lambda item: (-item[0], item[1])) def build_context(job_path: str, project_id: str, goal: str): project = registered_project(project_id) candidates = context_candidates(project, goal) excerpts, manifest_files, characters = [], [], 0 max_files, max_characters, per_file = 16, 7_000, 1_200 for score, relative, raw in candidates: if len(excerpts) >= max_files or characters >= max_characters: break excerpt = redact_context(raw[:min(per_file, max_characters - characters)]).strip() if not excerpt or context_excerpt_is_sensitive(excerpt): continue excerpts.append(f"### FILE: {relative}\n{excerpt}") characters += len(excerpt) manifest_files.append({"path": relative, "sha256": sha(raw), "characters": len(excerpt), "relevance": score}) bundle = "\n\n".join(excerpts) allowed, safe_bundle = scan(bundle, "input") if not allowed: raise ValueError("goal_context_security_blocked") bundle = safe_bundle base = os.path.splitext(job_path)[0] bundle_path, manifest_path = base + ".context.txt", base + ".context.json" with open(bundle_path, "w", encoding="utf-8") as handle: handle.write(bundle + "\n") os.chmod(bundle_path, 0o600) manifest = { "project_id": project_id, "domain": project["domain"], "domain_root": project["domain_root"], "generated_at": now(), "files": manifest_files, "file_count": len(manifest_files), "characters": characters, "truncated": len(manifest_files) < len(candidates), "bundle_sha256": sha(bundle), } atomic_json(manifest_path, manifest) return bundle, manifest, os.path.relpath(manifest_path, ROOT) def requests_side_effect(goal: str) -> bool: normalized = " ".join(goal.lower().split()) patterns = [ r"^(hãy\s+)?(làm luôn|sửa|thay đổi|triển khai|thực hiện|hoàn thành|chạy|tạo|xóa|cài đặt|commit|push)\b", r"^(please\s+)?(implement|fix|change|deploy|run|create|delete|install|commit|push)\b", r"\b(thực hiện thao tác|sửa code|ghi file|mở pull request|create a pull request)\b", ] return any(re.search(pattern, normalized) for pattern in patterns) def extract_patch(text: str) -> str: match = re.search(r"```(?:diff|patch)\s*\n(.*?)```", text, re.DOTALL | re.IGNORECASE) candidate = match.group(1) if match else text start = candidate.find("diff --git ") if start < 0: raise ValueError("goal_patch_missing") patch = candidate[start:].strip() + "\n" if len(patch.encode("utf-8")) > 200_000: raise ValueError("goal_patch_too_large") return patch def validate_patch_check(patch: str) -> None: """Require a candidate diff to apply before it can pass H2 or H3.""" with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".patch") as handle: handle.write(patch) handle.flush() try: check = subprocess.run( ["git", "apply", "--check", "--whitespace=error", handle.name], cwd=ROOT, capture_output=True, text=True, timeout=30, ) except FileNotFoundError as error: raise ValueError("goal_patch_validator_unavailable") from error if check.returncode != 0: detail = (check.stderr or check.stdout or "git_apply_check_failed").strip().replace("\n", " ")[:160] raise ValueError(f"goal_patch_check_failed:{detail}") def validate_write_output(text: str) -> str: """Fail at the producing harness when a write-intent reply is not a diff. Keeping this separate from storage avoids creating an artifact before the reviewer has completed, while ensuring H2/H3 accurately identify a model that violated the required output contract. """ patch = extract_patch(text) validate_patch_check(patch) return patch def patch_repair_attempts() -> int: """Return the bounded number of chances to repair an invalid model diff.""" try: configured = int(os.environ.get("CASAN_GOAL_PATCH_REPAIR_ATTEMPTS", "3")) except ValueError: configured = 3 return min(max(configured, 0), 3) def patch_output_tokens() -> int: """Keep a multi-hunk patch from being cut at the generic chat limit.""" try: configured = int(os.environ.get("CASAN_GOAL_PATCH_MAX_OUTPUT_TOKENS", "6000")) except ValueError: configured = 6000 return min(max(configured, 512), 8192) def patch_repair_models(primary_model: str): """Use direct cloud -> gateway -> local order for one bounded H2 recovery.""" configured = [item.strip() for item in os.environ.get("CASAN_GOAL_PATCH_REPAIR_MODELS", "").split(",") if item.strip()] unique, seen = [], set() for candidate in configured + [primary_model]: if candidate and candidate not in seen: seen.add(candidate) unique.append(candidate) return unique def provider_for_model(model: str) -> str: if model.startswith("openai:"): return "openai" if model.startswith("anthropic:"): return "anthropic" if model.startswith("openai-compatible:"): return "omniroute" if model.startswith("ollama:"): return "ollama" return "model" def repair_write_output(model: str, original_prompt: str, invalid_output: str, contract_error="goal_patch_missing"): """Ask the producing model to repair format only, without relaxing H2. A write-intent goal may never advance to approval without a checked unified diff. Models occasionally answer with an implementation plan despite the contract, so one bounded repair avoids treating a recoverable formatting lapse as a final H2 failure. The caller still H4-scans and validates the result before accepting it. """ if patch_repair_attempts() == 0: return False, "", {}, "goal_patch_missing" repair_prompt = ( f"Your previous response violated the required write-output contract: {contract_error}. " "Return ONLY one complete, applicable unified git diff inside a ```diff fence. The first non-empty line inside the fence MUST be `diff --git `. " "Every hunk must be complete and the patch must pass `git apply --check`. Do not emit `index` lines, placeholder hashes, commentary, plans, summaries, or side effects. Preserve the original objective and workspace restrictions.\n\n" f"ORIGINAL CONTRACT:\n{original_prompt}\n\nPREVIOUS INVALID RESPONSE:\n{invalid_output[:8000]}" ) candidates = patch_repair_models(model)[:patch_repair_attempts()] last_metadata, last_reason = {}, "goal_patch_missing" attempts = [] for attempt_number, candidate in enumerate(candidates, start=1): ok, output, metadata, reason = call_model(candidate, repair_prompt, candidate.startswith(("openai:", "anthropic:", "openai-compatible:")), max_output_tokens=patch_output_tokens()) metadata = dict(metadata) metadata["repair_model"] = candidate metadata["repair_attempt"] = attempt_number last_metadata = metadata if not ok: last_reason = f"goal_patch_repair_failed:{reason}" attempts.append({ "attempt": attempt_number, "provider": provider_for_model(candidate), "model": candidate, "status": "failed", "reason": last_reason, }) continue try: validate_write_output(output) except ValueError as error: last_reason = str(error) attempts.append({ "attempt": attempt_number, "provider": provider_for_model(candidate), "model": candidate, "status": "failed", "reason": last_reason, }) continue attempts.append({ "attempt": attempt_number, "provider": provider_for_model(candidate), "model": candidate, "status": "pass", "reason": "ok", }) metadata["repair_attempts"] = attempts return True, output, metadata, "ok" last_metadata["repair_attempts"] = attempts return False, "", last_metadata, f"goal_patch_repair_invalid:{last_reason}" def merge_usage(primary: dict, additional: dict) -> dict: """Preserve provider metadata while adding numeric usage from a repair call.""" merged = dict(primary) for key, value in additional.items(): if isinstance(value, (int, float)) and isinstance(merged.get(key), (int, float)): merged[key] = merged[key] + value elif key not in merged: merged[key] = value return merged def validate_and_store_patch(job_path: str, job: dict, patch: str) -> dict: roots = [str(value).strip("/") for value in job.get("workspace", {}).get("context_roots", [])] changed = [] header_paths = [] for line in patch.splitlines(): if line.startswith("diff --git "): fields = shlex.split(line) if len(fields) != 4: raise ValueError("goal_patch_header_invalid") header_paths.extend(fields[2:]) elif line.startswith(("--- ", "+++ ", "rename from ", "rename to ", "copy from ", "copy to ")): value = line.split(" ", 1)[1].split("\t", 1)[0] header_paths.append(value) for raw in header_paths: path = raw.strip() if path == "/dev/null": continue if path.startswith(("a/", "b/")): path = path[2:] if path.startswith("/") or ".." in path.split("/"): raise ValueError("goal_patch_path_denied") if not any(path == root or path.startswith(root + "/") for root in roots): raise ValueError(f"goal_patch_outside_workspace:{path}") changed.append(path) if not changed or len(set(changed)) > 20: raise ValueError("goal_patch_file_count_invalid") artifact = job_path[:-5] + ".patch" with open(artifact, "x", encoding="utf-8") as handle: handle.write(patch) handle.flush() os.fsync(handle.fileno()) os.chmod(artifact, 0o600) check = subprocess.run(["git", "apply", "--check", "--whitespace=error", artifact], cwd=ROOT, capture_output=True, text=True, timeout=30) if check.returncode != 0: os.unlink(artifact) raise ValueError("goal_patch_check_failed:" + (check.stderr or check.stdout).strip()[:160]) return {"path": os.path.relpath(artifact, ROOT), "sha256": sha(patch), "files": sorted(set(changed)), "bytes": len(patch.encode("utf-8")), "status": "awaiting_approval", "preview": patch[:50_000]} def submit_side_effect(job: dict, manifest: dict, patch_artifact: dict) -> dict: payload = { "goal_id": job["id"], "project_id": job["project"], "context_manifest_hash": manifest["bundle_sha256"], "requested_operation": job["goal"], "patch_sha256": patch_artifact["sha256"], "patch_path": patch_artifact["path"], "changed_files": patch_artifact["files"], } result = subprocess.run([ "python3", APPROVAL_INBOX, "submit", "--project", job["project"], "--action", "goal.workspace.execute", "--target", job["project"], "--risk", "high", "--sensitive", "--proposer", job["actor"], "--reason", "Goal requests a workspace side effect; execution remains disabled until approval", "--payload", json.dumps(payload, ensure_ascii=False), ], cwd=ROOT, capture_output=True, text=True, env=os.environ.copy(), timeout=30) if result.returncode != 0: raise RuntimeError((result.stderr or result.stdout or "goal_approval_submit_failed").strip()) return json.loads(result.stdout) def scan(text: str, mode: str): with tempfile.TemporaryDirectory() as directory: source = os.path.join(directory, "input.txt") output = os.path.join(directory, "output.txt") with open(source, "w", encoding="utf-8") as handle: handle.write(text) environment = os.environ.copy() # The goal workflow already records and enforces its H4 boundary. Keep # deterministic injection/secret/PII checks active, but do not turn a # temporary semantic-classifier outage into a false-positive block. environment["CASAN_SECURITY_STRICT"] = "0" result = subprocess.run( ["bash", SECURITY, source, output, mode], cwd=ROOT, capture_output=True, text=True, env=environment, timeout=60, ) safe = "" if os.path.isfile(output): with open(output, encoding="utf-8") as handle: safe = handle.read().strip() return result.returncode == 0, safe def call_model(model: str, prompt: str, cloud: bool, timeout_seconds=None, max_output_tokens=None): if not model: return False, "", {}, "model_unconfigured" with tempfile.TemporaryDirectory() as directory: prompt_path = os.path.join(directory, "prompt.txt") output_path = os.path.join(directory, "output.json") with open(prompt_path, "w", encoding="utf-8") as handle: handle.write(prompt) environment = os.environ.copy() environment["CASAN_PREFLIGHT"] = "1" if cloud else "0" # Long Markdown objectives need enough time for a local 9B model to # ingest the brief and produce a bounded plan. The outer goal timeout # remains the hard ceiling; this only raises the router's 60s default. environment.setdefault("CASAN_MODEL_TIMEOUT_SEC", os.environ.get("CASAN_GOAL_LOCAL_TIMEOUT_SEC", "240") if not cloud else "120") configured_output = os.environ.get("CASAN_GOAL_MAX_OUTPUT_TOKENS", "1400") if max_output_tokens is not None: configured_output = str(max_output_tokens) environment["CASAN_MODEL_GENERATE_MAX_TOKENS"] = configured_output timeout = max(1, int(timeout_seconds or os.environ.get("CASAN_GOAL_MODEL_TIMEOUT", "300"))) environment["CASAN_MODEL_TIMEOUT_SEC"] = str(min( int(environment.get("CASAN_MODEL_TIMEOUT_SEC", timeout)), timeout )) try: result = subprocess.run( ["bash", MODEL_ROUTER, prompt_path, output_path, "--role", "generate", "--model", model], cwd=ROOT, capture_output=True, text=True, env=environment, timeout=timeout, ) except subprocess.TimeoutExpired: return False, "", {}, "model_timeout" if result.returncode != 0 or not os.path.isfile(output_path): raw = (result.stderr or result.stdout or "").strip().splitlines() detail = raw[-1][:180] if raw else "no_detail" detail = re.sub(r"(?i)(bearer|api[_-]?key|token)[=: ]+\S+", r"\1=[redacted]", detail) return False, "", {}, f"model_exit_{result.returncode}:{detail}" try: output = load_json(output_path) except (OSError, ValueError, json.JSONDecodeError): return False, "", {}, "model_output_invalid" text = str(output.get("text") or "").strip() if not text: return False, "", {}, "model_output_empty" metadata = { "input_tokens": int(output.get("input_tokens") or 0), "output_tokens": int(output.get("output_tokens") or 0), "latency_ms": int(output.get("latency_ms") or 0), } return True, text, metadata, "ok" def call_account_model(provider: str, prompt: str, timeout_seconds=None): base_url = os.environ.get("CASAN_AUTH_BRIDGE_URL", "").rstrip("/") token = os.environ.get("CASAN_AUTH_BRIDGE_TOKEN", "") if provider not in {"codex", "claude"} or not base_url or not token: return False, "", {}, "account_bridge_unavailable" request = urllib.request.Request( f"{base_url}/v1/models/{provider}/generate", data=json.dumps({"prompt": prompt}).encode("utf-8"), headers={"Content-Type": "application/json", "X-CASAN-Bridge-Token": token}, method="POST", ) try: with urllib.request.urlopen(request, timeout=max(1, int(timeout_seconds or 310))) as response: payload = json.loads(response.read().decode("utf-8")) except (urllib.error.URLError, TimeoutError, ValueError): return False, "", {}, "account_bridge_failed" text = str(payload.get("text") or "").strip() if not payload.get("success") or not text: return False, "", {}, str(payload.get("reason") or "account_model_failed")[:120] usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {} metadata = { "input_tokens": int(usage.get("input_tokens") or 0), "output_tokens": int(usage.get("output_tokens") or 0), "latency_ms": int(payload.get("latency_ms") or 0), "model": str(payload.get("model") or f"{provider}-account-default"), } return True, text, metadata, "ok" def reviewer_failure_retryable(reason: str) -> bool: """Only transient/provider failures may advance to another reviewer.""" normalized = str(reason or "").lower() non_retryable = ( "endpoint_not_allowed", "model_unconfigured", "model_not_discovered", "invalid_request", "content_policy", "security_blocked", ) return not any(marker in normalized for marker in non_retryable) def reviewer_candidates(account_provider: str, cloud_model: str, local_model: str): """Build a stable, de-duplicated account/cloud -> OmniRoute -> local chain.""" candidates = [] if account_provider: candidates.append({ "kind": "account", "provider": f"{account_provider}-account", "model": f"{account_provider}-account-default", "value": account_provider, }) fallback_model = os.environ.get("CASAN_GOAL_CLOUD_FALLBACK_MODEL", "").strip() if fallback_model: fallback_provider = "omniroute" if fallback_model.startswith("openai-compatible:") else "cloud-fallback" candidates.append({"kind": "model", "provider": fallback_provider, "model": fallback_model, "value": fallback_model}) elif cloud_model: candidates.append({"kind": "model", "provider": os.environ.get("CASAN_GOAL_CLOUD_PROVIDER", "cloud"), "model": cloud_model, "value": cloud_model}) for model in [item.strip() for item in os.environ.get("CASAN_GOAL_CLOUD_MODELS", "").split(",") if item.strip()]: provider = "anthropic" if model.startswith("anthropic:") else "openai" if model.startswith("openai:") else "cloud" candidates.append({"kind": "model", "provider": provider, "model": model, "value": model}) raw_omniroute = os.environ.get("CASAN_GOAL_OMNIROUTE_MODELS", "") omniroute_models = [item.strip() for item in raw_omniroute.split(",") if item.strip()] for model in omniroute_models: value = model if ":" in model else f"openai-compatible:{model}" candidates.append({"kind": "model", "provider": "omniroute", "model": value, "value": value}) local_reviewer = os.environ.get("CASAN_GOAL_LOCAL_REVIEWER_MODEL", "").strip() or local_model if local_reviewer: candidates.append({"kind": "model", "provider": os.environ.get("CASAN_GOAL_LOCAL_PROVIDER", "local-policy"), "model": local_reviewer, "value": local_reviewer}) unique, seen = [], set() for candidate in candidates: key = (candidate["kind"], candidate["value"]) if key not in seen: seen.add(key) unique.append(candidate) return unique def run_reviewer_chain(job_path: str, prompt: str, account_provider: str, cloud_model: str, local_model: str): max_attempts = max(1, min(int(os.environ.get("CASAN_GOAL_REVIEWER_MAX_ATTEMPTS", "5")), 10)) deadline_seconds = max(1, min(int(os.environ.get("CASAN_GOAL_REVIEWER_DEADLINE_SEC", "360")), 900)) deadline = time.monotonic() + deadline_seconds ledger = [] update_job(job_path, reviewer_attempts=ledger) last_reason = "reviewer_candidates_unavailable" candidates = reviewer_candidates(account_provider, cloud_model, local_model) if len(candidates) > max_attempts and candidates[-1].get("provider") == os.environ.get("CASAN_GOAL_LOCAL_PROVIDER", "local-policy"): candidates = candidates[:max_attempts - 1] + [candidates[-1]] if max_attempts > 1 else [candidates[-1]] else: candidates = candidates[:max_attempts] for candidate in candidates: remaining = int(deadline - time.monotonic()) if remaining <= 0: last_reason = "reviewer_deadline_exceeded" break attempt_number = len(ledger) + 1 stage(job_path, "cloud-reviewer", "running", f"Reviewer attempt {attempt_number}/{max_attempts}", candidate["provider"], candidate["model"]) started_at, started_clock = now(), time.monotonic() if candidate["kind"] == "account": ok, result, metadata, reason = call_account_model(candidate["value"], prompt, remaining) else: ok, result, metadata, reason = call_model(candidate["value"], prompt, candidate["provider"] != os.environ.get("CASAN_GOAL_LOCAL_PROVIDER", "local-policy"), remaining) retryable = False if ok else reviewer_failure_retryable(reason) ledger.append({ "attempt": attempt_number, "provider": candidate["provider"], "model": candidate["model"], "status": "pass" if ok else "failed", "reason": reason, "retryable": retryable, "started_at": started_at, "finished_at": now(), "latency_ms": int((time.monotonic() - started_clock) * 1000), }) update_job(job_path, reviewer_attempts=ledger) if ok: return True, result, metadata, "ok", candidate last_reason = reason if not retryable and any(marker in str(reason).lower() for marker in ("security_blocked", "content_policy")): break return False, "", {}, last_reason, (ledger[-1] if ledger else {}) def audit(job: dict, status: str) -> str: path = os.path.join(STATE_ROOT, "logs", "audit", "goal-orchestrator.jsonl") head_path = os.path.join(STATE_ROOT, "logs", "audit", "goal-orchestrator-head.txt") lock_path = os.path.join(STATE_ROOT, "logs", "audit", "goal-orchestrator.lock") os.makedirs(os.path.dirname(path), exist_ok=True) with open(lock_path, "a", encoding="utf-8") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) previous = "0" * 64 try: with open(head_path, encoding="utf-8") as handle: previous = handle.read().strip() or previous except OSError: pass core = { "timestamp": now(), "harness": "H5-governance", "goal_id": job["id"], "tenant": job.get("tenant", "default"), "actor": job.get("actor", "unknown"), "goal_hash": sha(str(job.get("goal", ""))), "status": status, "local_provider": job.get("local_provider", ""), "cloud_provider": job.get("cloud_provider", ""), "prev_hash": previous, } record_hash = sha(json.dumps(core, sort_keys=True, ensure_ascii=False)) append_jsonl(path, {**core, "record_hash": record_hash}) with open(head_path, "w", encoding="utf-8") as handle: handle.write(record_hash + "\n") handle.flush() os.fsync(handle.fileno()) os.chmod(head_path, 0o600) fcntl.flock(lock.fileno(), fcntl.LOCK_UN) return record_hash def metric(job: dict, status: str, started: float, local_meta: dict, cloud_meta: dict) -> None: append_jsonl(os.path.join(STATE_ROOT, "logs", "cost", "metrics.jsonl"), { "timestamp": now(), "trace_id": job["id"], "harness": "H6-agentops", "agent": "goal.orchestrator", "step": "local-worker-cloud-reviewer", "status": status, "exit_code": 0 if status in {"success", "degraded"} else 2, "latency_ms": int((time.monotonic() - started) * 1000), "input_tokens": int(local_meta.get("input_tokens", 0)) + int(cloud_meta.get("input_tokens", 0)), "output_tokens": int(local_meta.get("output_tokens", 0)) + int(cloud_meta.get("output_tokens", 0)), "total_tokens": sum(int(meta.get(key, 0)) for meta in (local_meta, cloud_meta) for key in ("input_tokens", "output_tokens")), "cost_estimate": 0.0, "cost_source": "provider_usage_logs", "input_hash": sha(str(job.get("goal", ""))), "output_hash": sha(str(job.get("result", ""))), }) def run(job_path: str) -> int: started = time.monotonic() job = load_json(job_path) goal_id = str(job["id"]) goal = str(job.get("goal") or "").strip() local_model = os.environ.get("CASAN_GOAL_LOCAL_MODEL", "") cloud_model = os.environ.get("CASAN_GOAL_CLOUD_MODEL", "") account_provider = os.environ.get("CASAN_GOAL_ACCOUNT_PROVIDER", "") local_meta, cloud_meta = {}, {} try: update_job(job_path, status="running", started_at=now()) emit(goal_id, "H1-context", "running", "Validating objective contract") if len(goal) < 10 or len(goal) > 8000: raise ValueError("goal_length_invalid") project_id = str(job.get("project") or "") emit(goal_id, "H4-security", "running", "Scanning objective before model routing") allowed, safe_goal = scan(goal, "input") if not allowed: emit(goal_id, "H4-security", "blocked", "Objective rejected by security boundary") raise ValueError("goal_security_blocked") context_bundle, context_manifest, context_manifest_path = build_context(job_path, project_id, safe_goal) context_summary = { "files": context_manifest["file_count"], "characters": context_manifest["characters"], "truncated": context_manifest["truncated"], "path": context_manifest_path, } update_job(job_path, context_manifest=context_summary) emit(goal_id, "H1-context", "pass", "Allowlisted workspace snapshot prepared", { "goal_hash": sha(goal), "project_id": project_id, **context_summary, "bundle_sha256": context_manifest["bundle_sha256"], }) emit(goal_id, "H4-security", "running", "Objective and workspace snapshot passed; model outputs pending") write_intent = requests_side_effect(safe_goal) stage(job_path, "local-worker", "running", "Local model is developing the primary solution", job.get("local_provider", ""), local_model) emit(goal_id, "H2-tool", "running", "Local worker is developing a solution", {"provider": job.get("local_provider", ""), "model": local_model}) output_contract = ( "Produce ONLY one complete unified git diff inside a ```diff fence. The diff must implement the objective, include every required file, and may only touch paths visible in the workspace snapshot. Do not include commentary outside the diff. " if write_intent else "Produce: clarified outcome, assumptions, ordered implementation plan, risks, and verifiable acceptance checks. Respond in the same language as the objective. " ) local_prompt = ( "You are the local CASAN worker. Solve the user's objective concretely. " + output_contract + "Use only the bounded, redacted workspace snapshot below as repository evidence. " "Do not claim to inspect any filesystem outside this snapshot and do not perform side effects.\n\n" f"OBJECTIVE:\n{safe_goal}\n\nWORKSPACE SNAPSHOT ({project_id}):\n{context_bundle}" ) ok, local_draft, local_meta, reason = call_model(local_model, local_prompt, False) if not ok: stage(job_path, "local-worker", "error", reason, job.get("local_provider", ""), local_model) emit(goal_id, "H2-tool", "error", "Local worker failed", {"reason": reason}) raise RuntimeError(f"local_worker_failed:{reason}") allowed, safe_local = scan(local_draft, "output") if not allowed: emit(goal_id, "H4-security", "blocked", "Local worker output rejected") raise ValueError("local_output_security_blocked") if write_intent: try: validate_write_output(safe_local) except ValueError as first_error: repair_candidates = patch_repair_models(local_model)[:patch_repair_attempts()] repair_model = repair_candidates[0] if repair_candidates else local_model repair_provider = provider_for_model(repair_model) stage(job_path, "local-worker", "running", "Repairing invalid patch output contract", repair_provider, repair_model) emit(goal_id, "H2-tool", "running", "Worker is repairing patch output contract", {"reason": str(first_error), "max_attempts": patch_repair_attempts(), "repair_provider": repair_provider, "repair_model": repair_model, "repair_candidates": repair_candidates}) repaired, repaired_output, repaired_meta, repair_reason = repair_write_output(local_model, local_prompt, safe_local, str(first_error)) local_meta = merge_usage(local_meta, repaired_meta) if not repaired: reason = repair_reason else: allowed, safe_local = scan(repaired_output, "output") if not allowed: emit(goal_id, "H4-security", "blocked", "Repaired local worker output rejected") raise ValueError("local_output_security_blocked") try: validate_write_output(safe_local) reason = "" except ValueError as repair_error: reason = f"goal_patch_repair_invalid:{repair_error}" if reason: actual_repair_model = str(repaired_meta.get("repair_model") or repair_model) update_job(job_path, local_usage=local_meta, patch_repair_attempts=repaired_meta.get("repair_attempts", [])) stage(job_path, "local-worker", "error", reason, provider_for_model(actual_repair_model), actual_repair_model) emit(goal_id, "H2-tool", "error", "Worker violated patch output contract after repair", {"reason": reason, "repair_provider": provider_for_model(actual_repair_model), "repair_model": actual_repair_model, "repair_attempts": repaired_meta.get("repair_attempts", [])}) raise ValueError(reason) stage(job_path, "local-worker", "pass", "Primary solution prepared", job.get("local_provider", ""), local_model) update_job(job_path, local_draft=safe_local, local_usage=local_meta) emit(goal_id, "H2-tool", "pass", "Local solution prepared", {"provider": job.get("local_provider", ""), "model": local_model, **local_meta}) stage(job_path, "cloud-reviewer", "running", "Independent reviewer is challenging and improving the local solution", job.get("cloud_provider", ""), cloud_model) emit(goal_id, "H3-eval", "running", "Cloud reviewer is evaluating the local solution", {"provider": job.get("cloud_provider", ""), "model": cloud_model}) reviewer_contract = ( "Return ONLY the corrected, complete unified git diff inside a ```diff fence. Preserve valid implementation details, fix defects, and include no prose or `index` lines outside the diff. Every hunk must be complete and pass `git apply --check`. " if write_intent else "Return one final actionable solution with ordered steps and acceptance checks. Respond in the same language as the objective. " ) review_prompt = ( "You are the cloud CASAN reviewer. Critically review the local worker's proposal " "against the objective. Correct gaps, remove unsafe or unverifiable claims, and " + reviewer_contract + "Use only the exact bounded, redacted " "workspace snapshot provided below; your execution directory is intentionally empty. " "Do not inspect or infer from any other filesystem and do not perform side effects.\n\n" f"OBJECTIVE:\n{safe_goal}\n\nWORKSPACE SNAPSHOT ({project_id}):\n{context_bundle}\n\n" f"LOCAL WORKER PROPOSAL:\n{safe_local[:5000]}" ) cloud_ok, cloud_result, cloud_meta, cloud_reason, reviewer = run_reviewer_chain( job_path, review_prompt, account_provider, cloud_model, local_model ) reviewer_provider = str(reviewer.get("provider") or job.get("cloud_provider", "")) reviewer_model = str(reviewer.get("model") or cloud_model) if cloud_ok: allowed, safe_result = scan(cloud_result, "output") if not allowed: emit(goal_id, "H4-security", "blocked", "Cloud reviewer output rejected") raise ValueError("cloud_output_security_blocked") if write_intent: try: validate_write_output(safe_result) except ValueError as error: cloud_ok = False cloud_reason = f"reviewer_output_contract_invalid:{error}" if cloud_ok: stage(job_path, "cloud-reviewer", "pass", "Independent review incorporated", reviewer_provider, reviewer_model) emit(goal_id, "H3-eval", "pass", "Independent review incorporated", {"provider": reviewer_provider, "model": reviewer_model, **cloud_meta}) final_status = "completed" metric_status = "success" else: safe_result = safe_local stage(job_path, "cloud-reviewer", "warning", cloud_reason, reviewer_provider, reviewer_model) emit(goal_id, "H3-eval", "warning", "Cloud reviewer output contract rejected; local solution retained", {"reason": cloud_reason}) final_status = "degraded" metric_status = "degraded" else: safe_result = safe_local stage(job_path, "cloud-reviewer", "warning", cloud_reason, reviewer_provider, reviewer_model) emit(goal_id, "H3-eval", "warning", "Cloud reviewer unavailable; local solution retained", {"reason": cloud_reason}) final_status = "degraded" metric_status = "degraded" emit(goal_id, "H4-security", "pass", "Objective and all released outputs passed security scans") if write_intent: patch = extract_patch(safe_result) patch_artifact = validate_and_store_patch(job_path, load_json(job_path), patch) proposal = submit_side_effect(load_json(job_path), context_manifest, patch_artifact) approval = {"id": proposal["id"], "status": proposal["status"], "action": proposal["action"]} patch_artifact["approval_id"] = proposal["id"] final_status = "requires_approval" metric_status = "degraded" safe_result = "Implementation patch generated and independently reviewed. Approval is required before applying it to the workspace." emit(goal_id, "H7-orchestration", "blocked", "Reviewed patch awaits approval", {"proposal_id": proposal["id"], "patch_sha256": patch_artifact["sha256"]}) job = update_job(job_path, status=final_status, result=safe_result, patch_artifact=patch_artifact, approval=approval, cloud_usage=cloud_meta, finished_at=now()) else: job = update_job(job_path, status=final_status, result=safe_result, cloud_usage=cloud_meta, finished_at=now()) emit(goal_id, "H5-governance", "running", "Anchoring orchestration decision") audit_hash = audit(job, final_status) emit(goal_id, "H5-governance", "pass", "Orchestration decision anchored", {"audit_hash": audit_hash, "status": final_status}) emit(goal_id, "H6-agentops", "running", "Recording orchestration telemetry") metric(job, metric_status, started, local_meta, cloud_meta) emit(goal_id, "H6-agentops", "pass", "Orchestration telemetry recorded", {"latency_ms": int((time.monotonic() - started) * 1000), "status": metric_status}) if not write_intent: emit(goal_id, "H7-orchestration", "pass", "Local worker and cloud review workflow completed", {"status": final_status, "cloud_incorporated": cloud_ok}) update_job(job_path, audit_hash=audit_hash) return 0 except Exception as error: reason = str(error)[:120] or error.__class__.__name__ update_job(job_path, status="failed", error=reason, finished_at=now()) emit(goal_id, "H5-governance", "running", "Anchoring failed orchestration decision") job = load_json(job_path) audit_hash = audit(job, "failed") emit(goal_id, "H5-governance", "pass", "Failed decision anchored", {"audit_hash": audit_hash}) # Keep gate semantics distinct. H2 may report a patch-contract failure, # while H6/H7 must report their own lifecycle state and retain the H2 # error only as causal evidence. Reusing `reason` here made the UI show # goal_patch_missing as if telemetry and orchestration independently # failed the patch contract. emit(goal_id, "H6-agentops", "error", "goal_orchestration_failed", {"root_reason": reason, "latency_ms": int((time.monotonic() - started) * 1000)}) emit(goal_id, "H7-orchestration", "blocked", "goal_workflow_stopped", {"root_reason": reason}) update_job(job_path, audit_hash=audit_hash) return 2 def verify_audit() -> int: path = os.path.join(STATE_ROOT, "logs", "audit", "goal-orchestrator.jsonl") previous = "0" * 64 records = 0 try: handle = open(path, encoding="utf-8") except OSError: print(json.dumps({"ok": True, "records": 0, "head": previous})) return 0 with handle: for line in handle: if not line.strip(): continue records += 1 try: record = json.loads(line) except ValueError: print(json.dumps({"ok": False, "records": records, "reason": "invalid_json"})) return 3 record_hash = str(record.pop("record_hash", "")) expected = sha(json.dumps(record, sort_keys=True, ensure_ascii=False)) if record.get("prev_hash") != previous or record_hash != expected: print(json.dumps({"ok": False, "records": records, "reason": "chain_break"})) return 3 previous = record_hash print(json.dumps({"ok": True, "records": records, "head": previous})) return 0 def main() -> int: parser = argparse.ArgumentParser() group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--job-file") group.add_argument("--verify-audit", action="store_true") args = parser.parse_args() if args.verify_audit: return verify_audit() path = os.path.abspath(args.job_file) state = os.path.abspath(STATE_ROOT) + os.sep if not path.startswith(state): raise SystemExit("GOAL_JOB_PATH_DENIED") return run(path) if __name__ == "__main__": raise SystemExit(main())