diff --git a/agent/agent.ts b/agent/agent.ts new file mode 100644 index 0000000000..b6c4ad8266 --- /dev/null +++ b/agent/agent.ts @@ -0,0 +1,203 @@ +/** + * Built-in Harper Agent component (#626). + * + * `startOnMainThread` is invoked once on the main thread by `componentLoader`. + * When `agent.enabled` is `false` (default) the component registers nothing + * and returns immediately — opt-in keeps surprise LLM costs at bay. When + * enabled, the six operations land on the operations API, the session table + * is realized lazily on first use, and the loop runs in-process. + * + * The component intentionally avoids `handleApplication`: it has nothing + * worker-thread-shaped to do. Operator-only tools (FS, schedule, fetch) are + * inline; registry-backed tools (#615/#617/#618) will fold in via toolset.ts + * once those land. + */ + +import { dirname, isAbsolute, resolve as resolvePath } from 'node:path'; +import { CONFIG_PARAMS } from '../utility/hdbTerms.ts'; +import harperLogger from '../utility/logging/harper_logger.ts'; +import { Models } from '../resources/models/Models.ts'; +import { composeToolset } from './toolset.ts'; +import { buildOperations } from './operations.ts'; +import { runAgent, _resetInFlightForTests } from './loop.ts'; +import { appendMessage, getSession } from './session.ts'; +import type { AgentConfig, AgentScopes, AgentTool } from './types.ts'; + +const log = harperLogger.loggerWithTag('agent'); + +const DEFAULT_CONFIG: AgentConfig = { + enabled: false, + maxTurns: 50, + maxCostUsd: 5, + autoApprove: false, + allowDestructive: false, + user: 'hdb_agent', +}; + +interface StartOpts { + server: { + registerOperation: (def: { name: string; execute: (op: any) => any | Promise }) => void; + }; + // Component-level config plumbed by componentLoader (`...componentConfig`). + enabled?: boolean; + provider?: string; + model?: string; + maxTurns?: number; + maxCostUsd?: number; + autoApprove?: boolean; + allowDestructive?: boolean; + user?: string; + componentsScope?: string; +} + +export async function startOnMainThread(opts: StartOpts): Promise { + const config = mergeConfig(opts); + if (!config.enabled) { + log.info?.('Agent component disabled (agent.enabled=false); skipping registration'); + return; + } + + // Lazily required to avoid pulling configUtils at module-eval time for tests. + const { getConfigPath, getConfigFilePath } = require('../config/configUtils.js'); + const scopes = resolveScopes(config, getConfigPath, getConfigFilePath); + + const models = new Models(); + const abortControllers = new Map(); + let liveConfig: AgentConfig = config; + let composed = composeToolset({ + allowDestructive: liveConfig.allowDestructive, + onFollowup: handleFollowup, + }); + + // Only warn when the operator explicitly configured `maxCostUsd`. Logging on the default + // every boot would flood the log without telling anyone anything actionable. + if (opts.maxCostUsd !== undefined) { + log.warn?.( + `agent.maxCostUsd=${liveConfig.maxCostUsd} is advertised but not yet enforced; cost-cap wiring depends on #612 telemetry.` + ); + } + + async function handleFollowup(sessionId: string, prompt: string): Promise { + const session = await getSession(sessionId); + if (!session) { + log.warn?.(`schedule_followup target session ${sessionId} not found`); + return; + } + await appendMessage(sessionId, { role: 'user', content: prompt, createdAt: Date.now() }); + startRun(sessionId); + } + + function currentTools(): AgentTool[] { + return composed.tools; + } + + function startRun(sessionId: string): void { + // A run is already active for this session. `runAgent` coalesces concurrent starts onto the + // existing in-flight promise (bound to the existing controller), so creating a second + // controller here would orphan it — `cancelRun` would then abort a controller nothing is + // listening to, leaving the live run uncancellable. Any message appended before this call + // (e.g. by a scheduled followup) is picked up by the in-flight loop on its next turn. + if (abortControllers.has(sessionId)) return; + const controller = new AbortController(); + abortControllers.set(sessionId, controller); + runAgent({ + sessionId, + models, + tools: currentTools(), + scopes, + maxTurns: liveConfig.maxTurns, + autoApprove: liveConfig.autoApprove, + signal: controller.signal, + generateOpts: { model: liveConfig.model }, + }) + .catch((err) => log.error?.(`Agent run failed for ${sessionId}: ${(err as Error)?.message ?? err}`)) + .finally(() => { + if (abortControllers.get(sessionId) === controller) abortControllers.delete(sessionId); + }); + } + + function cancelRun(sessionId: string): boolean { + // Clear any scheduled followups first. Without this, a timer set via `schedule_followup` + // would fire after the operator cancelled, silently re-injecting a user prompt and kicking + // the loop off again — surprising behavior and avoidable LLM cost. + for (const [id, followup] of composed.scheduled.entries()) { + if (followup.sessionId === sessionId) { + clearTimeout(followup.timer); + composed.scheduled.delete(id); + } + } + const controller = abortControllers.get(sessionId); + if (!controller) return false; + controller.abort(new Error('cancelled by operator')); + abortControllers.delete(sessionId); + return true; + } + + function setConfig(patch: Partial): AgentConfig { + const previousAllowDestructive = liveConfig.allowDestructive; + liveConfig = { ...liveConfig, ...patch }; + if (liveConfig.allowDestructive !== previousAllowDestructive) { + composed = composeToolset({ + allowDestructive: liveConfig.allowDestructive, + onFollowup: handleFollowup, + }); + } + // NOTE: an already in-flight run captured its toolset (and autoApprove) at start, so flipping + // allowDestructive here only affects subsequent runs — the live loop finishes on its existing + // toolset. Acceptable: the approval gate still applies on the next turn's run, and operators + // who need to halt a run immediately use cancel_agent_run. Tightening to per-turn re-evaluation + // would require threading a config getter into the loop; deferred until there's a need. + return liveConfig; + } + + const operations = buildOperations({ + getConfig: () => liveConfig, + setConfig, + startRun, + cancelRun, + }); + for (const op of operations) opts.server.registerOperation(op); + + log.info?.(`Agent component initialized with ${composed.tools.length} tools`); +} + +function resolveScopes( + config: AgentConfig, + getConfigPath: (param: string) => string | undefined, + getConfigFilePath?: () => string +): AgentScopes { + const componentsRoot = getConfigPath(CONFIG_PARAMS.COMPONENTSROOT) ?? process.cwd(); + const logDir = getConfigPath(CONFIG_PARAMS.LOGGING_ROOT) ?? process.cwd(); + const rootPath = getConfigPath(CONFIG_PARAMS.ROOTPATH) ?? componentsRoot; + const configFile = getConfigFilePath?.(); + const configDir = configFile ? dirname(configFile) : process.cwd(); + // A relative `componentsScope` is resolved against rootPath, as documented in the schema — + // NOT against componentsRoot, which would double-nest (`./components` → componentsRoot/components). + // With no scope set, the full componentsRoot is the FS write scope. + const scopedComponents = config.componentsScope + ? isAbsolute(config.componentsScope) + ? config.componentsScope + : resolvePath(rootPath, config.componentsScope) + : componentsRoot; + return { componentsRoot: scopedComponents, logDir, configDir }; +} + +function mergeConfig(opts: StartOpts): AgentConfig { + return { + ...DEFAULT_CONFIG, + ...(opts.enabled !== undefined && { enabled: !!opts.enabled }), + ...(opts.provider !== undefined && { provider: String(opts.provider) }), + ...(opts.model !== undefined && { model: String(opts.model) }), + ...(opts.maxTurns !== undefined && { maxTurns: Number(opts.maxTurns) }), + ...(opts.maxCostUsd !== undefined && { maxCostUsd: Number(opts.maxCostUsd) }), + ...(opts.autoApprove !== undefined && { autoApprove: !!opts.autoApprove }), + ...(opts.allowDestructive !== undefined && { allowDestructive: !!opts.allowDestructive }), + ...(opts.user !== undefined && { user: String(opts.user) }), + ...(opts.componentsScope !== undefined && { componentsScope: String(opts.componentsScope) }), + }; +} + +/** Test-only: reset module state between specs. */ +export function _resetForTests(): void { + _resetInFlightForTests(); +} diff --git a/agent/loop.ts b/agent/loop.ts new file mode 100644 index 0000000000..0f638c7ec1 --- /dev/null +++ b/agent/loop.ts @@ -0,0 +1,205 @@ +/** + * Manual agent loop for the built-in agent (#626). + * + * Wraps `scope.models.generate({ ..., toolMode: 'return' })` and dispatches + * any tool calls the model returns. This is a temporary stand-in for the + * unified `toolMode: 'auto'` orchestrator landing in #612 — when that ships, + * tool-call dispatch and the per-turn loop collapse into a single + * `generate({ ..., toolMode: 'auto' })` call. The approval/abort gates here + * still live in the component (the orchestrator won't know about + * `destructive` or operator approval semantics). + * + * Per-session serialization (one concurrent run per session) is handled + * here via {@link runAgent}'s in-flight map. Multiple sessions interleave + * on the event loop naturally because each turn is mostly awaiting the LLM + * or a tool's I/O. + */ + +import type { GenerateOpts, GenerateResult, Message, Models, ToolCall, ToolDef } from '../resources/models/types.ts'; +import { addPendingApproval, appendMessage, getSession, markApprovalConsumed, setStatus } from './session.ts'; +import type { AgentMessage, AgentScopes, AgentTool, AgentToolContext } from './types.ts'; +import { toolMapByName } from './toolset.ts'; + +export interface RunAgentOpts { + sessionId: string; + models: Pick; + tools: AgentTool[]; + scopes: AgentScopes; + maxTurns: number; + /** When false, destructive tools pause the loop with a pending approval instead of executing. */ + autoApprove?: boolean; + signal?: AbortSignal; + generateOpts?: Omit; + /** System prompt injected as the first turn when the transcript is empty. */ + systemPrompt?: string; +} + +const inFlight = new Map>(); + +export function runAgent(opts: RunAgentOpts): Promise { + const existing = inFlight.get(opts.sessionId); + if (existing) return existing; + const run = doRun(opts).finally(() => { + if (inFlight.get(opts.sessionId) === run) inFlight.delete(opts.sessionId); + }); + inFlight.set(opts.sessionId, run); + return run; +} + +async function doRun(opts: RunAgentOpts): Promise { + const toolMap = toolMapByName(opts.tools); + const toolDefs: ToolDef[] = opts.tools.map((t) => t.def); + await setStatus(opts.sessionId, 'running'); + const ctx: AgentToolContext = { sessionId: opts.sessionId, signal: opts.signal, scopes: opts.scopes }; + + try { + // First, drain any resolved-but-unconsumed approvals from a prior pause. Either execute + // or refuse each saved call, recording an observation, so the next model turn sees the + // result of the operator decision. + await consumeResolvedApprovals(opts.sessionId, toolMap, ctx); + + // If a turn produced multiple gated tool calls and the operator has only resolved some of + // them, the remaining approvals are still pending — meaning the assistant's tool_calls do + // not yet all have tool responses. Re-entering the generate loop now would send an + // incomplete tool-response set and the provider would 400. Stay paused until every gated + // call for this turn is resolved (each `approve_agent_action` re-runs this path). + const afterConsume = await getSession(opts.sessionId); + if (afterConsume?.pendingApprovals.some((a) => !a.resolved)) { + await setStatus(opts.sessionId, 'awaiting_approval'); + return; + } + + for (let turn = 0; turn < opts.maxTurns; turn++) { + if (opts.signal?.aborted) return; // status was already set to `aborted` by cancelRun + const session = await getSession(opts.sessionId); + if (!session) throw new Error(`Session ${opts.sessionId} vanished mid-run`); + const messages = toModelMessages(session.messages, opts.systemPrompt); + const result: GenerateResult = await opts.models.generate( + { messages, tools: toolDefs, system: opts.systemPrompt }, + { ...opts.generateOpts, toolMode: 'return', signal: opts.signal } + ); + + await appendMessage(opts.sessionId, { + role: 'assistant', + content: result.content ?? '', + toolCalls: result.toolCalls, + createdAt: Date.now(), + }); + + if (!result.toolCalls || result.toolCalls.length === 0) { + await setStatus(opts.sessionId, 'completed'); + return; + } + + const paused = await dispatchToolCalls(result.toolCalls, toolMap, ctx, opts); + if (paused || opts.signal?.aborted) return; + } + await setStatus(opts.sessionId, 'completed', `Reached maxTurns=${opts.maxTurns} without a final answer.`); + } catch (err) { + // If the abort signal fired, the cancel path already set the session to `aborted` — + // don't clobber that with `error`. The rejection here is just the awaited generate/tool + // honoring the signal, not a real failure. + if (opts.signal?.aborted) return; + await setStatus(opts.sessionId, 'error', err instanceof Error ? err.message : String(err)); + throw err; + } +} + +/** + * Returns `true` when the loop should pause (any destructive tool call required approval). + * Non-destructive calls execute inline and their observations are appended. Destructive calls + * register pending approvals but do NOT append a tool message — `consumeResolvedApprovals` + * writes the single tool response on the next run. This keeps the 1:1 mapping between + * assistant tool_calls and tool responses that LLM APIs enforce, including when the assistant + * message mixes destructive and non-destructive calls in the same turn. + */ +async function dispatchToolCalls( + calls: ToolCall[], + toolMap: Map, + ctx: AgentToolContext, + opts: RunAgentOpts +): Promise { + let needsApproval = false; + for (const call of calls) { + if (opts.signal?.aborted) return true; + const tool = toolMap.get(call.name); + const destructiveAndGated = tool?.destructive && !opts.autoApprove; + if (destructiveAndGated) { + await addPendingApproval(opts.sessionId, { + toolName: call.name, + arguments: call.arguments ?? {}, + toolCallId: call.id, + reason: 'destructive', + }); + needsApproval = true; + // Don't break — keep processing remaining calls so non-destructive ones in the same + // turn still execute and write their tool responses. Their results may be useful + // context for the operator deciding whether to approve. + continue; + } + const observation = await invokeTool(call, toolMap, ctx); + await appendMessage(opts.sessionId, { + role: 'tool', + content: observation, + toolCallId: call.id, + createdAt: Date.now(), + }); + } + return needsApproval; +} + +async function consumeResolvedApprovals( + sessionId: string, + toolMap: Map, + ctx: AgentToolContext +): Promise { + const session = await getSession(sessionId); + if (!session) return; + const toConsume = session.pendingApprovals.filter((a) => a.resolved && !a.consumed); + for (const approval of toConsume) { + const observation = approval.approved + ? await invokeTool( + { id: approval.toolCallId, name: approval.toolName, arguments: approval.arguments }, + toolMap, + ctx + ) + : JSON.stringify({ ok: false, error: 'denied_by_operator', tool: approval.toolName }); + await appendMessage(sessionId, { + role: 'tool', + content: observation, + toolCallId: approval.toolCallId, + createdAt: Date.now(), + }); + await markApprovalConsumed(sessionId, approval.id); + } +} + +async function invokeTool(call: ToolCall, toolMap: Map, ctx: AgentToolContext): Promise { + const tool = toolMap.get(call.name); + if (!tool) return JSON.stringify({ error: 'unknown_tool', name: call.name }); + try { + const result = await tool.handler(call.arguments ?? {}, ctx); + return JSON.stringify({ ok: true, result }); + } catch (err) { + return JSON.stringify({ ok: false, error: err instanceof Error ? err.message : String(err) }); + } +} + +function toModelMessages(items: AgentMessage[], systemPrompt: string | undefined): Message[] { + const out: Message[] = []; + if (systemPrompt && !items.some((m) => m.role === 'system')) { + out.push({ role: 'system', content: systemPrompt }); + } + for (const item of items) { + const message: Message = { role: item.role, content: item.content }; + if (item.toolCalls) message.toolCalls = item.toolCalls; + if (item.toolCallId) message.toolCallId = item.toolCallId; + out.push(message); + } + return out; +} + +/** Test-only: clear the in-flight tracking map. */ +export function _resetInFlightForTests(): void { + inFlight.clear(); +} diff --git a/agent/operations.ts b/agent/operations.ts new file mode 100644 index 0000000000..728c0c0f27 --- /dev/null +++ b/agent/operations.ts @@ -0,0 +1,148 @@ +/** + * Operations API surface for the built-in agent (#626). + * + * Six handlers, all super_user-only — the auth check is inline because none + * of these ops are registered in `utility/operation_authorization.ts`'s + * `requiredPermissions` map. Without the inline gate, a non-SU request would + * fall through the standard flow and be allowed. + */ + +import type { OperationDefinition } from '../server/serverHelpers/serverUtilities.ts'; +import { OPERATIONS_ENUM } from '../utility/hdbTerms.ts'; +import { ClientError } from '../utility/errors/hdbError.ts'; +import { createSession, getSession, listSessions, appendMessage, resolveApproval, setStatus } from './session.ts'; +import type { AgentConfig, AgentMessage, AgentRunStatus } from './types.ts'; + +export interface OperationDeps { + getConfig: () => AgentConfig; + setConfig: (patch: Partial) => AgentConfig; + startRun: (sessionId: string) => void; + cancelRun: (sessionId: string) => boolean; +} + +export function buildOperations(deps: OperationDeps): OperationDefinition[] { + return [ + { + name: OPERATIONS_ENUM.AGENT_PROMPT, + execute: async (op) => agentPrompt(op, deps), + }, + { + name: OPERATIONS_ENUM.GET_AGENT_SESSION, + execute: async (op) => getAgentSession(op), + }, + { + name: OPERATIONS_ENUM.LIST_AGENT_SESSIONS, + execute: async (op) => listAgentSessions(op), + }, + { + name: OPERATIONS_ENUM.CANCEL_AGENT_RUN, + execute: async (op) => cancelAgentRun(op, deps), + }, + { + name: OPERATIONS_ENUM.APPROVE_AGENT_ACTION, + execute: async (op) => approveAgentAction(op, deps), + }, + { + name: OPERATIONS_ENUM.SET_AGENT_CONFIG, + execute: async (op) => setAgentConfig(op, deps), + }, + ]; +} + +function requireSuperUser(op: any): void { + if (!op?.hdb_user?.role?.permission?.super_user) { + throw new ClientError('Agent operations require super_user', 403); + } +} + +async function agentPrompt(op: any, deps: OperationDeps) { + requireSuperUser(op); + const config = deps.getConfig(); + if (!config.enabled) { + throw new ClientError('Agent component is disabled (agent.enabled=false)', 409); + } + const message = String(op?.message ?? '').trim(); + if (!message) throw new ClientError('message is required', 400); + + let sessionId: string = op?.session_id; + if (sessionId) { + const existing = await getSession(sessionId); + if (!existing) throw new ClientError(`Unknown session ${sessionId}`, 404); + if (existing.status === 'running' || existing.status === 'awaiting_approval') { + throw new ClientError( + `Session ${sessionId} is ${existing.status}; resolve or cancel before sending another prompt`, + 409 + ); + } + await appendMessage(sessionId, asUserMessage(message)); + } else { + const created = await createSession({ + user: op?.hdb_user?.username ?? config.user, + model: config.model, + provider: config.provider, + initialMessage: asUserMessage(message), + }); + sessionId = created.session_id; + } + + deps.startRun(sessionId); + return { session_id: sessionId, status: 'running' satisfies AgentRunStatus }; +} + +async function getAgentSession(op: any) { + requireSuperUser(op); + const sessionId = String(op?.session_id ?? ''); + if (!sessionId) throw new ClientError('session_id is required', 400); + const session = await getSession(sessionId); + if (!session) throw new ClientError(`Unknown session ${sessionId}`, 404); + return session; +} + +async function listAgentSessions(op: any) { + requireSuperUser(op); + const limit = Number.isFinite(op?.limit) ? Number(op.limit) : undefined; + return { sessions: await listSessions({ limit }) }; +} + +async function cancelAgentRun(op: any, deps: OperationDeps) { + requireSuperUser(op); + const sessionId = String(op?.session_id ?? ''); + if (!sessionId) throw new ClientError('session_id is required', 400); + const session = await getSession(sessionId); + if (!session) throw new ClientError(`Unknown session ${sessionId}`, 404); + // Abort any active controller (best-effort — there may not be one if the loop is paused + // in `awaiting_approval` or sitting `idle` between turns). Always update the persisted + // status so a paused session can still be terminated by the operator. + const signalledLiveRun = deps.cancelRun(sessionId); + const wasTerminal = session.status === 'completed' || session.status === 'aborted' || session.status === 'error'; + if (!wasTerminal) await setStatus(sessionId, 'aborted', 'Cancelled by operator'); + return { cancelled: !wasTerminal, signalledLiveRun }; +} + +async function approveAgentAction(op: any, deps: OperationDeps) { + requireSuperUser(op); + const sessionId = String(op?.session_id ?? ''); + const approvalId = String(op?.approval_id ?? ''); + if (!sessionId || !approvalId) { + throw new ClientError('session_id and approval_id are required', 400); + } + const approved = op?.approved !== false; + const resolved = await resolveApproval(sessionId, approvalId, approved); + // Either decision should resume the loop: an approval means execute the saved tool call; + // a denial means hand the operator-rejection observation back to the model so it can adjust. + deps.startRun(sessionId); + return resolved; +} + +async function setAgentConfig(op: any, deps: OperationDeps) { + requireSuperUser(op); + const patch: Partial = {}; + for (const key of ['enabled', 'provider', 'model', 'maxTurns', 'maxCostUsd', 'autoApprove', 'allowDestructive']) { + if (op?.[key] !== undefined) (patch as any)[key] = op[key]; + } + return deps.setConfig(patch); +} + +function asUserMessage(content: string): AgentMessage { + return { role: 'user', content, createdAt: Date.now() }; +} diff --git a/agent/session.ts b/agent/session.ts new file mode 100644 index 0000000000..18514a2040 --- /dev/null +++ b/agent/session.ts @@ -0,0 +1,187 @@ +/** + * Built-in agent session storage. Backed by `system.hdb_agent_session` so + * transcripts and pending approvals survive restarts. The exported helpers + * are async to keep the surface uniform — once the underlying store gains + * non-blocking paths the call sites won't have to change. + * + * Intentionally separate from #511's `ConversationResource`: that primitive + * is app-facing (per-tenant, multi-user, user-defined schema) while this + * table is server-local and operator-owned. + */ + +import { randomUUID } from 'node:crypto'; +import { table } from '../resources/databases.ts'; +import { SYSTEM_SCHEMA_NAME, SYSTEM_TABLE_NAMES } from '../utility/hdbTerms.ts'; +import type { AgentMessage, AgentRunStatus, AgentSessionRow, ApprovalRequest } from './types.ts'; + +let cachedTable: any; + +export function getAgentSessionTable(): any { + if (cachedTable) return cachedTable; + cachedTable = table({ + table: SYSTEM_TABLE_NAMES.AGENT_SESSION_TABLE_NAME, + database: SYSTEM_SCHEMA_NAME, + audit: true, + trackDeletes: false, + attributes: [ + { name: 'session_id', isPrimaryKey: true }, + { name: 'user', type: 'string', indexed: true }, + { name: 'status', type: 'string', indexed: true }, + { name: 'messages' }, + { name: 'pendingApprovals' }, + { name: 'model', type: 'string' }, + { name: 'provider', type: 'string' }, + { name: 'createdAt', type: 'number', indexed: true }, + { name: 'updatedAt', type: 'number', indexed: true }, + { name: 'lastError', type: 'string' }, + ], + }); + return cachedTable; +} + +export interface CreateSessionOpts { + sessionId?: string; + user: string; + model?: string; + provider?: string; + initialMessage?: AgentMessage; +} + +export async function createSession(opts: CreateSessionOpts): Promise { + const now = Date.now(); + const row: AgentSessionRow = { + session_id: opts.sessionId ?? randomUUID(), + user: opts.user, + status: 'idle', + messages: opts.initialMessage ? [opts.initialMessage] : [], + pendingApprovals: [], + model: opts.model, + provider: opts.provider, + createdAt: now, + updatedAt: now, + }; + await getAgentSessionTable().primaryStore.put(row.session_id, row); + return row; +} + +export async function getSession(sessionId: string): Promise { + return getAgentSessionTable().primaryStore.get(sessionId); +} + +export async function listSessions(opts: { limit?: number } = {}): Promise { + const limit = opts.limit ?? 100; + const out: AgentSessionRow[] = []; + for (const entry of getAgentSessionTable().primaryStore.getRange({ reverse: true, limit })) { + if (entry.value) out.push(entry.value as AgentSessionRow); + } + return out; +} + +/** + * Per-session mutation lock. Each row mutation is a read-modify-write against the table; without + * serialization, two concurrent mutations on the same session (e.g. the loop appending an assistant + * message while the operator resolves an approval) both read the same row and the second `put` + * clobbers the first — a lost update. This chains all mutations for a given session id so they + * apply sequentially. Reads (`getSession`/`listSessions`) intentionally don't take the lock. + */ +const sessionLocks = new Map>(); + +function withSessionLock(sessionId: string, fn: () => Promise): Promise { + const prev = (sessionLocks.get(sessionId) ?? Promise.resolve()).catch(() => {}); + const result = prev.then(() => fn()); + const tail = result.catch(() => {}); + sessionLocks.set(sessionId, tail); + void tail.then(() => { + if (sessionLocks.get(sessionId) === tail) sessionLocks.delete(sessionId); + }); + return result; +} + +export function appendMessage(sessionId: string, message: AgentMessage): Promise { + return withSessionLock(sessionId, async () => { + const session = await requireSession(sessionId); + session.messages.push(message); + session.updatedAt = Date.now(); + await getAgentSessionTable().primaryStore.put(sessionId, session); + return session; + }); +} + +export function setStatus(sessionId: string, status: AgentRunStatus, lastError?: string): Promise { + return withSessionLock(sessionId, async () => { + const session = await requireSession(sessionId); + session.status = status; + session.lastError = lastError; + session.updatedAt = Date.now(); + await getAgentSessionTable().primaryStore.put(sessionId, session); + return session; + }); +} + +export function addPendingApproval( + sessionId: string, + approval: Omit +): Promise { + return withSessionLock(sessionId, async () => { + const session = await requireSession(sessionId); + const entry: ApprovalRequest = { ...approval, id: randomUUID(), createdAt: Date.now() }; + session.pendingApprovals.push(entry); + session.status = 'awaiting_approval'; + session.updatedAt = Date.now(); + await getAgentSessionTable().primaryStore.put(sessionId, session); + return entry; + }); +} + +export function markApprovalConsumed(sessionId: string, approvalId: string): Promise { + return withSessionLock(sessionId, async () => { + const session = await requireSession(sessionId); + const entry = session.pendingApprovals.find((a) => a.id === approvalId); + if (!entry) throw new Error(`No pending approval ${approvalId} on session ${sessionId}`); + if (!entry.resolved) throw new Error(`Approval ${approvalId} not yet resolved`); + if (entry.consumed) return; + entry.consumed = true; + session.updatedAt = Date.now(); + await getAgentSessionTable().primaryStore.put(sessionId, session); + }); +} + +export function resolveApproval(sessionId: string, approvalId: string, approved: boolean): Promise { + return withSessionLock(sessionId, async () => { + const session = await requireSession(sessionId); + const entry = session.pendingApprovals.find((a) => a.id === approvalId); + if (!entry) throw new Error(`No pending approval ${approvalId} on session ${sessionId}`); + if (entry.resolved) throw new Error(`Approval ${approvalId} already resolved`); + entry.resolved = true; + entry.approved = approved; + entry.resolvedAt = Date.now(); + // Either decision (approve or deny) returns the session to a resumable `idle` state so the + // loop can run again and deliver the resulting observation to the model. Operators who want + // to terminate the whole run should use `cancel_agent_run` instead of denying. + if (!session.pendingApprovals.some((a) => !a.resolved)) { + session.status = 'idle'; + } + session.updatedAt = Date.now(); + await getAgentSessionTable().primaryStore.put(sessionId, session); + return entry; + }); +} + +async function requireSession(sessionId: string): Promise { + const session = await getSession(sessionId); + if (!session) throw new Error(`No agent session ${sessionId}`); + return session; +} + +/** + * Reset the cached table reference. Test-only seam; production code lets the + * lazy-getter initialize once per process. + */ +export function _resetForTests(): void { + cachedTable = undefined; +} + +/** Inject a mock table accessor for unit tests. Pass `undefined` to restore the lazy default. */ +export function _setTableForTests(mock: any): void { + cachedTable = mock; +} diff --git a/agent/tools/fsTools.ts b/agent/tools/fsTools.ts new file mode 100644 index 0000000000..81d8018595 --- /dev/null +++ b/agent/tools/fsTools.ts @@ -0,0 +1,276 @@ +/** + * Operator-only filesystem tools for the built-in agent (#626). + * + * Every path is resolved against the configured scopes (componentsRoot, + * logDir, configDir) and rejected if it escapes them. Writes are restricted + * to `componentsRoot` — `logDir` and `configDir` are observation-only so + * the agent can read logs and inspect config without rewriting either. + * + * Lifted in spirit from the external `harper-agent` CLI's file tools; the + * sandboxing rules are tightened here because the in-process agent can + * reach more of the filesystem than a remote CLI. + */ + +import { readFile, writeFile, readdir, stat, mkdir, realpath, lstat, open } from 'node:fs/promises'; +import { resolve, dirname, relative, sep, isAbsolute } from 'node:path'; +import type { AgentTool, AgentToolContext, AgentScopes } from '../types.ts'; + +const MAX_READ_BYTES = 5 * 1024 * 1024; // 5 MiB +const MAX_WRITE_BYTES = 5 * 1024 * 1024; +const MAX_GREP_RESULTS = 500; +const MAX_PATTERN_LENGTH = 1000; +const DEFAULT_TAIL_LINES = 200; +const TAIL_READ_BYTES = 1 * 1024 * 1024; // 1 MiB — enough for thousands of normal log lines + +type Access = 'read' | 'write'; + +async function resolveScoped(scopes: AgentScopes, path: string, access: Access): Promise { + const absolute = resolve(path); + const candidates = [scopes.componentsRoot]; + if (access === 'read') { + candidates.push(scopes.logDir, scopes.configDir); + } + // Reject a symlink leaf. `safeRealPath` resolves existing symlinks via `realpath` (so a link to + // an out-of-scope *existing* file is caught by the isInside check below) — but a link whose + // target does NOT exist makes `realpath` throw, and the fallback returns the link's own in-scope + // path. `writeFile`/`readFile` then follow the link out of scope. An explicit lstat closes that + // gap: a legitimate component/log/config file is never a symlink. + try { + const linkStat = await lstat(absolute); + if (linkStat.isSymbolicLink()) { + throw new Error(`Refusing to ${access} through a symlink: ${path}`); + } + } catch (err) { + // ENOENT (path doesn't exist yet — normal for a new-file write) is fine; rethrow anything else + // (including our own symlink rejection). + if ((err as NodeJS.ErrnoException)?.code !== 'ENOENT') throw err; + } + const realAbsolute = await safeRealPath(absolute); + for (const root of candidates) { + const realRoot = await safeRealPath(root); + if (isInside(realAbsolute, realRoot)) return realAbsolute; + } + throw new Error(`Path is outside the agent's ${access} scope: ${path}`); +} + +async function safeRealPath(p: string): Promise { + try { + return await realpath(p); + } catch { + // Missing leaf is fine — resolve the deepest existing ancestor and join. + const parent = dirname(p); + if (parent === p) return p; + const parentReal = await safeRealPath(parent); + return resolve(parentReal, p.slice(parent.length + 1)); + } +} + +function isInside(child: string, parent: string): boolean { + const rel = relative(parent, child); + // On Windows, `path.relative` returns an absolute path when the two arguments are on + // different drive letters (e.g. C:\components vs D:\etc). Without this check the agent + // could escape its scope by naming a path on another drive. + if (isAbsolute(rel)) return false; + return rel === '' || (!rel.startsWith('..') && !rel.includes(`..${sep}`)); +} + +export const readFileTool: AgentTool = { + def: { + name: 'read_file', + description: 'Read a UTF-8 text file within componentsRoot, logDir, or configDir.', + parameters: { + type: 'object', + properties: { + path: { type: 'string', description: 'Absolute filesystem path.' }, + }, + required: ['path'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const path = await resolveScoped(ctx.scopes, args.path, 'read'); + const st = await stat(path); + if (st.size > MAX_READ_BYTES) { + throw new Error(`File ${path} exceeds ${MAX_READ_BYTES}-byte read cap (size ${st.size})`); + } + const content = await readFile(path, 'utf8'); + return { path, size: st.size, content }; + }, +}; + +export const writeFileTool: AgentTool = { + def: { + name: 'write_file', + description: 'Write a UTF-8 text file within componentsRoot. Creates parent directories as needed.', + parameters: { + type: 'object', + properties: { + path: { type: 'string', description: 'Absolute filesystem path under componentsRoot.' }, + content: { type: 'string', description: 'UTF-8 file contents.' }, + }, + required: ['path', 'content'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const content = String(args.content ?? ''); + if (Buffer.byteLength(content, 'utf8') > MAX_WRITE_BYTES) { + throw new Error(`Write exceeds ${MAX_WRITE_BYTES}-byte cap`); + } + const path = await resolveScoped(ctx.scopes, args.path, 'write'); + await mkdir(dirname(path), { recursive: true }); + await writeFile(path, content, 'utf8'); + return { path, bytesWritten: Buffer.byteLength(content, 'utf8') }; + }, + destructive: true, +}; + +export const listDirTool: AgentTool = { + def: { + name: 'list_dir', + description: 'List the immediate entries in a directory within an allowed scope.', + parameters: { + type: 'object', + properties: { + path: { type: 'string', description: 'Absolute filesystem path.' }, + }, + required: ['path'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const path = await resolveScoped(ctx.scopes, args.path, 'read'); + const entries = await readdir(path, { withFileTypes: true }); + return { + path, + entries: entries.map((e) => ({ + name: e.name, + kind: e.isDirectory() ? 'directory' : e.isFile() ? 'file' : 'other', + })), + }; + }, +}; + +export const grepFilesTool: AgentTool = { + def: { + name: 'grep_files', + description: 'Search recursively under a scoped directory for a regex pattern. Returns matched lines.', + parameters: { + type: 'object', + properties: { + root: { type: 'string', description: 'Directory to search under.' }, + pattern: { type: 'string', description: 'JavaScript-compatible regular expression source.' }, + flags: { type: 'string', description: 'Regex flags (default: "i").' }, + maxResults: { type: 'integer', minimum: 1, maximum: MAX_GREP_RESULTS }, + }, + required: ['root', 'pattern'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const root = await resolveScoped(ctx.scopes, args.root, 'read'); + // Cap pattern length. A maliciously crafted regex (e.g. nested quantifiers) can backtrack + // catastrophically and block the main thread; JS has no native per-match timeout. The agent + // is super_user-gated so this is self-inflicted DoS rather than a privilege boundary, but a + // length cap removes the easiest footgun without a worker-thread regex sandbox. + const patternSource = String(args.pattern ?? ''); + if (patternSource.length > MAX_PATTERN_LENGTH) { + throw new Error(`grep pattern exceeds ${MAX_PATTERN_LENGTH}-char cap`); + } + const pattern = new RegExp(patternSource, args.flags ?? 'i'); + const cap = Math.min(args.maxResults ?? MAX_GREP_RESULTS, MAX_GREP_RESULTS); + const results: Array<{ path: string; line: number; text: string }> = []; + await walk(root, async (file) => { + if (results.length >= cap) return false; + // `stat` first so a multi-GB log or database file can't be slurped into memory by a + // well-formed grep request. Anything over the read cap is silently skipped. + let size = 0; + try { + const st = await stat(file); + size = st.size; + } catch { + return true; + } + if (size > MAX_READ_BYTES) return true; + const text = await readFile(file, 'utf8').catch(() => ''); + if (!text) return true; + const lines = text.split('\n'); + for (let i = 0; i < lines.length; i++) { + if (results.length >= cap) return false; + if (pattern.test(lines[i])) results.push({ path: file, line: i + 1, text: lines[i] }); + } + return true; + }); + return { root, count: results.length, results }; + }, +}; + +export const tailFileTool: AgentTool = { + def: { + name: 'tail_file', + description: 'Return the last N lines of a UTF-8 file. Useful for log tails.', + parameters: { + type: 'object', + properties: { + path: { type: 'string', description: 'Absolute filesystem path.' }, + lines: { type: 'integer', minimum: 1, maximum: 5000 }, + }, + required: ['path'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const path = await resolveScoped(ctx.scopes, args.path, 'read'); + const wanted = Math.min(args.lines ?? DEFAULT_TAIL_LINES, 5000); + // Read only the trailing TAIL_READ_BYTES — a multi-GB log file otherwise OOMs the process. + const st = await stat(path); + const start = Math.max(0, st.size - TAIL_READ_BYTES); + const truncated = start > 0; + const fh = await open(path, 'r'); + try { + const buf = Buffer.alloc(st.size - start); + await fh.read(buf, 0, buf.length, start); + const text = buf.toString('utf8'); + const all = text.split('\n'); + // `split('\n')` on a file ending with `\n` leaves a trailing empty entry — drop it so the + // "last N lines" the agent sees matches what a human reading the file would see. + if (all.length > 0 && all[all.length - 1] === '') all.pop(); + // When we read from a mid-file offset the first "line" is almost certainly a partial + // fragment of a real line. Drop it so we don't hand the agent a misleading prefix. + if (truncated && all.length > 0) all.shift(); + const sliceStart = Math.max(0, all.length - wanted); + return { path, lines: all.slice(sliceStart), truncated }; + } finally { + await fh.close(); + } + }, +}; + +export const fsTools: AgentTool[] = [readFileTool, writeFileTool, listDirTool, grepFilesTool, tailFileTool]; + +async function walk(root: string, visit: (file: string) => Promise): Promise { + // Resolve the scope root once via realpath so the per-entry symlink check below has a + // stable comparison anchor; otherwise a symlink in the root itself could shift the anchor. + const realRoot = await safeRealPath(root); + const stack: string[] = [root]; + while (stack.length) { + const dir = stack.pop()!; + let entries; + try { + entries = await readdir(dir, { withFileTypes: true }); + } catch { + continue; + } + for (const entry of entries) { + const full = resolve(dir, entry.name); + if (entry.isDirectory()) { + if (entry.name === 'node_modules' || entry.name === '.git') continue; + // Re-resolve via realpath so a symlinked directory pointing outside the scope is rejected. + // Without this, `componentsRoot/escape -> /etc` would let grep walk into /etc. + const realFull = await safeRealPath(full); + if (!isInside(realFull, realRoot)) continue; + stack.push(full); + } else if (entry.isFile()) { + const realFull = await safeRealPath(full); + if (!isInside(realFull, realRoot)) continue; + const proceed = await visit(full); + if (proceed === false) return; + } + } + } +} diff --git a/agent/tools/httpFetchTool.ts b/agent/tools/httpFetchTool.ts new file mode 100644 index 0000000000..f75f80ca52 --- /dev/null +++ b/agent/tools/httpFetchTool.ts @@ -0,0 +1,112 @@ +/** + * `http_fetch` for the built-in agent (#626). Wraps the platform `fetch` + * with a size cap, an inactivity timeout, and a metadata/loopback blocklist + * so the agent can probe its own deployed components and pull lightweight + * web pages for context without becoming an SSRF vector against cloud + * instance-metadata endpoints or unrelated internal services. + */ + +import { isIP } from 'node:net'; +import type { AgentTool, AgentToolContext } from '../types.ts'; + +const MAX_BYTES = 2 * 1024 * 1024; // 2 MiB cap on response bodies +const DEFAULT_TIMEOUT_MS = 30_000; +// Hard-blocked literal hosts. Cloud-metadata services live on these IPs and exposing them +// to a prompt-controlled fetch is a credential-leak vector. Loopback to the local Harper +// instance is allowed via `localhost`/`127.0.0.1` for self-testing — those are NOT blocked. +const BLOCKED_HOSTS = new Set([ + '169.254.169.254', // AWS / GCP / Azure IMDS + 'fd00:ec2::254', // AWS IMDSv2 IPv6 + 'metadata.google.internal', + 'metadata.goog', +]); + +export const httpFetchTool: AgentTool = { + def: { + name: 'http_fetch', + description: + "Issue an HTTP request from the Harper server. Useful for hitting the agent's own components on localhost and pulling reference pages.", + parameters: { + type: 'object', + properties: { + url: { type: 'string', description: 'Absolute URL.' }, + method: { type: 'string', enum: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD'] }, + headers: { type: 'object', additionalProperties: { type: 'string' } }, + body: { type: 'string', description: 'Request body as a string (JSON or form-encoded).' }, + timeoutMs: { type: 'integer', minimum: 1, maximum: 120_000 }, + }, + required: ['url'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const url = String(args.url ?? ''); + if (!/^https?:\/\//i.test(url)) throw new Error('http_fetch requires an http(s) URL'); + let parsed: URL; + try { + parsed = new URL(url); + } catch { + throw new Error(`http_fetch could not parse URL: ${url}`); + } + const host = parsed.hostname.toLowerCase(); + if (BLOCKED_HOSTS.has(host)) { + throw new Error(`http_fetch blocked by metadata-host policy: ${host}`); + } + // IPv4 link-local (169.254.0.0/16) covers IMDS variants beyond the canonical 169.254.169.254. + if (isIP(host) === 4 && host.startsWith('169.254.')) { + throw new Error(`http_fetch blocked by link-local policy: ${host}`); + } + const timeoutMs = Math.min(args.timeoutMs ?? DEFAULT_TIMEOUT_MS, 120_000); + const localAbort = new AbortController(); + const timer = setTimeout(() => localAbort.abort(new Error(`http_fetch timed out after ${timeoutMs}ms`)), timeoutMs); + const signal = combineSignals(ctx.signal, localAbort.signal); + try { + const response = await fetch(url, { + method: args.method ?? 'GET', + headers: args.headers, + body: args.body, + signal, + }); + const buffer = await readCapped(response, MAX_BYTES); + return { + status: response.status, + headers: Object.fromEntries(response.headers.entries()), + body: buffer.toString('utf8'), + truncated: buffer.length === MAX_BYTES, + }; + } finally { + clearTimeout(timer); + } + }, +}; + +async function readCapped(response: Response, cap: number): Promise { + const reader = response.body?.getReader(); + if (!reader) return Buffer.alloc(0); + const chunks: Buffer[] = []; + let total = 0; + while (total < cap) { + const { value, done } = await reader.read(); + if (done) break; + const chunk = Buffer.from(value); + const room = cap - total; + if (chunk.length > room) { + chunks.push(chunk.subarray(0, room)); + total += room; + await reader.cancel().catch(() => {}); + break; + } + chunks.push(chunk); + total += chunk.length; + } + return Buffer.concat(chunks, total); +} + +function combineSignals(...signals: Array): AbortSignal | undefined { + const present = signals.filter((s): s is AbortSignal => Boolean(s)); + if (present.length === 0) return undefined; + if (present.length === 1) return present[0]; + // `AbortSignal.any` (Node 20+) manages listener cleanup internally; the manual + // `addEventListener` approach leaked listeners on `ctx.signal` for the lifetime of the agent + // run when a fetch completed normally. + return AbortSignal.any(present); +} diff --git a/agent/tools/scheduleTool.ts b/agent/tools/scheduleTool.ts new file mode 100644 index 0000000000..f0db3b93c2 --- /dev/null +++ b/agent/tools/scheduleTool.ts @@ -0,0 +1,68 @@ +/** + * `schedule_followup` for the built-in agent (#626). + * + * Lives on the main thread so it survives worker-thread restarts that happen + * on code reload. The followup callback is injected from the entry point — + * this module only schedules the timer and tracks pending entries. + */ + +import { randomUUID } from 'node:crypto'; +import type { AgentTool, AgentToolContext } from '../types.ts'; + +const MAX_DELAY_MS = 24 * 60 * 60 * 1000; // 24h +const MIN_DELAY_MS = 1_000; + +export interface ScheduledFollowup { + id: string; + sessionId: string; + prompt: string; + fireAt: number; + timer: NodeJS.Timeout; +} + +export interface ScheduleToolDeps { + onFollowup: (sessionId: string, prompt: string) => Promise | void; +} + +export function buildScheduleTool(deps: ScheduleToolDeps): { + tool: AgentTool; + pending: Map; +} { + const pending = new Map(); + const tool: AgentTool = { + def: { + name: 'schedule_followup', + description: 'Re-invoke the agent on the current session after the given delay with a new prompt.', + parameters: { + type: 'object', + properties: { + delayMs: { type: 'integer', minimum: MIN_DELAY_MS, maximum: MAX_DELAY_MS }, + prompt: { type: 'string', description: 'Prompt the agent should re-enter on with.' }, + }, + required: ['delayMs', 'prompt'], + }, + }, + handler: async (args: any, ctx: AgentToolContext) => { + const delay = Number(args.delayMs); + if (!Number.isFinite(delay) || delay < MIN_DELAY_MS || delay > MAX_DELAY_MS) { + throw new Error(`delayMs must be between ${MIN_DELAY_MS} and ${MAX_DELAY_MS}`); + } + const prompt = String(args.prompt ?? '').trim(); + if (!prompt) throw new Error('prompt is required'); + const id = randomUUID(); + const fireAt = Date.now() + delay; + const timer = setTimeout(() => { + pending.delete(id); + Promise.resolve() + .then(() => deps.onFollowup(ctx.sessionId, prompt)) + .catch(() => { + /* swallow — caller logs */ + }); + }, delay); + timer.unref?.(); + pending.set(id, { id, sessionId: ctx.sessionId, prompt, fireAt, timer }); + return { id, fireAt }; + }, + }; + return { tool, pending }; +} diff --git a/agent/toolset.ts b/agent/toolset.ts new file mode 100644 index 0000000000..48342ee271 --- /dev/null +++ b/agent/toolset.ts @@ -0,0 +1,43 @@ +/** + * Tool composer for the built-in agent (#626). + * + * Operator-only tools (FS, schedule, fetch) are inline today. Once the + * unified MCP tool registry (#615) lands with the Operations (#617) and + * Application (#618) profiles, this is the seam where RBAC-filtered + * registry tools get folded in for the agent's configured user. The shape + * of {@link composeToolset} won't change — only the body gains a registry + * lookup. + */ + +import { fsTools } from './tools/fsTools.ts'; +import { httpFetchTool } from './tools/httpFetchTool.ts'; +import { buildScheduleTool, type ScheduleToolDeps, type ScheduledFollowup } from './tools/scheduleTool.ts'; +import type { AgentTool } from './types.ts'; + +export interface ComposeToolsetOpts extends ScheduleToolDeps { + /** When `false`, destructive tools are filtered out at composition time. */ + allowDestructive?: boolean; + /** Operator-injected extras (tests, custom plugins). */ + extraTools?: AgentTool[]; +} + +export interface ComposedToolset { + tools: AgentTool[]; + scheduled: Map; +} + +export function composeToolset(opts: ComposeToolsetOpts): ComposedToolset { + const schedule = buildScheduleTool(opts); + const all: AgentTool[] = [...fsTools, httpFetchTool, schedule.tool, ...(opts.extraTools ?? [])]; + const tools = opts.allowDestructive === false ? all.filter((t) => !t.destructive) : all; + return { tools, scheduled: schedule.pending }; +} + +export function toolMapByName(tools: AgentTool[]): Map { + const map = new Map(); + for (const tool of tools) { + if (map.has(tool.def.name)) throw new Error(`Duplicate tool registered: ${tool.def.name}`); + map.set(tool.def.name, tool); + } + return map; +} diff --git a/agent/types.ts b/agent/types.ts new file mode 100644 index 0000000000..3b8d989d18 --- /dev/null +++ b/agent/types.ts @@ -0,0 +1,85 @@ +/** + * Shared types for the built-in Harper Agent component (#626). + * + * `AgentMessage` mirrors `resources/models/types.ts:Message` rather than + * re-exporting it: the session row is durable storage and shouldn't trail + * shape changes in the model-access surface. + */ + +import type { ToolCall, ToolDef } from '../resources/models/types.ts'; + +export type AgentRunStatus = 'idle' | 'running' | 'awaiting_approval' | 'completed' | 'aborted' | 'error'; + +export interface AgentMessage { + role: 'system' | 'user' | 'assistant' | 'tool'; + content: string; + toolCalls?: ToolCall[]; + toolCallId?: string; + /** Wall-clock timestamp of when this item entered the transcript. */ + createdAt: number; +} + +export interface ApprovalRequest { + id: string; + toolName: string; + arguments: object; + /** Tool-call id from the assistant message that produced this request. Used to resume execution. */ + toolCallId: string; + /** Human-readable rationale for why approval is required (e.g. 'destructive', 'first_use'). */ + reason: string; + createdAt: number; + resolved?: boolean; + approved?: boolean; + resolvedAt?: number; + /** Set to true once the loop has consumed the resolved approval (executed or refused). */ + consumed?: boolean; +} + +export interface AgentSessionRow { + session_id: string; + user: string; + status: AgentRunStatus; + messages: AgentMessage[]; + pendingApprovals: ApprovalRequest[]; + model?: string; + provider?: string; + createdAt: number; + updatedAt: number; + lastError?: string; +} + +export interface AgentToolHandler { + (args: object, ctx: AgentToolContext): Promise; +} + +export interface AgentTool { + def: ToolDef; + handler: AgentToolHandler; + /** When true, invocations of this tool require an approval gate unless `autoApprove` is set. */ + destructive?: boolean; +} + +export interface AgentToolContext { + sessionId: string; + signal?: AbortSignal; + /** Filesystem scope roots accessible to FS tools. Read-only after composition. */ + scopes: AgentScopes; +} + +export interface AgentScopes { + componentsRoot: string; + logDir: string; + configDir: string; +} + +export interface AgentConfig { + enabled: boolean; + provider?: string; + model?: string; + maxTurns: number; + maxCostUsd: number; + autoApprove: boolean; + allowDestructive: boolean; + user: string; + componentsScope?: string; +} diff --git a/components/componentLoader.ts b/components/componentLoader.ts index 7d944cadb6..19bffcc390 100644 --- a/components/componentLoader.ts +++ b/components/componentLoader.ts @@ -117,6 +117,9 @@ export const TRUSTED_RESOURCE_PLUGINS: any = { }; if (isMainThread) { TRUSTED_RESOURCE_PLUGINS.operationsApi = require('../server/operationsServer'); + // Built-in agent component (#626). Only loads if the root config carries an `agent:` block; + // the block's `enabled: false` default keeps it inert even when the key is present. + TRUSTED_RESOURCE_PLUGINS.agent = require('../agent/agent'); } else { // The HTTP operations API itself only binds in the main thread, but worker threads still // dispatch operations — most notably, the replication WebSocket handler in workers receives diff --git a/config-root.schema.json b/config-root.schema.json index 73c2c52265..f5eba64281 100644 --- a/config-root.schema.json +++ b/config-root.schema.json @@ -561,6 +561,49 @@ } }, + "agent": { + "type": "object", + "description": "Built-in Harper Agent component. Disabled by default to avoid surprise LLM costs.", + "additionalProperties": false, + "properties": { + "enabled": { "type": "boolean", "description": "Enable the built-in agent component. Default: false" }, + "provider": { + "type": "string", + "description": "Optional model provider override. Falls back to scope.models default when omitted." + }, + "model": { + "type": "string", + "description": "Optional model id override. Falls back to scope.models default when omitted." + }, + "maxTurns": { + "type": "integer", + "minimum": 1, + "description": "Maximum tool-call iterations per agent run. Default: 50" + }, + "maxCostUsd": { + "type": "number", + "minimum": 0, + "description": "Per-session cost budget. NOT YET ENFORCED — surfacing operator intent until #612 ships per-call cost telemetry. Default: 5.00" + }, + "autoApprove": { + "type": "boolean", + "description": "Run without per-action approval gates. Destructive ops still require allowDestructive and approval. Default: false" + }, + "allowDestructive": { + "type": "boolean", + "description": "Allow destructive operator tools (drop_component, restart, set_configuration, ...). Default: false" + }, + "user": { + "type": "string", + "description": "Harper user the agent acts as. Default: hdb_agent (created at startup if missing)." + }, + "componentsScope": { + "type": "string", + "description": "Filesystem scope for component edits, relative to rootPath. Default: ./components" + } + } + }, + "analytics": { "type": "object", "additionalProperties": false, diff --git a/package.json b/package.json index 6f3c7266b1..98c7a38b7e 100644 --- a/package.json +++ b/package.json @@ -22,6 +22,7 @@ "harper": "./dist/bin/harper.js" }, "files": [ + "agent", "bin", "CODE_OF_CONDUCT.md", "components", diff --git a/tsconfig.json b/tsconfig.json index eea8c5dcb3..a41dc0e8c9 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,5 +1,6 @@ { "include": [ + "agent/**/*", "bin/**/*", "components/**/*", "config/**/*", diff --git a/unitTests/agent/fsTools.test.js b/unitTests/agent/fsTools.test.js new file mode 100644 index 0000000000..fe5acda584 --- /dev/null +++ b/unitTests/agent/fsTools.test.js @@ -0,0 +1,133 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { mkdtempSync, writeFileSync, mkdirSync, readFileSync, existsSync } = require('node:fs'); +const { tmpdir } = require('node:os'); +const { join } = require('node:path'); +const { readFileTool, writeFileTool, listDirTool, grepFilesTool, tailFileTool } = require('#src/agent/tools/fsTools'); + +function mkScopes() { + const root = mkdtempSync(join(tmpdir(), 'agent-fs-')); + const componentsRoot = join(root, 'components'); + const logDir = join(root, 'logs'); + const configDir = join(root, 'config'); + mkdirSync(componentsRoot); + mkdirSync(logDir); + mkdirSync(configDir); + return { componentsRoot, logDir, configDir, root }; +} + +function ctx(scopes) { + return { sessionId: 'sess', scopes }; +} + +describe('agent/fsTools', () => { + let scopes; + beforeEach(() => { + scopes = mkScopes(); + }); + + it('read_file returns contents inside componentsRoot', async () => { + writeFileSync(join(scopes.componentsRoot, 'a.txt'), 'hello'); + const result = await readFileTool.handler({ path: join(scopes.componentsRoot, 'a.txt') }, ctx(scopes)); + assert.equal(result.content, 'hello'); + }); + + it('read_file rejects paths outside scope roots', async () => { + writeFileSync(join(scopes.root, 'outside.txt'), 'nope'); + await assert.rejects( + readFileTool.handler({ path: join(scopes.root, 'outside.txt') }, ctx(scopes)), + /outside the agent's read scope/ + ); + }); + + it('write_file refuses writes to logDir', async () => { + await assert.rejects( + writeFileTool.handler({ path: join(scopes.logDir, 'evil.txt'), content: 'x' }, ctx(scopes)), + /outside the agent's write scope/ + ); + }); + + it('write_file creates parents and writes within componentsRoot', async () => { + const target = join(scopes.componentsRoot, 'nested', 'b.txt'); + const result = await writeFileTool.handler({ path: target, content: 'x' }, ctx(scopes)); + assert.equal(result.bytesWritten, 1); + assert.equal(readFileSync(target, 'utf8'), 'x'); + }); + + it('write_file is marked destructive', () => { + assert.equal(writeFileTool.destructive, true); + }); + + it('list_dir enumerates direct children of an allowed scope', async () => { + writeFileSync(join(scopes.componentsRoot, 'a.txt'), '1'); + mkdirSync(join(scopes.componentsRoot, 'sub')); + const { entries } = await listDirTool.handler({ path: scopes.componentsRoot }, ctx(scopes)); + const names = entries.map((e) => e.name).sort(); + assert.deepEqual(names, ['a.txt', 'sub']); + }); + + it('grep_files finds matches and respects maxResults', async () => { + writeFileSync(join(scopes.componentsRoot, 'a.txt'), 'apple\nbanana\nApple'); + const { results } = await grepFilesTool.handler({ root: scopes.componentsRoot, pattern: 'apple' }, ctx(scopes)); + assert.equal(results.length, 2); + assert.equal(results[0].line, 1); + }); + + it('tail_file returns the last N lines', async () => { + writeFileSync(join(scopes.logDir, 'srv.log'), 'a\nb\nc\nd\n'); + const { lines } = await tailFileTool.handler({ path: join(scopes.logDir, 'srv.log'), lines: 2 }, ctx(scopes)); + assert.deepEqual(lines, ['c', 'd']); + }); + + it('grep_files refuses to traverse symlinked dirs that escape scope', async () => { + const { symlinkSync } = require('node:fs'); + // Create an out-of-scope dir with a file, then link into componentsRoot. + const escapeTarget = join(scopes.root, 'escape-target'); + mkdirSync(escapeTarget); + writeFileSync(join(escapeTarget, 'secret.txt'), 'PRIVATE'); + try { + symlinkSync(escapeTarget, join(scopes.componentsRoot, 'gateway'), 'dir'); + } catch (err) { + // Symlink not supported (e.g. some CI envs without permission) — skip the assertion + // rather than fail the suite. Real environments support it. + if (err.code === 'EPERM' || err.code === 'ENOTSUP') return; + throw err; + } + writeFileSync(join(scopes.componentsRoot, 'a.txt'), 'PRIVATE'); + const { results } = await grepFilesTool.handler({ root: scopes.componentsRoot, pattern: 'PRIVATE' }, ctx(scopes)); + // Should only find the file in componentsRoot, not the file behind the symlink. + assert.equal(results.length, 1); + assert.match(results[0].path, /a\.txt$/); + }); + + it('write_file refuses to write through a symlink whose target is outside scope (incl. non-existent target)', async () => { + const { symlinkSync } = require('node:fs'); + const outsideTarget = join(scopes.root, 'outside-secret.txt'); // does NOT exist → realpath would throw + try { + symlinkSync(outsideTarget, join(scopes.componentsRoot, 'escape-link'), 'file'); + } catch (err) { + if (err.code === 'EPERM' || err.code === 'ENOTSUP') return; + throw err; + } + await assert.rejects( + writeFileTool.handler({ path: join(scopes.componentsRoot, 'escape-link'), content: 'pwned' }, ctx(scopes)), + /through a symlink/ + ); + assert.equal(existsSync(outsideTarget), false); + }); + + it('refuses paths that resolve outside scope via ..', async () => { + const escape = join(scopes.componentsRoot, '..', '..', 'etc', 'passwd'); + await assert.rejects(readFileTool.handler({ path: escape }, ctx(scopes)), /outside the agent's read scope/); + }); + + it('write_file enforces the byte cap', async () => { + const big = 'x'.repeat(6 * 1024 * 1024); + await assert.rejects( + writeFileTool.handler({ path: join(scopes.componentsRoot, 'big.txt'), content: big }, ctx(scopes)), + /exceeds/ + ); + assert.equal(existsSync(join(scopes.componentsRoot, 'big.txt')), false); + }); +}); diff --git a/unitTests/agent/httpFetchTool.test.js b/unitTests/agent/httpFetchTool.test.js new file mode 100644 index 0000000000..70c8e9771d --- /dev/null +++ b/unitTests/agent/httpFetchTool.test.js @@ -0,0 +1,43 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { httpFetchTool } = require('#src/agent/tools/httpFetchTool'); + +const ctx = { sessionId: 'sess', scopes: { componentsRoot: '/tmp', logDir: '/tmp', configDir: '/tmp' } }; + +describe('agent/httpFetchTool', () => { + it('rejects non-http(s) URLs', async () => { + await assert.rejects(httpFetchTool.handler({ url: 'file:///etc/passwd' }, ctx), /http\(s\) URL/); + }); + + it('blocks the AWS/GCP cloud-metadata IP', async () => { + await assert.rejects( + httpFetchTool.handler({ url: 'http://169.254.169.254/latest/meta-data/' }, ctx), + /metadata-host policy|link-local policy/ + ); + }); + + it('blocks the GCP metadata hostname', async () => { + await assert.rejects( + httpFetchTool.handler({ url: 'http://metadata.google.internal/computeMetadata/v1/' }, ctx), + /metadata-host policy/ + ); + }); + + it('blocks the IPv4 link-local range beyond the canonical IMDS IP', async () => { + await assert.rejects(httpFetchTool.handler({ url: 'http://169.254.42.42/probe' }, ctx), /link-local policy/); + }); + + it('does not block localhost (operators self-test against their own server)', async () => { + // Use port 1 so the request fails fast with a connection error rather than hitting any + // real service. We only care that the URL passes the policy check. + await assert.rejects( + httpFetchTool.handler({ url: 'http://127.0.0.1:1/', timeoutMs: 500 }, ctx), + (err) => !/metadata-host policy|link-local policy/.test(err.message) + ); + }); + + it('rejects malformed URLs with a clear error', async () => { + await assert.rejects(httpFetchTool.handler({ url: 'http://[invalid' }, ctx), /could not parse URL/); + }); +}); diff --git a/unitTests/agent/loop.test.js b/unitTests/agent/loop.test.js new file mode 100644 index 0000000000..57bc51a3d9 --- /dev/null +++ b/unitTests/agent/loop.test.js @@ -0,0 +1,486 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { runAgent, _resetInFlightForTests } = require('#src/agent/loop'); +const session = require('#src/agent/session'); + +function makeMockTable() { + const store = new Map(); + return { + store, + primaryStore: { + async put(key, value) { + store.set(key, structuredClone(value)); + }, + async get(key) { + const value = store.get(key); + return value ? structuredClone(value) : undefined; + }, + getRange() { + return []; + }, + }, + }; +} + +function stubModels(turns) { + let i = 0; + return { + async generate() { + const turn = turns[i++]; + if (!turn) throw new Error('stubModels exhausted'); + return turn; + }, + }; +} + +const scopes = { componentsRoot: '/tmp', logDir: '/tmp', configDir: '/tmp' }; +const noTools = []; + +describe('agent/loop runAgent', () => { + beforeEach(() => { + session._setTableForTests(makeMockTable()); + _resetInFlightForTests(); + }); + + afterEach(() => { + session._setTableForTests(undefined); + }); + + it('terminates on a no-tool-call response and marks the session completed', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'hi', createdAt: Date.now() }); + const models = stubModels([{ content: 'done', finishReason: 'stop' }]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: noTools, + scopes, + maxTurns: 5, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'completed'); + const lastMessage = reloaded.messages[reloaded.messages.length - 1]; + assert.equal(lastMessage.role, 'assistant'); + assert.equal(lastMessage.content, 'done'); + }); + + it('dispatches tool calls and appends tool messages between turns', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'echo', createdAt: Date.now() }); + const calls = []; + const tool = { + def: { name: 'echo', description: 'echo', parameters: { type: 'object' } }, + handler: async (args) => { + calls.push(args); + return { echoed: args.value }; + }, + }; + const models = stubModels([ + { + content: 'calling tool', + finishReason: 'tool_calls', + toolCalls: [{ id: 'c1', name: 'echo', arguments: { value: 7 } }], + }, + { content: 'all done', finishReason: 'stop' }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'completed'); + assert.deepEqual(calls, [{ value: 7 }]); + const toolMessage = reloaded.messages.find((m) => m.role === 'tool'); + assert.ok(toolMessage); + assert.equal(toolMessage.toolCallId, 'c1'); + assert.match(toolMessage.content, /echoed/); + }); + + it('records a tool failure as a structured observation without aborting', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + const tool = { + def: { name: 'broken', description: 'broken', parameters: { type: 'object' } }, + handler: async () => { + throw new Error('handler boom'); + }, + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [{ id: 'c1', name: 'broken', arguments: {} }], + }, + { content: 'recovered', finishReason: 'stop' }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'completed'); + const toolMessage = reloaded.messages.find((m) => m.role === 'tool'); + assert.match(toolMessage.content, /handler boom/); + }); + + it('completes with an explanatory error when maxTurns is hit', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'loop', createdAt: Date.now() }); + const tool = { + def: { name: 'spin', description: 'spin', parameters: { type: 'object' } }, + handler: async () => ({ ok: true }), + }; + const turns = Array.from({ length: 5 }, (_, i) => ({ + content: `t${i}`, + finishReason: 'tool_calls', + toolCalls: [{ id: `c${i}`, name: 'spin', arguments: {} }], + })); + const models = stubModels(turns); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 3, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'completed'); + assert.match(reloaded.lastError ?? '', /maxTurns=3/); + }); + + it('halts on a destructive tool call when autoApprove is false', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + let executed = false; + const tool = { + def: { name: 'restart', description: 'restart', parameters: { type: 'object' } }, + destructive: true, + handler: async () => { + executed = true; + return { ok: true }; + }, + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [{ id: 'c1', name: 'restart', arguments: {} }], + }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + autoApprove: false, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(executed, false); + assert.equal(reloaded.status, 'awaiting_approval'); + assert.equal(reloaded.pendingApprovals.length, 1); + assert.equal(reloaded.pendingApprovals[0].toolName, 'restart'); + // No placeholder tool response: LLM APIs reject duplicate tool responses for the same + // tool_call_id. The tool response is only written when the operator resolves the approval. + const toolMessages = reloaded.messages.filter((m) => m.role === 'tool'); + assert.equal(toolMessages.length, 0); + }); + + it('preserves 1:1 tool-call mapping when a turn mixes destructive and non-destructive calls', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + const reads = []; + const readTool = { + def: { name: 'read', description: 'read', parameters: { type: 'object' } }, + handler: async (args) => { + reads.push(args); + return { value: 'data' }; + }, + }; + const dropTool = { + def: { name: 'drop', description: 'drop', parameters: { type: 'object' } }, + destructive: true, + handler: async () => ({ dropped: true }), + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [ + { id: 'c1', name: 'read', arguments: { what: 'first' } }, + { id: 'c2', name: 'drop', arguments: { table: 'x' } }, + { id: 'c3', name: 'read', arguments: { what: 'second' } }, + ], + }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [readTool, dropTool], + scopes, + maxTurns: 5, + autoApprove: false, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'awaiting_approval'); + assert.equal(reads.length, 2, 'both non-destructive reads should execute'); + const toolMessages = reloaded.messages.filter((m) => m.role === 'tool'); + // Two tool responses (for c1 and c3); c2 is awaiting approval — no placeholder. + assert.equal(toolMessages.length, 2); + const toolCallIds = toolMessages.map((m) => m.toolCallId).sort(); + assert.deepEqual(toolCallIds, ['c1', 'c3']); + assert.equal(reloaded.pendingApprovals.length, 1); + assert.equal(reloaded.pendingApprovals[0].toolCallId, 'c2'); + }); + + it('executes a destructive tool when autoApprove is true', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + let executed = false; + const tool = { + def: { name: 'restart', description: 'restart', parameters: { type: 'object' } }, + destructive: true, + handler: async () => { + executed = true; + return { ok: true }; + }, + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [{ id: 'c1', name: 'restart', arguments: {} }], + }, + { content: 'done', finishReason: 'stop' }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + autoApprove: true, + }); + + assert.equal(executed, true); + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'completed'); + }); + + it('consumes an approved approval on the next run and executes the saved call', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + let executed = 0; + const tool = { + def: { name: 'restart', description: 'restart', parameters: { type: 'object' } }, + destructive: true, + handler: async () => { + executed++; + return { restarted: true }; + }, + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [{ id: 'c1', name: 'restart', arguments: { force: true } }], + }, + { content: 'done after approval', finishReason: 'stop' }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + autoApprove: false, + }); + + // First run halts at awaiting_approval. Operator approves, loop resumes. + const halted = await session.getSession(created.session_id); + const approval = halted.pendingApprovals[0]; + await session.resolveApproval(created.session_id, approval.id, true); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + autoApprove: false, + }); + + assert.equal(executed, 1); + const final = await session.getSession(created.session_id); + assert.equal(final.status, 'completed'); + const toolMessages = final.messages.filter((m) => m.role === 'tool'); + // Exactly one tool response for the gated call (the executed one) — no placeholder. + assert.equal(toolMessages.length, 1); + assert.match(toolMessages[0].content, /restarted/); + assert.equal(toolMessages[0].toolCallId, 'c1'); + assert.equal(final.pendingApprovals[0].consumed, true); + }); + + it('records a denied approval as denied_by_operator without executing', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + let executed = 0; + const tool = { + def: { name: 'restart', description: 'restart', parameters: { type: 'object' } }, + destructive: true, + handler: async () => { + executed++; + return { restarted: true }; + }, + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [{ id: 'c1', name: 'restart', arguments: {} }], + }, + { content: 'pivoted', finishReason: 'stop' }, + ]); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + autoApprove: false, + }); + + const halted = await session.getSession(created.session_id); + await session.resolveApproval(created.session_id, halted.pendingApprovals[0].id, false); + + await runAgent({ + sessionId: created.session_id, + models, + tools: [tool], + scopes, + maxTurns: 5, + autoApprove: false, + }); + + assert.equal(executed, 0); + const final = await session.getSession(created.session_id); + assert.equal(final.status, 'completed'); + const toolMessages = final.messages.filter((m) => m.role === 'tool'); + assert.equal(toolMessages.length, 1); + assert.match(toolMessages[0].content, /denied_by_operator/); + }); + + it('stays paused until ALL gated calls in a turn are resolved (no partial-approval 400)', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + let executed = 0; + const dropTool = { + def: { name: 'drop', description: 'drop', parameters: { type: 'object' } }, + destructive: true, + handler: async () => { + executed++; + return { dropped: true }; + }, + }; + const models = stubModels([ + { + content: '', + finishReason: 'tool_calls', + toolCalls: [ + { id: 'c1', name: 'drop', arguments: { t: 'A' } }, + { id: 'c2', name: 'drop', arguments: { t: 'B' } }, + ], + }, + { content: 'done', finishReason: 'stop' }, + ]); + const run = { sessionId: created.session_id, models, tools: [dropTool], scopes, maxTurns: 5, autoApprove: false }; + + await runAgent(run); + let s = await session.getSession(created.session_id); + assert.equal(s.status, 'awaiting_approval'); + assert.equal(s.pendingApprovals.length, 2); + + // Approve only the first. The loop must NOT advance to generate() with one tool response missing. + await session.resolveApproval(created.session_id, s.pendingApprovals[0].id, true); + await runAgent(run); + s = await session.getSession(created.session_id); + assert.equal(executed, 1, 'first approved call executed'); + assert.equal(s.status, 'awaiting_approval', 'still paused on the second pending approval'); + + // Approve the second; now the loop can complete. + await session.resolveApproval(created.session_id, s.pendingApprovals[1].id, true); + await runAgent(run); + s = await session.getSession(created.session_id); + assert.equal(executed, 2); + assert.equal(s.status, 'completed'); + const toolMsgs = s.messages.filter((m) => m.role === 'tool'); + assert.deepEqual(toolMsgs.map((m) => m.toolCallId).sort(), ['c1', 'c2']); + }); + + it('preserves aborted status when signal aborts mid-generate', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'go', createdAt: Date.now() }); + const controller = new AbortController(); + const models = { + async generate(_input, _opts) { + // Caller aborts mid-call; honor the signal as a real backend would. + controller.abort(); + await session.setStatus(created.session_id, 'aborted'); + const err = new Error('AbortError'); + err.name = 'AbortError'; + throw err; + }, + }; + + await runAgent({ + sessionId: created.session_id, + models, + tools: noTools, + scopes, + maxTurns: 5, + signal: controller.signal, + }); + + const reloaded = await session.getSession(created.session_id); + assert.equal(reloaded.status, 'aborted'); + }); + + it('coalesces concurrent runs against the same session', async () => { + const created = await session.createSession({ user: 'admin' }); + await session.appendMessage(created.session_id, { role: 'user', content: 'one', createdAt: Date.now() }); + let calls = 0; + const models = { + async generate() { + calls++; + return { content: 'ok', finishReason: 'stop' }; + }, + }; + const a = runAgent({ sessionId: created.session_id, models, tools: noTools, scopes, maxTurns: 1 }); + const b = runAgent({ sessionId: created.session_id, models, tools: noTools, scopes, maxTurns: 1 }); + assert.equal(a, b); + await a; + assert.equal(calls, 1); + }); +}); diff --git a/unitTests/agent/session.test.js b/unitTests/agent/session.test.js new file mode 100644 index 0000000000..423433d83b --- /dev/null +++ b/unitTests/agent/session.test.js @@ -0,0 +1,160 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { + createSession, + getSession, + listSessions, + appendMessage, + setStatus, + addPendingApproval, + resolveApproval, + _setTableForTests, +} = require('#src/agent/session'); + +function makeMockTable() { + const store = new Map(); + return { + store, + primaryStore: { + async put(key, value) { + store.set(key, structuredClone(value)); + }, + async get(key) { + const value = store.get(key); + return value ? structuredClone(value) : undefined; + }, + getRange({ limit = Infinity, reverse } = {}) { + const entries = Array.from(store.entries()); + if (reverse) entries.reverse(); + return entries.slice(0, limit).map(([key, value]) => ({ key, value: structuredClone(value) })); + }, + }, + }; +} + +describe('agent/session', () => { + let mock; + + beforeEach(() => { + mock = makeMockTable(); + _setTableForTests(mock); + }); + + afterEach(() => { + _setTableForTests(undefined); + }); + + it('creates a session with an initial user message', async () => { + const session = await createSession({ + user: 'admin', + initialMessage: { role: 'user', content: 'hello', createdAt: Date.now() }, + }); + assert.equal(session.user, 'admin'); + assert.equal(session.status, 'idle'); + assert.equal(session.messages.length, 1); + assert.equal(session.messages[0].content, 'hello'); + assert.deepEqual(session.pendingApprovals, []); + const reloaded = await getSession(session.session_id); + assert.equal(reloaded.session_id, session.session_id); + }); + + it('appends messages and updates updatedAt', async () => { + const session = await createSession({ user: 'admin' }); + const initialUpdatedAt = session.updatedAt; + await new Promise((r) => setTimeout(r, 5)); + const updated = await appendMessage(session.session_id, { + role: 'assistant', + content: 'hi back', + createdAt: Date.now(), + }); + assert.equal(updated.messages.length, 1); + assert.equal(updated.messages[0].role, 'assistant'); + assert.ok(updated.updatedAt >= initialUpdatedAt); + }); + + it('rejects appendMessage for an unknown session', async () => { + await assert.rejects( + appendMessage('nope', { role: 'user', content: 'x', createdAt: Date.now() }), + /No agent session/ + ); + }); + + it('transitions through approval lifecycle', async () => { + const session = await createSession({ user: 'admin' }); + const approval = await addPendingApproval(session.session_id, { + toolName: 'drop_component', + arguments: { name: 'demo' }, + reason: 'destructive', + }); + assert.ok(approval.id); + assert.equal(approval.resolved, undefined); + + const afterAdd = await getSession(session.session_id); + assert.equal(afterAdd.status, 'awaiting_approval'); + assert.equal(afterAdd.pendingApprovals.length, 1); + + const resolved = await resolveApproval(session.session_id, approval.id, true); + assert.equal(resolved.resolved, true); + assert.equal(resolved.approved, true); + + const afterResolve = await getSession(session.session_id); + assert.equal(afterResolve.status, 'idle'); + }); + + it('returns to idle even when an approval is denied (deny is not abort)', async () => { + const sess = await createSession({ user: 'admin' }); + const approval = await addPendingApproval(sess.session_id, { + toolName: 'restart', + arguments: {}, + toolCallId: 'c1', + reason: 'destructive', + }); + await resolveApproval(sess.session_id, approval.id, false); + const reloaded = await getSession(sess.session_id); + assert.equal(reloaded.status, 'idle'); + }); + + it('rejects double-resolution of an approval', async () => { + const session = await createSession({ user: 'admin' }); + const approval = await addPendingApproval(session.session_id, { + toolName: 'restart', + arguments: {}, + reason: 'destructive', + }); + await resolveApproval(session.session_id, approval.id, true); + await assert.rejects(resolveApproval(session.session_id, approval.id, true), /already resolved/); + }); + + it('lists sessions in reverse insertion order', async () => { + const a = await createSession({ user: 'admin' }); + const b = await createSession({ user: 'admin' }); + const sessions = await listSessions({ limit: 10 }); + const ids = sessions.map((s) => s.session_id); + assert.ok(ids.includes(a.session_id)); + assert.ok(ids.includes(b.session_id)); + assert.equal(sessions[0].session_id, b.session_id); + }); + + it('serializes concurrent mutations on the same session (no lost updates)', async () => { + const session = await createSession({ user: 'admin' }); + // Fire several mutations concurrently. Without per-session serialization each would read the + // same snapshot and the last put would clobber the rest, losing messages. + await Promise.all([ + appendMessage(session.session_id, { role: 'user', content: 'a', createdAt: Date.now() }), + appendMessage(session.session_id, { role: 'assistant', content: 'b', createdAt: Date.now() }), + appendMessage(session.session_id, { role: 'user', content: 'c', createdAt: Date.now() }), + ]); + const reloaded = await getSession(session.session_id); + assert.equal(reloaded.messages.length, 3); + assert.deepEqual(reloaded.messages.map((m) => m.content).sort(), ['a', 'b', 'c']); + }); + + it('setStatus persists the new status and optional error', async () => { + const session = await createSession({ user: 'admin' }); + await setStatus(session.session_id, 'error', 'boom'); + const reloaded = await getSession(session.session_id); + assert.equal(reloaded.status, 'error'); + assert.equal(reloaded.lastError, 'boom'); + }); +}); diff --git a/utility/hdbTerms.ts b/utility/hdbTerms.ts index d1599190c2..a7426b2487 100644 --- a/utility/hdbTerms.ts +++ b/utility/hdbTerms.ts @@ -178,6 +178,7 @@ export const SYSTEM_TABLE_NAMES = { USER_TABLE_NAME: 'hdb_user', INFO_TABLE_NAME: 'hdb_info', DEPLOYMENT_TABLE_NAME: 'hdb_deployment', + AGENT_SESSION_TABLE_NAME: 'hdb_agent_session', } as const; /** Hash attribute for the system info table */ @@ -302,6 +303,12 @@ export const OPERATIONS_ENUM = { GET_DEPLOYMENT: 'get_deployment', GET_DEPLOYMENT_PAYLOAD: 'get_deployment_payload', DELETE_DEPLOYMENT_PAYLOAD: 'delete_deployment_payload', + AGENT_PROMPT: 'agent_prompt', + GET_AGENT_SESSION: 'get_agent_session', + LIST_AGENT_SESSIONS: 'list_agent_sessions', + CANCEL_AGENT_RUN: 'cancel_agent_run', + APPROVE_AGENT_ACTION: 'approve_agent_action', + SET_AGENT_CONFIG: 'set_agent_config', } as const; /** Defines valid file types that we are able to handle in 'import_from_s3' ops */ @@ -550,6 +557,15 @@ export const CONFIG_PARAMS = { MCP_APPLICATION_RATELIMIT_SESSIONPERSECOND: 'mcp_application_rateLimit_sessionPerSecond', MCP_SESSION_IDLETIMEOUTSECONDS: 'mcp_session_idleTimeoutSeconds', MCP_SESSION_ALLOWCLIENTDELETE: 'mcp_session_allowClientDelete', + AGENT_ENABLED: 'agent_enabled', + AGENT_PROVIDER: 'agent_provider', + AGENT_MODEL: 'agent_model', + AGENT_MAXTURNS: 'agent_maxTurns', + AGENT_MAXCOSTUSD: 'agent_maxCostUsd', + AGENT_AUTOAPPROVE: 'agent_autoApprove', + AGENT_ALLOWDESTRUCTIVE: 'agent_allowDestructive', + AGENT_USER: 'agent_user', + AGENT_COMPONENTSSCOPE: 'agent_componentsScope', REPLICATION: 'replication', REPLICATION_HOSTNAME: 'replication_hostname', REPLICATION_URL: 'replication_url',