Record a run and observations

self.client = client or WyrdClient()
self.state = WyrdState.from_path(BUNDLE, self.client)
self.state.start_bifrost()
# Spans export to Wyrd through the SDK's OTLP exporter; the Agents SDK's
# own traces become spans too, under the request's Run.
self.state.start_telemetry()
set_trace_processors([WyrdAgentsTraceProcessor()])

let state = WyrdState::from_path_with(bundle, client.clone())?;
// Starts the buffered writer the ticket rows go through.
state.start_bifrost().await?;

// WYRD_SERVER_URL and WYRD_API_KEY, else the settings `wyrd server dev` saved, say which server to use and who the app is.
const client = WyrdClient.connect();
// The declared Service, pulled by `npm run deploy`.
const state = WyrdState.fromPath(BUNDLE_DIR, { client });
await state.startBifrost(); // writes tickets to Bifrost
await state.startTelemetry(); // exports spans to Wyrd

run = self.state.run()
with run, tracer.start_as_current_span("support-desk.ask"):
    # These headers tie the gateway's record of the call to this Run.
    headers = {
        "wyrd-run-id": run.run_id,
        "wyrd-card-uid": str(self.state.card_ref("agent").uid),
    }
    model = OpenAIChatCompletionsModel(
        self.model, self.gateway.with_options(default_headers=headers)
    )
    agent = Agent(name="support-agent", model=model)
    result = await Runner.run(agent, self.template.replace("{{question}}", question))
    answer = str(result.final_output)
    run.observe.record(TICKETS, {"question": question, "answer": answer})

let agent = self.state.agent(run.alias())?;
let ExecutionRoute::WyrdGateway { deployment } = &agent.card().spec.execution_route else {
    return Err(ApiError::internal(
        "the Agent does not route through the Wyrd gateway",
    ));
};
let prompt = &agent.prompt().metadata.prompt;
let rendered = prompt
    .render(&[("question", question)])
    .map_err(|_| ApiError::internal("the Agent's Prompt does not render"))?;
let mut body = serde_json::to_value(rendered)
    .map_err(|_| ApiError::internal("the rendered Prompt does not serialize"))?["body"]
    .take();
body["model"] = json!(format!("{deployment}/{}", prompt.model));
let client = self.state.client()?;
// These headers tie the gateway's record of the call to this Run.
let reply = reqwest::Client::new()
    .post(format!("{}/v1/chat/completions", client.server_url()))
    .bearer_auth(client.access_token().await?.expose())
    .header("wyrd-run-id", run.run_id().as_str())
    .header("wyrd-card-uid", run.subject_uid().as_str())
    .json(&body)
    .timeout(ANSWER_TIMEOUT)
    .send()
    .await
    .map_err(|_| ApiError::internal("the Wyrd gateway call failed"))?;
let status = reply.status();
let reply: Option<Value> = reply.json().await.ok();
if !status.is_success() {
    return Err(ApiError::refusal(status, reply.as_ref()));
}
let reply =
    reply.ok_or_else(|| ApiError::internal("the gateway reply could not be read"))?;
let answer = reply["choices"][0]["message"]["content"]
    .as_str()
    .ok_or_else(|| ApiError::internal("the gateway reply carries no text"))?
    .to_owned();
run.observe()
    .record(
        "vala.datasets.tickets",
        &json!({ "question": question, "answer": answer }),
    )
    .await?;

const wyrdRun = support.state.run();
// Spans started inside the scope carry the Run's ID.
return wyrdRun.scope(async () => {
  const text = await answer(support.gateway, support.state, wyrdRun, question);
  await wyrdRun.observe.record(TICKETS, { question, answer: text });
  return check(wyrdRun, text);
});

return {
    "no-refund-promise": run.observe.verify("no-refund-promise", answer=ticket["answer"]),
    "ticket-complete": run.for_card("agent").observe.verify(
        "ticket-complete", ticket=ticket
    ),
}

let run = desk.state.run();
let agent = run.for_card("agent")?;
let work = async {
    let answer = desk.answer(&agent, &body.question).await?;
    let promise = run
        .observe()
        .verify(
            "no-refund-promise",
            Inputs::named([("answer", Input::from(json!(answer)))]),
        )
        .await?;
    let ticket = json!({
        "question": body.question,
        "answer": answer,
        "order_id": body.order_id,
    });
    let complete = agent
        .observe()
        .verify(
            "ticket-complete",
            Inputs::named([("ticket", Input::from(ticket))]),
        )
        .await?;
    Ok::<_, ApiError>((answer, promise, complete))
};
let (answer, promise, complete) = run
    .scope(work.instrument(tracing::info_span!("support-desk.ask")))
    .await?;

