Skip to content

fix(agents): emit model call context on project agent runs #3373

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kwakayama merged 1 commit into from
Aug 5, 2026
Merged

fix(agents): emit model call context on project agent runs #3373

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 129 additions & 1 deletion src/internal-agents/run-stream.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,22 @@ import type {
import { registerSkill } from "#veryfront/skill/registry.ts";
import type { RemoteToolSource, Tool } from "#veryfront/tool";
import { __resetLoggerConfigForTests, type LogEntry } from "#veryfront/utils/logger/logger.ts";
import type { AgentRunEventSink } from "#veryfront/runtime/model-call-context.ts";
import { getActiveRunEventSinks } from "#veryfront/runtime/run-event-sink-context.ts";
import { AgentRunSessionManager } from "./session-manager.ts";
import { buildMergedTools, createRuntimeAgentStreamResponse } from "./run-stream.ts";
import {
buildMergedTools,
createRuntimeAgentStreamResponse,
MODEL_CALL_CONTEXT_SSE_EVENT_NAME,
} from "./run-stream.ts";

function parseSseFrames(body: string): Array<{ event: string; data: unknown }> {
return body.split("\n\n").flatMap((frame) => {
const event = /^event: (.+)$/m.exec(frame)?.[1];
const data = /^data: (.+)$/m.exec(frame)?.[1];
return event && data ? [{ event, data: JSON.parse(data) as unknown }] : [];
});
}

class RecordingSpan implements Span {
readonly attributes: Record<string, AttributeValue> = {};
Expand Down Expand Up @@ -2561,4 +2575,118 @@ describe("internal-agents/run-stream", () => {
);
assertEquals(debugEntry?.component, "internal-agent-run-stream");
});
describe("model call context", () => {
const modelCallContextEvent = {
type: "AGENT_RUN_MODEL_CALL_CONTEXT",
messages: [{ role: "system", content: "test system prompt" }],
tools: [{ type: "function", name: "granted_tool", inputSchema: {} }],
};

function contextAgent(): Agent {
return {
id: "context-agent",
config: { id: "context-agent", model: "anthropic/claude-opus-4-6", system: "test" },
} as unknown as Agent;
}

function contextRunInput(runId: string) {
return {
agentId: "context-agent",
threadId: crypto.randomUUID(),
runId,
messages: [],
tools: [],
context: [],
} as Parameters<typeof createRuntimeAgentStreamResponse>[0];
}

it("streams a context emitted while the runtime stream is created", async () => {
let sinkDuringCreate: AgentRunEventSink | undefined;

const response = await createRuntimeAgentStreamResponse(
contextRunInput("run_context_setup"),
contextAgent(),
{
sessionManager: new AgentRunSessionManager(),
createRuntime: () => ({
stream: async () => {
// The real runtime dispatches its first model call here, so the
// sink has to already be scoped by the time stream() runs.
sinkDuringCreate = getActiveRunEventSinks().mandatory;
await sinkDuringCreate?.(modelCallContextEvent as never);
return new ReadableStream<Uint8Array>({
start(controller) {
controller.close();
},
});
},
}),
},
);

const frames = parseSseFrames(await response.text());
assertEquals(Boolean(sinkDuringCreate), true);
assertEquals(
frames
.filter((frame) => frame.event === MODEL_CALL_CONTEXT_SSE_EVENT_NAME)
.map((frame) => frame.data),
[modelCallContextEvent],
);
});

it("keeps the context ahead of the step it describes", async () => {
const response = await createRuntimeAgentStreamResponse(
contextRunInput("run_context_order"),
contextAgent(),
{
sessionManager: new AgentRunSessionManager(),
createRuntime: () => ({
stream: async () => {
await getActiveRunEventSinks().mandatory?.(modelCallContextEvent as never);
return new ReadableStream<Uint8Array>({
start(controller) {
controller.close();
},
});
},
}),
},
);

const names = parseSseFrames(await response.text()).map((frame) => frame.event);
assertEquals(names[0], "RunStarted");
assertEquals(names[1], MODEL_CALL_CONTEXT_SSE_EVENT_NAME);
});
Comment on lines +2637 to +2659

@coderabbitai coderabbitai Bot Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Test ordering against a runtime step event.

Lines 2644-2650 return a stream with no mapped runtime step. The assertion only proves that model-call context follows RunStarted. It does not prove that model-call context precedes the step it describes. Emit a runtime SSE step frame and assert that the context frame occurs before that mapped frame.

As per coding guidelines, “For behavior changes, add or update a focused failing test before changing the implementation.”

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/internal-agents/run-stream.test.ts` around lines 2637 - 2659, Update the
test “keeps the context ahead of the step it describes” to emit a mapped runtime
step SSE event after modelCallContextEvent, rather than closing the stream
without a step. Parse the resulting frames and assert the model-call context
event appears before the corresponding runtime step event, while retaining the
RunStarted ordering assertion.

Source: Coding guidelines


it("streams a context emitted for a later step while the client reads", async () => {
let sinkDuringConsume: AgentRunEventSink | undefined;

const response = await createRuntimeAgentStreamResponse(
contextRunInput("run_context_step_two"),
contextAgent(),
{
sessionManager: new AgentRunSessionManager(),
createRuntime: () => ({
stream: async () =>
new ReadableStream<Uint8Array>({
// Multi-step runs dispatch later model calls as the stream is
// pulled, long after stream() returned.
async pull(controller) {
sinkDuringConsume = getActiveRunEventSinks().mandatory;
await sinkDuringConsume?.(modelCallContextEvent as never);
controller.close();
},
}),
}),
},
);

const frames = parseSseFrames(await response.text());
assertEquals(Boolean(sinkDuringConsume), true);
assertEquals(
frames.filter((frame) => frame.event === MODEL_CALL_CONTEXT_SSE_EVENT_NAME).length,
1,
);
});
});
});
99 changes: 77 additions & 22 deletions src/internal-agents/run-stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@ import {
mapRuntimeEventToAgUi,
parseSseJsonEvents,
} from "./ag-ui-sse.ts";
import type { AgentRunEvent, AgentRunEventSink } from "#veryfront/runtime/model-call-context.ts";
import { runWithMandatoryRunEventSink } from "#veryfront/runtime/run-event-sink-context.ts";
import { AgentRunCancelledError, type AgentRunSessionManager } from "./session-manager.ts";
import { composeInternalAgentRunSystemPrompt } from "./run-system-prompt.ts";
import type { RuntimeRunAgentInput } from "./schema.ts";
Expand All @@ -64,6 +66,12 @@ const INTERNAL_AGENT_RUNTIME_HEARTBEAT_INTERVAL_MS = 25_000;
const INTERNAL_AGENT_RUNTIME_HEARTBEAT_FRAME = new TextEncoder().encode(
": internal-agent-runtime-heartbeat\n\n",
);
/**
* SSE frame name carrying AGENT_RUN_MODEL_CALL_CONTEXT to veryfront-api. Not an
* AG-UI event: veryfront-api persists it under its own event type rather than
* folding it into the run's public event sequence.
*/
export const MODEL_CALL_CONTEXT_SSE_EVENT_NAME = "AgentRunModelCallContext";

type RuntimeFilteredAgent = Agent & {
config: Agent["config"] & {
Expand Down Expand Up @@ -638,6 +646,31 @@ function compactRuntimeMessagesForStream(
) as Message[];
}

/**
* Relays run events produced by the runtime into the run's SSE stream.
*
* The first model call is dispatched while `runtime.stream()` is still being
* awaited, before there is a controller to enqueue into, so events raised
* before `attach` are buffered and replayed once the stream opens.
*/
function createModelCallContextRelay(): {
sink: AgentRunEventSink;
attach: (emit: (event: AgentRunEvent) => void) => void;
} {
const buffered: AgentRunEvent[] = [];
let emit: ((event: AgentRunEvent) => void) | undefined;
return {
sink: (event) => {
if (emit) emit(event);
else buffered.push(event);
},
attach: (next) => {
emit = next;
for (const event of buffered.splice(0)) next(event);
},
};
}
Comment on lines +656 to +672

export async function createRuntimeAgentStreamResponse(
input: RuntimeRunAgentInput,
agent: Agent,
Expand All @@ -659,6 +692,7 @@ export async function createRuntimeAgentStreamResponse(
let completedResponse: AgentResponse | null = null;
let runtimeStream: ReadableStream<Uint8Array>;
let closeSandbox = createIdempotentAsyncCleanup();
const modelCallContextRelay = createModelCallContextRelay();
try {
const forwardedAllowedRemoteToolNames = getAllowedRemoteToolNames(input.forwardedProps);
const sourceAllowedRemoteToolNames = getAgentAllowedRemoteToolNames(agent);
Expand Down Expand Up @@ -788,27 +822,34 @@ export async function createRuntimeAgentStreamResponse(
runtimeToolNames.length,
);
const maxOutputTokens = getForwardedMaxOutputTokens(input.forwardedProps);
const candidateRuntimeStream = await runtime.stream(
runtimeMessages,
{
threadId: input.threadId,
runId: input.runId,
...(deps.projectAgentSandbox?.authToken
? { authToken: deps.projectAgentSandbox.authToken }
: {}),
...(input.parentRunId ? { parentRunId: input.parentRunId } : {}),
...(input.state !== undefined ? { state: input.state } : {}),
context: input.context,
forwardedProps: input.forwardedProps,
},
{
onFinish: (response) => {
completedResponse = response;
},
},
undefined,
maxOutputTokens,
abortSignal,
// Scoped here because the runtime dispatches the run's first model call
// before stream() resolves. Later steps inherit this scope through the
// stream they are pumped from.
const candidateRuntimeStream = await runWithMandatoryRunEventSink(
modelCallContextRelay.sink,
() =>
runtime.stream(
runtimeMessages,
{
threadId: input.threadId,
runId: input.runId,
...(deps.projectAgentSandbox?.authToken
? { authToken: deps.projectAgentSandbox.authToken }
: {}),
...(input.parentRunId ? { parentRunId: input.parentRunId } : {}),
...(input.state !== undefined ? { state: input.state } : {}),
context: input.context,
forwardedProps: input.forwardedProps,
},
{
onFinish: (response) => {
completedResponse = response;
},
},
undefined,
maxOutputTokens,
abortSignal,
),
);
if (candidateRuntimeStream.locked) {
throw new TypeError("Internal agent runtime returned a locked stream");
Expand Down Expand Up @@ -960,6 +1001,14 @@ export async function createRuntimeAgentStreamResponse(
threadId: input.threadId,
agentId: agent.id,
});
// Replays whatever the first model call already produced, then
// forwards later steps as they happen. RunStarted stays first.
modelCallContextRelay.attach((event) =>
enqueueIfAttached(
MODEL_CALL_CONTEXT_SSE_EVENT_NAME,
event as unknown as Record<string, unknown>,
)
);
Comment on lines +1004 to +1011

@coderabbitai coderabbitai Bot Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔒 Security & Privacy | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -eu

printf '%s\n' '--- run-stream structure ---'
ast-grep outline src/internal-agents/run-stream.ts | sed -n '1,220p'

printf '%s\n' '--- stream relay and response headers ---'
sed -n '960,1040p' src/internal-agents/run-stream.ts
sed -n '1180,1250p' src/internal-agents/run-stream.ts

printf '%s\n' '--- response callers and cache directives ---'
rg -n -C 4 'createRuntimeAgentStreamResponse|Cache-Control|cache-control|no-cache|no-store|text/event-stream' --glob '*.{ts,tsx}' .

printf '%s\n' '--- SSE helper ---'
sed -n '1,240p' src/internal-agents/ag-ui-sse.ts

Repository: veryfront/veryfront-code

Length of output: 50381

🏁 Script executed:

#!/bin/bash
set -eu

printf '%s\n' '--- target response implementation ---'
sed -n '980,1030p' src/internal-agents/run-stream.ts
sed -n '1200,1240p' src/internal-agents/run-stream.ts

printf '%s\n' '--- exact response symbol references ---'
rg -n -C 3 'createRuntimeAgentStreamResponse' src --glob '*.{ts,tsx}' --glob '!**/*.generated.*'

printf '%s\n' '--- cache directives in relevant server paths ---'
rg -n -C 3 'Cache-Control|cache-control|no-cache|no-store' src/internal-agents src/server --glob '*.{ts,tsx}' --glob '!**/*.generated.*' | head -n 500

printf '%s\n' '--- SSE helper implementation ---'
wc -l src/internal-agents/ag-ui-sse.ts
cat -n src/internal-agents/ag-ui-sse.ts | sed -n '1,220p'

printf '%s\n' '--- cache-control semantic probe ---'
python3 - <<'PY'
from email.message import Message

for value in ("no-cache", "no-store"):
    m = Message()
    m["Cache-Control"] = value
    print(value, "=", [x.strip().lower() for x in m["Cache-Control"].split(",")])
PY

Repository: veryfront/veryfront-code

Length of output: 50380

🏁 Script executed:

#!/bin/bash
set -eu

printf '%s\n' '--- production caller ---'
sed -n '820,930p' src/server/handlers/request/agent-stream.handler.ts

printf '%s\n' '--- request handler return and middleware path ---'
rg -n -C 8 'runAgentStream|internalFetch|toNativeResponse|agent-stream' src/server/handlers/request/agent-stream.handler.ts src/server/index.ts src/server --glob '*.{ts,tsx}' --glob '!**/*.generated.*' | head -n 350

printf '%s\n' '--- response mutation utilities on the exact path ---'
rg -n -C 5 'new Response|Response\(|toNativeResponse|headers\.set|headers\.append|headers\.delete|Cache-Control|cache-control' src/server/index.ts src/server/handlers/request/agent-stream.handler.ts src/server/middleware src/server/runtime-handler --glob '*.{ts,tsx}' 2>/dev/null | head -n 500

printf '%s\n' '--- focused tests for stream response headers ---'
rg -n -C 8 'content-type|Cache-Control|cache-control|text/event-stream|keep-alive|headers\.get' src/internal-agents/run-stream.test.ts src/server/handlers/request --glob '*.{ts,tsx}' | head -n 300

printf '%s\n' '--- read-only source verifier ---'
python3 - <<'PY'
from pathlib import Path

stream = Path("src/internal-agents/run-stream.ts").read_text()
handler = Path("src/server/handlers/request/agent-stream.handler.ts").read_text()

assert '"Cache-Control": "no-cache"' in stream
assert '"Content-Type": "text/event-stream"' in stream
assert "createRuntimeAgentStreamResponse(runtimeInput, runtimeAgent" in handler

start = stream.index('return new Response(response, {')
end = stream.index('  });', start) + len('  });')
response_block = stream[start:end]
print("direct_response_block_contains_no_cache=", '"Cache-Control": "no-cache"' in response_block)
print("direct_response_block_contains_no_store=", '"Cache-Control": "no-store"' in response_block)
print("production_caller_count=", handler.count("createRuntimeAgentStreamResponse(runtimeInput, runtimeAgent"))
PY

Repository: veryfront/veryfront-code

Length of output: 50383

🏁 Script executed:

#!/bin/bash
set -eu

printf '%s\n' '--- agent-stream trust boundary ---'
sed -n '120,190p' src/server/handlers/request/agent-stream.handler.ts
sed -n '730,825p' src/server/handlers/request/agent-stream.handler.ts
sed -n '930,980p' src/server/handlers/request/agent-stream.handler.ts

printf '%s\n' '--- event name and context payload definitions ---'
rg -n -C 8 'MODEL_CALL_CONTEXT_SSE_EVENT_NAME|modelCallContext|ModelCallContext|MODEL_CALL_CONTEXT' src/internal-agents src --glob '*.{ts,tsx}' --glob '!**/*.generated.*' | head -n 350

printf '%s\n' '--- request lifecycle stream handling ---'
sed -n '100,165p' src/server/runtime-handler/request-lifecycle.ts

printf '%s\n' '--- focused header verifier ---'
python3 - <<'PY'
from pathlib import Path

stream = Path("src/internal-agents/run-stream.ts").read_text()
handler = Path("src/server/handlers/request/agent-stream.handler.ts").read_text()
server = Path("src/server/index.ts").read_text()
lifecycle = Path("src/server/runtime-handler/request-lifecycle.ts").read_text()

start = stream.index('return new Response(response, {')
end = stream.index('  });', start) + len('  });')
block = stream[start:end]

assert '"Cache-Control": "no-cache"' in block
assert '"Cache-Control": "no-store"' not in block
assert 'if (!headers.has(key))' in handler
assert 'headers: src.headers' in server
assert 'return response;' in lifecycle

print("stream_header=", "no-cache")
print("wrapper_overrides_existing_cache_control=", False)
print("native_adapter_preserves_headers=", True)
print("lifecycle_returns_event_stream_without_header_rewrite=", True)

Repository: veryfront/veryfront-code

Length of output: 36274

Sensitive Data Exposure (CWE-525): Use of Web Browser Cache Containing Sensitive Information

Reachability: Internal · Exploitability: Difficult

Reachability path 
● Entry
  src/internal-agents/run-stream.test.ts
│
▼
● Sink
  src/internal-agents/run-stream.ts

Set Cache-Control to no-store for the SSE response. no-cache permits storage and only requires revalidation. The control-plane wrapper preserves this header, so it does not prevent cache retention of model-call prompts and tool definitions.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/internal-agents/run-stream.ts` around lines 1004 - 1011, Update the SSE
response setup surrounding modelCallContextRelay.attach and the control-plane
response wrapper to set Cache-Control explicitly to no-store rather than
no-cache. Preserve this header through the wrapper so model-call prompts and
tool definitions cannot be stored by intermediaries.

heartbeatTimer = setInterval(
enqueueHeartbeatIfAttached,
INTERNAL_AGENT_RUNTIME_HEARTBEAT_INTERVAL_MS,
Expand All @@ -968,7 +1017,13 @@ export async function createRuntimeAgentStreamResponse(
while (true) {
throwIfAborted();

const { done, value } = await reader.read();
// A runtime that dispatches later model calls from its pull()
// rather than from a continuation of stream() needs the sink in
// scope on the read that triggers the pull.
const { done, value } = await runWithMandatoryRunEventSink(
modelCallContextRelay.sink,
() => reader.read(),
);
throwIfAborted();

if (done) {
Expand Down