From 8721ae8a102b3ebbe8719c09efde4984c76a365f Mon Sep 17 00:00:00 2001 From: hkonsti <45428767+hkonsti@users.noreply.github.com> Date: Thu, 22 Oct 2026 01:22:10 +0000 Subject: [PATCH 1/5] add a built-in agent: chat with an agent proxy runs itself, behind its own gate A Chat page, behind the `harness` experiment, talks to an agent login Personal Agent Proxy runs itself. It has grants and an activity log like any agent, and every call it makes goes through the same MCP tool dispatch as Claude's, for the conversation's task. The model is Claude Sonnet 5.5 over OpenRouter. Everything lives in the database: a thread per conversation, an append-only log of entries (the person, the model with its tool calls, each call's result), and a job per unit of work, inference on a log or one tool call. A worker in each api instance claims jobs with a lease and finishes them in the same transaction as what they write, fenced by the lease, so an instance that dies mid-job loses nothing: the next worker picks the job up. A write that was interrupted is never made again; the model is told its outcome is unknown. One tool beyond the connections, sleep_until, is a job due later. The web app gets a second sidebar beside the icon rail for the conversations, and a page mode that scrolls a part of itself so the header, sidebars and input stay put. --- packages/api/src/activity/route.ts | 6 +- packages/api/src/agent-side/tools.ts | 5 +- packages/api/src/agents/create-agent-login.ts | 4 + packages/api/src/harness/built-in-agent.ts | 34 ++ packages/api/src/harness/inference.test.ts | 126 ++++++ packages/api/src/harness/inference.ts | 134 ++++++ packages/api/src/harness/model.test.ts | 65 +++ packages/api/src/harness/model.ts | 83 ++++ packages/api/src/harness/route.test.ts | 92 ++++ packages/api/src/harness/route.ts | 97 ++++ packages/api/src/harness/thread-task.ts | 43 ++ packages/api/src/harness/tool-job.test.ts | 102 +++++ packages/api/src/harness/tool-job.ts | 94 ++++ packages/api/src/harness/tool-surface.test.ts | 41 ++ packages/api/src/harness/tool-surface.ts | 56 +++ packages/api/src/harness/worker.test.ts | 67 +++ packages/api/src/harness/worker.ts | 95 ++++ packages/api/src/index.ts | 6 + packages/api/src/mcp/tools.ts | 28 +- packages/api/src/server/create-api-router.ts | 9 + packages/api/src/utils/env-schema.ts | 2 +- packages/db/package.json | 1 + .../migration.sql | 83 ++++ packages/db/prisma/schema.prisma | 104 ++++- packages/db/src/agent.ts | 15 + packages/db/src/harness.ts | 427 ++++++++++++++++++ packages/docs/content/agents/chat.mdx | 40 ++ packages/docs/src/nav.ts | 2 +- .../integrations/src/agent-providers.test.ts | 19 +- packages/integrations/src/agent-providers.ts | 26 +- packages/integrations/src/index.ts | 1 + packages/ui/src/styles.css | 2 +- packages/utils/package.json | 1 + packages/utils/src/experiments.ts | 3 + packages/utils/src/harness.test.ts | 68 +++ packages/utils/src/harness.ts | 74 +++ packages/web/src/client/harness-client.ts | 64 +++ packages/web/src/client/schemas.ts | 6 +- packages/web/src/components/app-shell.tsx | 71 ++- .../web/src/components/audit-entry-dialog.tsx | 5 +- packages/web/src/components/nav.tsx | 14 +- packages/web/src/loaders.ts | 18 + packages/web/src/pages/agent-page.tsx | 7 +- packages/web/src/pages/chat-page.tsx | 347 ++++++++++++++ packages/web/src/router.tsx | 6 + 45 files changed, 2561 insertions(+), 32 deletions(-) create mode 100644 packages/api/src/harness/built-in-agent.ts create mode 100644 packages/api/src/harness/inference.test.ts create mode 100644 packages/api/src/harness/inference.ts create mode 100644 packages/api/src/harness/model.test.ts create mode 100644 packages/api/src/harness/model.ts create mode 100644 packages/api/src/harness/route.test.ts create mode 100644 packages/api/src/harness/route.ts create mode 100644 packages/api/src/harness/thread-task.ts create mode 100644 packages/api/src/harness/tool-job.test.ts create mode 100644 packages/api/src/harness/tool-job.ts create mode 100644 packages/api/src/harness/tool-surface.test.ts create mode 100644 packages/api/src/harness/tool-surface.ts create mode 100644 packages/api/src/harness/worker.test.ts create mode 100644 packages/api/src/harness/worker.ts create mode 100644 packages/db/prisma/migrations/v024_add_built_in_agent_threads/migration.sql create mode 100644 packages/db/src/harness.ts create mode 100644 packages/docs/content/agents/chat.mdx create mode 100644 packages/utils/src/harness.test.ts create mode 100644 packages/utils/src/harness.ts create mode 100644 packages/web/src/client/harness-client.ts create mode 100644 packages/web/src/pages/chat-page.tsx diff --git a/packages/api/src/activity/route.ts b/packages/api/src/activity/route.ts index 71914980..d4f14af0 100644 --- a/packages/api/src/activity/route.ts +++ b/packages/api/src/activity/route.ts @@ -22,8 +22,8 @@ export type AuditEntryResponse = { recordTitle: string | null; query: string | null; outcome: 'allowed' | 'denied'; - // The web app's agent pages, or the MCP server. - via: 'web' | 'mcp'; + // The web app's agent pages, the MCP server, or the built-in agent's own loop. + via: 'web' | 'mcp' | 'harness'; // The agent's task the request was made for, among the answer's `tasks`; none for entries from // before tasks, and for what Personal Agent Proxy did itself. taskId: string | null; @@ -35,7 +35,7 @@ export type ActivityTaskResponse = { agentId: string; title: string; goal: string; - via: 'web' | 'mcp'; + via: 'web' | 'mcp' | 'harness'; startedAt: string; // Null while the agent hasn't called it finished, abandoned (@proxy/utils/tasks) or not. finishedAt: string | null; diff --git a/packages/api/src/agent-side/tools.ts b/packages/api/src/agent-side/tools.ts index 42ce3166..54893d6b 100644 --- a/packages/api/src/agent-side/tools.ts +++ b/packages/api/src/agent-side/tools.ts @@ -110,8 +110,9 @@ export function listAgentTools(signedIn: SignedInAgent, connectionId: string): P }); } -// Who calls and which way it came in: the web app's agent pages, or the MCP server. -export type AgentCaller = {agent: SignedInAgent; via: 'web' | 'mcp'}; +// Who calls and which way it came in: the web app's agent pages, the MCP server, or the built-in +// agent's own loop. +export type AgentCaller = {agent: SignedInAgent; via: 'web' | 'mcp' | 'harness'}; /** Who runs a tool, and the task it is run for, by its id. */ // PRE-TASK CLIENTS: null for a pre-task client's call (mcp/pre-task-clients.ts). diff --git a/packages/api/src/agents/create-agent-login.ts b/packages/api/src/agents/create-agent-login.ts index dbf3b087..d7062879 100644 --- a/packages/api/src/agents/create-agent-login.ts +++ b/packages/api/src/agents/create-agent-login.ts @@ -32,6 +32,10 @@ function nameFor(provider: AgentProvider, name: string | undefined): Result> { return Do(async ($) => { const provider = $(requirePresent(findAgentProvider(providerId), ApiErr.parseError(`Unknown agent ${providerId}`))); + // The built-in agent is made by Personal Agent Proxy itself (harness/built-in-agent.ts), never handed out. + if (provider.connects === 'built-in') { + return $(Err(ApiErr.parseError(`${provider.name} is not an agent a login is made for`))); + } const name = $(nameFor(provider, givenName)); const password = generatePassword(); const passwordHash = await Bun.password.hash(password); diff --git a/packages/api/src/harness/built-in-agent.ts b/packages/api/src/harness/built-in-agent.ts new file mode 100644 index 00000000..22b13dc7 --- /dev/null +++ b/packages/api/src/harness/built-in-agent.ts @@ -0,0 +1,34 @@ +import {randomBytes} from 'node:crypto'; + +import type {Result} from 'ts-results-es'; + +import {ApiErr, type ApiError, Do, requirePresent} from '@proxy/utils'; +import {createAgent, getSignedInAgentByProvider, isProviderConflict, type SignedInAgent} from '@proxy/db/agent'; +import {BUILT_IN_AGENT_PROVIDER} from '@proxy/integrations'; + +/** + * The organization's built-in agent: an Agent like any other, with grants and an activity log of + * its own, so the person sets what it may do where they set it for Claude. Made the first time it + * is needed, with a password nobody knows, since it never signs in: its loop runs in the api. + */ +export function requireBuiltInAgent(orgId: string): Promise> { + return Do(async ($) => { + const existing = $(await getSignedInAgentByProvider(orgId, BUILT_IN_AGENT_PROVIDER.id)); + if (existing) { + return existing; + } + const created = await createAgent({ + orgId, + providerId: BUILT_IN_AGENT_PROVIDER.id, + name: BUILT_IN_AGENT_PROVIDER.name, + username: `proxy-${randomBytes(6).toString('hex')}`, + passwordHash: await Bun.password.hash(randomBytes(32).toString('base64url')), + }); + // Two first chats at once: the other one made it. + if (created.isErr() && !isProviderConflict(created.error)) { + return $(created); + } + const made = $(await getSignedInAgentByProvider(orgId, BUILT_IN_AGENT_PROVIDER.id)); + return $(requirePresent(made, ApiErr.internalError(new Error('The built-in agent was made and is not there')))); + }); +} diff --git a/packages/api/src/harness/inference.test.ts b/packages/api/src/harness/inference.test.ts new file mode 100644 index 00000000..cd9a1a13 --- /dev/null +++ b/packages/api/src/harness/inference.test.ts @@ -0,0 +1,126 @@ +import {beforeEach, describe, expect, mock, test} from 'bun:test'; +import {Err, Ok} from 'ts-results-es'; + +import {ApiErr} from '@proxy/utils'; + +const mockEnv: {OPENROUTER_KEY?: string} = {}; +mock.module('../utils/env', () => ({env: mockEnv})); + +const getJobThread = mock(); +const listEntries = mock(); +const finishIdleInference = mock(async () => Ok(true)); +const finishJob = mock(async () => Ok(true)); +const completeInference = mock(async () => Ok(true)); +const requeueJob = mock(async () => Ok(true)); +mock.module('@proxy/db/harness', () => ({getJobThread, listEntries, finishIdleInference, finishJob, completeInference, requeueJob})); + +const requireBuiltInAgent = mock(async () => Ok({agentId: 'agent-1', orgId: 'org-1', providerId: 'proxy', name: 'Personal Agent Proxy', experiments: []})); +mock.module('./built-in-agent', () => ({requireBuiltInAgent})); + +const ensureThreadTask = mock(async () => Ok('task-1')); +mock.module('./thread-task', () => ({ensureThreadTask})); + +const agentInstructions = mock(async (_agent: unknown, flow: string) => Ok(`You are signed in.\n\n${flow}`)); +mock.module('../mcp/tools', () => ({agentInstructions, MCP_TOOLS: []})); + +const askModel = mock(); +mock.module('./model', () => ({askModel})); + +mock.module('./tool-surface', () => ({MODEL_TOOLS: [{name: 'read', description: 'Reads', parameters: {}}]})); + +const now = new Date('2026-10-21T10:00:00Z'); +const thread = {id: 'thread-1', orgId: 'org-1', agentId: 'agent-1', taskId: 'task-1', createdAt: now}; +const job = {id: 'job-1', threadId: 'thread-1', kind: 'inference' as const, status: 'running' as const, entryId: null, callId: null, runAt: now, claimedBy: 'w1', attempts: 1}; + +function entry(kind: 'person' | 'assistant' | 'tool_result', text: string, extra: Partial<{toolCalls: Array<{id: string; name: string; arguments: string}>; callId: string}> = {}) { + return {id: `e-${text}`, seq: 1, threadId: 'thread-1', kind, text, toolCalls: extra.toolCalls ?? null, callId: extra.callId ?? null, isError: false, inputTokens: null, outputTokens: null, createdAt: now}; +} + +beforeEach(() => { + mock.clearAllMocks(); + mockEnv.OPENROUTER_KEY = 'sk-test'; + getJobThread.mockResolvedValue(Ok(thread)); + listEntries.mockResolvedValue(Ok([entry('person', 'Hi')])); + askModel.mockResolvedValue(Ok({text: 'Hello', toolCalls: [], inputTokens: 10, outputTokens: 2})); +}); + +describe('runInference', () => { + test('runs the model on the log, as the chat form, and writes its answer with a job per call', async () => { + const {runInference} = await import('./inference'); + listEntries.mockResolvedValue( + Ok([ + entry('person', 'Hi'), + entry('assistant', '', {toolCalls: [{id: 'c1', name: 'read', arguments: '{}'}]}), + entry('tool_result', '{"records":[]}', {callId: 'c1'}), + ]), + ); + askModel.mockResolvedValue(Ok({text: '', toolCalls: [{id: 'c2', name: 'sleep_until', arguments: '{"seconds":60}'}, {id: 'c3', name: 'read', arguments: '{}'}], inputTokens: 10, outputTokens: 2})); + + const result = await runInference(job, 'w1', now); + + expect(result.isOk()).toBe(true); + expect(ensureThreadTask).toHaveBeenCalledWith(thread, now); + const request = askModel.mock.calls[0]?.[0] as {messages: Array>; tools: unknown[]}; + expect(request.messages[0]).toMatchObject({role: 'system'}); + expect(String(request.messages[0]?.content)).toContain('You are talking with the person directly'); + expect(String(request.messages[0]?.content)).toContain('The time now is 2026-10-21T10:00:00.000Z'); + expect(request.messages.slice(1)).toEqual([ + {role: 'user', content: 'Hi'}, + {role: 'assistant', content: null, tool_calls: [{id: 'c1', type: 'function', function: {name: 'read', arguments: '{}'}}]}, + {role: 'tool', tool_call_id: 'c1', content: '{"records":[]}'}, + ]); + expect(request.tools).toEqual([{name: 'read', description: 'Reads', parameters: {}}]); + expect(completeInference).toHaveBeenCalledWith( + 'job-1', + 'w1', + 'thread-1', + {text: '', toolCalls: [{id: 'c2', name: 'sleep_until', arguments: '{"seconds":60}'}, {id: 'c3', name: 'read', arguments: '{}'}], inputTokens: 10, outputTokens: 2}, + {c2: new Date('2026-10-21T10:01:00Z')}, + ); + }); + + test('a log with a call still running is left alone, and a thread that is gone ends the job', async () => { + const {runInference} = await import('./inference'); + listEntries.mockResolvedValue(Ok([entry('person', 'Hi'), entry('assistant', '', {toolCalls: [{id: 'c1', name: 'read', arguments: '{}'}]})])); + await runInference(job, 'w1', now); + expect(finishIdleInference).toHaveBeenCalledWith('job-1', 'w1', 'thread-1'); + expect(askModel).not.toHaveBeenCalled(); + + getJobThread.mockResolvedValue(Ok(null)); + await runInference(job, 'w1', now); + expect(finishJob).toHaveBeenCalledWith('job-1', 'w1'); + }); + + test('without a key, the person is told the server is not set up', async () => { + const {runInference} = await import('./inference'); + delete mockEnv.OPENROUTER_KEY; + + await runInference(job, 'w1', now); + + expect(askModel).not.toHaveBeenCalled(); + expect(completeInference).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', expect.objectContaining({text: expect.stringContaining('OPENROUTER_KEY'), toolCalls: []}), {}); + }); + + test('a model that can’t be reached is tried again later, and the person told after the last try', async () => { + const {runInference} = await import('./inference'); + askModel.mockResolvedValue(Err(ApiErr.providerUnreachable(new Error('down')))); + + await runInference(job, 'w1', now); + expect(requeueJob).toHaveBeenCalledWith('job-1', 'w1', new Date('2026-10-21T10:00:05Z')); + expect(completeInference).not.toHaveBeenCalled(); + + await runInference({...job, attempts: 3}, 'w1', now); + expect(completeInference).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', expect.objectContaining({text: expect.stringContaining('could not be reached')}), {}); + }); + + test('after thirty answers without the person, it stops and says so', async () => { + const {runInference} = await import('./inference'); + const alone = [entry('person', 'Go'), ...Array.from({length: 30}, (_, i) => entry('assistant', `step ${i}`))]; + listEntries.mockResolvedValue(Ok([...alone, entry('tool_result', 'x', {callId: 'none'})])); + + await runInference(job, 'w1', now); + + expect(askModel).not.toHaveBeenCalled(); + expect(completeInference).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', expect.objectContaining({text: expect.stringContaining('30 steps')}), {}); + }); +}); diff --git a/packages/api/src/harness/inference.ts b/packages/api/src/harness/inference.ts new file mode 100644 index 00000000..966db755 --- /dev/null +++ b/packages/api/src/harness/inference.ts @@ -0,0 +1,134 @@ +import {Err, Result} from 'ts-results-es'; + +import {ApiErr, type ApiError, Do} from '@proxy/utils'; +import {readyForInference, SLEEP_TOOL, sleepUntil, type ToolCall} from '@proxy/utils/harness'; +import {completeInference, type EntryRow, finishIdleInference, finishJob, getJobThread, type JobRow, listEntries, requeueJob, type ThreadRow} from '@proxy/db/harness'; + +import {agentInstructions} from '../mcp/tools'; +import {log, serializeError} from '../observability/log'; +import {env} from '../utils/env'; +import {requireBuiltInAgent} from './built-in-agent'; +import {askModel, type ModelMessage} from './model'; +import {ensureThreadTask} from './thread-task'; +import {MODEL_TOOLS} from './tool-surface'; + +/** + * An inference job: the model, run once on a thread's log. Its answer goes in the log as an + * assistant entry, with a tool job for each call it made; a sleep's job is due when the sleep + * ends. The log not ready, as when the person spoke while a call was still running, is left for + * the job the last result queues. A model that can't be reached is tried again a few times, then + * the person is told. + */ + +// How the built-in agent is to go about it, where an MCP client is told about tasks. +const FLOW = + 'You are talking with the person directly, on the Chat page of Personal Agent Proxy; what you write is shown to them as it is, so write plain text, short, and answer what they asked. list_connections lists the accounts you can use and what you may do with each. list_tools with a connection gives that connection’s tools and their params. read runs the tools that only read; run runs the ones that change something or reach other people. Every call is made for this conversation’s task, filled in for you: there is nothing to start or finish. When you have called sleep_until, you are woken at that time with nothing but this conversation, so say in your message what you will do then.'; + +// How many answers the model may give in a row without the person saying anything, calls and +// wake-ups included, before it stops and waits for them. +const MAX_TURNS_ALONE = 30; + +// How often a model that can't be reached is tried, and how long to wait between tries. +const MAX_MODEL_ATTEMPTS = 3; +const RETRY_MS = 5_000; + +const MAX_TOKENS = 4000; +const TIMEOUT_MS = 120_000; + +const NOT_SET_UP = 'Chat is not set up on this server: it needs an OpenRouter key (OPENROUTER_KEY). Everything else works as before.'; +const UNREACHABLE = 'The model could not be reached. Say something to try again.'; +const TOO_LONG = `I have taken ${MAX_TURNS_ALONE} steps without hearing from you, so I have stopped. Tell me to go on if you want me to.`; + +// The log as the model reads it: a tool result as the answer to its call, in the chat form. +function toMessages(entries: EntryRow[]): ModelMessage[] { + return entries.map((entry): ModelMessage => { + if (entry.kind === 'person') { + return {role: 'user', content: entry.text}; + } + if (entry.kind === 'tool_result') { + return {role: 'tool', tool_call_id: entry.callId ?? '', content: entry.text}; + } + const calls = entry.toolCalls ?? []; + return { + role: 'assistant', + content: entry.text === '' && calls.length > 0 ? null : entry.text, + ...(calls.length > 0 ? {tool_calls: calls.map((call) => ({id: call.id, type: 'function' as const, function: {name: call.name, arguments: call.arguments}}))} : {}), + }; + }); +} + +// How many answers the model has given since the person last spoke. +function turnsAlone(entries: EntryRow[]): number { + const lastPerson = entries.findLastIndex((entry) => entry.kind === 'person'); + return entries.slice(lastPerson + 1).filter((entry) => entry.kind === 'assistant').length; +} + +// When each call's job is due: a sleep's when it ends, every other's now. +function dueTimes(calls: ToolCall[], now: Date): Record { + return Object.fromEntries( + calls.flatMap((call) => { + if (call.name !== SLEEP_TOOL) { + return []; + } + const parsed = Result.wrap((): unknown => JSON.parse(call.arguments)); + return [[call.id, (parsed.isOk() ? sleepUntil(parsed.value, now) : null) ?? now]]; + }), + ); +} + +// Answers the person with `text` alone, as the model would, when the model can't be run. +function answerWith(job: JobRow, workerId: string, text: string): Promise> { + return completeInference(job.id, workerId, job.threadId, {text, toolCalls: [], inputTokens: 0, outputTokens: 0}, {}); +} + +function systemPrompt(thread: ThreadRow, now: Date): Promise> { + return Do(async ($) => { + const signedIn = $(await requireBuiltInAgent(thread.orgId)); + const instructions = $(await agentInstructions(signedIn, FLOW)); + return `${instructions}\n\nThe time now is ${now.toISOString()}.`; + }); +} + +/** Runs the job: one turn of the model on its thread's log, written back; nothing when the log isn't ready. */ +export function runInference(job: JobRow, workerId: string, now = new Date()): Promise> { + return Do(async ($) => { + const thread = $(await getJobThread(job.threadId)); + if (!thread) { + $(await finishJob(job.id, workerId)); + return; + } + const entries = $(await listEntries(thread.id)); + if (!readyForInference(entries)) { + $(await finishIdleInference(job.id, workerId, thread.id)); + return; + } + if (turnsAlone(entries) >= MAX_TURNS_ALONE) { + $(await answerWith(job, workerId, TOO_LONG)); + return; + } + const key = env.OPENROUTER_KEY; + if (!key) { + $(await answerWith(job, workerId, NOT_SET_UP)); + return; + } + $(await ensureThreadTask(thread, now)); + const system = $(await systemPrompt(thread, now)); + const turn = await askModel({key, messages: [{role: 'system', content: system}, ...toMessages(entries)], tools: MODEL_TOOLS, maxTokens: MAX_TOKENS, timeoutMs: TIMEOUT_MS}); + if (turn.isErr() && turn.error.kind !== 'provider_unreachable') { + return $(Err(turn.error)); + } + if (turn.isErr()) { + log.warn('The built-in agent’s model could not be reached', {attempt: job.attempts, ...serializeError('cause' in turn.error ? turn.error.cause : turn.error)}); + if (job.attempts < MAX_MODEL_ATTEMPTS) { + $(await requeueJob(job.id, workerId, new Date(now.getTime() + RETRY_MS * job.attempts))); + return; + } + $(await answerWith(job, workerId, UNREACHABLE)); + return; + } + const written = $(await completeInference(job.id, workerId, thread.id, turn.value, dueTimes(turn.value.toolCalls, now))); + if (!written) { + return $(Err(ApiErr.conflict('The job was no longer this worker’s'))); + } + }); +} diff --git a/packages/api/src/harness/model.test.ts b/packages/api/src/harness/model.test.ts new file mode 100644 index 00000000..ceabef41 --- /dev/null +++ b/packages/api/src/harness/model.test.ts @@ -0,0 +1,65 @@ +import {afterAll, beforeEach, describe, expect, spyOn, test} from 'bun:test'; + +import {askModel, MODEL} from './model'; + +// What OpenRouter answers, and with what status. +let status = 200; +let message: Record = {content: 'Hello'}; +const fetchSpy = spyOn(globalThis, 'fetch').mockImplementation((async () => + status === 200 ? Response.json({choices: [{message}], usage: {prompt_tokens: 120, completion_tokens: 8}}) : Response.json({error: {message: 'down'}}, {status})) as unknown as typeof fetch); + +afterAll(() => { + fetchSpy.mockRestore(); +}); + +beforeEach(() => { + fetchSpy.mockClear(); + status = 200; + message = {content: 'Hello'}; +}); + +const request = { + key: 'sk-test', + messages: [ + {role: 'system' as const, content: 'Be brief.'}, + {role: 'user' as const, content: 'Hi'}, + ], + tools: [{name: 'read', description: 'Reads', parameters: {type: 'object', properties: {}}}], + maxTokens: 100, + timeoutMs: 1000, +}; + +// The body of the request made to OpenRouter. +function sent(): {model: string; messages: unknown[]; tools: unknown[]; max_tokens: number; provider: {data_collection: string}} { + return JSON.parse(String(fetchSpy.mock.calls[0]?.[1]?.body)); +} + +describe('askModel', () => { + test('asks Claude Sonnet 5.5 over OpenRouter with the tools, on a host that keeps nothing', async () => { + const turn = await askModel(request); + + expect(turn.unwrap()).toEqual({text: 'Hello', toolCalls: [], inputTokens: 120, outputTokens: 8}); + expect(String(fetchSpy.mock.calls[0]?.[0])).toBe('https://openrouter.ai/api/v1/chat/completions'); + expect((fetchSpy.mock.calls[0]?.[1]?.headers as Record).authorization).toBe('Bearer sk-test'); + const body = sent(); + expect(body.model).toBe(MODEL); + expect(body.messages).toEqual(request.messages); + expect(body.tools).toEqual([{type: 'function', function: {name: 'read', description: 'Reads', parameters: {type: 'object', properties: {}}}}]); + expect(body.max_tokens).toBe(100); + expect(body.provider.data_collection).toBe('deny'); + }); + + test('reads the calls the model made, with their arguments as it wrote them', async () => { + message = {content: null, tool_calls: [{id: 'call-1', type: 'function', function: {name: 'read', arguments: '{"connection":"c1"}'}}]}; + const turn = await askModel(request); + + expect(turn.unwrap()).toEqual({text: '', toolCalls: [{id: 'call-1', name: 'read', arguments: '{"connection":"c1"}'}], inputTokens: 120, outputTokens: 8}); + }); + + test('a provider that answers with an error is unreachable', async () => { + status = 502; + const turn = await askModel(request); + + expect(turn.unwrapErr().kind).toBe('provider_unreachable'); + }); +}); diff --git a/packages/api/src/harness/model.ts b/packages/api/src/harness/model.ts new file mode 100644 index 00000000..187cddca --- /dev/null +++ b/packages/api/src/harness/model.ts @@ -0,0 +1,83 @@ +import {Err, type Result} from 'ts-results-es'; +import {z} from 'zod'; + +import {ApiErr, type ApiError, Do, httpRequest} from '@proxy/utils'; +import type {ToolCall} from '@proxy/utils/harness'; + +/** + * The built-in agent's model: Claude Sonnet 5.5 over OpenRouter, answering a conversation with + * text, tool calls, or both, in the chat completions form. One call per turn; the loop + * (inference.ts) runs the calls and asks again. + */ + +const OPENROUTER_URL = 'https://openrouter.ai/api/v1/chat/completions'; + +export const MODEL = 'anthropic/claude-sonnet-5.5'; + +/** A tool as the model is given it: its name, what it does, and the JSON Schema of its arguments. */ +export type ModelTool = {name: string; description: string; parameters: object}; + +/** The conversation as the model reads it, in the chat completions form. */ +export type ModelMessage = + | {role: 'system'; content: string} + | {role: 'user'; content: string} + | {role: 'assistant'; content: string | null; tool_calls?: Array<{id: string; type: 'function'; function: {name: string; arguments: string}}>} + | {role: 'tool'; tool_call_id: string; content: string}; + +export type ModelRequest = { + key: string; + messages: ModelMessage[]; + tools: ModelTool[]; + maxTokens: number; + timeoutMs: number; +}; + +/** What the model answered: its text, the calls it made, and what the turn cost. */ +export type ModelTurn = {text: string; toolCalls: ToolCall[]; inputTokens: number; outputTokens: number}; + +const completionSchema = z.object({ + choices: z + .array( + z.object({ + message: z.object({ + content: z.string().nullable().optional(), + tool_calls: z.array(z.object({id: z.string(), function: z.object({name: z.string(), arguments: z.string()})})).optional(), + }), + }), + ) + .min(1), + usage: z.object({prompt_tokens: z.number(), completion_tokens: z.number()}).optional(), +}); + +/** One turn of the model on the conversation. */ +export function askModel(request: ModelRequest): Promise> { + return Do(async ($) => { + const answered = await httpRequest( + OPENROUTER_URL, + { + method: 'POST', + headers: {'content-type': 'application/json', authorization: `Bearer ${request.key}`}, + body: JSON.stringify({ + model: MODEL, + messages: request.messages, + tools: request.tools.map((tool) => ({type: 'function', function: {name: tool.name, description: tool.description, parameters: tool.parameters}})), + max_tokens: request.maxTokens, + // Only a host that neither keeps prompts nor trains on them: the conversation is + // the person's, and what their email and notes say is in it. + provider: {data_collection: 'deny'}, + }), + }, + {schema: completionSchema, timeoutMs: request.timeoutMs}, + ); + if (answered.isErr()) { + return $(Err(ApiErr.providerUnreachable(answered.error))); + } + const {message} = answered.value.choices[0] ?? {}; + return { + text: message?.content ?? '', + toolCalls: (message?.tool_calls ?? []).map((call) => ({id: call.id, name: call.function.name, arguments: call.function.arguments})), + inputTokens: answered.value.usage?.prompt_tokens ?? 0, + outputTokens: answered.value.usage?.completion_tokens ?? 0, + }; + }); +} diff --git a/packages/api/src/harness/route.test.ts b/packages/api/src/harness/route.test.ts new file mode 100644 index 00000000..d3110037 --- /dev/null +++ b/packages/api/src/harness/route.test.ts @@ -0,0 +1,92 @@ +import {beforeEach, describe, expect, mock, test} from 'bun:test'; +import {Ok} from 'ts-results-es'; + +const listThreads = mock(); +const createThread = mock(); +const getThread = mock(); +const listEntries = mock(); +const listOpenJobs = mock(); +const appendPersonMessage = mock(); +mock.module('@proxy/db/harness', () => ({listThreads, createThread, getThread, listEntries, listOpenJobs, appendPersonMessage})); + +mock.module('../utils/user-org', () => ({requireUserOrgId: mock(async () => Ok('org-1'))})); + +const requireBuiltInAgent = mock(async () => Ok({agentId: 'agent-1', orgId: 'org-1', providerId: 'proxy', name: 'Personal Agent Proxy', experiments: ['harness']})); +mock.module('./built-in-agent', () => ({requireBuiltInAgent})); + +const createThreadTask = mock(async () => Ok('task-1')); +mock.module('./thread-task', () => ({createThreadTask})); + +const now = new Date('2026-10-21T10:00:00Z'); +const thread = {id: 'thread-1', orgId: 'org-1', agentId: 'agent-1', taskId: 'task-1', createdAt: now}; + +function request(params: Record = {}, body: unknown = undefined, experiments: string[] = ['harness']) { + return {user: {userId: 'user-1', email: 'a@example.com', experiments}, params, body} as never; +} + +beforeEach(() => { + mock.clearAllMocks(); + listThreads.mockResolvedValue(Ok([{...thread, title: 'Hi'}])); + createThread.mockResolvedValue(Ok(thread)); + getThread.mockResolvedValue(Ok(thread)); + listEntries.mockResolvedValue(Ok([{id: 'e1', seq: 1, threadId: 'thread-1', kind: 'person', text: 'Hi', toolCalls: null, callId: null, isError: false, inputTokens: null, outputTokens: null, createdAt: now}])); + listOpenJobs.mockResolvedValue(Ok([])); + appendPersonMessage.mockResolvedValue(Ok({id: 'e2', seq: 2, threadId: 'thread-1', kind: 'person', text: 'Hello', toolCalls: null, callId: null, isError: false, inputTokens: null, outputTokens: null, createdAt: now})); +}); + +describe('the chat routes', () => { + test('are refused without the experiment', async () => { + const {handleListThreadsRoute, handlePostMessageRoute} = await import('./route'); + + expect((await handleListThreadsRoute(request({}, undefined, []))).unwrapErr().kind).toBe('forbidden'); + expect((await handlePostMessageRoute(request({threadId: 'thread-1'}, {text: 'Hi'}, []))).unwrapErr().kind).toBe('forbidden'); + expect(appendPersonMessage).not.toHaveBeenCalled(); + }); + + test('list the threads with what the person first said as the title', async () => { + const {handleListThreadsRoute} = await import('./route'); + + expect((await handleListThreadsRoute(request())).unwrap()).toEqual({threads: [{id: 'thread-1', createdAt: '2026-10-21T10:00:00.000Z', taskId: 'task-1', title: 'Hi'}]}); + }); + + test('make a thread for the built-in agent, with a task for its calls', async () => { + const {handleCreateThreadRoute} = await import('./route'); + + const made = await handleCreateThreadRoute(request()); + + expect(requireBuiltInAgent).toHaveBeenCalledWith('org-1'); + expect(createThreadTask).toHaveBeenCalledWith('org-1', 'agent-1'); + expect(createThread).toHaveBeenCalledWith({orgId: 'org-1', agentId: 'agent-1', taskId: 'task-1'}); + expect(made.unwrap().thread).toEqual({id: 'thread-1', createdAt: '2026-10-21T10:00:00.000Z', taskId: 'task-1', title: null}); + }); + + test('give the log with whether the agent is busy or asleep', async () => { + const {handleGetThreadRoute} = await import('./route'); + const soon = new Date(Date.now() + 60_000); + const base = {id: 'j1', threadId: 'thread-1', kind: 'tool' as const, status: 'queued' as const, entryId: 'e1', callId: 'c1', claimedBy: null, attempts: 0}; + + const idle = (await handleGetThreadRoute(request({threadId: 'thread-1'}))).unwrap(); + expect(idle.entries).toEqual([{id: 'e1', kind: 'person', text: 'Hi', toolCalls: null, callId: null, isError: false, at: '2026-10-21T10:00:00.000Z'}]); + expect(idle).toMatchObject({busy: false, sleepingUntil: null}); + + listOpenJobs.mockResolvedValue(Ok([{...base, runAt: new Date(Date.now() - 1000)}])); + expect((await handleGetThreadRoute(request({threadId: 'thread-1'}))).unwrap()).toMatchObject({busy: true, sleepingUntil: null}); + + listOpenJobs.mockResolvedValue(Ok([{...base, runAt: soon}])); + expect((await handleGetThreadRoute(request({threadId: 'thread-1'}))).unwrap()).toMatchObject({busy: false, sleepingUntil: soon.toISOString()}); + + getThread.mockResolvedValue(Ok(null)); + expect((await handleGetThreadRoute(request({threadId: 'nope'}))).unwrapErr()).toMatchObject({kind: 'not_found', resource: 'thread'}); + }); + + test('append what the person says, trimmed, and refuse nothing or too much', async () => { + const {handlePostMessageRoute} = await import('./route'); + + const posted = await handlePostMessageRoute(request({threadId: 'thread-1'}, {text: ' Hello \n'})); + expect(appendPersonMessage).toHaveBeenCalledWith('thread-1', 'Hello'); + expect(posted.unwrap().entry).toMatchObject({id: 'e2', kind: 'person', text: 'Hello'}); + + expect((await handlePostMessageRoute(request({threadId: 'thread-1'}, {text: ' '}))).unwrapErr().kind).toBe('parse_error'); + expect((await handlePostMessageRoute(request({threadId: 'thread-1'}, {text: 'x'.repeat(20_001)}))).unwrapErr().kind).toBe('parse_error'); + }); +}); diff --git a/packages/api/src/harness/route.ts b/packages/api/src/harness/route.ts new file mode 100644 index 00000000..e52df3d4 --- /dev/null +++ b/packages/api/src/harness/route.ts @@ -0,0 +1,97 @@ +import {Err, type Result} from 'ts-results-es'; +import {z} from 'zod'; + +import {ApiErr, type ApiError, Do, hasExperiment, parseSchema, requirePresent} from '@proxy/utils'; +import type {ToolCall} from '@proxy/utils/harness'; +import {appendPersonMessage, createThread, type EntryRow, getThread, listEntries, listOpenJobs, listThreads, type ThreadRow} from '@proxy/db/harness'; + +import type {AuthenticatedRequest} from '../server/middleware/require-auth'; +import {requireUserOrgId} from '../utils/user-org'; +import {requireBuiltInAgent} from './built-in-agent'; +import {createThreadTask} from './thread-task'; + +/** + * The Chat page's routes: the person's threads with the built-in agent, one thread's log, and + * what they say in it. Behind the `harness` experiment. Nothing here runs the model: a message + * queues a job, and the worker (worker.ts) does the rest; the page reads the log again as it goes. + */ + +const MAX_MESSAGE = 20_000; + +export type ThreadResponse = {id: string; createdAt: string; taskId: string; title: string | null}; + +export type EntryResponse = {id: string; kind: 'person' | 'assistant' | 'tool_result'; text: string; toolCalls: ToolCall[] | null; callId: string | null; isError: boolean; at: string}; + +/** The thread's log, and what the agent is up to: busy on it, asleep until a time, or waiting for the person. */ +export type ThreadLogResponse = {thread: ThreadResponse; entries: EntryResponse[]; busy: boolean; sleepingUntil: string | null}; + +function toThread(thread: ThreadRow & {title?: string | null}): ThreadResponse { + return {id: thread.id, createdAt: thread.createdAt.toISOString(), taskId: thread.taskId, title: thread.title ?? null}; +} + +function toEntry(entry: EntryRow): EntryResponse { + return {id: entry.id, kind: entry.kind, text: entry.text, toolCalls: entry.toolCalls, callId: entry.callId, isError: entry.isError, at: entry.createdAt.toISOString()}; +} + +function requireHarness(request: AuthenticatedRequest): Promise> { + return Do(async ($) => { + if (!hasExperiment(request.user, 'harness')) { + return $(Err(ApiErr.forbidden())); + } + return $(await requireUserOrgId(request)); + }); +} + +/** `GET /api/chat/threads`: the organization's threads, newest first. */ +export function handleListThreadsRoute(request: AuthenticatedRequest): Promise> { + return Do(async ($) => { + const orgId = $(await requireHarness(request)); + return {threads: $(await listThreads(orgId)).map(toThread)}; + }); +} + +/** `POST /api/chat/threads`: a new thread, with the built-in agent made where the organization has none yet. */ +export function handleCreateThreadRoute(request: AuthenticatedRequest): Promise> { + return Do(async ($) => { + const orgId = $(await requireHarness(request)); + const agent = $(await requireBuiltInAgent(orgId)); + const taskId = $(await createThreadTask(orgId, agent.agentId)); + return {thread: toThread($(await createThread({orgId, agentId: agent.agentId, taskId})))}; + }); +} + +/** `GET /api/chat/threads/:threadId`: the thread's log and whether the agent is busy on it. */ +export function handleGetThreadRoute(request: AuthenticatedRequest): Promise> { + return Do(async ($) => { + const orgId = $(await requireHarness(request)); + const threadId = request.params.threadId ?? ''; + const thread = $(requirePresent($(await getThread(orgId, threadId)), ApiErr.notFound('thread', threadId))); + const [entries, jobs] = await Promise.all([listEntries(thread.id), listOpenJobs(thread.id)]); + const open = $(jobs); + const now = Date.now(); + // Asleep when every open job is a tool job due later; the first to be due is when it wakes. + const later = open.filter((job) => job.runAt.getTime() > now); + const asleep = open.length > 0 && later.length === open.length; + const first = later.map((job) => job.runAt.getTime()).sort((a, b) => a - b)[0]; + return { + thread: toThread(thread), + entries: $(entries).map(toEntry), + busy: open.length > 0 && !asleep, + sleepingUntil: asleep && first !== undefined ? new Date(first).toISOString() : null, + }; + }); +} + +const messageBodySchema = z.object({text: z.string().trim().min(1).max(MAX_MESSAGE)}); + +/** `POST /api/chat/threads/:threadId/messages` with `{text}`: what the person says; the agent is woken. */ +export function handlePostMessageRoute(request: AuthenticatedRequest): Promise> { + return Do(async ($) => { + const orgId = $(await requireHarness(request)); + const {text} = $(parseSchema(messageBodySchema, request.body)); + const threadId = request.params.threadId ?? ''; + $(requirePresent($(await getThread(orgId, threadId)), ApiErr.notFound('thread', threadId))); + const entry = $(requirePresent($(await appendPersonMessage(threadId, text)), ApiErr.notFound('thread', threadId))); + return {entry: toEntry(entry)}; + }); +} diff --git a/packages/api/src/harness/thread-task.ts b/packages/api/src/harness/thread-task.ts new file mode 100644 index 00000000..d81414b6 --- /dev/null +++ b/packages/api/src/harness/thread-task.ts @@ -0,0 +1,43 @@ +import type {Result} from 'ts-results-es'; + +import {type ApiError, Do} from '@proxy/utils'; +import {taskEndedAt} from '@proxy/utils/tasks'; +import {createAgentTask, getAgentTask} from '@proxy/db/task'; +import {listEntries, setThreadTask, type ThreadRow} from '@proxy/db/harness'; + +/** + * The task a thread's calls are made for, as every agent's calls are, so the activity log says + * which conversation each request was for. A thread starts with one, and gets a new one when it + * was abandoned, a day after it started, since the gate refuses calls for one that was. + */ + +const TITLE = 'Chat'; +const MAX_GOAL = 2000; + +// What the task is for: what the person first said in the thread, as far as a goal holds. +function goalOf(entries: Array<{kind: string; text: string}>): string { + const first = entries.find((entry) => entry.kind === 'person')?.text.trim() ?? ''; + if (first === '') { + return 'A conversation on the Chat page'; + } + return first.length > MAX_GOAL ? `${first.slice(0, MAX_GOAL - 1)}…` : first; +} + +/** A thread's first task, before it has a log. */ +export function createThreadTask(orgId: string, agentId: string): Promise> { + return Do(async ($) => $(await createAgentTask({orgId, agentId, title: TITLE, goal: 'A conversation on the Chat page', via: 'harness'})).id); +} + +/** The thread's task, still under way: the one it has, or a new one in its place. */ +export function ensureThreadTask(thread: ThreadRow, now: Date): Promise> { + return Do(async ($) => { + const task = $(await getAgentTask(thread.orgId, thread.agentId, thread.taskId)); + if (task && taskEndedAt({startedAt: task.createdAt, finishedAt: task.finishedAt}, now) === null) { + return task.id; + } + const entries = $(await listEntries(thread.id)); + const made = $(await createAgentTask({orgId: thread.orgId, agentId: thread.agentId, title: TITLE, goal: goalOf(entries), via: 'harness'})); + $(await setThreadTask(thread.id, made.id)); + return made.id; + }); +} diff --git a/packages/api/src/harness/tool-job.test.ts b/packages/api/src/harness/tool-job.test.ts new file mode 100644 index 00000000..0403fd65 --- /dev/null +++ b/packages/api/src/harness/tool-job.test.ts @@ -0,0 +1,102 @@ +import {beforeEach, describe, expect, mock, test} from 'bun:test'; +import {Ok} from 'ts-results-es'; + +const getJobThread = mock(); +const getEntry = mock(); +const completeToolJob = mock(async () => Ok(true)); +const failJob = mock(async () => Ok(true)); +const finishJob = mock(async () => Ok(true)); +mock.module('@proxy/db/harness', () => ({getJobThread, getEntry, completeToolJob, failJob, finishJob})); + +const signedIn = {agentId: 'agent-1', orgId: 'org-1', providerId: 'proxy', name: 'Personal Agent Proxy', experiments: []}; +mock.module('./built-in-agent', () => ({requireBuiltInAgent: mock(async () => Ok(signedIn))})); + +const ensureThreadTask = mock(async () => Ok('task-2')); +mock.module('./thread-task', () => ({ensureThreadTask})); + +const callMcpTool = mock(); +mock.module('../mcp/tools', () => ({callMcpTool})); + +mock.module('./tool-surface', () => ({isGateTool: (name: string) => ['read', 'run', 'list_connections', 'list_tools', 'send_feedback'].includes(name)})); + +const now = new Date('2026-10-21T10:00:00Z'); +const thread = {id: 'thread-1', orgId: 'org-1', agentId: 'agent-1', taskId: 'task-1', createdAt: now}; + +function job(callId: string, attempts = 1) { + return {id: 'job-1', threadId: 'thread-1', kind: 'tool' as const, status: 'running' as const, entryId: 'entry-1', callId, runAt: now, claimedBy: 'w1', attempts}; +} + +function entryWith(calls: Array<{id: string; name: string; arguments: string}>) { + return {id: 'entry-1', seq: 2, threadId: 'thread-1', kind: 'assistant' as const, text: '', toolCalls: calls, callId: null, isError: false, inputTokens: 1, outputTokens: 1, createdAt: now}; +} + +beforeEach(() => { + mock.clearAllMocks(); + getJobThread.mockResolvedValue(Ok(thread)); + getEntry.mockResolvedValue(Ok(entryWith([{id: 'c1', name: 'read', arguments: '{"connection":"mail-1","tool":"emails_list"}'}]))); + callMcpTool.mockResolvedValue({content: [{type: 'text', text: '{"records":[]}'}], structuredContent: {records: []}}); +}); + +describe('runToolJob', () => { + test('runs a gate tool through the MCP dispatch as the built-in agent, for the thread’s task', async () => { + const {runToolJob} = await import('./tool-job'); + + const result = await runToolJob(job('c1'), 'w1', now); + + expect(result.isOk()).toBe(true); + expect(callMcpTool).toHaveBeenCalledWith(signedIn, 'read', {connection: 'mail-1', tool: 'emails_list', task: 'task-2'}, false, 'harness'); + expect(completeToolJob).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', {callId: 'c1', text: '{"records":[]}', isError: false}); + }); + + test('a refused call is answered as an error, in the gate’s words', async () => { + const {runToolJob} = await import('./tool-job'); + callMcpTool.mockResolvedValue({content: [{type: 'text', text: 'This login may not run emails_list with this connection.'}], isError: true}); + + await runToolJob(job('c1'), 'w1', now); + + expect(completeToolJob).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', {callId: 'c1', text: 'This login may not run emails_list with this connection.', isError: true}); + }); + + test('a sleep answers with the time, since its job was due when the sleep ended', async () => { + const {runToolJob} = await import('./tool-job'); + getEntry.mockResolvedValue(Ok(entryWith([{id: 'c1', name: 'sleep_until', arguments: '{"seconds":60}'}]))); + + await runToolJob(job('c1'), 'w1', now); + + expect(callMcpTool).not.toHaveBeenCalled(); + expect(completeToolJob).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', {callId: 'c1', text: 'You slept, and it is now 2026-10-21T10:00:00.000Z.', isError: false}); + }); + + test('a write a worker was interrupted in is not made again: the model is told the outcome is unknown', async () => { + const {runToolJob} = await import('./tool-job'); + getEntry.mockResolvedValue(Ok(entryWith([{id: 'c1', name: 'run', arguments: '{"connection":"mail-1","tool":"emails_send","params":{"id":"m1"}}'}]))); + + await runToolJob(job('c1', 2), 'w1', now); + + expect(callMcpTool).not.toHaveBeenCalled(); + expect(completeToolJob).toHaveBeenCalledWith('job-1', 'w1', 'thread-1', {callId: 'c1', text: expect.stringContaining('outcome of this call is unknown'), isError: true}); + + // A read is simply made again. + getEntry.mockResolvedValue(Ok(entryWith([{id: 'c1', name: 'read', arguments: '{}'}]))); + await runToolJob(job('c1', 2), 'w1', now); + expect(callMcpTool).toHaveBeenCalledTimes(1); + }); + + test('bad arguments, an unknown tool and a missing call are answered or failed, never thrown', async () => { + const {runToolJob} = await import('./tool-job'); + getEntry.mockResolvedValue(Ok(entryWith([{id: 'c1', name: 'read', arguments: 'not json'}, {id: 'c2', name: 'fly', arguments: '{}'}]))); + + await runToolJob(job('c1'), 'w1', now); + expect(completeToolJob).toHaveBeenLastCalledWith('job-1', 'w1', 'thread-1', {callId: 'c1', text: 'The arguments were not a JSON object.', isError: true}); + + await runToolJob(job('c2'), 'w1', now); + expect(completeToolJob).toHaveBeenLastCalledWith('job-1', 'w1', 'thread-1', {callId: 'c2', text: 'There is no tool fly.', isError: true}); + + await runToolJob(job('c9'), 'w1', now); + expect(failJob).toHaveBeenCalledWith('job-1', 'w1', 'No call c9 in entry entry-1'); + + getJobThread.mockResolvedValue(Ok(null)); + await runToolJob(job('c1'), 'w1', now); + expect(finishJob).toHaveBeenCalledWith('job-1', 'w1'); + }); +}); diff --git a/packages/api/src/harness/tool-job.ts b/packages/api/src/harness/tool-job.ts new file mode 100644 index 00000000..be6fd131 --- /dev/null +++ b/packages/api/src/harness/tool-job.ts @@ -0,0 +1,94 @@ +import {Result} from 'ts-results-es'; + +import {type ApiError, Do} from '@proxy/utils'; +import {SLEEP_TOOL, sleepUntil} from '@proxy/utils/harness'; +import {completeToolJob, failJob, finishJob, getEntry, getJobThread, type JobRow, type ThreadRow} from '@proxy/db/harness'; + +import {callMcpTool} from '../mcp/tools'; +import {requireBuiltInAgent} from './built-in-agent'; +import {ensureThreadTask} from './thread-task'; +import {isGateTool} from './tool-surface'; + +/** + * A tool job: one call the model made, run and answered in the log. The MCP server's tools go + * through the gate as the built-in agent, for the thread's task, logged like any agent's call. A + * sleep answers with the time, since its job was due when the sleep ended. A call that changes + * something is never run twice: a worker that was interrupted mid-call leaves the outcome unknown, + * and the model is told so rather than the call being made again. + */ + +// How much of a result the log keeps, and the model reads. +const MAX_RESULT_CHARS = 60_000; + +type ToolAnswer = {text: string; isError: boolean}; + +function error(text: string): ToolAnswer { + return {text, isError: true}; +} + +function parseArgs(json: string): Record | null { + const parsed = Result.wrap((): unknown => JSON.parse(json)); + if (parsed.isErr() || typeof parsed.value !== 'object' || parsed.value === null || Array.isArray(parsed.value)) { + return null; + } + return parsed.value as Record; +} + +function cut(text: string): string { + if (text.length <= MAX_RESULT_CHARS) { + return text; + } + return `${text.slice(0, MAX_RESULT_CHARS)}\n\n[Cut: the result was ${text.length} characters long.]`; +} + +// Runs the call through the gate, for the thread's task. +function runGateTool(thread: ThreadRow, name: string, args: Record, now: Date): Promise> { + return Do(async ($) => { + const signedIn = $(await requireBuiltInAgent(thread.orgId)); + const taskId = $(await ensureThreadTask(thread, now)); + const result = await callMcpTool(signedIn, name, {...args, task: taskId}, false, 'harness'); + return {text: cut(result.content.map((part) => part.text).join('\n')), isError: result.isError === true}; + }); +} + +function answer(thread: ThreadRow, job: JobRow, call: {name: string; arguments: string}, now: Date): Promise> { + return Do(async ($) => { + const args = parseArgs(call.arguments); + if (args === null) { + return error('The arguments were not a JSON object.'); + } + if (call.name === SLEEP_TOOL) { + if (sleepUntil(args, now) === null) { + return error('Give until, a time in ISO 8601, or seconds from now.'); + } + return {text: `You slept, and it is now ${now.toISOString()}.`, isError: false}; + } + if (!isGateTool(call.name)) { + return error(`There is no tool ${call.name}.`); + } + // A write run by a worker that lost the job mid-call may have gone through. + if (call.name === 'run' && job.attempts > 1) { + return error('The outcome of this call is unknown: the worker running it was interrupted before it could record the answer. Check what it changed before making it again.'); + } + return $(await runGateTool(thread, call.name, args, now)); + }); +} + +/** Runs the job: the call it is for, its answer written to the log, and the model woken once every call is answered. */ +export function runToolJob(job: JobRow, workerId: string, now = new Date()): Promise> { + return Do(async ($) => { + const thread = $(await getJobThread(job.threadId)); + if (!thread) { + $(await finishJob(job.id, workerId)); + return; + } + const entry = $(await getEntry(job.entryId ?? '')); + const call = entry.toolCalls?.find((candidate) => candidate.id === job.callId); + if (!call) { + $(await failJob(job.id, workerId, `No call ${job.callId ?? ''} in entry ${entry.id}`)); + return; + } + const answered = $(await answer(thread, job, call, now)); + $(await completeToolJob(job.id, workerId, thread.id, {callId: call.id, ...answered})); + }); +} diff --git a/packages/api/src/harness/tool-surface.test.ts b/packages/api/src/harness/tool-surface.test.ts new file mode 100644 index 00000000..4171f0aa --- /dev/null +++ b/packages/api/src/harness/tool-surface.test.ts @@ -0,0 +1,41 @@ +import {describe, expect, test} from 'bun:test'; + +import {MCP_TOOLS} from '../mcp/tools'; +import {isGateTool, MODEL_TOOLS} from './tool-surface'; + +type Params = {properties: Record; required: string[]}; + +function tool(name: string) { + const found = MODEL_TOOLS.find((candidate) => candidate.name === name); + if (!found) { + throw new Error(`No model tool ${name}`); + } + return {...found, parameters: found.parameters as Params}; +} + +describe('the built-in agent’s tools', () => { + test('are the MCP server’s without the two about tasks, and a sleep', () => { + expect(MODEL_TOOLS.map((candidate) => candidate.name)).toEqual(['list_connections', 'list_tools', 'read', 'run', 'send_feedback', 'sleep_until']); + }); + + test('take no task: the loop fills it in', () => { + for (const name of ['list_connections', 'list_tools', 'read', 'run']) { + expect(tool(name).parameters.properties).not.toHaveProperty('task'); + expect(tool(name).parameters.required).not.toContain('task'); + expect(tool(name).description).not.toContain('from start_task'); + } + expect(tool('read').parameters.required).toEqual(['connection', 'tool']); + }); + + test('keep every other param the MCP server’s have', () => { + const run = MCP_TOOLS.find((candidate) => candidate.name === 'run'); + expect(Object.keys(tool('run').parameters.properties)).toEqual(Object.keys(run?.inputSchema.properties ?? {}).filter((key) => key !== 'task')); + }); + + test('the gate’s tools are the MCP server’s but the task ones', () => { + expect(isGateTool('read')).toBe(true); + expect(isGateTool('send_feedback')).toBe(true); + expect(isGateTool('start_task')).toBe(false); + expect(isGateTool('sleep_until')).toBe(false); + }); +}); diff --git a/packages/api/src/harness/tool-surface.ts b/packages/api/src/harness/tool-surface.ts new file mode 100644 index 00000000..c55c4a11 --- /dev/null +++ b/packages/api/src/harness/tool-surface.ts @@ -0,0 +1,56 @@ +import {SLEEP_TOOL} from '@proxy/utils/harness'; + +import {MCP_TOOLS, type McpToolDefinition} from '../mcp/tools'; +import type {ModelTool} from './model'; + +/** + * The tools the built-in agent has: the MCP server's, as any agent gets them, less the two about + * tasks, since every call in a thread is for the thread's task and the loop fills the task in + * (inference.ts, tool-job.ts); and one of its own, to be woken later. Nothing else: everything it + * does to the person's accounts goes through the same gate as Claude's calls. + */ + +// The MCP tools the built-in agent doesn't need: its task is the thread's. +const TASK_TOOLS = new Set(['start_task', 'finish_task']); + +const SLEEP: ModelTool = { + name: SLEEP_TOOL, + description: + 'Stops until a time, then wakes you with a note of the time. For checking back on something later, or doing what the person asked for at a time they named. Give `until`, the time to wake at, in ISO 8601 with its offset, or `seconds` from now; at most a week. Say what you will do when you wake in your message, since it is all you will have then.', + parameters: { + type: 'object', + properties: { + until: {type: 'string', description: 'When to wake, in ISO 8601, such as 2026-10-22T07:30:00-07:00'}, + seconds: {type: 'number', description: 'How many seconds from now to wake, instead of until'}, + }, + additionalProperties: false, + }, +}; + +// What the tools that take a task are for, said without it; every other tool keeps the MCP +// server's words. +const DESCRIPTIONS: Record = { + list_connections: + 'Call this before telling the person a service isn’t connected: their email, meeting notes, documents or issues may well be here. The accounts and information you can use, such as a mailbox or the person’s vault, each with its id and what you may do with it.', + list_tools: 'The tools you have with one connection: each one’s name, what it does, whether it runs with read or run, its risk, and the JSON Schema of its params. Take `connection` from list_connections.', + read: 'Runs a tool that only reads, such as listing or opening emails: never changes anything. Take `connection` from list_connections and `tool` from list_tools.', + run: 'Runs a tool that changes something or reaches other people, such as archiving an email, saving a draft or sending one. Take `connection` from list_connections and `tool` from list_tools.', +}; + +// The MCP tool without its `task` param, which the loop fills in. +function withoutTask(tool: McpToolDefinition): ModelTool { + const {task: _task, ...properties} = tool.inputSchema.properties; + return { + name: tool.name, + description: DESCRIPTIONS[tool.name] ?? tool.description, + parameters: {...tool.inputSchema, properties, required: tool.inputSchema.required.filter((name) => name !== 'task')}, + }; +} + +/** The tools the model is given, in the order the MCP server lists its. */ +export const MODEL_TOOLS: ModelTool[] = [...MCP_TOOLS.filter((tool) => !TASK_TOOLS.has(tool.name)).map(withoutTask), SLEEP]; + +/** Whether a tool call is for one of the MCP server's tools, run through the gate. */ +export function isGateTool(name: string): boolean { + return MCP_TOOLS.some((tool) => tool.name === name && !TASK_TOOLS.has(tool.name)); +} diff --git a/packages/api/src/harness/worker.test.ts b/packages/api/src/harness/worker.test.ts new file mode 100644 index 00000000..4975e918 --- /dev/null +++ b/packages/api/src/harness/worker.test.ts @@ -0,0 +1,67 @@ +import {beforeEach, describe, expect, mock, test} from 'bun:test'; +import {Err, Ok, type Result} from 'ts-results-es'; + +import {ApiErr, type ApiError} from '@proxy/utils'; +import type {JobRow} from '@proxy/db/harness'; + +type Run = (job: JobRow, workerId: string) => Promise>; + +const claimJob = mock(); +const failJob = mock(async () => Ok(true)); +const renewLease = mock(async () => Ok(true)); +mock.module('@proxy/db/harness', () => ({claimJob, failJob, renewLease})); + +const runInference = mock(async () => Ok(undefined)); +mock.module('./inference', () => ({runInference})); + +const runToolJob = mock(async () => Ok(undefined)); +mock.module('./tool-job', () => ({runToolJob})); + +const now = new Date(); +const inference = {id: 'j1', threadId: 't1', kind: 'inference' as const, status: 'running' as const, entryId: null, callId: null, runAt: now, claimedBy: 'w', attempts: 1}; +const tool = {...inference, id: 'j2', kind: 'tool' as const, entryId: 'e1', callId: 'c1'}; + +// Waits for the worker to get through what is queued. +const settle = () => new Promise((resolve) => setTimeout(resolve, 50)); + +beforeEach(() => { + mock.clearAllMocks(); +}); + +describe('the worker', () => { + test('claims jobs one after another and runs each by its kind, then stops claiming', async () => { + const {startHarnessWorker} = await import('./worker'); + claimJob.mockResolvedValueOnce(Ok(inference)).mockResolvedValueOnce(Ok(tool)).mockResolvedValue(Ok(null)); + + const worker = startHarnessWorker(); + await settle(); + await worker.stop(); + const claims = claimJob.mock.calls.length; + await settle(); + + expect(runInference).toHaveBeenCalledWith(inference, expect.any(String)); + expect(runToolJob).toHaveBeenCalledWith(tool, expect.any(String)); + expect(runInference.mock.calls[0]?.[1]).toBe(runToolJob.mock.calls[0]?.[1]); + expect(claimJob.mock.calls.length).toBe(claims); + }); + + test('a job that fails, or throws, is marked failed with why, and one claimed too often is given up', async () => { + const {startHarnessWorker} = await import('./worker'); + runInference.mockResolvedValueOnce(Err(ApiErr.conflict('no longer mine'))); + runToolJob.mockRejectedValueOnce(new Error('boom')); + claimJob + .mockResolvedValueOnce(Ok(inference)) + .mockResolvedValueOnce(Ok(tool)) + .mockResolvedValueOnce(Ok({...inference, id: 'j3', attempts: 5})) + .mockResolvedValue(Ok(null)); + + const worker = startHarnessWorker(); + await settle(); + await worker.stop(); + + expect(failJob).toHaveBeenCalledWith('j1', expect.any(String), expect.stringContaining('no longer mine')); + expect(failJob).toHaveBeenCalledWith('j2', expect.any(String), 'boom'); + expect(failJob).toHaveBeenCalledWith('j3', expect.any(String), 'Given up after 4 attempts'); + expect(runInference).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/api/src/harness/worker.ts b/packages/api/src/harness/worker.ts new file mode 100644 index 00000000..e458caf1 --- /dev/null +++ b/packages/api/src/harness/worker.ts @@ -0,0 +1,95 @@ +import {randomBytes} from 'node:crypto'; +import {hostname} from 'node:os'; +import {setTimeout as sleep} from 'node:timers/promises'; + +import {Result} from 'ts-results-es'; + +import {claimJob, failJob, type JobRow, renewLease} from '@proxy/db/harness'; + +import {log, serializeError} from '../observability/log'; +import {runInference} from './inference'; +import {runToolJob} from './tool-job'; + +/** + * The worker that runs the built-in agent's jobs, one at a time, in each api instance: it claims + * the next job due, keeps its lease while the job runs, and claims the next as soon as it is done, + * so a turn's calls and the next turn follow each other without waiting; nothing due, it looks + * again in a moment. The queue is the database, so instances need nothing of each other: one that + * goes down mid-job loses its lease, and another picks the job up. Stopping, as for a deploy, the + * worker claims nothing more and finishes the job it is on, up to a point. + */ + +const POLL_MS = 500; +const LEASE_MS = 120_000; +const RENEW_MS = 30_000; +// After this many claims, a job that keeps failing is given up. +const MAX_ATTEMPTS = 4; +// How long stopping waits for the job under way. +const STOP_MS = 8_000; + +export type HarnessWorker = {stop: () => Promise}; + +// One name per worker, so a lease says who holds it. +function workerName(): string { + return `${hostname()}-${process.pid}-${randomBytes(3).toString('hex')}`; +} + +function run(job: JobRow, workerId: string): Promise> { + if (job.kind === 'inference') { + return runInference(job, workerId); + } + return runToolJob(job, workerId); +} + +// Runs the job with its lease kept, and marks it failed where it throws or fails. +async function runClaimed(job: JobRow, workerId: string): Promise { + if (job.attempts > MAX_ATTEMPTS) { + await failJob(job.id, workerId, `Given up after ${job.attempts - 1} attempts`); + return; + } + const renewal = setInterval(() => { + void renewLease(job.id, workerId, LEASE_MS); + }, RENEW_MS); + const result = await Result.wrapAsync(() => run(job, workerId)); + clearInterval(renewal); + const failure = result.isErr() ? result.error : result.value.isErr() ? result.value.error : null; + if (failure === null) { + return; + } + log.error('Built-in agent job failed', {jobId: job.id, kind: job.kind, attempt: job.attempts, ...serializeError(failure)}); + const message = failure instanceof Error ? failure.message : JSON.stringify(failure); + await failJob(job.id, workerId, message.slice(0, 1000)); +} + +/** Starts the worker; it runs until stopped. */ +export function startHarnessWorker(): HarnessWorker { + const workerId = workerName(); + let stopping = false; + let current: Promise = Promise.resolve(); + + const loop = async (): Promise => { + while (!stopping) { + const claimed = await claimJob(workerId, LEASE_MS); + if (claimed.isErr()) { + log.error('Built-in agent worker could not claim a job', serializeError(claimed.error)); + await sleep(POLL_MS * 4); + continue; + } + if (claimed.value === null) { + await sleep(POLL_MS); + continue; + } + current = runClaimed(claimed.value, workerId); + await current; + } + }; + void loop(); + log.info('Built-in agent worker started', {workerId}); + + return { + stop: async () => { + stopping = true; + await Promise.race([current, sleep(STOP_MS)]); + }, + }; +} diff --git a/packages/api/src/index.ts b/packages/api/src/index.ts index 9793c2b5..c5139bd5 100644 --- a/packages/api/src/index.ts +++ b/packages/api/src/index.ts @@ -5,6 +5,7 @@ import {Result} from 'ts-results-es'; import {connectDb, disconnectDb} from '@proxy/db'; import {takeOverCustomConnections} from './connections/take-over-custom'; +import {startHarnessWorker, type HarnessWorker} from './harness/worker'; import {log, serializeError} from './observability/log'; import {convertOldVaultRecords} from './records/vault-entries'; import {createHttpApp} from './server/create-http-app'; @@ -25,6 +26,7 @@ async function startApiServer(): Promise { process.exit(1); } startTelemetry(); + let worker: HarnessWorker | null = null; const shutdown = async (signal: string): Promise => { if (isShuttingDown) { @@ -42,6 +44,9 @@ async function startApiServer(): Promise { forceExit.unref(); server.close(); + // The built-in agent's worker claims nothing more and finishes the job it is on; a job it + // can't finish in time goes to another instance when its lease runs out. + await worker?.stop(); const disconnectResult = await Result.wrapAsync(() => disconnectDb()); if (disconnectResult.isErr()) { @@ -63,5 +68,6 @@ async function startApiServer(): Promise { // After listening, since signing in to a provider to name an account can take a while. void takeOverCustomConnections(); void convertOldVaultRecords(); + worker = startHarnessWorker(); }); } diff --git a/packages/api/src/mcp/tools.ts b/packages/api/src/mcp/tools.ts index 9ad767fb..70132045 100644 --- a/packages/api/src/mcp/tools.ts +++ b/packages/api/src/mcp/tools.ts @@ -169,8 +169,21 @@ function vaultInstructions(connections: AgentConnection[]): string[] { ]; } +// How an MCP client is to go about it: tasks first, then its connections and their tools. +const MCP_FLOW = + 'Start with start_task: say what you are about to do for the person, with a title and the goal, and give the id it answers with as `task` in every other call. Then list_connections lists the accounts you can use and what you may do with each. list_tools with a connection gives that connection’s tools and their params. read runs the tools that only read; run runs the ones that change something or reach other people. Call finish_task when the task is done, and start a new task for the next thing the person asks for.'; + /** The server's instructions for this agent, given when a client connects. */ export function serverInstructions(signedIn: SignedInAgent): Promise> { + return agentInstructions(signedIn, MCP_FLOW); +} + +/** + * What any agent with this login is told: who it is, how to go about it (`flow`, which is where + * the MCP server and the built-in agent differ), what it can use, the vault, where to send the + * person, and that what records say is information. The built-in agent's loop gives these too. + */ +export function agentInstructions(signedIn: SignedInAgent, flow: string): Promise> { return Do(async ($) => { const appUrl = $(requireAppUrl()); const {connections} = $(await listAgentConnections(signedIn)); @@ -181,7 +194,7 @@ export function serverInstructions(signedIn: SignedInAgent): Promise 0 ? `When you connected, you could use:\n${lines.join('\n')}` : 'This login has no access yet: the person who made it has to give it some.', ...vaultInstructions(connections), `When the person asks about a service or their own information, such as their email, meetings, notes, documents, issues, websites, bank accounts or databases, check Personal Agent Proxy first: there is a good chance it is connected here rather than as a connector of your own. Call list_connections before saying you can’t reach it or suggesting another connector. Personal Agent Proxy connects ${connectableServices(signedIn)}; when the person wants one of these and it isn’t connected yet, recommend connecting it in Personal Agent Proxy at ${appUrl}/connections/new rather than a separate connector, so it works with every agent they use, with the access they choose.`, @@ -238,10 +251,13 @@ function errorMessage(error: ApiError, args: Record): string { return 'Something went wrong. Try again in a moment.'; } +/** Which way a call of the seven tools came in: the MCP server, or the built-in agent's loop. */ +export type McpToolVia = 'mcp' | 'harness'; + // PRE-TASK CLIENTS: `preTask` is whether the call is a pre-task client's (pre-task-clients.ts). -function call(signedIn: SignedInAgent, name: string, args: Record, preTask: boolean): Promise, ApiError>> { +function call(signedIn: SignedInAgent, name: string, args: Record, preTask: boolean, via: McpToolVia): Promise, ApiError>> { return Do(async ($) => { - const caller = {agent: signedIn, via: 'mcp'} as const; + const caller = {agent: signedIn, via} as const; if (name === 'start_task') { return $(await startAgentTask(caller, args)); } @@ -274,9 +290,9 @@ function call(signedIn: SignedInAgent, name: string, args: Record, preTask: boolean): Promise { - const result = await call(signedIn, name, args, preTask); +/** Calls one of the seven tools as the signed-in agent, the way it came in logged with it. */ +export async function callMcpTool(signedIn: SignedInAgent, name: string, args: Record, preTask: boolean, via: McpToolVia = 'mcp'): Promise { + const result = await call(signedIn, name, args, preTask, via); if (result.isErr()) { return {content: [{type: 'text', text: errorMessage(result.error, args)}], isError: true}; } diff --git a/packages/api/src/server/create-api-router.ts b/packages/api/src/server/create-api-router.ts index b4310006..3c59c017 100644 --- a/packages/api/src/server/create-api-router.ts +++ b/packages/api/src/server/create-api-router.ts @@ -36,6 +36,7 @@ import {handleConnectMysqlRoute, handleTestMysqlRoute} from '../connections/mysq import {handleConnectPlaidRoute, handleCreatePlaidLinkTokenRoute} from '../connections/plaid/route'; import {handleConnectPostgresRoute, handleTestPostgresRoute} from '../connections/postgres/route'; import {handleDeleteConnectionRoute, handleGetConnectionRoute, handleListConnectionsRoute, handleRenameConnectionRoute, handleSetConnectionDefaultsRoute} from '../connections/route'; +import {handleCreateThreadRoute, handleGetThreadRoute, handleListThreadsRoute, handlePostMessageRoute} from '../harness/route'; import {handleFillIntentRoute, handlePickIntentRoute} from '../intent/route'; import {handleMcpMethodNotAllowedRoute, handleMcpRoute} from '../mcp/route'; import {handleDeleteMeRoute, handleFinishOnboardingRoute, handleMeRoute} from '../me/route'; @@ -158,6 +159,14 @@ export function createApiRouter(): express.Router { router.get('/api/widgets/accounts', withAuthResult('Widget bank accounts route', handleBankAccountsRoute)); router.get('/api/widgets/transactions', withAuthResult('Widget bank transactions route', handleBankTransactionsRoute)); + // The Chat page: the person's threads with the built-in agent, whose loop runs in the api + // (harness/worker.ts); a message queues its work. + router.use('/api/chat', requireBrowserOrigin); + router.get('/api/chat/threads', withAuthResult('List chat threads route', handleListThreadsRoute)); + router.post('/api/chat/threads', withAuthResult('Create chat thread route', handleCreateThreadRoute)); + router.get('/api/chat/threads/:threadId', withAuthResult('Get chat thread route', handleGetThreadRoute)); + router.post('/api/chat/threads/:threadId/messages', withAuthResult('Post chat message route', handlePostMessageRoute)); + // The agent side: its own sign-in and cookie, and every request made for one of the agent's // tasks, checked against the agent's access and logged with the task. router.post('/agent-auth/login', requireBrowserOrigin, wrapAsyncRoute('Agent login route', handleAgentLoginRoute)); diff --git a/packages/api/src/utils/env-schema.ts b/packages/api/src/utils/env-schema.ts index 72483b7d..b573a28e 100644 --- a/packages/api/src/utils/env-schema.ts +++ b/packages/api/src/utils/env-schema.ts @@ -107,7 +107,7 @@ export const envSchema = z.object({ OPENROUTER_KEY: documented(z.string().min(1).optional(), { group: 'Core', description: - 'An OpenRouter API key, for the model that fills in the fields of the widget Jev picks on the home page (gpt-oss-120b, on Cerebras or Groq). What is typed there is sent to it. Without it, or without `TYPESAFE_API_KEY`, the home page says it is not set up; everything else works as before.', + 'An OpenRouter API key, for the model that fills in the fields of the widget Jev picks on the home page (gpt-oss-120b, on Cerebras or Groq), and for the built-in agent on the Chat page (Claude Sonnet 5.5). What is typed there is sent to it. Without it, or without `TYPESAFE_API_KEY`, the home page says it is not set up, and without it the Chat page does; everything else works as before.', }), NODE_ENV: documented(z.enum(['development', 'test', 'production']).default('production'), { diff --git a/packages/db/package.json b/packages/db/package.json index a7dc8d64..5627c7fc 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -10,6 +10,7 @@ "./auth": "./src/auth.ts", "./connection": "./src/connection.ts", "./feedback": "./src/feedback.ts", + "./harness": "./src/harness.ts", "./vault": "./src/vault.ts", "./instance-meta": "./src/instance-meta.ts", "./oauth": "./src/oauth.ts", diff --git a/packages/db/prisma/migrations/v024_add_built_in_agent_threads/migration.sql b/packages/db/prisma/migrations/v024_add_built_in_agent_threads/migration.sql new file mode 100644 index 00000000..ab934417 --- /dev/null +++ b/packages/db/prisma/migrations/v024_add_built_in_agent_threads/migration.sql @@ -0,0 +1,83 @@ +-- The built-in agent: a thread per conversation with its log of entries (what the person said, +-- what the model answered with its tool calls, and what each call answered), and the jobs that +-- run it (the model on a log, one tool call), claimed by one api instance at a time with a lease. +-- Its requests come in a third way, `harness`, beside the agent pages and the MCP server. + +-- CreateEnum +CREATE TYPE "AgentEntryKind" AS ENUM ('person', 'assistant', 'tool_result'); + +-- CreateEnum +CREATE TYPE "AgentJobKind" AS ENUM ('inference', 'tool'); + +-- CreateEnum +CREATE TYPE "AgentJobStatus" AS ENUM ('queued', 'running', 'done', 'failed'); + +-- AlterEnum +ALTER TYPE "AuditVia" ADD VALUE 'harness'; + +-- CreateTable +CREATE TABLE "AgentThread" ( + "id" TEXT NOT NULL, + "orgId" TEXT NOT NULL, + "agentId" TEXT NOT NULL, + "taskId" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "AgentThread_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "AgentEntry" ( + "id" TEXT NOT NULL, + "seq" SERIAL NOT NULL, + "threadId" TEXT NOT NULL, + "kind" "AgentEntryKind" NOT NULL, + "text" TEXT NOT NULL, + "toolCalls" TEXT, + "callId" TEXT, + "isError" BOOLEAN NOT NULL DEFAULT false, + "inputTokens" INTEGER, + "outputTokens" INTEGER, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "AgentEntry_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "AgentJob" ( + "id" TEXT NOT NULL, + "threadId" TEXT NOT NULL, + "kind" "AgentJobKind" NOT NULL, + "status" "AgentJobStatus" NOT NULL DEFAULT 'queued', + "entryId" TEXT, + "callId" TEXT, + "runAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "claimedBy" TEXT, + "leaseExpiresAt" TIMESTAMP(3), + "attempts" INTEGER NOT NULL DEFAULT 0, + "error" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "AgentJob_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "AgentThread_orgId_createdAt_idx" ON "AgentThread"("orgId", "createdAt"); + +-- CreateIndex +CREATE UNIQUE INDEX "AgentEntry_seq_key" ON "AgentEntry"("seq"); + +-- CreateIndex +CREATE INDEX "AgentEntry_threadId_seq_idx" ON "AgentEntry"("threadId", "seq"); + +-- CreateIndex +CREATE INDEX "AgentJob_status_runAt_idx" ON "AgentJob"("status", "runAt"); + +-- CreateIndex +CREATE INDEX "AgentJob_threadId_idx" ON "AgentJob"("threadId"); + +-- CreateIndex +CREATE UNIQUE INDEX "AgentJob_entryId_callId_key" ON "AgentJob"("entryId", "callId"); + diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 28f3cb6b..e6ce7bab 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -108,6 +108,7 @@ model Organization { audit AuditEntry[] feedback AgentFeedback[] tasks AgentTask[] + threads AgentThread[] } model OrgMember { @@ -280,10 +281,12 @@ enum AuditOutcome { denied } -// How an agent came in: the web app's agent pages, or the MCP server. +// How an agent came in: the web app's agent pages, the MCP server, or the built-in agent's own +// loop (`harness`). enum AuditVia { web mcp + harness } // One request an agent made, allowed or denied, or one thing Personal Agent Proxy did itself for a @@ -428,3 +431,102 @@ model OAuthGrant { @@index([clientId]) @@index([agentId]) } + +// The built-in agent's conversation with a person: one thread, holding its log of entries. Every +// call made in it is for the thread's task, as any agent's calls are; a task abandoned a day on is +// replaced by a new one. Everything that happens to the agent is an entry, and everything that +// makes it do something is a job (AgentJob). +model AgentThread { + id String @id @default(cuid()) + orgId String + // The built-in agent of the organization, an Agent like any other, with grants of its own. + agentId String + taskId String + + createdAt DateTime @default(now()) + updatedAt DateTime @default(now()) @updatedAt + + org Organization @relation(fields: [orgId], references: [id], onDelete: Cascade) + entries AgentEntry[] + jobs AgentJob[] + + @@index([orgId, createdAt]) +} + +// What an entry of a thread's log is: what the person said, what the model answered, with the tool +// calls it made, or what one of those calls answered. +enum AgentEntryKind { + person + assistant + tool_result +} + +// One entry of a thread's log, in the order of `seq`. The log is the transcript the model reads: +// an assistant entry carries its tool calls, and each gets a tool_result entry once run. +model AgentEntry { + id String @id @default(cuid()) + seq Int @unique @default(autoincrement()) + threadId String + kind AgentEntryKind + // What the person said, what the model said (maybe nothing beside its calls), or what a call + // answered. + text String + // An assistant entry's tool calls as JSON (`@proxy/utils/harness` ToolCall[]); null otherwise. + toolCalls String? + // A tool_result entry's call, by the id the model gave it; null otherwise. + callId String? + // A tool_result that is an error, in words the model can act on. + isError Boolean @default(false) + // An assistant entry's cost. + inputTokens Int? + outputTokens Int? + + createdAt DateTime @default(now()) + + thread AgentThread @relation(fields: [threadId], references: [id], onDelete: Cascade) + + @@index([threadId, seq]) +} + +enum AgentJobKind { + inference + tool +} + +enum AgentJobStatus { + queued + running + done + failed +} + +// A unit of the built-in agent's work, claimed by one api instance at a time: run the model on a +// thread's log (inference), or run one tool call of an assistant entry (tool). `runAt` is when it +// may run, which is how a call to sleep is kept. A worker claims a job due with a lease, renews +// the lease while it works, and finishes the job in the same transaction as the entries it writes, +// only while it still holds the lease; a lease that runs out gives the job to the next worker. +model AgentJob { + id String @id @default(cuid()) + threadId String + kind AgentJobKind + status AgentJobStatus @default(queued) + // A tool job's call: the assistant entry it is in and the id the model gave it. + entryId String? + callId String? + runAt DateTime @default(now()) + // Which worker holds it, and until when, while it is running. + claimedBy String? + leaseExpiresAt DateTime? + attempts Int @default(0) + // Why it failed, where it did. + error String? + + createdAt DateTime @default(now()) + updatedAt DateTime @default(now()) @updatedAt + + thread AgentThread @relation(fields: [threadId], references: [id], onDelete: Cascade) + + @@unique([entryId, callId]) + @@index([status, runAt]) + @@index([threadId]) +} diff --git a/packages/db/src/agent.ts b/packages/db/src/agent.ts index a373f034..99c4a088 100644 --- a/packages/db/src/agent.ts +++ b/packages/db/src/agent.ts @@ -164,6 +164,21 @@ export async function createAgentSession(data: { export type {SignedInAgent} from './signed-in-agent'; +/** + * The organization's agent of one provider, as a signed-in agent would be read, for the built-in + * agent, which signs in nowhere; `Ok(null)` for none, or one revoked. + */ +export async function getSignedInAgentByProvider( + orgId: string, + providerId: string, +): Promise> { + return ( + await wrapDb(() => + db.agent.findFirst({where: {orgId, providerId}, select: signedInAgentSelect}), + ) + ).map((agent) => (agent ? toSignedInAgent(agent) : null)); +} + /** `Ok(null)` means no live session: unknown, expired, or the agent has been revoked. */ export async function getSignedInAgent( tokenHash: string, diff --git a/packages/db/src/harness.ts b/packages/db/src/harness.ts new file mode 100644 index 00000000..1c466ea8 --- /dev/null +++ b/packages/db/src/harness.ts @@ -0,0 +1,427 @@ +import type {AgentEntryKind, AgentJobKind, AgentJobStatus, Prisma} from '@prisma/client'; +import {Err, Ok, type Result} from 'ts-results-es'; + +import {ApiErr, type ApiError, wrapDb} from '@proxy/utils'; +import {type LogEntry, pendingToolCalls, type ToolCall} from '@proxy/utils/harness'; + +import {db} from '.'; + +/** + * The built-in agent's threads, their logs and their jobs. Every change to a thread's log or its + * jobs is made with the thread's row locked, so a result landing and the model being run on the + * log never race: either the inference job sees the result, or the result sees no job and queues + * one. A worker's write is fenced by its lease: it goes through only while the job is still its. + */ + +export type ThreadRow = { + id: string; + orgId: string; + agentId: string; + taskId: string; + createdAt: Date; +}; + +export type EntryRow = LogEntry & { + id: string; + seq: number; + threadId: string; + isError: boolean; + inputTokens: number | null; + outputTokens: number | null; + createdAt: Date; +}; + +export type JobRow = { + id: string; + threadId: string; + kind: AgentJobKind; + status: AgentJobStatus; + entryId: string | null; + callId: string | null; + runAt: Date; + claimedBy: string | null; + attempts: number; +}; + +const threadSelect = {id: true, orgId: true, agentId: true, taskId: true, createdAt: true} as const; + +const entrySelect = { + id: true, + seq: true, + threadId: true, + kind: true, + text: true, + toolCalls: true, + callId: true, + isError: true, + inputTokens: true, + outputTokens: true, + createdAt: true, +} as const; + +const jobSelect = { + id: true, + threadId: true, + kind: true, + status: true, + entryId: true, + callId: true, + runAt: true, + claimedBy: true, + attempts: true, +} as const; + +type EntryRecord = Prisma.AgentEntryGetPayload<{select: typeof entrySelect}>; + +type Tx = Prisma.TransactionClient; + +function toEntry(entry: EntryRecord): EntryRow { + return { + ...entry, + toolCalls: entry.toolCalls === null ? null : (JSON.parse(entry.toolCalls) as ToolCall[]), + }; +} + +/** A new thread for the person's organization, its calls made for `taskId`. */ +export async function createThread(data: { + orgId: string; + agentId: string; + taskId: string; +}): Promise> { + return wrapDb(() => db.agentThread.create({data, select: threadSelect})); +} + +/** The organization's threads, newest first, each with what the person first said, as its title. */ +export async function listThreads( + orgId: string, +): Promise, ApiError>> { + return ( + await wrapDb(() => + db.agentThread.findMany({ + where: {orgId}, + select: { + ...threadSelect, + entries: {where: {kind: 'person'}, orderBy: {seq: 'asc'}, take: 1, select: {text: true}}, + }, + orderBy: {createdAt: 'desc'}, + }), + ) + ).map((threads) => + threads.map(({entries, ...thread}) => ({...thread, title: entries[0]?.text ?? null})), + ); +} + +/** `Ok(null)` means no such thread in the organization. */ +export async function getThread( + orgId: string, + threadId: string, +): Promise> { + return wrapDb(() => + db.agentThread.findFirst({where: {id: threadId, orgId}, select: threadSelect}), + ); +} + +/** The thread a job is for, by the job; `Ok(null)` when it is gone. */ +export async function getJobThread(threadId: string): Promise> { + return wrapDb(() => db.agentThread.findUnique({where: {id: threadId}, select: threadSelect})); +} + +/** The thread's log, in order. */ +export async function listEntries(threadId: string): Promise> { + return ( + await wrapDb(() => + db.agentEntry.findMany({where: {threadId}, select: entrySelect, orderBy: {seq: 'asc'}}), + ) + ).map((entries) => entries.map(toEntry)); +} + +/** Whether the agent is busy on the thread: a job of it queued or running, a sleep among them. */ +export async function isThreadBusy(threadId: string): Promise> { + return ( + await wrapDb(() => db.agentJob.count({where: {threadId, status: {in: ['queued', 'running']}}})) + ).map((count) => count > 0); +} + +/** The jobs still to run or running on the thread: what the agent is up to, for the person. */ +export async function listOpenJobs(threadId: string): Promise> { + return wrapDb(() => + db.agentJob.findMany({ + where: {threadId, status: {in: ['queued', 'running']}}, + select: jobSelect, + orderBy: {runAt: 'asc'}, + }), + ); +} + +/** A new task for the thread's calls, once the one before was abandoned. */ +export async function setThreadTask( + threadId: string, + taskId: string, +): Promise> { + return (await wrapDb(() => db.agentThread.update({where: {id: threadId}, data: {taskId}}))).map( + () => undefined, + ); +} + +// Locks the thread's row for the rest of the transaction; false when there is no such thread. +async function lockThread(tx: Tx, threadId: string): Promise { + const rows = await tx.$queryRaw< + Array<{id: string}> + >`SELECT "id" FROM "AgentThread" WHERE "id" = ${threadId} FOR UPDATE`; + return rows.length > 0; +} + +// Queues an inference job for the thread unless one is queued or running already. With the +// thread locked. +async function queueInference(tx: Tx, threadId: string): Promise { + const active = await tx.agentJob.count({ + where: {threadId, kind: 'inference', status: {in: ['queued', 'running']}}, + }); + if (active > 0) { + return; + } + await tx.agentJob.create({data: {threadId, kind: 'inference'}}); +} + +/** Appends what the person said and wakes the agent. `Ok(null)` means no such thread. */ +export async function appendPersonMessage( + threadId: string, + text: string, +): Promise> { + return wrapDb(() => + db.$transaction(async (tx) => { + if (!(await lockThread(tx, threadId))) { + return null; + } + const entry = await tx.agentEntry.create({ + data: {threadId, kind: 'person' satisfies AgentEntryKind, text}, + select: entrySelect, + }); + await queueInference(tx, threadId); + return toEntry(entry); + }), + ); +} + +/** + * Claims the next job due for `workerId`: one queued whose time has come, or one running whose + * worker's lease ran out, the earliest due first. `Ok(null)` when none is due. The claim holds for + * `leaseMs`, and `renewLease` keeps it. + */ +export async function claimJob( + workerId: string, + leaseMs: number, +): Promise> { + return ( + await wrapDb( + () => + db.$queryRaw>` + UPDATE "AgentJob" SET + "status" = 'running', + "claimedBy" = ${workerId}, + "leaseExpiresAt" = now() + make_interval(secs => ${leaseMs / 1000}), + "attempts" = "attempts" + 1, + "updatedAt" = now() + WHERE "id" = ( + SELECT "id" FROM "AgentJob" + WHERE ("status" = 'queued' AND "runAt" <= now()) + OR ("status" = 'running' AND "leaseExpiresAt" < now()) + ORDER BY "runAt" ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + RETURNING "id", "threadId", "kind", "status", "entryId", "callId", "runAt", "claimedBy", "attempts"`, + ) + ).map((rows) => rows[0] ?? null); +} + +/** Keeps the worker's claim on the job; `Ok(false)` when the job is no longer its. */ +export async function renewLease( + jobId: string, + workerId: string, + leaseMs: number, +): Promise> { + return ( + await wrapDb( + () => db.$executeRaw` + UPDATE "AgentJob" SET "leaseExpiresAt" = now() + make_interval(secs => ${leaseMs / 1000}), "updatedAt" = now() + WHERE "id" = ${jobId} AND "claimedBy" = ${workerId} AND "status" = 'running'`, + ) + ).map((count) => count > 0); +} + +// Marks the worker's running job as `status`; false when the job is no longer the worker's, in +// which case nothing else of the transaction should go through. +async function fence( + tx: Tx, + jobId: string, + workerId: string, + status: AgentJobStatus, + error: string | null = null, +): Promise { + const {count} = await tx.agentJob.updateMany({ + where: {id: jobId, claimedBy: workerId, status: 'running'}, + data: {status, error, claimedBy: null, leaseExpiresAt: null}, + }); + return count > 0; +} + +// Gives the job back to the queue, to run at `runAt`; false when it is no longer the worker's. +async function requeue(tx: Tx, jobId: string, workerId: string, runAt: Date): Promise { + const {count} = await tx.agentJob.updateMany({ + where: {id: jobId, claimedBy: workerId, status: 'running'}, + data: {status: 'queued', runAt, claimedBy: null, leaseExpiresAt: null}, + }); + return count > 0; +} + +/** Marks the worker's job done with nothing written; `Ok(false)` when it was no longer the worker's. */ +export async function finishJob( + jobId: string, + workerId: string, +): Promise> { + return wrapDb(() => db.$transaction((tx) => fence(tx, jobId, workerId, 'done'))); +} + +/** Marks the worker's job failed, saying why; `Ok(false)` when it was no longer the worker's. */ +export async function failJob( + jobId: string, + workerId: string, + error: string, +): Promise> { + return wrapDb(() => db.$transaction((tx) => fence(tx, jobId, workerId, 'failed', error))); +} + +/** Puts the worker's job back in the queue for later; `Ok(false)` when it was no longer the worker's. */ +export async function requeueJob( + jobId: string, + workerId: string, + runAt: Date, +): Promise> { + return wrapDb(() => db.$transaction((tx) => requeue(tx, jobId, workerId, runAt))); +} + +/** What the model answered: its text, its tool calls, and what they cost. */ +export type AssistantTurn = { + text: string; + toolCalls: ToolCall[]; + inputTokens: number; + outputTokens: number; +}; + +/** + * Finishes an inference job with the model's answer: the assistant entry, and a tool job for each + * of its calls, due at `runAt` where a call says when (a sleep), else now. `Ok(false)` when the + * job was no longer the worker's, and nothing is written. + */ +export async function completeInference( + jobId: string, + workerId: string, + threadId: string, + turn: AssistantTurn, + runAt: Record, +): Promise> { + return wrapDb(() => + db.$transaction(async (tx) => { + await lockThread(tx, threadId); + if (!(await fence(tx, jobId, workerId, 'done'))) { + return false; + } + const entry = await tx.agentEntry.create({ + data: { + threadId, + kind: 'assistant' satisfies AgentEntryKind, + text: turn.text, + toolCalls: turn.toolCalls.length === 0 ? null : JSON.stringify(turn.toolCalls), + inputTokens: turn.inputTokens, + outputTokens: turn.outputTokens, + }, + select: {id: true}, + }); + if (turn.toolCalls.length > 0) { + await tx.agentJob.createMany({ + data: turn.toolCalls.map((call) => ({ + threadId, + kind: 'tool' satisfies AgentJobKind, + entryId: entry.id, + callId: call.id, + runAt: runAt[call.id] ?? new Date(), + })), + }); + } + return true; + }), + ); +} + +/** + * Finishes a tool job with what the call answered, and wakes the agent once every call of the + * answer has its result. `Ok(false)` when the job was no longer the worker's, and nothing is + * written. + */ +export async function completeToolJob( + jobId: string, + workerId: string, + threadId: string, + result: {callId: string; text: string; isError: boolean}, +): Promise> { + return wrapDb(() => + db.$transaction(async (tx) => { + await lockThread(tx, threadId); + if (!(await fence(tx, jobId, workerId, 'done'))) { + return false; + } + await tx.agentEntry.create({ + data: { + threadId, + kind: 'tool_result' satisfies AgentEntryKind, + text: result.text, + callId: result.callId, + isError: result.isError, + }, + }); + const entries = ( + await tx.agentEntry.findMany({ + where: {threadId}, + select: entrySelect, + orderBy: {seq: 'asc'}, + }) + ).map(toEntry); + if (pendingToolCalls(entries).length === 0) { + await queueInference(tx, threadId); + } + return true; + }), + ); +} + +/** + * Finishes an inference job that found nothing to do, with the thread locked, so a result landing + * meanwhile queues the next one. `Ok(false)` when the job was no longer the worker's. + */ +export async function finishIdleInference( + jobId: string, + workerId: string, + threadId: string, +): Promise> { + return wrapDb(() => + db.$transaction(async (tx) => { + await lockThread(tx, threadId); + return fence(tx, jobId, workerId, 'done'); + }), + ); +} + +/** The assistant entry a tool job's call is in; `Err` not found when it is gone. */ +export async function getEntry(entryId: string): Promise> { + const found = await wrapDb(() => + db.agentEntry.findUnique({where: {id: entryId}, select: entrySelect}), + ); + if (found.isErr()) { + return found; + } + if (found.value === null) { + return Err(ApiErr.notFound('entry', entryId)); + } + return Ok(toEntry(found.value)); +} diff --git a/packages/docs/content/agents/chat.mdx b/packages/docs/content/agents/chat.mdx new file mode 100644 index 00000000..c9fe8909 --- /dev/null +++ b/packages/docs/content/agents/chat.mdx @@ -0,0 +1,40 @@ +--- +title: Chat with the built-in agent +description: Talk to an agent Personal Agent Proxy runs itself, with the same access settings and activity log as any agent you connect. +--- + +**Chat** in the sidebar is a conversation with an agent Personal Agent Proxy runs itself. It is an +agent login like any other, named Personal Agent Proxy on the Agents page: what it may do with each +connection is set there, as for Claude or ChatGPT, and every call it makes is in the activity log, +by the conversation it was made for. It can use what you connected, and the vault as its memory. + +What it does differently from an agent you connect is where it runs: in Personal Agent Proxy, not +on another company's servers. The model is Claude Sonnet 5.5, reached through OpenRouter with a +host that neither keeps prompts nor trains on them, so a self-hosted Personal Agent Proxy needs an +`OPENROUTER_KEY`. + +## How a conversation runs + +Say what you want done. The agent answers, or makes calls first: each shows as a card under its +message, with what it asked and, once it has it, what came back. A call it is not allowed is +refused and logged as blocked, as it would be for any agent. It keeps working after you leave the +page, and the page shows the conversation as it is when you come back. + +It can also wait: asked to check on something later, or to do something at a time, it sleeps until +then, and the page says until when. It wakes with the conversation and nothing else, so it says +what it will do before it sleeps. + +**Activity for this conversation** under the input opens the activity log narrowed to the +conversation's task. + + +Chat is being tried out with some organizations first, so it is only offered to an organization +that has it turned on. On a self-hosted Personal Agent Proxy, turn it on for every organization in +its database: + +```sql +UPDATE "Organization" SET experiments = array_append(experiments, 'harness') +WHERE NOT 'harness' = ANY(experiments); +``` + + diff --git a/packages/docs/src/nav.ts b/packages/docs/src/nav.ts index 568cb7fb..7adf1ccd 100644 --- a/packages/docs/src/nav.ts +++ b/packages/docs/src/nav.ts @@ -33,7 +33,7 @@ export const NAV: {title: string; pages: NavEntry[]}[] = [ 'connections/custom-mcp', ], }, - {title: 'Agents', pages: ['agents/mcp']}, + {title: 'Agents', pages: ['agents/mcp', 'agents/chat']}, { title: 'Self-hosting', pages: [ diff --git a/packages/integrations/src/agent-providers.test.ts b/packages/integrations/src/agent-providers.test.ts index 49706ff6..f5c53609 100644 --- a/packages/integrations/src/agent-providers.test.ts +++ b/packages/integrations/src/agent-providers.test.ts @@ -1,6 +1,11 @@ import {describe, expect, test} from 'bun:test'; -import {AGENT_PROVIDERS, findAgentProvider, suggestAgentProvider} from './agent-providers'; +import { + AGENT_PROVIDERS, + BUILT_IN_AGENT_PROVIDER, + findAgentProvider, + suggestAgentProvider, +} from './agent-providers'; describe('agent providers', () => { test('have unique ids', () => { @@ -30,11 +35,21 @@ describe('agent providers', () => { } }); - test('finds a provider by id', () => { + test('finds a provider by id, the built-in agent too, which is not on offer', () => { expect(findAgentProvider('dot')?.name).toBe('Dots'); + expect(findAgentProvider('proxy')).toBe(BUILT_IN_AGENT_PROVIDER); + expect(AGENT_PROVIDERS).not.toContain(BUILT_IN_AGENT_PROVIDER); expect(findAgentProvider('other')).toBeUndefined(); }); + test('the built-in agent uses the app’s own icon', async () => { + expect( + await Bun.file( + new URL(`../../web/public${BUILT_IN_AGENT_PROVIDER.faviconUrl}`, import.meta.url), + ).exists(), + ).toBe(true); + }); + test("suggests the provider an MCP client's name points at", () => { expect(suggestAgentProvider('Claude')?.id).toBe('claude'); expect(suggestAgentProvider('Claude Code')?.id).toBe('claude-code'); diff --git a/packages/integrations/src/agent-providers.ts b/packages/integrations/src/agent-providers.ts index 82b685aa..ace2d571 100644 --- a/packages/integrations/src/agent-providers.ts +++ b/packages/integrations/src/agent-providers.ts @@ -9,7 +9,8 @@ export type AgentProviderId = | 'claude-desktop' | 'hermes' | 'chatgpt' - | 'custom'; + | 'custom' + | 'proxy'; export type AgentProvider = { id: AgentProviderId; @@ -18,9 +19,10 @@ export type AgentProvider = { company: string; /** * How the agent gets its access: `mcp` adds Personal Agent Proxy as a connector and signs in - * over OAuth, which makes its login; `login` is handed a username and password. + * over OAuth, which makes its login; `login` is handed a username and password; `built-in` is + * Personal Agent Proxy's own agent, which needs neither. */ - connects: 'mcp' | 'login'; + connects: 'mcp' | 'login' | 'built-in'; /** * Runs on the person's own computer, so it reaches a Personal Agent Proxy there, such as one * in development at `http://localhost`. Every other agent needs a public https address. @@ -40,6 +42,20 @@ export type AgentProvider = { faviconUrl?: string; }; +/** + * The agent Personal Agent Proxy runs itself, on the Chat page: an agent login like any other for + * its access and its activity, made by Personal Agent Proxy the first time the person chats, never + * handed out, so it is not among the providers a person can make a login for. + */ +export const BUILT_IN_AGENT_PROVIDER: AgentProvider = { + id: 'proxy', + name: 'Personal Agent Proxy', + company: 'Personal Agent Proxy', + connects: 'built-in', + local: true, + faviconUrl: '/favicon.svg', +}; + // The agents an organization can make a login for: one each, and any number of custom ones. export const AGENT_PROVIDERS: AgentProvider[] = [ { @@ -135,7 +151,11 @@ export const AGENT_PROVIDERS: AgentProvider[] = [ }, ]; +/** A provider by id, the built-in agent's among them. */ export function findAgentProvider(id: string): AgentProvider | undefined { + if (id === BUILT_IN_AGENT_PROVIDER.id) { + return BUILT_IN_AGENT_PROVIDER; + } return AGENT_PROVIDERS.find((provider) => provider.id === id); } diff --git a/packages/integrations/src/index.ts b/packages/integrations/src/index.ts index 0d222458..dc029408 100644 --- a/packages/integrations/src/index.ts +++ b/packages/integrations/src/index.ts @@ -1,5 +1,6 @@ export { AGENT_PROVIDERS, + BUILT_IN_AGENT_PROVIDER, CUSTOM_AGENT_NAME_MAX, findAgentProvider, suggestAgentProvider, diff --git a/packages/ui/src/styles.css b/packages/ui/src/styles.css index ea570e82..5df76e29 100644 --- a/packages/ui/src/styles.css +++ b/packages/ui/src/styles.css @@ -116,7 +116,7 @@ --sidebar-foreground: oklch(0.145 0 0); --sidebar-primary: oklch(0.205 0 0); --sidebar-primary-foreground: oklch(0.985 0 0); - --sidebar-accent: oklch(0.97 0 0); + --sidebar-accent: oklch(0.95 0 0); --sidebar-accent-foreground: oklch(0.205 0 0); --sidebar-border: oklch(0.922 0 0); --sidebar-ring: oklch(0.708 0 0); diff --git a/packages/utils/package.json b/packages/utils/package.json index 430545d1..37f584a0 100644 --- a/packages/utils/package.json +++ b/packages/utils/package.json @@ -6,6 +6,7 @@ "exports": { ".": "./src/index.ts", "./days": "./src/days.ts", + "./harness": "./src/harness.ts", "./tasks": "./src/tasks.ts", "./units": "./src/units.ts", "./transaction-categories": "./src/transaction-categories.ts" diff --git a/packages/utils/src/experiments.ts b/packages/utils/src/experiments.ts index 33dc1c99..2b5c326a 100644 --- a/packages/utils/src/experiments.ts +++ b/packages/utils/src/experiments.ts @@ -8,6 +8,9 @@ export const EXPERIMENTS = { // The Ask quick action: what a meeting, document, email or issue says, answered from reading it. ask: 'Ask quick action', + // The built-in agent: a conversation with an agent Personal Agent Proxy runs itself, behind its + // own gate, on the Chat page. + harness: 'Built-in agent', } as const satisfies Record; export type Experiment = keyof typeof EXPERIMENTS; diff --git a/packages/utils/src/harness.test.ts b/packages/utils/src/harness.test.ts new file mode 100644 index 00000000..d3e057ec --- /dev/null +++ b/packages/utils/src/harness.test.ts @@ -0,0 +1,68 @@ +import {describe, expect, test} from 'bun:test'; + +import {type LogEntry, pendingToolCalls, readyForInference, sleepUntil} from './harness'; + +const person = (text: string): LogEntry => ({kind: 'person', text, toolCalls: null, callId: null}); +const assistant = (text: string, calls: Array<{id: string; name: string}> = []): LogEntry => ({ + kind: 'assistant', + text, + toolCalls: calls.map((call) => ({...call, arguments: '{}'})), + callId: null, +}); +const result = (callId: string): LogEntry => ({ + kind: 'tool_result', + text: 'ok', + toolCalls: null, + callId, +}); + +describe('pendingToolCalls', () => { + test('is the last answer’s calls without a result, and none before an answer', () => { + expect(pendingToolCalls([person('hi')])).toEqual([]); + const log = [ + person('hi'), + assistant('', [ + {id: 'a', name: 'read'}, + {id: 'b', name: 'read'}, + ]), + result('a'), + ]; + expect(pendingToolCalls(log).map((call) => call.id)).toEqual(['b']); + expect(pendingToolCalls([...log, result('b')])).toEqual([]); + }); + + test('only the last answer’s calls count', () => { + const log = [assistant('', [{id: 'a', name: 'read'}]), assistant('done')]; + expect(pendingToolCalls(log)).toEqual([]); + }); +}); + +describe('readyForInference', () => { + test('after the person spoke, or every call has its result, but not after the model’s own answer', () => { + expect(readyForInference([])).toBe(false); + expect(readyForInference([person('hi')])).toBe(true); + expect(readyForInference([person('hi'), assistant('hello')])).toBe(false); + const calling = [person('hi'), assistant('', [{id: 'a', name: 'read'}])]; + expect(readyForInference(calling)).toBe(false); + expect(readyForInference([...calling, result('a')])).toBe(true); + }); +}); + +describe('sleepUntil', () => { + const now = new Date('2026-10-21T10:00:00Z'); + + test('takes a time or seconds from now, never in the past and at most a week', () => { + expect(sleepUntil({until: '2026-10-21T10:05:00Z'}, now)).toEqual( + new Date('2026-10-21T10:05:00Z'), + ); + expect(sleepUntil({seconds: 120}, now)).toEqual(new Date('2026-10-21T10:02:00Z')); + expect(sleepUntil({until: '2026-10-21T09:00:00Z'}, now)).toEqual(now); + expect(sleepUntil({seconds: 60 * 60 * 24 * 30}, now)).toEqual(new Date('2026-10-28T10:00:00Z')); + }); + + test('is null for anything else', () => { + expect(sleepUntil({until: 'tomorrow'}, now)).toBeNull(); + expect(sleepUntil({}, now)).toBeNull(); + expect(sleepUntil('soon', now)).toBeNull(); + }); +}); diff --git a/packages/utils/src/harness.ts b/packages/utils/src/harness.ts new file mode 100644 index 00000000..cb8be80b --- /dev/null +++ b/packages/utils/src/harness.ts @@ -0,0 +1,74 @@ +/** + * The built-in agent's log, as the api writes it and the web app reads it: what is shared between + * them. The log is the transcript: the person's messages, the model's answers with the tool calls + * it made, and what each call answered. A call is pending until its result is in the log, and the + * model is run again only once none is. + */ + +/** A tool call the model made, as the model API gives it: its arguments are the JSON it wrote. */ +export type ToolCall = {id: string; name: string; arguments: string}; + +export type LogEntry = { + kind: 'person' | 'assistant' | 'tool_result'; + text: string; + toolCalls: ToolCall[] | null; + callId: string | null; +}; + +/** The calls of the last assistant entry that have no result yet; none when the log is ready for the model. */ +export function pendingToolCalls(entries: LogEntry[]): ToolCall[] { + const last = entries.findLastIndex((entry) => entry.kind === 'assistant'); + if (last === -1) { + return []; + } + const answered = new Set( + entries + .slice(last + 1) + .filter((entry) => entry.kind === 'tool_result') + .map((entry) => entry.callId), + ); + return (entries[last]?.toolCalls ?? []).filter((call) => !answered.has(call.id)); +} + +/** + * Whether the model should be run on the log: something has come since it last answered, and + * every call it made has its result. + */ +export function readyForInference(entries: LogEntry[]): boolean { + if (pendingToolCalls(entries).length > 0) { + return false; + } + const last = entries.at(-1); + if (!last) { + return false; + } + return last.kind !== 'assistant'; +} + +/** The name of the one tool the built-in agent has beyond its connections: a wake-up later. */ +export const SLEEP_TOOL = 'sleep_until'; + +// How long a sleep may be at most. +const MAX_SLEEP_MS = 7 * 24 * 60 * 60 * 1000; + +/** + * When a call to sleep wakes: `{until}` as an ISO time, or `{seconds}` from now; in the past is + * now, and more than a week is cut to a week. Null when the arguments say neither. + */ +export function sleepUntil(args: unknown, now: Date): Date | null { + if (typeof args !== 'object' || args === null) { + return null; + } + const {until, seconds} = args as {until?: unknown; seconds?: unknown}; + const requested = + typeof seconds === 'number' + ? new Date(now.getTime() + seconds * 1000) + : typeof until === 'string' + ? new Date(until) + : null; + if (requested === null || Number.isNaN(requested.getTime())) { + return null; + } + const earliest = Math.max(requested.getTime(), now.getTime()); + return new Date(Math.min(earliest, now.getTime() + MAX_SLEEP_MS)); +} diff --git a/packages/web/src/client/harness-client.ts b/packages/web/src/client/harness-client.ts new file mode 100644 index 00000000..72547697 --- /dev/null +++ b/packages/web/src/client/harness-client.ts @@ -0,0 +1,64 @@ +import {z} from 'zod'; +import {apiRequest} from '@/client/request'; + +// The Chat page's api: the person's threads with the built-in agent, a thread's log, and what +// they say in it. + +const threadSchema = z.object({ + id: z.string(), + createdAt: z.string(), + // The task every call in the thread is made for, which the activity log can be narrowed to. + taskId: z.string(), + // What the person first said, as the thread's name; null before they said anything. + title: z.string().nullable(), +}); + +const toolCallSchema = z.object({id: z.string(), name: z.string(), arguments: z.string()}); + +const entrySchema = z.object({ + id: z.string(), + kind: z.enum(['person', 'assistant', 'tool_result']), + text: z.string(), + // An assistant entry's calls; null otherwise. + toolCalls: z.array(toolCallSchema).nullable(), + // A tool result's call; null otherwise. + callId: z.string().nullable(), + isError: z.boolean(), + at: z.string(), +}); + +const threadLogSchema = z.object({ + thread: threadSchema, + entries: z.array(entrySchema), + // Whether the agent is at work on the thread, or asleep until a time. + busy: z.boolean(), + sleepingUntil: z.string().nullable(), +}); + +export type ChatThread = z.infer; +export type ChatEntry = z.infer; +export type ChatToolCall = z.infer; +export type ThreadLog = z.infer; + +export function listThreads() { + return apiRequest('GET', '/api/chat/threads', z.object({threads: z.array(threadSchema)})); +} + +export function createThread() { + return apiRequest('POST', '/api/chat/threads', z.object({thread: threadSchema})); +} + +export function getThread(threadId: string) { + return apiRequest('GET', `/api/chat/threads/${threadId}`, threadLogSchema); +} + +export function postMessage(threadId: string, text: string) { + return apiRequest( + 'POST', + `/api/chat/threads/${threadId}/messages`, + z.object({entry: entrySchema}), + { + text, + }, + ); +} diff --git a/packages/web/src/client/schemas.ts b/packages/web/src/client/schemas.ts index 92777644..d6842ba3 100644 --- a/packages/web/src/client/schemas.ts +++ b/packages/web/src/client/schemas.ts @@ -53,8 +53,8 @@ export const auditEntrySchema = z.object({ recordTitle: z.string().nullable(), query: z.string().nullable(), outcome: z.enum(['allowed', 'denied']), - // The web app's agent pages, or the MCP server. - via: z.enum(['web', 'mcp']), + // The web app's agent pages, the MCP server, or the built-in agent's own loop. + via: z.enum(['web', 'mcp', 'harness']), // The agent's task it was made for, among the answer's tasks; none for entries from before // tasks, and for what Personal Agent Proxy did itself. taskId: z.string().nullable(), @@ -66,7 +66,7 @@ export const activityTaskSchema = z.object({ agentId: z.string(), title: z.string(), goal: z.string(), - via: z.enum(['web', 'mcp']), + via: z.enum(['web', 'mcp', 'harness']), startedAt: z.string(), // Null while the agent hasn't called it finished. finishedAt: z.string().nullable(), diff --git a/packages/web/src/components/app-shell.tsx b/packages/web/src/components/app-shell.tsx index a726a0cb..b851709a 100644 --- a/packages/web/src/components/app-shell.tsx +++ b/packages/web/src/components/app-shell.tsx @@ -1,7 +1,12 @@ -import type {ReactNode} from 'react'; +import {PanelLeftIcon} from 'lucide-react'; +import {type ReactNode, useState} from 'react'; +import {useLocation} from 'react-router'; import {AppSidebar} from '@/components/nav'; +import {Button} from '@proxy/ui/components/button'; +import {Sheet, SheetContent, SheetTitle} from '@proxy/ui/components/sheet'; import {SidebarTrigger} from '@proxy/ui/components/sidebar'; import {useDesktop} from '@proxy/ui/hooks/use-media-query'; +import {cn} from '@proxy/ui/lib/utils'; type Props = { children: ReactNode; @@ -9,21 +14,72 @@ type Props = { // button alone in its corner. title?: ReactNode; actions?: ReactNode; + // A second sidebar beside the first, as Chat's conversations: a column on desktop, and on a + // phone a sheet opened from the header, named by `panelLabel`. + panel?: ReactNode; + panelLabel?: string; + // `inside`: the page is as tall as the window and scrolls a part of itself, as Chat scrolls its + // log, so the header, the sidebars and the rest stay put, the overscroll bounce included. + // Otherwise the whole page scrolls. + scroll?: 'page' | 'inside'; }; +// The second sidebar on a phone: a sheet from the left. It is keyed by the path, so a move +// elsewhere closes it. +function PanelSheet({panel, label}: {panel: ReactNode; label: string}) { + const [open, setOpen] = useState(false); + + return ( + + + + {label} + {panel} + + + ); +} + // The human side's frame: the sidebar, a header with the page's title and actions, and the page // in one centered column. -export function AppShell({children, title, actions}: Props) { +export function AppShell({ + children, + title, + actions, + panel, + panelLabel = 'Panel', + scroll = 'page', +}: Props) { const desktop = useDesktop(); + const {pathname} = useLocation(); return ( <> -
+ {panel !== undefined && desktop && ( + + )} +
{title !== undefined && (
{!desktop && } + {!desktop && panel !== undefined && ( + + )} {title}
{actions &&
{actions}
} @@ -34,7 +90,14 @@ export function AppShell({children, title, actions}: Props) {
)} -
+
{children}
diff --git a/packages/web/src/components/audit-entry-dialog.tsx b/packages/web/src/components/audit-entry-dialog.tsx index 564a6f3c..80a5459f 100644 --- a/packages/web/src/components/audit-entry-dialog.tsx +++ b/packages/web/src/components/audit-entry-dialog.tsx @@ -36,6 +36,9 @@ function formatMoment(iso: string): string { }); } +// How the request came in, as the entry says. +const THROUGH = {web: 'Agent web pages', mcp: 'MCP', harness: 'Built-in agent'} as const; + function Detail({label, children}: {label: string; children: ReactNode}) { return (
@@ -174,7 +177,7 @@ function AuditEntryContent({ {action ? action.label : Unknown} - {entry.via === 'mcp' ? 'MCP' : 'Agent web pages'} + {THROUGH[entry.via]} ); diff --git a/packages/web/src/components/nav.tsx b/packages/web/src/components/nav.tsx index 3d916e0d..408b899c 100644 --- a/packages/web/src/components/nav.tsx +++ b/packages/web/src/components/nav.tsx @@ -3,15 +3,18 @@ import { HomeIcon, LibraryIcon, LogOutIcon, + MessageSquareIcon, PlugIcon, ScrollTextIcon, SettingsIcon, type LucideIcon, } from 'lucide-react'; +import type {Experiment} from '@proxy/utils'; import {Link, useLocation, useRouteLoaderData} from 'react-router'; import {signOut} from '@/client/auth-client'; import type {Me} from '@/client/me-client'; import {ThemeMenu} from '@/components/theme-menu'; +import {useExperiments} from '@/hooks/use-experiment'; import { Sidebar, SidebarContent, @@ -23,10 +26,12 @@ import { useSidebar, } from '@proxy/ui/components/sidebar'; -type Destination = {to: string; label: string; icon: LucideIcon}; +// `experiment` is the experiment a destination is behind, shown only with it on. +type Destination = {to: string; label: string; icon: LucideIcon; experiment?: Experiment}; const DESTINATIONS: Destination[] = [ {to: '/', label: 'Home', icon: HomeIcon}, + {to: '/chat', label: 'Chat', icon: MessageSquareIcon, experiment: 'harness'}, {to: '/connections', label: 'Connections', icon: PlugIcon}, {to: '/vault', label: 'Vault', icon: LibraryIcon}, {to: '/agents', label: 'Agents', icon: BotIcon}, @@ -44,14 +49,19 @@ function isActive(pathname: string, to: string): boolean { export function AppSidebar() { const {pathname} = useLocation(); const me = useRouteLoaderData('human'); + const experiments = useExperiments(); const {setOpenMobile} = useSidebar(); + const destinations = DESTINATIONS.filter( + (destination) => + destination.experiment === undefined || experiments.includes(destination.experiment), + ); return ( - {DESTINATIONS.map((destination) => ( + {destinations.map((destination) => ( (change: Promise>) => { const result = await change; if (result.isErr()) { @@ -95,7 +98,7 @@ function AgentDetail({id}: {id: string}) { } actions={ <> - {!revoked && ( + {!revoked && !builtIn && ( diff --git a/packages/web/src/pages/chat-page.tsx b/packages/web/src/pages/chat-page.tsx new file mode 100644 index 00000000..f993989c --- /dev/null +++ b/packages/web/src/pages/chat-page.tsx @@ -0,0 +1,347 @@ +import {ChevronRightIcon, MoonIcon, PlusIcon, SearchIcon, SendIcon} from 'lucide-react'; +import {type KeyboardEvent, useEffect, useRef, useState} from 'react'; +import {Link, useLoaderData, useNavigate, useRevalidator} from 'react-router'; +import { + type ChatEntry, + type ChatThread, + type ChatToolCall, + createThread, + getThread, + postMessage, + type ThreadLog, +} from '@/client/harness-client'; +import {AppShell, PageTitle} from '@/components/app-shell'; +import {Button} from '@proxy/ui/components/button'; +import {Input} from '@proxy/ui/components/input'; +import {Textarea} from '@proxy/ui/components/textarea'; +import {cn} from '@proxy/ui/lib/utils'; +import {formatDateTime} from '@/lib/format'; +import {describeFetchError} from '@/lib/loader-utils'; +import type {chatLoader} from '@/loaders'; + +// How often the log is read again while the agent is at work, and while it is not: a reply, or a +// wake-up, shows within a moment either way. +const BUSY_POLL_MS = 1000; +const IDLE_POLL_MS = 4000; + +// How much of a call's arguments the card shows on its line. +const ARGS_PREVIEW = 80; + +function preview(json: string): string { + const text = json.replace(/\s+/g, ' '); + return text.length > ARGS_PREVIEW ? `${text.slice(0, ARGS_PREVIEW - 1)}…` : text; +} + +// One call the agent made, with what it answered once it has: opened for the whole of both. +function ToolCallCard({call, result}: {call: ChatToolCall; result: ChatEntry | undefined}) { + const [open, setOpen] = useState(false); + return ( +
+ + {open && ( +
+ 
{call.arguments}
+ {result && ( + 
+							{result.text}
+
+ )} +
+ )} +
+ ); +} + +// The log as a conversation: what the person said on the right, what the agent said on the left +// with its calls under it, each call's answer folded into its card. +function Log({entries}: {entries: ChatEntry[]}) { + const results = new Map( + entries.filter((entry) => entry.kind === 'tool_result').map((entry) => [entry.callId, entry]), + ); + return ( +
    + {entries + .filter((entry) => entry.kind !== 'tool_result') + .map((entry) => ( +
  1. + {entry.text !== '' && ( +
    + {entry.text} +
    + )} + {entry.toolCalls?.map((call) => ( +
    + +
    + ))} + {formatDateTime(entry.at)} +
    2. + ))} +
