CASAN/packages/casan-harness/scripts/bash/goal-orchestrator.py

#!/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 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|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 submit_side_effect(job: dict, manifest: dict) -> dict:
    payload = {
        "goal_id": job["id"],
        "project_id": job["project"],
        "context_manifest_hash": manifest["bundle_sha256"],
        "requested_operation": job["goal"],
    }
    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):
    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"))
        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=int(os.environ.get("CASAN_GOAL_MODEL_TIMEOUT", "300")),
            )
        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):
    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=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 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")

        if requests_side_effect(safe_goal):
            proposal = submit_side_effect(job, context_manifest)
            approval = {"id": proposal["id"], "status": proposal["status"], "action": proposal["action"]}
            stage(job_path, "local-worker", "blocked", "Workspace execution requires approval; no model or tool was allowed to write", job.get("local_provider", ""), local_model)
            stage(job_path, "cloud-reviewer", "blocked", "Reviewer is not an execution channel", job.get("cloud_provider", ""), cloud_model)
            update_job(job_path, status="requires_approval", approval=approval, result="This objective requests a workspace side effect. CASAN created a governed approval proposal and did not execute or modify files.", finished_at=now())
            emit(goal_id, "H2-tool", "blocked", "Side effect withheld pending approval", {"proposal_id": proposal["id"], "action": proposal["action"]})
            emit(goal_id, "H3-eval", "blocked", "Cloud reviewer cannot bypass the approval boundary")
            emit(goal_id, "H4-security", "pass", "Workspace remained read-only")
            gated_job = load_json(job_path)
            audit_hash = audit(gated_job, "requires_approval")
            emit(goal_id, "H5-governance", "pass", "Approval proposal and decision anchored", {"audit_hash": audit_hash, "proposal_id": proposal["id"]})
            metric(gated_job, "degraded", started, {}, {})
            emit(goal_id, "H6-agentops", "pass", "Approval routing telemetry recorded")
            emit(goal_id, "H7-orchestration", "blocked", "Awaiting governed approval", {"proposal_id": proposal["id"]})
            update_job(job_path, audit_hash=audit_hash)
            return 0

        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})
        local_prompt = (
            "You are the local CASAN worker. Solve the user's objective concretely. "
            "Produce: clarified outcome, assumptions, ordered implementation plan, risks, "
            "and verifiable acceptance checks. Respond in the same language as the objective. "
            "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", "Cloud model 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})
        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 "
            "return one final actionable solution with ordered steps and acceptance checks. "
            "Respond in the same language as the objective. 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]}"
        )
        if account_provider:
            cloud_ok, cloud_result, cloud_meta, cloud_reason = call_account_model(account_provider, review_prompt)
            if not cloud_ok and os.environ.get("CASAN_GOAL_CLOUD_FALLBACK_MODEL"):
                cloud_ok, cloud_result, cloud_meta, cloud_reason = call_model(
                    os.environ["CASAN_GOAL_CLOUD_FALLBACK_MODEL"], review_prompt, True
                )
        else:
            cloud_ok, cloud_result, cloud_meta, cloud_reason = call_model(cloud_model, review_prompt, True)
        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", "Cloud review incorporated", job.get("cloud_provider", ""), cloud_model)
            emit(goal_id, "H3-eval", "pass", "Cloud review incorporated", {"provider": job.get("cloud_provider", ""), "model": cloud_model, **cloud_meta})
            final_status = "completed"
            metric_status = "success"
        else:
            safe_result = safe_local
            stage(job_path, "cloud-reviewer", "warning", cloud_reason, job.get("cloud_provider", ""), cloud_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")
        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})
        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())