use noxid_agent_ir::{AgentDefinition, AgentEvent};
use noxid_source::js_escape;
#[derive(Clone, Debug, Default)]
pub struct AgentRuntimeOptions {
pub agents: Vec<AgentDefinition>,
}
impl AgentRuntimeOptions {
pub fn is_empty(&self) -> bool {
!self.agents.iter().any(|agent| agent.engine.is_some())
}
}
fn event_javascript(event: &AgentEvent) -> String {
let payload_type = event
.payload_type
.as_ref()
.map(|ty| format!("\"{}\"", js_escape(&ty.to_string())))
.unwrap_or_else(|| "null".into());
let validators = event
.payload_type_id
.as_ref()
.map(|id| format!("\"{}\"", js_escape(id.as_str())))
.into_iter()
.chain(event.payload_type.as_ref().and_then(|ty| match ty {
noxid_types::Type::Named(name) => Some(format!("\"type:{}\"", js_escape(name))),
_ => None,
}))
.collect::<Vec<_>>()
.join(", ");
format!(
"Object.freeze({{ name: \"{}\", role: \"{}\", external: {}, payloadType: {payload_type}, validators: Object.freeze([{validators}]) }})",
js_escape(&event.name),
event.role.as_str(),
event.external,
)
}
fn contract_javascript(contract: &noxid_agent_ir::AgentContract) -> String {
let validators = contract
.type_id
.as_ref()
.map(|id| format!("\"{}\"", js_escape(id.as_str())))
.into_iter()
.chain(match &contract.ty {
noxid_types::Type::Named(name) => Some(format!("\"type:{}\"", js_escape(name))),
_ => None,
})
.collect::<Vec<_>>()
.join(", ");
format!(
"Object.freeze({{ id: \"{}\", type: \"{}\", validators: Object.freeze([{validators}]) }})",
js_escape(contract.id.as_str()),
js_escape(&contract.ty.to_string()),
)
}
pub(crate) fn agents_runtime_javascript(
options: &AgentRuntimeOptions,
base_path: &str,
timeout_ms: u64,
) -> String {
if options.is_empty() {
return String::new();
}
let prefix = if base_path == "/" {
"/_noxid/agents/".to_string()
} else {
format!("{}/_noxid/agents/", base_path.trim_end_matches('/'))
};
let declarations = options
.agents
.iter()
.filter_map(|agent| {
let engine = agent.engine.as_ref()?;
let events = agent
.events
.iter()
.map(event_javascript)
.collect::<Vec<_>>()
.join(", ");
let tools = engine
.tools
.iter()
.map(|tool| {
let capabilities = tool
.capabilities
.iter()
.map(|capability| format!("\"{}\"", js_escape(capability)))
.collect::<Vec<_>>()
.join(", ");
format!(
"Object.freeze({{ endpoint: \"{}\", version: {}, capabilities: Object.freeze([{capabilities}]), schemaHash: \"{}\" }})",
js_escape(&tool.endpoint),
tool.version,
js_escape(&tool.schema_hash),
)
})
.collect::<Vec<_>>()
.join(", ");
Some(format!(
" \"{}\": Object.freeze({{ id: \"{}\", name: \"{}\", agentId: \"{}\", model: \"{}\", modelId: \"{}\", instructions: \"{}\", maxTurns: {}, timeoutMs: {timeout_ms}, runCapability: \"{}\", resumeCapability: \"{}\", input: {}, output: {}, events: Object.freeze([{events}]), tools: Object.freeze([{tools}]) }}),",
js_escape(&agent.name),
js_escape(agent.id.as_str()),
js_escape(&agent.name),
js_escape(&engine.agent_id),
js_escape(&engine.model),
js_escape(engine.model_id.as_str()),
js_escape(&engine.instructions.text),
engine.max_turns,
js_escape(&engine.run_capability),
js_escape(&engine.resume_capability),
contract_javascript(&agent.input),
contract_javascript(&agent.output),
))
})
.collect::<Vec<_>>()
.join("\n");
let runtime = AGENT_RUNTIME.replace(
"__AGENT_EVENT_VOCABULARY__",
&format!(
"[{}]",
noxid_ir::AGENT_EVENT_VOCABULARY
.iter()
.map(|field| format!("\"{field}\""))
.collect::<Vec<_>>()
.join(", ")
),
);
format!(
"\n// noxid-runtime:feature-start:agents\nconst agentRunPrefix = \"{}\";\nconst agentDeclarations = Object.freeze({{\n{declarations}\n}});\n{runtime}// noxid-runtime:feature-end:agents\n",
js_escape(&prefix),
)
}
pub(crate) const AGENT_RUNTIME: &str = r##"
const AGENT_RUN_RECORD_SCHEMA = "noxid.agent.run.v1";
const AGENT_RUN_ID = /^[A-Za-z0-9_-]{16,128}$/;
const AGENT_RUN_TTL_SECONDS = 604_800;
const AGENT_RUN_STATES = new Set(["Running", "Paused", "Completed", "Failed", "Cancelled"]);
const AGENT_FINAL_ANSWER_TOOL = "noxid_final_answer";
const AGENT_TOOL_RESULT_MAX_BYTES = 65_536;
// The compiler-driven events (`ToolStarted`, `ToolCompleted`,
// `PermissionRequired`) carry a developer-declared payload type, so the engine
// projects a canonical record onto whatever fields that type declares. A
// declared field outside this vocabulary is one the engine has no value for,
// and the run refuses rather than inventing one.
const AGENT_EVENT_VOCABULARY = Object.freeze(__AGENT_EVENT_VOCABULARY__);
const agentRunStorage = __noxidStorage("agent_runs");
const agentRunControllers = new Map();
let agentReconciliation = null;
function agentEngineError(code, message) {
return Object.assign(new Error(message), { code, agentEngine: true });
}
function agentDeclarationFor(name) {
return Object.hasOwn(agentDeclarations, name) ? agentDeclarations[name] : null;
}
function agentValidatorFor(contract, label, declaration) {
for (const key of contract.validators) {
const validator = typeValidators[key];
if (typeof validator === "function") return validator;
}
throw agentEngineError(
"AGENT_VALIDATOR_MISSING",
`agent \`${declaration.name}\` has no boundary validator for its ${label} type \`${contract.type}\`; the engine never accepts a value it cannot validate`,
);
}
function agentEventDefinition(declaration, name) {
return declaration.events.find((event) => event.name === name) ?? null;
}
function agentRunId() {
const value = globalThis.crypto?.randomUUID?.();
if (typeof value === "string" && AGENT_RUN_ID.test(value)) return value;
throw agentEngineError("AGENT_RUN_ID_UNAVAILABLE", "agent runs require crypto.randomUUID");
}
// ---------------------------------------------------------------------------
// The derived registry, as the provider request sees it.
// ---------------------------------------------------------------------------
function agentToolEndpointSchema(tool) {
return endpointSchemas.find((schema) => schema.name === tool.endpoint && schema.version === tool.version) ?? null;
}
function agentJsonSchemaForType(type) {
if (type.startsWith("Optional<") && type.endsWith(">")) {
const inner = agentJsonSchemaForType(type.slice(9, -1));
return { anyOf: [inner, { type: "null" }] };
}
if (type.startsWith("Array<") && type.endsWith(">")) return { type: "array", items: agentJsonSchemaForType(type.slice(6, -1)) };
if (type === "String") return { type: "string" };
if (type === "Date") return { type: "string" };
if (type === "Boolean") return { type: "boolean" };
if (type === "Int") return { type: "integer" };
if (type === "Number" || type === "Float") return { type: "number" };
const declared = Object.hasOwn(modelTypeSchemas, `type:${type}`) ? modelTypeSchemas[`type:${type}`] : null;
if (declared !== null) return declared.schema;
return { type: "object" };
}
function agentToolInputSchema(schema) {
const properties = Object.create(null);
const required = [];
for (const field of [...schema.params, ...schema.query, ...schema.body]) {
properties[field.name] = agentJsonSchemaForType(field.type);
if (!field.type.startsWith("Optional<")) required.push(field.name);
}
return { type: "object", properties, required, additionalProperties: false };
}
function agentFinalAnswerSchema(declaration) {
const type = declaration.output.type;
const schema = agentJsonSchemaForType(type);
return schema.type === "object" || schema.properties !== undefined
? schema
: { type: "object", properties: { value: schema }, required: ["value"], additionalProperties: false };
}
function agentFinalAnswerWrapped(declaration) {
const schema = agentFinalAnswerSchema(declaration);
return schema.properties !== undefined && Object.keys(schema.properties).length === 1 && Object.hasOwn(schema.properties, "value")
&& !Object.hasOwn(modelTypeSchemas, `type:${declaration.output.type}`);
}
function agentToolRegistry(declaration) {
const entries = new Map();
for (const tool of declaration.tools) {
const schema = agentToolEndpointSchema(tool);
if (schema === null) {
throw agentEngineError(
"AGENT_TOOL_ENDPOINT_MISSING",
`agent \`${declaration.name}\` lists tool \`${tool.endpoint}@${tool.version}\`, which this build emits no endpoint for`,
);
}
entries.set(schema.name, Object.freeze({ tool, schema, capabilities: tool.capabilities }));
}
return entries;
}
function agentProviderTools(declaration, registry) {
const tools = [...registry.values()].map((entry) => Object.freeze({
name: entry.schema.name,
description: entry.schema.description ?? `Call the ${entry.schema.name} endpoint (${entry.schema.method} ${entry.schema.path}).`,
schema: agentToolInputSchema(entry.schema),
}));
tools.push(Object.freeze({
name: AGENT_FINAL_ANSWER_TOOL,
description: `Return the run's final ${declaration.output.type} answer and end the run.`,
schema: agentFinalAnswerSchema(declaration),
}));
return Object.freeze(tools);
}
// ---------------------------------------------------------------------------
// The persisted run record. It is read back through the same validator the
// engine wrote it with: a drifted record fails the resume rather than feeding
// the loop a shape it never produced.
// ---------------------------------------------------------------------------
function agentRunKey(agent, runId) {
return `${agent}:${runId}`;
}
function agentValidTurn(turn) {
if (turn === null || typeof turn !== "object" || Array.isArray(turn)) return false;
if (!Number.isSafeInteger(turn.index) || turn.index < 0) return false;
if (typeof turn.text !== "string") return false;
if (!Array.isArray(turn.toolCalls) || !Array.isArray(turn.results)) return false;
for (const call of turn.toolCalls) {
if (call === null || typeof call !== "object") return false;
if (typeof call.id !== "string" || typeof call.name !== "string") return false;
if (call.arguments === null || typeof call.arguments !== "object") return false;
}
for (const result of turn.results) {
if (result === null || typeof result !== "object") return false;
if (typeof result.id !== "string" || typeof result.name !== "string") return false;
if (typeof result.ok !== "boolean") return false;
}
return true;
}
function agentValidRunRecord(record, agent) {
if (record === null || typeof record !== "object" || Array.isArray(record)) return null;
if (record.schema !== AGENT_RUN_RECORD_SCHEMA) return null;
if (typeof record.runId !== "string" || !AGENT_RUN_ID.test(record.runId)) return null;
if (record.agent !== agent) return null;
if (!AGENT_RUN_STATES.has(record.state)) return null;
if (record.principal !== null && typeof record.principal !== "string") return null;
if (!Number.isSafeInteger(record.version) || record.version < 1) return null;
if (!Array.isArray(record.turns) || !record.turns.every(agentValidTurn)) return null;
if (record.pending !== null && (typeof record.pending !== "object" || typeof record.pending?.name !== "string")) return null;
return record;
}
async function agentPersistRun(run) {
// Every write moves the version. A resume's compare-and-swap names the
// version it read, so a record that moved under it is a record it no
// longer owns.
run.version = Number.isSafeInteger(run.version) ? run.version + 1 : 1;
run.updatedAt = new Date().toISOString();
await agentRunStorage.set(agentRunKey(run.agent, run.runId), run, { ttl: AGENT_RUN_TTL_SECONDS });
}
// Acquiring a paused run is a decision, not a write. The compare-and-swap
// names the state *and* the version the caller read, so of two concurrent
// resumes exactly one moves the record to `Running` and dispatches the pending
// tool; the loser never reaches the endpoint path, the provider, or the
// event stream. A store that cannot swap conditionally refuses the resume
// rather than dispatching a paused tool twice.
async function agentAcquirePausedRun(run) {
if (typeof agentRunStorage.compareAndSet !== "function") {
throw agentEngineError("AGENT_RUN_STORE_UNAVAILABLE", "the agent run store cannot claim a paused run exclusively, so the resume is refused rather than risking a second dispatch of the same tool");
}
const next = {
...run,
state: "Running",
approved: run.pending.id,
pending: null,
version: run.version + 1,
updatedAt: new Date().toISOString(),
};
const acquired = await agentRunStorage.compareAndSet(
agentRunKey(run.agent, run.runId),
{ state: "Paused", version: run.version },
next,
{ ttl: AGENT_RUN_TTL_SECONDS },
);
return acquired ? next : null;
}
async function agentLoadRun(agent, runId) {
const stored = await agentRunStorage.get(agentRunKey(agent, runId));
if (stored === null) return null;
const valid = agentValidRunRecord(stored, agent);
if (valid === null) {
await agentRunStorage.delete(agentRunKey(agent, runId));
throw agentEngineError("AGENT_RUN_RECORD_DRIFT", `agent \`${agent}\` run \`${runId}\` is persisted in a shape this build did not write; it was discarded rather than resumed`);
}
return valid;
}
/// Startup reconciliation, owned by the WO-24 queue worker. A pause is durable,
/// so a paused run survives the process that created it; a run still marked
/// `Running` when a process starts was orphaned by the previous process's exit
/// and is failed with `AGENT_TIMEOUT` rather than left waiting forever.
async function __noxidReconcileAgentRuns() {
if (agentReconciliation !== null) return agentReconciliation;
agentReconciliation = (async () => {
let paused = 0;
let orphaned = 0;
let keys;
try { keys = await agentRunStorage.list(""); }
catch { agentReconciliation = null; throw agentEngineError("AGENT_RUN_STORE_UNAVAILABLE", "agent run reconciliation cannot read the agent_runs namespace"); }
for (const key of keys) {
const separator = key.indexOf(":");
if (separator <= 0) continue;
const agent = key.slice(0, separator);
const declaration = agentDeclarationFor(agent);
if (declaration === null) continue;
let record;
try { record = await agentLoadRun(agent, key.slice(separator + 1)); }
catch { continue; }
if (record === null) continue;
if (record.state === "Paused") { paused += 1; continue; }
if (record.state !== "Running") continue;
record.state = "Failed";
record.error = { code: "AGENT_TIMEOUT", message: "The run was interrupted by a process restart and exceeded its declared timeout" };
await agentPersistRun(record);
orphaned += 1;
}
return Object.freeze({ paused, orphaned });
})();
return agentReconciliation;
}
// ---------------------------------------------------------------------------
// Events.
// ---------------------------------------------------------------------------
function agentProjectPayload(declaration, eventName, typeName, canonical) {
const entry = Object.hasOwn(modelTypeSchemas, `type:${typeName}`) ? modelTypeSchemas[`type:${typeName}`] : null;
const properties = entry?.schema?.properties ?? null;
if (properties === null) {
throw agentEngineError(
"AGENT_EVENT_UNREPRESENTABLE",
`agent \`${declaration.name}\` declares \`event ${eventName}(${typeName})\`, and \`${typeName}\` is not a declared record type the engine can fill; declare it as a type whose fields are among ${AGENT_EVENT_VOCABULARY.join(", ")}`,
);
}
const payload = Object.create(null);
for (const key of Object.keys(properties)) {
if (!Object.hasOwn(canonical, key)) {
throw agentEngineError(
"AGENT_EVENT_UNREPRESENTABLE",
`agent \`${declaration.name}\` declares \`event ${eventName}(${typeName})\` with field \`${key}\`, which the engine has no value for; a compiler-driven ${eventName} payload may declare ${AGENT_EVENT_VOCABULARY.join(", ")}`,
);
}
payload[key] = canonical[key];
}
return payload;
}
function agentEvent(declaration, name, canonical = null, raw = undefined) {
const definition = agentEventDefinition(declaration, name);
if (definition === null) return null;
if (definition.payloadType === null) return Object.freeze({ tag: name });
const value = canonical === null
? raw
: (definition.payloadType === "String" ? canonical.summary ?? canonical.name : agentProjectPayload(declaration, name, definition.payloadType, canonical));
// Every value that reaches here is compiler-constructed (`Failed`,
// `Paused`, `Token`), already validated (`Completed`, through the output
// validator), or projected field-by-field onto a declared type from the
// canonical record. When the build emits a validator for the payload type it
// runs; `AgentError` and the compiler-owned `Paused` payload have no declared
// type and therefore no validator to run.
let validator = null;
for (const key of definition.validators) {
if (typeof typeValidators[key] === "function") { validator = typeValidators[key]; break; }
}
const trusted = validator === null ? value : validator(value, true);
return Object.freeze({ tag: name, value: trusted === undefined ? null : trusted });
}
// ---------------------------------------------------------------------------
// The provider turn. One streaming call per turn, with the registry's tool
// schemas attached; text deltas surface as `Token`, tool-use blocks accumulate
// into the turn's calls.
// ---------------------------------------------------------------------------
function agentAnthropicMessages(declaration, run) {
const messages = [{ role: "user", content: JSON.stringify(run.input) }];
for (const turn of run.turns) {
const content = [];
if (turn.text.length !== 0) content.push({ type: "text", text: turn.text });
for (const call of turn.toolCalls) content.push({ type: "tool_use", id: call.id, name: call.name, input: call.arguments });
if (content.length !== 0) messages.push({ role: "assistant", content });
if (turn.results.length !== 0) {
messages.push({
role: "user",
content: turn.results.map((result) => ({ type: "tool_result", tool_use_id: result.id, content: JSON.stringify(result.content), is_error: result.ok === false })),
});
}
}
return messages;
}
function agentOpenAiMessages(declaration, run) {
const messages = [{ role: "system", content: declaration.instructions }, { role: "user", content: JSON.stringify(run.input) }];
for (const turn of run.turns) {
const message = { role: "assistant", content: turn.text.length === 0 ? null : turn.text };
if (turn.toolCalls.length !== 0) {
message.tool_calls = turn.toolCalls.map((call) => ({ id: call.id, type: "function", function: { name: call.name, arguments: JSON.stringify(call.arguments) } }));
}
messages.push(message);
for (const result of turn.results) messages.push({ role: "tool", tool_call_id: result.id, content: JSON.stringify(result.content) });
}
return messages;
}
function agentParsedArguments(raw) {
if (raw.length === 0) return Object.create(null);
try {
const value = JSON.parse(raw);
return value !== null && typeof value === "object" && !Array.isArray(value) ? value : Object.create(null);
} catch { return Object.create(null); }
}
async function* agentAnthropicTurn(declaration, definition, run, tools, signal) {
const body = {
model: definition.modelId,
max_tokens: definition.maxTokens === null ? 1024 : definition.maxTokens,
system: declaration.instructions,
messages: agentAnthropicMessages(declaration, run),
tools: tools.map((tool) => ({ name: tool.name, description: tool.description, input_schema: tool.schema })),
stream: true,
};
if (definition.temperature !== null) body.temperature = definition.temperature;
const response = await modelSend(definition, "/v1/messages", anthropicHeaders(definition), body, signal, true);
let text = "";
const blocks = new Map();
let input = 0;
let output = 0;
for await (const event of modelSseEvents(definition, response, signal)) {
if (event?.type === "message_start") input = event?.message?.usage?.input_tokens ?? input;
if (event?.type === "content_block_start" && event?.content_block?.type === "tool_use") {
blocks.set(event.index, { id: event.content_block.id, name: event.content_block.name, raw: "" });
}
if (event?.type === "content_block_delta" && typeof event?.delta?.text === "string") {
text += event.delta.text;
yield event.delta.text;
}
if (event?.type === "content_block_delta" && typeof event?.delta?.partial_json === "string") {
const block = blocks.get(event.index);
if (block !== undefined) block.raw += event.delta.partial_json;
}
if (event?.type === "message_delta") output = event?.usage?.output_tokens ?? output;
}
return Object.freeze({
text,
toolCalls: [...blocks.values()].map((block) => Object.freeze({ id: block.id, name: block.name, arguments: agentParsedArguments(block.raw) })),
usage: modelUsage(input, output),
});
}
async function* agentOpenAiTurn(declaration, definition, run, tools, signal) {
const body = {
model: definition.modelId,
messages: agentOpenAiMessages(declaration, run),
max_tokens: definition.maxTokens === null ? 1024 : definition.maxTokens,
tools: tools.map((tool) => ({ type: "function", function: { name: tool.name, description: tool.description, parameters: tool.schema } })),
stream: true,
stream_options: { include_usage: true },
};
if (definition.temperature !== null) body.temperature = definition.temperature;
const response = await modelSend(definition, "/v1/chat/completions", openaiHeaders(definition), body, signal, true);
let text = "";
const calls = new Map();
let input = 0;
let output = 0;
for await (const event of modelSseEvents(definition, response, signal)) {
const delta = event?.choices?.[0]?.delta;
if (typeof delta?.content === "string" && delta.content.length !== 0) {
text += delta.content;
yield delta.content;
}
if (Array.isArray(delta?.tool_calls)) {
for (const call of delta.tool_calls) {
const index = Number.isSafeInteger(call?.index) ? call.index : 0;
const existing = calls.get(index) ?? { id: "", name: "", raw: "" };
if (typeof call?.id === "string" && call.id.length !== 0) existing.id = call.id;
if (typeof call?.function?.name === "string" && call.function.name.length !== 0) existing.name = call.function.name;
if (typeof call?.function?.arguments === "string") existing.raw += call.function.arguments;
calls.set(index, existing);
}
}
if (event?.usage) {
input = event.usage.prompt_tokens ?? input;
output = event.usage.completion_tokens ?? output;
}
}
return Object.freeze({
text,
toolCalls: [...calls.values()].filter((call) => call.name.length !== 0).map((call, index) => Object.freeze({ id: call.id.length === 0 ? `call_${index}` : call.id, name: call.name, arguments: agentParsedArguments(call.raw) })),
usage: modelUsage(input, output),
});
}
// The scenario side of the loop. `noxid test` installs the WO-30 controller
// with a per-agent turn script; while it is installed the engine performs no
// provider I/O at all, and a turn the script does not supply fails closed with
// `MODEL_STUB_REQUIRED` rather than reaching a provider. The tool-call shape a
// scripted turn produces is exactly the shape the two provider readers
// produce, so the loop below this point cannot tell the difference — which is
// the point: a scenario exercises the real loop.
async function* agentScenarioTurn(declaration, controller) {
const script = controller.takeAgentTurn(declaration.name);
if (script === null || script === undefined) {
throw modelError(
"MODEL_STUB_REQUIRED",
`scenario ran agent \`${declaration.name}\` past its scripted turns with no stub left; add another entry to \`given: agent ${declaration.name} = turns [ ... ]\` (\`text "..."\`, \`tool <Endpoint> { field = value }\`, or \`final { field = value }\`)`,
);
}
let text = "";
for (const chunk of script.text) { text += chunk; yield chunk; }
const toolCalls = [];
if (script.call !== null && script.call !== undefined) {
const isFinal = script.call.kind === "final";
const name = isFinal ? AGENT_FINAL_ANSWER_TOOL : script.call.endpoint;
const args = isFinal && agentFinalAnswerWrapped(declaration)
? { value: script.call.arguments?.value }
: script.call.arguments;
toolCalls.push(Object.freeze({ id: `scenario_call_${script.index}`, name, arguments: args }));
}
return Object.freeze({
text,
toolCalls: Object.freeze(toolCalls),
usage: modelUsage(script.inputTokens ?? 0, script.outputTokens ?? 0),
});
}
function agentProviderTurn(declaration, definition, run, tools, signal) {
const controller = globalThis.__NOXID_MODEL_SCENARIO__;
if (controller !== undefined && controller !== null && typeof controller.takeAgentTurn === "function") {
return agentScenarioTurn(declaration, controller);
}
return definition.provider === "anthropic"
? agentAnthropicTurn(declaration, definition, run, tools, signal)
: agentOpenAiTurn(declaration, definition, run, tools, signal);
}
// ---------------------------------------------------------------------------
// Tool dispatch: the full endpoint path, under the run's principal.
// ---------------------------------------------------------------------------
// Three-valued, because a deferred capability is not a denial: the host
// authorizer may answer `true`, `"defer"` / `{ defer: true }`, or anything
// else, and only the first runs the endpoint.
async function agentAuthorizeTool(declaration, entry, request, environment, executionContext, signal, route) {
if (entry.capabilities.length === 0) return Object.freeze({ kind: "allowed" });
if (typeof authorize !== "function") return Object.freeze({ kind: "denied", capability: entry.capabilities[0], reason: "no authorizer is configured" });
for (const capability of entry.capabilities) {
let decision;
try {
decision = await authorize(Object.freeze({
capability,
semanticId: entry.schema.id,
traceId: __noxidTraceIdForRequest(request),
target: "agent",
agent: declaration.name,
route,
request,
environment,
executionContext,
signal,
}));
} catch { decision = false; }
if (decision === true) continue;
if (decision === "defer" || decision?.defer === true) return Object.freeze({ kind: "deferred", capability });
return Object.freeze({ kind: "denied", capability, reason: "the host authorizer denied it" });
}
return Object.freeze({ kind: "allowed" });
}
async function agentToolResponseBody(response) {
const text = await mcpBoundedResponseText(response);
if (text.length > AGENT_TOOL_RESULT_MAX_BYTES) {
throw agentEngineError("AGENT_TOOL_RESULT_TOO_LARGE", "the tool result exceeds the bounded agent tool-result size");
}
if (text.length === 0) return null;
try { return JSON.parse(text); } catch { return { text }; }
}
async function agentCallTool(declaration, entry, call, outerRequest, environment, executionContext) {
let endpointRequest;
try { endpointRequest = mcpEndpointRequest(outerRequest, entry.schema, call.arguments); }
catch (cause) {
return Object.freeze({ ok: false, content: Object.freeze({ code: "AGENT_TOOL_ARGUMENTS_INVALID", message: cause?.message ?? "the tool arguments could not be encoded for the endpoint" }) });
}
inheritNoxidRequestTrace(outerRequest, endpointRequest);
// The run's principal, erased to the runtime shape the kernel already
// builds: `agent:<AgentId>:acting:<session|system>`.
__noxidAgentRequests.set(endpointRequest, declaration.agentId);
const response = await handleEndpointRequest(endpointRequest, new URL(endpointRequest.url), environment, executionContext);
if (response === null) {
return Object.freeze({ ok: false, content: Object.freeze({ code: "AGENT_TOOL_DISPATCH_FAILED", message: `tool ${entry.schema.name} did not resolve to its declared endpoint` }) });
}
__noxidTraceResponseFailure(endpointRequest, response);
// Scenario observation only, and only of what the endpoint boundary already
// saw: the arguments the run dispatched and the status the endpoint's own
// validator produced. Nothing here changes the dispatch.
{
const observer = globalThis.__NOXID_MODEL_SCENARIO__;
if (observer !== undefined && observer !== null && typeof observer.recordAgentToolCall === "function") {
observer.recordAgentToolCall({ agent: declaration.name, tool: entry.schema.name, arguments: call.arguments, status: response.status, ok: response.ok });
}
}
let body;
try { body = await agentToolResponseBody(response); }
catch (cause) {
return Object.freeze({ ok: false, content: Object.freeze({ code: cause?.code ?? "AGENT_TOOL_RESULT_FAILED", message: cause?.message ?? "the tool result could not be represented safely" }) });
}
// The endpoint has already validated its own result against its declared
// type; an `ok: false` body is a refusal the model is told about verbatim,
// not a run failure.
if (!response.ok || body?.ok === false) {
return Object.freeze({ ok: false, content: Object.freeze({ status: response.status, error: body?.error ?? null }) });
}
return Object.freeze({ ok: true, content: body?.ok === true ? body.value ?? null : body });
}
// ---------------------------------------------------------------------------
// The loop.
// ---------------------------------------------------------------------------
async function* agentRunLoop(declaration, run, context) {
const definition = modelDefinitionFor(declaration.model);
const registry = agentToolRegistry(declaration);
const tools = agentProviderTools(declaration, registry);
const outputValidator = agentValidatorFor(declaration.output, "output", declaration);
const wrapped = agentFinalAnswerWrapped(declaration);
const runStarted = Date.now();
const runTrace = __noxidTraceForRequest(context.request) ?? (tracingMode === "full" ? __noxidTraceContext() : null);
let tokensInput = 0;
let tokensOutput = 0;
const finish = async (state, error, output) => {
run.state = state;
run.error = error;
run.output = output ?? null;
run.pending = null;
await agentPersistRun(run);
__noxidTraceEmit(runTrace, "agent.run", {
semanticId: declaration.id,
agent: declaration.name,
agentRun: run.runId,
state,
durationMs: Date.now() - runStarted,
tokensInput,
tokensOutput,
code: error?.code,
});
};
// One abort, two meanings: the WO-18 deadline fails the run with
// `AGENT_TIMEOUT`, a client disconnect or an explicit cancel ends it as
// `Cancelled`. Both stop the model call and the run.
async function* stopped() {
if (context.abortKind() === "timeout") {
const error = { code: "AGENT_TIMEOUT", message: `agent \`${declaration.name}\` exceeded its declared ${declaration.timeoutMs} ms run timeout` };
await finish("Failed", error, null);
yield agentEvent(declaration, "Failed", null, error);
return;
}
await finish("Cancelled", null, null);
const cancelled = agentEvent(declaration, "Cancelled");
if (cancelled !== null) yield cancelled;
}
const started = agentEvent(declaration, "Started");
if (started !== null) yield started;
while (true) {
if (context.signal.aborted) { yield* stopped(); return; }
// A resumed run finishes the turn the pause froze before it asks the model
// for another one: the approved call is the first work it does.
const last = run.turns[run.turns.length - 1] ?? null;
const resuming = last !== null && last.results.length < last.toolCalls.length;
let turn;
if (resuming) {
turn = last;
} else {
if (run.turns.length >= declaration.maxTurns) {
const error = { code: "AGENT_MAX_TURNS", message: `agent \`${declaration.name}\` reached its declared ceiling of ${declaration.maxTurns} turns without a final answer` };
await finish("Failed", error, null);
yield agentEvent(declaration, "Failed", null, error);
return;
}
const turnIndex = run.turns.length;
const turnStarted = Date.now();
let assistant;
try {
const source = agentProviderTurn(declaration, definition, run, tools, context.signal);
for (;;) {
const step = await source.next();
if (step.done) { assistant = step.value; break; }
const token = agentEvent(declaration, "Token", null, step.value);
if (token !== null) yield token;
}
} catch (cause) {
if (context.signal.aborted) { yield* stopped(); return; }
const error = {
code: "AGENT_MODEL_FAILED",
message: `agent \`${declaration.name}\` could not complete a model turn (${cause?.code ?? "MODEL_PROVIDER_ERROR"})`,
};
await finish("Failed", error, null);
yield agentEvent(declaration, "Failed", null, error);
return;
}
tokensInput += assistant.usage.inputTokens;
tokensOutput += assistant.usage.outputTokens;
__noxidTraceEmit(runTrace, "agent.turn", {
semanticId: declaration.id,
agent: declaration.name,
agentRun: run.runId,
agentTurn: turnIndex,
durationMs: Date.now() - turnStarted,
tokensInput: assistant.usage.inputTokens,
tokensOutput: assistant.usage.outputTokens,
});
turn = { index: turnIndex, text: assistant.text, toolCalls: assistant.toolCalls.map((call) => ({ id: call.id, name: call.name, arguments: call.arguments })), results: [] };
run.turns.push(turn);
await agentPersistRun(run);
const finalCall = turn.toolCalls.find((call) => call.name === AGENT_FINAL_ANSWER_TOOL) ?? null;
if (finalCall !== null || turn.toolCalls.length === 0) {
let candidate;
if (finalCall !== null) candidate = wrapped ? finalCall.arguments.value : finalCall.arguments;
else {
try { candidate = JSON.parse(turn.text); }
catch {
const error = { code: "AGENT_OUTPUT_INVALID", message: `agent \`${declaration.name}\` ended a turn without calling \`${AGENT_FINAL_ANSWER_TOOL}\`, and its text is not a \`${declaration.output.type}\` value` };
await finish("Failed", error, null);
yield agentEvent(declaration, "Failed", null, error);
return;
}
}
let output;
try { output = outputValidator(candidate, true); }
catch (cause) {
const error = { code: "AGENT_OUTPUT_INVALID", message: `agent \`${declaration.name}\` produced a final answer that violates its declared \`${declaration.output.type}\` output: ${cause?.message ?? String(cause)}` };
await finish("Failed", error, null);
yield agentEvent(declaration, "Failed", null, error);
return;
}
await finish("Completed", null, output === undefined ? null : output);
yield agentEvent(declaration, "Completed", null, output === undefined ? null : output);
return;
}
}
while (turn.results.length < turn.toolCalls.length) {
if (context.signal.aborted) { yield* stopped(); return; }
const call = turn.toolCalls[turn.results.length];
const entry = registry.get(call.name) ?? null;
if (entry === null) {
// A tool outside the derived registry is unreachable, and asking for it
// is not a failure: the model is told the name is not available and the
// loop continues with that as the tool result.
turn.results.push({ id: call.id, name: call.name, ok: false, content: { code: "AGENT_TOOL_NOT_AVAILABLE", message: `\`${call.name}\` is not one of this agent's tools; the available tools are ${[...registry.keys(), AGENT_FINAL_ANSWER_TOOL].join(", ")}` } });
await agentPersistRun(run);
continue;
}
const preapproved = run.approved === call.id;
const decision = preapproved
? Object.freeze({ kind: "allowed" })
: await agentAuthorizeTool(declaration, entry, context.request, context.environment, context.executionContext, context.signal, context.route);
if (preapproved) {
delete run.approved;
await agentPersistRun(run);
}
if (decision.kind === "deferred") {
run.state = "Paused";
run.pending = { id: call.id, name: call.name, endpoint: entry.schema.name, capability: decision.capability, arguments: call.arguments, turn: turn.index };
await agentPersistRun(run);
const permission = agentEvent(declaration, "PermissionRequired", agentToolCanonical(declaration, run, entry, call, turn, {
status: "deferred", ok: false, code: "AGENT_PERMISSION_REQUIRED",
message: `capability ${decision.capability} was deferred to a human`,
summary: `awaiting approval for ${decision.capability}`,
capability: decision.capability,
}));
if (permission !== null) yield permission;
__noxidTraceEmit(runTrace, "agent.run", {
semanticId: declaration.id, agent: declaration.name, agentRun: run.runId,
state: "Paused", durationMs: Date.now() - runStarted, tokensInput, tokensOutput,
});
yield agentEvent(declaration, "Paused", null, run.runId);
return;
}
const toolStarted = agentEvent(declaration, "ToolStarted", agentToolCanonical(declaration, run, entry, call, turn, { status: "started", ok: true }));
if (toolStarted !== null) yield toolStarted;
const toolBegan = Date.now();
let outcome;
if (decision.kind === "denied") {
outcome = Object.freeze({ ok: false, content: Object.freeze({ code: "AGENT_TOOL_DENIED", message: `capability ${decision.capability} was refused: ${decision.reason}` }) });
} else {
try { outcome = await agentCallTool(declaration, entry, call, context.request, context.environment, context.executionContext); }
catch (cause) {
if (context.signal.aborted) { yield* stopped(); return; }
const error = { code: "AGENT_TOOL_FAILED", message: `agent \`${declaration.name}\` could not dispatch tool \`${entry.schema.name}\` (${cause?.code ?? "unknown"})` };
await finish("Failed", error, null);
yield agentEvent(declaration, "Failed", null, error);
return;
}
}
__noxidTraceEmit(runTrace, "agent.tool", {
semanticId: declaration.id, agent: declaration.name, agentRun: run.runId, agentTurn: turn.index,
toolEndpoint: entry.schema.id, durationMs: Date.now() - toolBegan,
code: outcome.ok ? undefined : "AGENT_TOOL_FAILED",
});
turn.results.push({ id: call.id, name: call.name, ok: outcome.ok, content: outcome.content });
await agentPersistRun(run);
const completed = agentEvent(declaration, "ToolCompleted", agentToolCanonical(declaration, run, entry, call, turn, {
status: outcome.ok ? "completed" : "failed",
ok: outcome.ok,
summary: outcome.ok ? `${entry.schema.name} completed` : `${entry.schema.name} failed`,
code: outcome.ok ? "" : outcome.content?.code ?? "AGENT_TOOL_FAILED",
message: outcome.ok ? "" : outcome.content?.message ?? "the tool refused",
result: JSON.stringify(outcome.content ?? null),
}));
if (completed !== null) yield completed;
}
}
}
function agentToolCanonical(declaration, run, entry, call, turn, overrides) {
return {
name: entry.schema.name,
tool: entry.schema.name,
endpoint: entry.schema.id,
capability: entry.capabilities[0] ?? "",
arguments: JSON.stringify(call.arguments),
summary: entry.schema.name,
status: "started",
ok: true,
code: "",
message: "",
runId: run.runId,
turn: turn.index,
agent: declaration.name,
result: null,
...overrides,
};
}
// ---------------------------------------------------------------------------
// The SSE session.
// ---------------------------------------------------------------------------
function agentEndpointSchema(declaration, path) {
return Object.freeze({
id: declaration.id,
name: declaration.name,
version: 1,
description: null,
kind: "stream",
method: "POST",
path,
params: Object.freeze([]),
query: Object.freeze([]),
body: Object.freeze([]),
result: Object.freeze({ id: declaration.id, type: "String", typeId: null, validator: "", errorValidator: null }),
capabilities: Object.freeze([]),
timeoutMs: declaration.timeoutMs,
limit: null,
cache: null,
idempotent: false,
middleware: Object.freeze([]),
invalidates: Object.freeze([]),
});
}
function agentStreamResponse(schema, run, events, headers, release) {
const encoder = new TextEncoder();
let sequence = 0;
let iterator = null;
const body = new ReadableStream({
async start(controller) {
const emit = (frame) => {
try { controller.enqueue(encoder.encode(frame)); return true; }
catch { return false; }
};
try {
iterator = events[Symbol.asyncIterator]();
for (;;) {
const next = await iterator.next();
if (next.done) break;
if (next.value === null || next.value === undefined) continue;
sequence += 1;
if (!emit(streamFrame("message", next.value, `${run.runId}:${sequence}`))) break;
}
} catch (cause) {
const code = typeof cause?.code === "string" ? cause.code : "AGENT_RUN_FAILED";
emit(streamErrorFrame(schema, code, cause?.agentEngine === true ? cause.message : "The agent run failed"));
} finally {
release();
try { controller.close(); } catch {}
}
},
async cancel() {
const controller = agentRunControllers.get(run.runId);
if (controller !== undefined) controller.abort("disconnect");
if (iterator !== null && typeof iterator.return === "function") {
try { await iterator.return(); } catch {}
}
release();
},
});
return endpointResponseWithHeaders(new Response(body, { status: 200, headers: endpointStreamHeaders() }), headers);
}
async function agentSession(request, declaration, schema, capability, environment, executionContext, prepare) {
return withEndpointDeadline(schema, async (deadlineSignal, deadlineAt) => {
const middleware = await applyEndpointMiddleware(request, schema, Object.create(null), Object.create(null), environment, executionContext, deadlineSignal);
if (middleware.response) return endpointResponseWithHeaders(middleware.response, middleware.headers);
const guarded = Object.freeze({ ...schema, capabilities: Object.freeze([capability]) });
const authorization = await authorizeEndpoint(request, guarded, middleware.route, environment, executionContext, deadlineSignal);
if (authorization) return endpointResponseWithHeaders(authorization, middleware.headers);
const middlewareContext = middleware.context ?? EMPTY_MIDDLEWARE_CONTEXT;
// `Principal.Agent { id: AgentId(<agent>), actingFor }`, erased to the
// runtime shape the kernel already builds for an agent-attributed request.
const principal = __noxidPrincipal(middlewareContext, environment, declaration.agentId);
__noxidTraceBindPrincipal({ request }, principal);
__noxidTraceSemantic(request, "endpoint", declaration.id);
let prepared;
try { prepared = await prepare(principal, middlewareContext, middleware, deadlineSignal); }
catch (cause) {
const code = typeof cause?.code === "string" ? cause.code : "AGENT_RUN_FAILED";
return endpointResponseWithHeaders(failure(cause?.status ?? 500, code, cause?.agentEngine === true ? cause.message : "The agent run could not start", declaration.id), middleware.headers);
}
if (prepared.response) return endpointResponseWithHeaders(prepared.response, middleware.headers);
const run = prepared.run;
// `withEndpointDeadline` clears its own timer the moment this operation
// resolves with the streaming response, so the run owns the remainder of
// the declared timeout itself — exactly as a stream endpoint does.
const controller = new AbortController();
let abortKind = null;
const abort = (kind) => {
if (controller.signal.aborted) return;
abortKind = kind;
controller.abort(kind);
};
const onDisconnect = () => abort("disconnect");
request.signal.addEventListener("abort", onDisconnect, { once: true });
if (request.signal.aborted) abort("disconnect");
const timer = setTimeout(() => abort("timeout"), Math.max(0, deadlineAt - Date.now()));
agentRunControllers.set(run.runId, controller);
const release = () => {
clearTimeout(timer);
request.signal.removeEventListener("abort", onDisconnect);
if (agentRunControllers.get(run.runId) === controller) agentRunControllers.delete(run.runId);
};
const context = Object.freeze({
request,
environment,
executionContext,
signal: controller.signal,
abortKind: () => abortKind,
principal,
route: middleware.route,
});
return agentStreamResponse(schema, run, agentRunLoop(declaration, run, context), middleware.headers, release);
});
}
async function agentDecodeInput(declaration, request) {
let payload;
try { payload = JSON.parse(await request.text()); }
catch { throw Object.assign(agentEngineError("AGENT_INPUT_INVALID", "an agent run requires a JSON body of the form { \"input\": ... }"), { status: 400 }); }
if (payload === null || typeof payload !== "object" || Array.isArray(payload) || !Object.hasOwn(payload, "input")) {
throw Object.assign(agentEngineError("AGENT_INPUT_INVALID", "an agent run requires a JSON body of the form { \"input\": ... }"), { status: 400 });
}
const validator = agentValidatorFor(declaration.input, "input", declaration);
try { return validator(payload.input, true); }
catch (cause) {
throw Object.assign(agentEngineError("AGENT_INPUT_INVALID", `the run input violates the declared \`${declaration.input.type}\` type: ${cause?.message ?? String(cause)}`), { status: 422 });
}
}
async function handleAgentRunRequest(request, url, environment, executionContext) {
if (!url.pathname.startsWith(agentRunPrefix)) return null;
let segments;
try { segments = url.pathname.slice(agentRunPrefix.length).split("/").filter((segment) => segment.length !== 0).map(decodeURIComponent); }
catch { return failure(400, "AGENT_PATH_ENCODING_INVALID", "The agent run path contains invalid percent encoding"); }
if (segments.length !== 2 && segments.length !== 4) return failure(404, "AGENT_NOT_FOUND", "No agent run surface exists at this path");
if (segments[1] !== "runs") return failure(404, "AGENT_NOT_FOUND", "No agent run surface exists at this path");
if (segments.length === 4 && segments[3] !== "resume") return failure(404, "AGENT_NOT_FOUND", "No agent run surface exists at this path");
const declaration = agentDeclarationFor(segments[0]);
if (declaration === null) return failure(404, "AGENT_NOT_FOUND", `No engine agent named ${segments[0]} is declared`);
if (request.method !== "POST") return failure(405, "AGENT_METHOD_NOT_ALLOWED", "Agent runs require POST", declaration.id, null, { allow: "POST" });
if (segments.length === 2) {
const schema = agentEndpointSchema(declaration, `${agentRunPrefix}${declaration.name}/runs`);
return agentSession(request, declaration, schema, declaration.runCapability, environment, executionContext, async (principal) => {
const input = await agentDecodeInput(declaration, request);
const now = new Date().toISOString();
const run = {
schema: AGENT_RUN_RECORD_SCHEMA,
runId: agentRunId(),
agent: declaration.name,
agentSemanticId: declaration.id,
state: "Running",
principal: principal.canonical,
version: 0,
input,
turns: [],
pending: null,
output: null,
error: null,
startedAt: now,
updatedAt: now,
};
await agentPersistRun(run);
return { run };
});
}
const runId = segments[2];
if (!AGENT_RUN_ID.test(runId)) return failure(400, "AGENT_RUN_NOT_FOUND", "The run id is not a run this build could have created", declaration.id);
const schema = agentEndpointSchema(declaration, `${agentRunPrefix}${declaration.name}/runs/${runId}/resume`);
return agentSession(request, declaration, schema, declaration.resumeCapability, environment, executionContext, async (principal, middlewareContext, middleware, signal) => {
await __noxidReconcileAgentRuns().catch(() => {});
let run;
try { run = await agentLoadRun(declaration.name, runId); }
catch (cause) { return { response: failure(409, cause?.code ?? "AGENT_RUN_RECORD_DRIFT", cause?.message ?? "The persisted run could not be read", declaration.id) }; }
if (run === null) return { response: failure(404, "AGENT_RUN_NOT_FOUND", `Agent ${declaration.name} has no run ${runId}`, declaration.id) };
// Authority before state: a run belongs to the principal that started it.
// The stored canonical principal is the whole identity — the agent *and*
// the user it acts for — so a second user holding
// `agents.<name>.resume` cannot take over another user's paused run, and
// nothing about the record is disclosed or written before this check.
if (run.principal !== principal.canonical) {
return { response: failure(403, "AGENT_RUN_PRINCIPAL_MISMATCH", `Agent ${declaration.name} run ${runId} was started by another principal, and only the principal that started a run resumes it; resume it as that principal, or start a new run as this one`, declaration.id, { agent: declaration.name, runId }) };
}
if (run.state !== "Paused" || run.pending === null) {
// The conflict is structured for the same reason the acquisition
// conflict is: a caller that arrives after the run finished should read
// the outcome here rather than go looking for a second one. `output` is
// the recorded answer of a completed run and null in every other state,
// which is the only state that has one.
return { response: failure(409, "AGENT_RUN_NOT_PAUSED", `Agent ${declaration.name} run ${runId} is ${run.state}, and only a paused run resumes`, declaration.id, { runId, state: run.state, output: run.state === "Completed" ? run.output ?? null : null }) };
}
const registry = agentToolRegistry(declaration);
const entry = registry.get(run.pending.endpoint) ?? null;
if (entry === null) {
return { response: failure(409, "AGENT_RUN_TOOL_UNAVAILABLE", `The paused tool ${run.pending.endpoint} is no longer in this agent's registry`, declaration.id) };
}
// The deferred capability is re-checked with the *resuming* principal, not
// the one that paused: approval is an act by whoever is resuming.
const decision = await agentAuthorizeTool(declaration, entry, request, environment, executionContext, signal, middleware.route);
if (decision.kind !== "allowed") {
return { response: failure(403, "AGENT_PERMISSION_DENIED", `Capability ${run.pending.capability} is still not granted for agent ${declaration.name}`, declaration.id, { capability: run.pending.capability }) };
}
// The approved call is the first work of the resumed run: the pause froze
// the turn immediately before dispatch, and `approved` carries the fresh
// grant so the loop does not ask the authorizer a second time.
// `principal` is never rewritten: it is the ownership record the check
// above enforces, not a log of who touched the run last.
const acquired = await agentAcquirePausedRun(run);
if (acquired === null) {
// The winner may still be between its own writes, so the reported state
// is read with a bounded retry rather than guessed.
let current = null;
for (let attempt = 0; attempt < 3 && current === null; attempt += 1) {
try { current = await agentLoadRun(declaration.name, runId); } catch { break; }
if (current === null) await new Promise((resolve) => setTimeout(resolve, 5));
}
const state = current === null ? "Unknown" : current.state;
return { response: failure(409, "AGENT_RUN_ACQUIRED", `Agent ${declaration.name} run ${runId} was claimed by another resume and is now ${state}; a paused run dispatches exactly once, so read that resume's stream instead of starting a second one`, declaration.id, { runId, state, output: current?.output ?? null }) };
}
return { run: acquired };
});
}
"##;