Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
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
16 changes: 13 additions & 3 deletions cli/commands/push/command.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -88,15 +88,20 @@ interface GitProject {

async function withGitProject(test: (project: GitProject) => Promise<void>): Promise<void> {
const projectDir = await Deno.makeTempDir();
const isolatedGitConfigGlobal = await Deno.makeTempFile();
const originalGithubSha = Deno.env.get("GITHUB_SHA");
const originalGitConfigGlobal = Deno.env.get("GIT_CONFIG_GLOBAL");
const runGit = async (...args: string[]): Promise<string> => {
const result = await new Deno.Command("git", {
args,
cwd: projectDir,
clearEnv: true,
env: Object.fromEntries(
Object.entries(Deno.env.toObject()).filter(([key]) => !key.startsWith("GIT_")),
),
env: {
...Object.fromEntries(
Object.entries(Deno.env.toObject()).filter(([key]) => !key.startsWith("GIT_")),
),
GIT_CONFIG_GLOBAL: isolatedGitConfigGlobal,
},
stdout: "piped",
stderr: "piped",
}).output();
Expand All @@ -107,7 +112,10 @@ async function withGitProject(test: (project: GitProject) => Promise<void>): Pro

try {
Deno.env.delete("GITHUB_SHA");
await Deno.writeTextFile(isolatedGitConfigGlobal, "[init]\n\tdefaultBranch = main\n");
Deno.env.set("GIT_CONFIG_GLOBAL", isolatedGitConfigGlobal);
await runGit("init", "--quiet");
assertEquals(await runGit("symbolic-ref", "--short", "HEAD"), "main");
await runGit("config", "user.email", "test@veryfront.com");
await runGit("config", "user.name", "Veryfront Test");
await Deno.writeTextFile(`${projectDir}/app.ts`, "export const value = 1;\n");
Expand All @@ -117,7 +125,9 @@ async function withGitProject(test: (project: GitProject) => Promise<void>): Pro
} finally {
if (originalGithubSha === undefined) Deno.env.delete("GITHUB_SHA");
else Deno.env.set("GITHUB_SHA", originalGithubSha);
restoreEnv("GIT_CONFIG_GLOBAL", originalGitConfigGlobal);
await Deno.remove(projectDir, { recursive: true });
await Deno.remove(isolatedGitConfigGlobal);
}
}

Expand Down
68 changes: 34 additions & 34 deletions docs/api-reference/veryfront/agent.md

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion docs/api-reference/veryfront/embedding.md
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ export const { POST, GET, DELETE } = createUploadHandler(store, {
| `ragStore` | Creates a persistent RAG store with lazy embedding and similarity search. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/embedding/rag-store.ts#L212) |
| `registerEmbeddingProvider` | Register an embedding provider factory. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/embedding/resolve.ts#L25) |
| `resolveEmbeddingModel` | Resolve a "provider/model" string to an embedding runtime instance. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/embedding/resolve.ts#L116) |
| `similarity` | Compute cosine similarity between two numeric vectors. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/runtime/runtime-bridge.ts#L1128) |
| `similarity` | Compute cosine similarity between two numeric vectors. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/runtime/runtime-bridge.ts#L1258) |
| `vectorStore` | Creates an in-memory vector store with integrated embedding and similarity search. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/embedding/vector-store.ts#L46) |

### Types
Expand Down
40 changes: 20 additions & 20 deletions docs/api-reference/veryfront/provider.md

Large diffs are not rendered by default.

9 changes: 9 additions & 0 deletions src/agent/ag-ui/browser-chunk-encoder.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,15 @@ import { describe, it } from "#veryfront/testing/bdd.ts";
import { createAgUiBrowserChunkEncoder } from "./browser-chunk-encoder.ts";

describe("agent/ag-ui-browser-chunk-encoder", () => {
it("exposes its state as the browser response timing anchor", () => {
const encoder = createAgUiBrowserChunkEncoder({
getRuntimeEvents: () => [],
timing: { nowMs: null, epochMs: null },
});

assertEquals(encoder.timingState, encoder.state);
});

it("merges chunk metadata into the browser finalize response", () => {
const encoder = createAgUiBrowserChunkEncoder<{
id: string;
Expand Down
3 changes: 3 additions & 0 deletions src/agent/ag-ui/browser-chunk-encoder.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ import type { AgentResponse } from "../types.ts";
/** Public API contract for AG-UI browser chunk encoder. */
export interface AgUiBrowserChunkEncoder<TChunk> {
state: AgUiBrowserEncoderState;
/** Timing anchor consumed by the browser response composition root. */
timingState: AgUiBrowserEncoderState;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
encode: (chunk: TChunk) => AgUiBrowserEncodedEvent[];
finalize: (response: AgentResponse | null) => AgUiBrowserEncodedEvent[];
}
Expand Down Expand Up @@ -108,6 +110,7 @@ export function createAgUiBrowserChunkEncoder<TChunk>(

return {
state: runtimeEventEncoder.state,
timingState: runtimeEventEncoder.state,
encode: (chunk) => {
mergeMetadata(runtimeEventEncoder.state.metadata, options.getMetadataFromChunk?.(chunk));
return options.getRuntimeEvents(chunk).flatMap((event) => runtimeEventEncoder.encode(event));
Expand Down
57 changes: 56 additions & 1 deletion src/agent/ag-ui/browser-encoder.test.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import "#veryfront/schemas/_test-setup.ts";
import { assertEquals } from "#veryfront/testing/assert.ts";
import { assertEquals, assertThrows } from "#veryfront/testing/assert.ts";
import { describe, it } from "#veryfront/testing/bdd.ts";
import {
buildAgUiBrowserFinalizeResponse,
createAgUiBrowserEncoderState,
finalizeAgUiBrowserEvents,
mapRuntimeStreamEventToAgUiBrowserEvents,
stampAgUiBrowserEventTiming,
} from "./browser-encoder.ts";

describe("agent/ag-ui-browser-encoder", () => {
Expand Down Expand Up @@ -174,6 +175,60 @@ describe("agent/ag-ui-browser-encoder", () => {
assertEquals("elapsedMs" in (events[0]?.payload ?? {}), false);
});

it("preserves valid supplied timing and rejects invalid present timing", () => {
const state = createAgUiBrowserEncoderState({
nowMs: () => Number.NaN,
epochMs: () => -1,
});
const supplied = stampAgUiBrowserEventTiming(state, [{
event: "Custom",
payload: { elapsedMs: 12.5, emittedAt: 1_786_866_357_364 },
}]);
assertEquals(supplied[0]?.payload.elapsedMs, 12.5);
assertEquals(supplied[0]?.payload.emittedAt, 1_786_866_357_364);

assertThrows(
() =>
stampAgUiBrowserEventTiming(state, [{
event: "Custom",
payload: { elapsedMs: Number.POSITIVE_INFINITY, emittedAt: 1_786_866_357_364 },
}]),
TypeError,
"elapsedMs must be a finite non-negative number",
);
assertThrows(
() =>
stampAgUiBrowserEventTiming(state, [{
event: "Custom",
payload: { elapsedMs: 0, emittedAt: -1 },
}]),
TypeError,
"emittedAt must be a non-negative integer",
);
});

it("rejects timing generated by invalid clocks", () => {
const invalidElapsed = createAgUiBrowserEncoderState({
nowMs: (() => {
let reads = 0;
return () => reads++ === 0 ? 0 : Number.NaN;
})(),
epochMs: null,
});
assertThrows(
() => mapRuntimeStreamEventToAgUiBrowserEvents(invalidElapsed, { type: "start-step" }),
TypeError,
"elapsedMs must be a finite non-negative number",
);

const invalidEpoch = createAgUiBrowserEncoderState({ nowMs: null, epochMs: () => -1 });
assertThrows(
() => mapRuntimeStreamEventToAgUiBrowserEvents(invalidEpoch, { type: "start-step" }),
TypeError,
"emittedAt must be a non-negative integer",
);
});

it("clocks the state unless a caller explicitly opts out", () => {
// Three production composition roots build this state. An opt-in clock only
// has to be missed at one of them to lose elapsedMs for every hosted run,
Expand Down
57 changes: 45 additions & 12 deletions src/agent/ag-ui/browser-encoder.ts
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ export interface AgUiBrowserEncoderStateOptions {
* `Date.now`. Pass null to omit the stamp.
*/
epochMs?: (() => number) | null;
startedMs?: number;
}

/** Event emitted for AG-UI browser encoded. */
Expand All @@ -95,7 +96,7 @@ export function createAgUiBrowserEncoderState(
const nowMs = options.nowMs === null ? undefined : options.nowMs ?? (() => performance.now());
const epochMs = options.epochMs === null ? undefined : options.epochMs ?? (() => Date.now());
return {
...(nowMs ? { nowMs, startedMs: nowMs() } : {}),
...(nowMs ? { nowMs, startedMs: options.startedMs ?? nowMs() } : {}),
...(epochMs ? { epochMs } : {}),
messageId: null,
textOpen: false,
Expand Down Expand Up @@ -681,10 +682,13 @@ export function mapRuntimeStreamEventToAgUiBrowserEvents(
state: AgUiBrowserEncoderState,
event: AgUiRuntimeStreamEvent,
): AgUiBrowserEncodedEvent[] {
return stampTiming(state, mapRuntimeStreamEventToAgUiBrowserEventsUnstamped(state, event));
return stampAgUiBrowserEventTiming(
state,
mapRuntimeStreamEventToAgUiBrowserEventsUnstamped(state, event),
);
}

function stampTiming(
export function stampAgUiBrowserEventTiming(
state: AgUiBrowserEncoderState,
events: AgUiBrowserEncodedEvent[],
): AgUiBrowserEncodedEvent[] {
Expand All @@ -699,18 +703,47 @@ function stampTiming(
// wall-clock traces and logs, and turns ingest lag into `created_at -
// emittedAt`. Both are stamped because wall clocks can step backwards and
// the monotonic reading cannot.
const timing: Record<string, number> = {};
if (state.nowMs && state.startedMs !== undefined) {
timing.elapsedMs = Math.max(0, Math.round(state.nowMs() - state.startedMs));
}
if (state.epochMs) {
timing.emittedAt = Math.round(state.epochMs());
for (const { payload } of events) {
if (Object.hasOwn(payload, "elapsedMs")) assertValidElapsedMs(payload.elapsedMs);
if (Object.hasOwn(payload, "emittedAt")) assertValidEmittedAt(payload.emittedAt);
}
if (Object.keys(timing).length === 0) {

const needsElapsedMs = events.some(({ payload }) => !Object.hasOwn(payload, "elapsedMs"));
const needsEmittedAt = events.some(({ payload }) => !Object.hasOwn(payload, "emittedAt"));
const elapsedMs = needsElapsedMs && state.nowMs && state.startedMs !== undefined
? Math.max(0, Math.round(state.nowMs() - state.startedMs))
: undefined;
const emittedAt = needsEmittedAt && state.epochMs ? Math.round(state.epochMs()) : undefined;
if (elapsedMs !== undefined) assertValidElapsedMs(elapsedMs);
if (emittedAt !== undefined) assertValidEmittedAt(emittedAt);
if (elapsedMs === undefined && emittedAt === undefined) {
return events;
}

return events.map((entry) => ({ ...entry, payload: { ...entry.payload, ...timing } }));
return events.map((entry) => ({
...entry,
payload: {
...entry.payload,
...(elapsedMs !== undefined && !Object.hasOwn(entry.payload, "elapsedMs")
? { elapsedMs }
: {}),
...(emittedAt !== undefined && !Object.hasOwn(entry.payload, "emittedAt")
? { emittedAt }
: {}),
},
}));
}

function assertValidElapsedMs(value: unknown): asserts value is number {
if (typeof value !== "number" || !Number.isFinite(value) || value < 0) {
throw new TypeError("elapsedMs must be a finite non-negative number");
}
}

function assertValidEmittedAt(value: unknown): asserts value is number {
if (typeof value !== "number" || !Number.isInteger(value) || value < 0) {
throw new TypeError("emittedAt must be a non-negative integer");
}
}

function mapRuntimeStreamEventToAgUiBrowserEventsUnstamped(
Expand Down Expand Up @@ -952,7 +985,7 @@ export function finalizeAgUiBrowserEvents(
state: AgUiBrowserEncoderState,
response: AgentResponse | null,
): AgUiBrowserEncodedEvent[] {
return stampTiming(state, finalizeAgUiBrowserEventsUnstamped(state, response));
return stampAgUiBrowserEventTiming(state, finalizeAgUiBrowserEventsUnstamped(state, response));
}

function finalizeAgUiBrowserEventsUnstamped(
Expand Down
71 changes: 69 additions & 2 deletions src/agent/ag-ui/browser-response-stream.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import "#veryfront/schemas/_test-setup.ts";
import { assertEquals, assertStringIncludes } from "#veryfront/testing/assert.ts";
import { describe, it } from "#veryfront/testing/bdd.ts";
import { createAgUiBrowserResponseStream } from "./browser-response-stream.ts";
import { createAgUiChunkEncoderBridge } from "./chunk-encoder-bridge.ts";
import type { AgUiSseEvent } from "./host-support.ts";

async function collectStreamText(stream: ReadableStream<Uint8Array>): Promise<string> {
Expand All @@ -26,6 +27,14 @@ async function collectStreamText(stream: ReadableStream<Uint8Array>): Promise<st
return new TextDecoder().decode(merged);
}

function parseSseFrames(body: string): Array<{ event: string; data: Record<string, 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 Record<string, unknown> }] : [];
});
}

describe("agent/ag-ui-browser-response-stream", () => {
it("writes bootstrap events, encoded chunk events, and finalize events", async () => {
const stream = createAgUiBrowserResponseStream({
Expand Down Expand Up @@ -67,6 +76,53 @@ describe("agent/ag-ui-browser-response-stream", () => {
assertStringIncludes(text, "event: RunFinished");
});

it("shares the chunk encoder timing anchor with bootstrap and final events", async () => {
let now = 100;
const chunkEncoder = createAgUiChunkEncoderBridge<{ messageId: string }>({
getRuntimeEvents: (chunk) => [
{ type: "message-start", messageId: chunk.messageId },
{ type: "text-start", id: chunk.messageId },
],
timing: { nowMs: () => now, epochMs: null },
});

now = 150;
const stream = createAgUiBrowserResponseStream({
agUiInput: {
runId: "run-timing",
threadId: "thread-timing",
messages: [],
},
agentId: "assistant-1",
execution: {
agentUIStream: {
async *[Symbol.asyncIterator]() {
now = 175;
yield { messageId: "msg-1" };
},
},
fail: async () => {},
waitForFinish: async () => {
now = 200;
},
},
encoder: chunkEncoder,
initialState: {},
});

const frames = parseSseFrames(await collectStreamText(stream));
assertEquals(
frames.filter((frame) =>
["RunStarted", "TextMessageStart", "RunFinished"].includes(frame.event)
).map((frame) => [frame.event, frame.data.elapsedMs]),
[
["RunStarted", 50],
["TextMessageStart", 75],
["RunFinished", 100],
],
);
});

it("emits RunError and swallows execution.fail rejections", async () => {
const stream = createAgUiBrowserResponseStream({
agUiInput: {
Expand Down Expand Up @@ -165,8 +221,19 @@ describe("agent/ag-ui-browser-response-stream", () => {
});

const text = await collectStreamText(stream);
assertStringIncludes(text, "event: StateSnapshot");
assertStringIncludes(text, 'data: {"snapshot":{}}');
const stateSnapshot = parseSseFrames(text).find((frame) => frame.event === "StateSnapshot")
?.data;
assertEquals(stateSnapshot?.snapshot, {});
assertEquals(
typeof stateSnapshot?.elapsedMs === "number" &&
Number.isFinite(stateSnapshot.elapsedMs) && stateSnapshot.elapsedMs >= 0,
true,
);
assertEquals(
typeof stateSnapshot?.emittedAt === "number" &&
Number.isInteger(stateSnapshot.emittedAt) && stateSnapshot.emittedAt > 0,
true,
);
});

it("stops consuming chunks after the response stream is cancelled", async () => {
Expand Down
11 changes: 10 additions & 1 deletion src/agent/ag-ui/browser-response-stream.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
import type { AgentResponse } from "../types.ts";
import type { AgUiSseEvent } from "./host-support.ts";
import {
type AgUiBrowserEncoderState,
createAgUiBrowserEncoderState,
stampAgUiBrowserEventTiming,
} from "./browser-encoder.ts";

const encoder = new TextEncoder();

Expand Down Expand Up @@ -36,6 +41,8 @@ export interface AgUiBrowserResponseExecution<TChunk> {
export interface AgUiBrowserResponseEncoder<TChunk> {
encode: (chunk: TChunk) => AgUiSseEvent[];
finalize: (response: AgentResponse | null) => AgUiSseEvent[];
/** Shared timing anchor for bootstrap, chunk, and final events. */
timingState?: AgUiBrowserEncoderState;
}

/** Input payload for create AG-UI browser response stream. */
Expand Down Expand Up @@ -66,13 +73,15 @@ export function createAgUiBrowserResponseStream<TChunk, TState>(

return new ReadableStream<Uint8Array>({
start(controller) {
const timingState = input.encoder.timingState ?? createAgUiBrowserEncoderState();
const writeEvent = (event: AgUiSseEvent) => {
if (streamClosed) {
return false;
}

try {
controller.enqueue(formatAgUiSseEventWithId(event, nextEventId));
const [timed] = stampAgUiBrowserEventTiming(timingState, [event]);
controller.enqueue(formatAgUiSseEventWithId(timed ?? event, nextEventId));
nextEventId += 1;
return true;
} catch {
Expand Down
Loading