+ ); +} + +// What a thread is called: what the person first said in it, or when it was started. +function threadLabel(thread: ChatThread): string { + return thread.title ?? `New conversation · ${formatDateTime(thread.createdAt)}`; +} + +// The conversations in the second sidebar: a search over their names, a row to start a new one, +// and the recent ones newest first, the open one lit. +function ConversationList({ + threads, + current, + onNew, +}: { + threads: ChatThread[]; + current: string; + onNew: () => void; +}) { + const [search, setSearch] = useState(''); + const needle = search.trim().toLowerCase(); + const shown = threads.filter( + (thread) => needle === '' || threadLabel(thread).toLowerCase().includes(needle), + ); + const rowClass = + 'flex w-full items-center gap-2 rounded-md px-2 py-1.5 text-left text-sm hover:bg-sidebar-accent hover:text-sidebar-accent-foreground'; + + return ( +
+
+ + setSearch(event.target.value)} + placeholder="Search" + aria-label="Search conversations" + className="h-9 bg-background pl-8" + /> +
+ +
+ Recents + {shown.length === 0 && ( + + {needle === '' ? 'No conversations yet.' : 'Nothing matches.'} + + )} +
    + {shown.map((thread) => ( +
  • + + {threadLabel(thread)} + +
    • + ))} +
