Scenario Walkthrough
A live event enters Foundry from a coalition feed. The triage agent links it to Gotham entities, detects a confidence jump, and drafts an alert. The enrichment and correlation agents attach source packets, contradictions, and temporal context. AIP prepares an action package but cannot execute it. The commander approves a bounded recommendation; Apollo deploys the approved workflow version. If the outcome is rejected or later disproven, the feedback becomes an eval, the prompt optimizer proposes a safer rule, and the next release ships only after review and regression gates pass.
Code Examples
# Python FastAPI orchestration skeleton for ClearGlassInc Artemis
from dataclasses import dataclass
from enum import Enum
from typing import Literal
from fastapi import FastAPI, HTTPException
app = FastAPI(title="ClearGlassInc Artemis Control Plane")
class Risk(str, Enum):
green = "green"
amber = "amber"
red = "red"
@dataclass(frozen=True)
class MissionContext:
mission_id: str
user_id: str
compartments: set[str]
coalition: str
purpose: str
@dataclass(frozen=True)
class ActionPackage:
kind: Literal["intel_product", "case_update", "response_recommendation"]
target_id: str
rationale: str
evidence_ids: list[str]
risk: Risk
prompt_version: str
workflow_version: str
def enforce_need_to_know(ctx: MissionContext, entity_markings: set[str]) -> None:
if not entity_markings.issubset(ctx.compartments):
raise HTTPException(status_code=403, detail="Denied by compartment policy")
def requires_human_approval(pkg: ActionPackage) -> bool:
return pkg.risk in {Risk.amber, Risk.red} or pkg.kind == "response_recommendation"
@app.post("/missions/{mission_id}/triage")
def triage_event(mission_id: str, event_id: str, ctx: MissionContext) -> dict:
entity = foundry_ontology_get("Event", event_id) # ontology object action, not raw table access
enforce_need_to_know(ctx, set(entity["markings"]))
candidates = aip_agent_run(
agent="triage-correlation-v7",
inputs={"event_id": event_id, "mission_id": mission_id},
tools=["ontology.search", "gotham.case_graph", "vector.retrieve"],
max_tool_calls=12,
)
pkg = ActionPackage(
kind="response_recommendation",
target_id=event_id,
rationale=candidates["rationale"],
evidence_ids=candidates["evidence_ids"],
risk=Risk(candidates["risk"]),
prompt_version=candidates["prompt_version"],
workflow_version="triage-flow-2026.07.23",
)
audit_append("action_package_prepared", pkg.__dict__)
return {"status": "approval_required" if requires_human_approval(pkg) else "ready", "package": pkg.__dict__}
# Eval pipeline: convert feedback into gated self-upgrade proposals
def build_eval_from_feedback(feedback: dict) -> dict:
return {
"case_id": feedback["feedback_id"],
"input_event": feedback["event_id"],
"expected": feedback["operator_correction"],
"must_cite_evidence": True,
"forbidden": ["uncited claims", "cross-compartment leakage", "unapproved action"],
"metrics": ["precision", "recall", "latency_ms", "policy_pass_rate", "operator_trust"],
}
def propose_self_upgrade(feedback_batch: list[dict]) -> dict:
evals = [build_eval_from_feedback(item) for item in feedback_batch]
candidate = aip_prompt_optimizer(evals=evals, base_prompt="triage-correlation-v7")
report = run_offline_evals(candidate, evals)
if report["policy_pass_rate"] < 1.0 or report["precision_delta"] < 0:
return {"decision": "reject", "reason": "regression or policy failure", "report": report}
return {"decision": "human_review", "candidate": candidate, "report": report}