767 lines
37 KiB
Python
767 lines
37 KiB
Python
#!/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_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):
|
|
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")
|
|
environment.setdefault("CASAN_MODEL_GENERATE_MAX_TOKENS", os.environ.get("CASAN_GOAL_MAX_OUTPUT_TOKENS", "1400"))
|
|
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")
|
|
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 outside the diff. "
|
|
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")
|
|
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 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})
|
|
emit(goal_id, "H6-agentops", "error", "Orchestration failed", {"reason": reason, "latency_ms": int((time.monotonic() - started) * 1000)})
|
|
emit(goal_id, "H7-orchestration", "blocked", "Goal workflow stopped", {"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())
|