+
+
+ ); +} + +type Loaded = Awaited>; + +// The page: the conversation in its address, started over when the address changes, so what one +// conversation showed never lingers on another. +export function ChatPage() { + const loaded = useLoaderData(); + return ; +} + +// A conversation with the built-in agent: its log, read again as the agent works, and an input. +// The agent's calls go through the same gate as any agent's, so its activity is in the log by the +// conversation's task. +function Conversation({loaded}: {loaded: Loaded}) { + const navigate = useNavigate(); + const revalidator = useRevalidator(); + const [log, setLog] = useState(loaded.log); + const [text, setText] = useState(''); + const [sending, setSending] = useState(false); + const [error, setError] = useState(null); + // The log, which scrolls on its own; the input under it stays put. + const logView = useRef(null); + const threadId = loaded.log.thread.id; + // The thread as the list names it, which the log's own row doesn't. + const current = loaded.threads.find((thread) => thread.id === threadId) ?? loaded.log.thread; + + // The log as it is now, while the page is open. + useEffect(() => { + let cancelled = false; + const refresh = async () => { + const result = await getThread(threadId); + if (!cancelled && result.isOk()) { + setLog(result.value); + } + }; + const timer = setInterval(() => void refresh(), log.busy ? BUSY_POLL_MS : IDLE_POLL_MS); + return () => { + cancelled = true; + clearInterval(timer); + }; + }, [threadId, log.busy]); + + useEffect(() => { + logView.current?.scrollTo({top: logView.current.scrollHeight}); + }, [log.entries.length, log.busy]); + + const send = async () => { + const message = text.trim(); + if (message === '' || sending) { + return; + } + setSending(true); + setError(null); + const posted = await postMessage(threadId, message); + setSending(false); + if (posted.isErr()) { + setError(describeFetchError(posted.error)); + return; + } + setText(''); + const refreshed = await getThread(threadId); + if (refreshed.isOk()) { + setLog(refreshed.value); + } + // The first message names the thread in the list. + if (current.title === null) { + await revalidator.revalidate(); + } + }; + + const onKeyDown = (event: KeyboardEvent) => { + if (event.key === 'Enter' && !event.shiftKey) { + event.preventDefault(); + void send(); + } + }; + + const startThread = async () => { + const made = await createThread(); + if (made.isErr()) { + setError(describeFetchError(made.error)); + return; + } + void navigate(`/chat/${made.value.thread.id}`); + }; + + return ( + Chat} + panel={ + void startThread()} + /> + } + panelLabel="Conversations" + scroll="inside" + > +
+
+ {log.entries.length === 0 ? ( +

+ Say what you want done. The agent can use what you connected, as far as its access on + the Agents page allows, and every call it makes is in the activity log. +

+ ) : ( + + )} + {log.busy &&

Working…

} + {log.sleepingUntil && ( +

+ + Asleep until {formatDateTime(log.sleepingUntil)} +

+ )} +
+
+
+ {error &&

{error}

} +
+