feat(builtin-agent): persist ReAct loop turns as session JSONL

Each turn (assistant response / tool result) is appended to a JSONL file
at ~/.uncaged/workflow/sessions/<sessionId>.jsonl during the loop.

On completion, the JSONL is read back, each turn is stored as a CAS node,
and the detail payload references them as a flat turns[] array in
chronological order. The session file is then deleted.

Benefits:
- Real-time observability: tail -f the JSONL to watch loop progress
- Crash recovery: partial JSONL survives process death
- Zero write contention: one file per session
- Detail stays a flat array for easy consumption by CLI/dashboard

Changes:
- New session.ts: initSessionDir, appendSessionTurn, readSessionTurns, removeSession
- loop.ts: append JSONL each turn instead of accumulating in-memory
- detail.ts: reads session JSONL → persists turns to CAS → stores detail
- agent.ts: passes storageRoot/sessionId to loop, cleans up session on completion
- types.ts: remove index from TurnPayload (order is implicit in JSONL/array)
- schemas.ts: sync with type changes

Ref: #433
This commit is contained in:
2026-05-23 18:27:28 +08:00
parent 613793e128
commit 330db43b5f
7 changed files with 166 additions and 127 deletions
+31 -13
View File
@@ -7,17 +7,26 @@ import {
resolveModel, resolveModel,
resolveStorageRoot, resolveStorageRoot,
} from "@uncaged/workflow-agent-kit"; } from "@uncaged/workflow-agent-kit";
import { generateUlid } from "@uncaged/workflow-util"; import { createLogger, generateUlid } from "@uncaged/workflow-util";
import { storeBuiltinDetail } from "./detail.js"; import { storeBuiltinDetail } from "./detail.js";
import type { ChatMessage } from "./llm/index.js"; import type { ChatMessage } from "./llm/index.js";
import { BUILTIN_CONTINUE_MAX_TURNS, BUILTIN_MAX_TURNS, runBuiltinLoop } from "./loop.js"; import { BUILTIN_CONTINUE_MAX_TURNS, BUILTIN_MAX_TURNS, runBuiltinLoop } from "./loop.js";
import { buildBuiltinMessages } from "./prompt.js"; import { buildBuiltinMessages } from "./prompt.js";
import type { BuiltinSessionState } from "./types.js"; import { initSessionDir, removeSession } from "./session.js";
const sessions = new Map<string, BuiltinSessionState>(); const log = createLogger({ sink: { kind: "stderr" } });
function getSession(sessionId: string): BuiltinSessionState { type SessionRecord = {
sessionId: string;
model: string;
startedAtMs: number;
messages: ChatMessage[];
};
const sessions = new Map<string, SessionRecord>();
function getSession(sessionId: string): SessionRecord {
const session = sessions.get(sessionId); const session = sessions.get(sessionId);
if (session === undefined) { if (session === undefined) {
throw new Error(`builtin session not found: ${sessionId}`); throw new Error(`builtin session not found: ${sessionId}`);
@@ -36,7 +45,7 @@ async function runBuiltinWithMessages(
storageRoot: string, storageRoot: string,
provider: ReturnType<typeof resolveModel>, provider: ReturnType<typeof resolveModel>,
messages: ChatMessage[], messages: ChatMessage[],
session: BuiltinSessionState, session: SessionRecord,
store: Store, store: Store,
maxTurns: number, maxTurns: number,
): Promise<AgentRunResult> { ): Promise<AgentRunResult> {
@@ -45,22 +54,31 @@ async function runBuiltinWithMessages(
messages, messages,
toolCtx: buildToolContext(storageRoot), toolCtx: buildToolContext(storageRoot),
maxTurns, maxTurns,
existingTurns: session.turns, storageRoot,
sessionId: session.sessionId,
}); });
session.messages = loopResult.messages; session.messages = loopResult.messages;
session.turns = loopResult.turns;
const { detailHash, output } = await storeBuiltinDetail( if (loopResult.turnCount === 0) {
log("5RWTK9NB", "no turns produced, returning empty output");
await removeSession(storageRoot, session.sessionId);
return { output: "", detailHash: "", sessionId: session.sessionId };
}
// Read jsonl → persist turns to CAS → store detail
const { detailHash } = await storeBuiltinDetail(
store, store,
storageRoot,
session.sessionId, session.sessionId,
session.model, session.model,
session.startedAtMs, session.startedAtMs,
session.turns,
); );
const finalOutput = output !== "" ? output : loopResult.finalText; // Clean up session jsonl
return { output: finalOutput, detailHash, sessionId: session.sessionId }; await removeSession(storageRoot, session.sessionId);
return { output: loopResult.finalText, detailHash, sessionId: session.sessionId };
} }
async function runBuiltin(ctx: AgentContext): Promise<AgentRunResult> { async function runBuiltin(ctx: AgentContext): Promise<AgentRunResult> {
@@ -69,14 +87,14 @@ async function runBuiltin(ctx: AgentContext): Promise<AgentRunResult> {
const provider = resolveModel(config, config.defaultModel); const provider = resolveModel(config, config.defaultModel);
const sessionId = generateUlid(Date.now()); const sessionId = generateUlid(Date.now());
await initSessionDir(storageRoot);
const messages = buildBuiltinMessages(ctx); const messages = buildBuiltinMessages(ctx);
const session: BuiltinSessionState = { const session: SessionRecord = {
sessionId, sessionId,
model: provider.model, model: provider.model,
startedAtMs: Date.now(), startedAtMs: Date.now(),
messages, messages,
turns: [],
}; };
sessions.set(sessionId, session); sessions.set(sessionId, session);
+12 -78
View File
@@ -1,72 +1,15 @@
import { bootstrap, putSchema, type Store } from "@uncaged/json-cas"; import { bootstrap, putSchema, type Store } from "@uncaged/json-cas";
import { BUILTIN_DETAIL_SCHEMA, BUILTIN_TURN_SCHEMA } from "./schemas.js"; import { BUILTIN_DETAIL_SCHEMA, BUILTIN_TURN_SCHEMA } from "./schemas.js";
import type { import { readSessionTurns } from "./session.js";
BuiltinDetailPayload, import type { BuiltinDetailPayload } from "./types.js";
BuiltinLoopTurn,
BuiltinToolCall,
BuiltinTurnPayload,
BuiltinTurnRole,
} from "./types.js";
function mapToolCalls(calls: NonNullable<BuiltinLoopTurn["toolCalls"]>): BuiltinToolCall[] {
return calls.map((call) => ({
name: call.name,
args: call.args,
}));
}
function loopTurnToAssistantPayload(turn: BuiltinLoopTurn, index: number): BuiltinTurnPayload {
return {
index,
role: "assistant",
content: turn.assistantContent ?? "",
toolCalls:
turn.toolCalls !== null && turn.toolCalls.length > 0 ? mapToolCalls(turn.toolCalls) : null,
reasoning: null,
};
}
function loopTurnToToolPayloads(turn: BuiltinLoopTurn, startIndex: number): BuiltinTurnPayload[] {
if (turn.toolResults === null || turn.toolResults.length === 0) {
return [];
}
const payloads: BuiltinTurnPayload[] = [];
let index = startIndex;
for (const result of turn.toolResults) {
payloads.push({
index,
role: "tool" as BuiltinTurnRole,
content: result.content,
toolCalls: null,
reasoning: null,
});
index += 1;
}
return payloads;
}
/** Last assistant message with non-empty text. */
export function extractFinalAssistantText(turns: BuiltinLoopTurn[]): string {
for (let i = turns.length - 1; i >= 0; i--) {
const turn = turns[i];
if (turn === undefined) {
continue;
}
const text = turn.assistantContent;
if (text !== null && text.trim() !== "") {
return text;
}
}
return "";
}
type BuiltinSchemaHashes = { type BuiltinSchemaHashes = {
turn: string; turn: string;
detail: string; detail: string;
}; };
async function registerBuiltinSchemas(store: Store): Promise<BuiltinSchemaHashes> { export async function registerBuiltinSchemas(store: Store): Promise<BuiltinSchemaHashes> {
await bootstrap(store); await bootstrap(store);
const [turn, detail] = await Promise.all([ const [turn, detail] = await Promise.all([
putSchema(store, BUILTIN_TURN_SCHEMA), putSchema(store, BUILTIN_TURN_SCHEMA),
@@ -75,30 +18,22 @@ async function registerBuiltinSchemas(store: Store): Promise<BuiltinSchemaHashes
return { turn, detail }; return { turn, detail };
} }
/** Read session jsonl, persist each turn to CAS, return detail hash. */
export async function storeBuiltinDetail( export async function storeBuiltinDetail(
store: Store, store: Store,
storageRoot: string,
sessionId: string, sessionId: string,
model: string, model: string,
startedAtMs: number, startedAtMs: number,
turns: BuiltinLoopTurn[],
nowMs: number = Date.now(), nowMs: number = Date.now(),
): Promise<{ detailHash: string; output: string }> { ): Promise<{ detailHash: string; turnCount: number }> {
const schemas = await registerBuiltinSchemas(store); const schemas = await registerBuiltinSchemas(store);
const turns = await readSessionTurns(storageRoot, sessionId);
const turnHashes: string[] = []; const turnHashes: string[] = [];
let turnIndex = 0; for (const turn of turns) {
const hash = await store.put(schemas.turn, turn);
for (const loopTurn of turns) { turnHashes.push(hash);
const assistant = loopTurnToAssistantPayload(loopTurn, turnIndex);
const assistantHash = await store.put(schemas.turn, assistant);
turnHashes.push(assistantHash);
turnIndex += 1;
const toolPayloads = loopTurnToToolPayloads(loopTurn, turnIndex);
for (const toolPayload of toolPayloads) {
const toolHash = await store.put(schemas.turn, toolPayload);
turnHashes.push(toolHash);
turnIndex += 1;
}
} }
const duration = Math.max(0, nowMs - startedAtMs); const duration = Math.max(0, nowMs - startedAtMs);
@@ -110,6 +45,5 @@ export async function storeBuiltinDetail(
turns: turnHashes, turns: turnHashes,
}; };
const detailHash = await store.put(schemas.detail, detail); const detailHash = await store.put(schemas.detail, detail);
const output = extractFinalAssistantText(turns); return { detailHash, turnCount: turnHashes.length };
return { detailHash, output };
} }
+4 -2
View File
@@ -1,14 +1,16 @@
export { createBuiltinAgent } from "./agent.js"; export { createBuiltinAgent } from "./agent.js";
export { extractFinalAssistantText, storeBuiltinDetail } from "./detail.js"; export { registerBuiltinSchemas, storeBuiltinDetail } from "./detail.js";
export type { ChatMessage, LlmAssistantResponse, LlmToolCall } from "./llm/index.js"; export type { ChatMessage, LlmAssistantResponse, LlmToolCall } from "./llm/index.js";
export { chatCompletionWithTools } from "./llm/index.js"; export { chatCompletionWithTools } from "./llm/index.js";
export { BUILTIN_CONTINUE_MAX_TURNS, BUILTIN_MAX_TURNS, runBuiltinLoop } from "./loop.js"; export { BUILTIN_CONTINUE_MAX_TURNS, BUILTIN_MAX_TURNS, runBuiltinLoop } from "./loop.js";
export { buildBuiltinMessages } from "./prompt.js"; export { buildBuiltinMessages } from "./prompt.js";
export { appendSessionTurn, initSessionDir, readSessionTurns, removeSession } from "./session.js";
export type { BuiltinTool, ToolContext } from "./tools/index.js"; export type { BuiltinTool, ToolContext } from "./tools/index.js";
export { executeBuiltinTool, getBuiltinTools } from "./tools/index.js"; export { executeBuiltinTool, getBuiltinTools } from "./tools/index.js";
export type { export type {
BuiltinDetailPayload, BuiltinDetailPayload,
BuiltinLoopTurn, BuiltinLoopTurn,
BuiltinSessionState, BuiltinToolCallRecord,
BuiltinToolResultRecord,
BuiltinTurnPayload, BuiltinTurnPayload,
} from "./types.js"; } from "./types.js";
+59 -31
View File
@@ -2,13 +2,14 @@ import type { ResolvedLlmProvider } from "@uncaged/workflow-agent-kit";
import { createLogger } from "@uncaged/workflow-util"; import { createLogger } from "@uncaged/workflow-util";
import { type ChatMessage, chatCompletionWithTools, type LlmToolCall } from "./llm/index.js"; import { type ChatMessage, chatCompletionWithTools, type LlmToolCall } from "./llm/index.js";
import { appendSessionTurn } from "./session.js";
import { import {
builtinToolsToOpenAi, builtinToolsToOpenAi,
executeBuiltinTool, executeBuiltinTool,
getBuiltinTools, getBuiltinTools,
type ToolContext, type ToolContext,
} from "./tools/index.js"; } from "./tools/index.js";
import type { BuiltinLoopTurn, BuiltinToolCallRecord, BuiltinToolResultRecord } from "./types.js"; import type { BuiltinToolCall, BuiltinTurnPayload } from "./types.js";
const log = createLogger({ sink: { kind: "stderr" } }); const log = createLogger({ sink: { kind: "stderr" } });
@@ -20,31 +21,61 @@ export type RunBuiltinLoopOptions = {
messages: ChatMessage[]; messages: ChatMessage[];
toolCtx: ToolContext; toolCtx: ToolContext;
maxTurns: number; maxTurns: number;
existingTurns: BuiltinLoopTurn[]; storageRoot: string;
sessionId: string;
}; };
export type RunBuiltinLoopResult = { export type RunBuiltinLoopResult = {
finalText: string; finalText: string;
messages: ChatMessage[]; messages: ChatMessage[];
turns: BuiltinLoopTurn[]; turnCount: number;
}; };
function mapToolCalls(calls: LlmToolCall[]): BuiltinToolCallRecord[] { function mapToolCallsForPayload(calls: LlmToolCall[]): BuiltinToolCall[] {
return calls.map((call) => ({ return calls.map((call) => ({
id: call.id,
name: call.name, name: call.name,
args: call.arguments, args: call.arguments,
})); }));
} }
async function appendTurn(
storageRoot: string,
sessionId: string,
payload: BuiltinTurnPayload,
): Promise<void> {
await appendSessionTurn(storageRoot, sessionId, payload);
}
async function executeTurnTools(
calls: Array<{ id: string; name: string; arguments: string }>,
toolCtx: ToolContext,
messages: ChatMessage[],
storageRoot: string,
sessionId: string,
): Promise<number> {
let turnCount = 0;
for (const call of calls) {
const result = await executeBuiltinTool(call.name, call.arguments, toolCtx);
messages.push({ role: "tool", tool_call_id: call.id, content: result });
await appendTurn(storageRoot, sessionId, {
role: "tool",
content: result,
toolCalls: null,
reasoning: null,
});
turnCount += 1;
}
return turnCount;
}
/** Agent run loop: LLM ↔ tools until no tool_calls or maxTurns. */ /** Agent run loop: LLM ↔ tools until no tool_calls or maxTurns. */
export async function runBuiltinLoop( export async function runBuiltinLoop(
options: RunBuiltinLoopOptions, options: RunBuiltinLoopOptions,
): Promise<RunBuiltinLoopResult> { ): Promise<RunBuiltinLoopResult> {
const messages = [...options.messages]; const messages = [...options.messages];
const turns = [...options.existingTurns];
const openAiTools = builtinToolsToOpenAi(getBuiltinTools()); const openAiTools = builtinToolsToOpenAi(getBuiltinTools());
let finalText = ""; let finalText = "";
let turnCount = 0;
for (let turn = 0; turn < options.maxTurns; turn++) { for (let turn = 0; turn < options.maxTurns; turn++) {
log("8K2M4N7P", `builtin loop turn ${turn + 1}/${options.maxTurns}`); log("8K2M4N7P", `builtin loop turn ${turn + 1}/${options.maxTurns}`);
@@ -59,36 +90,33 @@ export async function runBuiltinLoop(
if (response.toolCalls === null || response.toolCalls.length === 0) { if (response.toolCalls === null || response.toolCalls.length === 0) {
finalText = response.content ?? ""; finalText = response.content ?? "";
turns.push({ await appendTurn(options.storageRoot, options.sessionId, {
assistantContent: response.content, role: "assistant",
content: response.content ?? "",
toolCalls: null, toolCalls: null,
toolResults: null, reasoning: null,
}); });
turnCount += 1;
break; break;
} }
const toolCallRecords = mapToolCalls(response.toolCalls); // Assistant turn with tool calls
const toolResults: BuiltinToolResultRecord[] = []; await appendTurn(options.storageRoot, options.sessionId, {
role: "assistant",
for (const call of response.toolCalls) { content: response.content ?? "",
const result = await executeBuiltinTool(call.name, call.arguments, options.toolCtx); toolCalls: mapToolCallsForPayload(response.toolCalls),
toolResults.push({ reasoning: null,
toolCallId: call.id,
name: call.name,
content: result,
});
messages.push({
role: "tool",
tool_call_id: call.id,
content: result,
});
}
turns.push({
assistantContent: response.content,
toolCalls: toolCallRecords,
toolResults,
}); });
turnCount += 1;
// Execute tools
turnCount += await executeTurnTools(
response.toolCalls,
options.toolCtx,
messages,
options.storageRoot,
options.sessionId,
);
} }
if (finalText === "" && messages.length > 0) { if (finalText === "" && messages.length > 0) {
@@ -106,5 +134,5 @@ export async function runBuiltinLoop(
} }
} }
return { finalText, messages, turns }; return { finalText, messages, turnCount };
} }
@@ -13,9 +13,8 @@ const BUILTIN_TOOL_CALL_SCHEMA: JSONSchema = {
export const BUILTIN_TURN_SCHEMA: JSONSchema = { export const BUILTIN_TURN_SCHEMA: JSONSchema = {
title: "builtin-turn", title: "builtin-turn",
type: "object", type: "object",
required: ["index", "role", "content"], required: ["role", "content"],
properties: { properties: {
index: { type: "integer" },
role: { type: "string", enum: ["assistant", "tool"] }, role: { type: "string", enum: ["assistant", "tool"] },
content: { type: "string" }, content: { type: "string" },
toolCalls: { toolCalls: {
@@ -0,0 +1,59 @@
import { appendFile, mkdir, readFile, rm } from "node:fs/promises";
import { join } from "node:path";
import { createLogger } from "@uncaged/workflow-util";
import type { BuiltinTurnPayload } from "./types.js";
const log = createLogger({ sink: { kind: "stderr" } });
function sessionsDir(storageRoot: string): string {
return join(storageRoot, "sessions");
}
function sessionFile(storageRoot: string, sessionId: string): string {
return join(sessionsDir(storageRoot), `${sessionId}.jsonl`);
}
/** Ensure sessions directory exists. */
export async function initSessionDir(storageRoot: string): Promise<void> {
await mkdir(sessionsDir(storageRoot), { recursive: true });
}
/** Append a turn to the session jsonl file. */
export async function appendSessionTurn(
storageRoot: string,
sessionId: string,
turn: BuiltinTurnPayload,
): Promise<void> {
const line = `${JSON.stringify(turn)}\n`;
await appendFile(sessionFile(storageRoot, sessionId), line, "utf-8");
log("3XQVN8KR", `session ${sessionId} appended ${turn.role} turn`);
}
/** Read all turns from session jsonl. Returns empty array if file does not exist. */
export async function readSessionTurns(
storageRoot: string,
sessionId: string,
): Promise<BuiltinTurnPayload[]> {
try {
const content = await readFile(sessionFile(storageRoot, sessionId), "utf-8");
const lines = content
.trim()
.split("\n")
.filter((l) => l.length > 0);
return lines.map((l) => JSON.parse(l) as BuiltinTurnPayload);
} catch {
return [];
}
}
/** Remove session jsonl file (called after detail is persisted to step CAS). */
export async function removeSession(storageRoot: string, sessionId: string): Promise<void> {
try {
await rm(sessionFile(storageRoot, sessionId));
log("7FWDP2MJ", `session ${sessionId} removed`);
} catch {
// already gone — fine
}
}
@@ -34,7 +34,6 @@ export type BuiltinToolCall = {
}; };
export type BuiltinTurnPayload = { export type BuiltinTurnPayload = {
index: number;
role: BuiltinTurnRole; role: BuiltinTurnRole;
content: string; content: string;
toolCalls: BuiltinToolCall[] | null; toolCalls: BuiltinToolCall[] | null;