const ticket = { question, answer: text, order_id: orderId ?? null };
const [promise, complete] = await Promise.all([
  wyrdRun.observe.verify("no-refund-promise", { answer: text }),
  wyrdRun.forCard("agent").observe.verify("ticket-complete", { ticket }),
]);

execution = run.for_card("agent").observe.submit(
    "follows-policy", letter=verify.File(letter.encode(), filename="letter.md")
)
assert isinstance(execution, verify.Execution)  # Run evidence always queues one

let submission = run
    .observe()
    .submit("follows-policy", letter_inputs(&letter))
    .await?;

const letter = verify.File.bytes(new TextEncoder().encode(text), { filename: "letter.md" });
const inputs = { letter };
const submitted = await wyrdRun.forCard("agent").observe.submit("follows-policy", inputs);
if (submitted.kind !== "queued") throw new Error("follows-policy judges Run evidence, so submit queues it");

# Any process can pick a check up from its execution ID alone.
state = verify.Execution.connect(execution_id, request.app.state.desk.client).status()

let state = Execution::connect_with(id, desk.state.client()?)
    .await?
    .status()
    .await?;

const execution = await verify.Execution.connect(executionId, { client: support.client });
const { status, error, result } = await execution.status();

app.state.desk = Desk()
yield
# Flush buffered tickets and spans to Wyrd before exiting.
app.state.desk.state.shutdown()

let telemetry = start_telemetry(&self.state)?;
let state = self.state.clone();
let served = axum::serve(listener, self.router())
    .with_graceful_shutdown(stop)
    .await;
// Flush spans first, then drain buffered tickets, before exiting.
telemetry.shutdown()?;
state.shutdown().await?;

// Flush every span and row before exiting: shutdown flushes the provider startTelemetry installed, then drains the writer.
for (const signal of ["SIGINT", "SIGTERM"]) {
  process.on(signal, async () => {
    server.close();
    await state.shutdown();
    process.exit(0);
  });
}

SQL = """
SELECT t.run_id, t.question, t.answer, s.trace_id, g.call_id, r.execution_id, r.verdict
FROM vala.datasets.tickets t
JOIN vala.traces.spans s ON s.run_id = t.run_id AND s.name = 'support-desk.ask'
JOIN vala.gateway.calls g ON g.run_id = t.run_id
JOIN vala.verification.results r ON r.run_id = t.run_id
WHERE t.run_id = $1
"""

/**
 * Read one Run's evidence back through Bifrost. Every row a Run produced
 * carries its `run_id`, so one join links the ticket to the agent's
 * span, its gateway call, and each verdict.
 *
 *   npm run evidence -- <run_id>
 */

import { Bifrost, type WyrdClient } from "wyrd-ai";
import { z } from "zod";

import { TICKETS } from "./wyrd.ts";

const Evidence = z.object({
  run_id: z.string(),
  question: z.string(),
  answer: z.string(),
  span: z.string(),
  trace_id: z.string(),
  call_id: z.string(),
  verdict: z.string(),
  execution_id: z.string(),
});

/** The joined evidence of Run `runId`, read as `client` (the ambient client by default). */
export async function evidence(runId: string, client?: WyrdClient) {
  const bifrost = await Bifrost.connect({ client });
  try {
    return await bifrost.sql(
      `SELECT c.run_id, c.question, c.answer, s.name AS span, encode(s.trace_id, 'hex') AS trace_id, g.call_id, r.verdict, r.execution_id
         FROM ${TICKETS} c
         JOIN vala.traces.spans s ON s.run_id = c.run_id AND s.name LIKE 'invoke_agent %'
         JOIN vala.gateway.calls g ON g.run_id = c.run_id
         JOIN vala.verification.results r ON r.run_id = c.run_id
        WHERE c.run_id = $1`,
      [runId],
      Evidence,
    );
  } finally {
    await bifrost.shutdown();
  }
}

if (import.meta.main) {
  const runId = process.argv[2];
  if (!runId) throw new Error("usage: npm run evidence -- <run_id>");
  console.table(await evidence(runId));
}