diff --git a/packages/app/src/electro-bridge/ipc/agent-host-api.ts b/packages/app/src/electro-bridge/ipc/agent-host-api.ts index 7955b1b2..885cf8d3 100644 --- a/packages/app/src/electro-bridge/ipc/agent-host-api.ts +++ b/packages/app/src/electro-bridge/ipc/agent-host-api.ts @@ -9,6 +9,9 @@ export const AGENT_HOST_CHANNELS = { ENQUEUE: "agent-host:enqueue", READY: "agent-host:ready", LIST_JOBS: "agent-host:list-jobs", + LIST_TASKS: "agent-host:list-tasks", + CONTROL_TASK: "agent-host:control-task", + REDIRECT_TASK: "agent-host:redirect-task", CANCEL: "agent-host:cancel", RESPOND: "agent-host:respond", REQUEST: "agent-host:request", @@ -26,6 +29,12 @@ export function createAgentHostAPI( ready: () => invoke(AGENT_HOST_CHANNELS.READY), enqueue: (dispatch) => invoke(AGENT_HOST_CHANNELS.ENQUEUE, dispatch), listJobs: () => invoke(AGENT_HOST_CHANNELS.LIST_JOBS), + listTasks: (agentMemberId) => + invoke(AGENT_HOST_CHANNELS.LIST_TASKS, agentMemberId), + controlTask: (taskId, action) => + invoke(AGENT_HOST_CHANNELS.CONTROL_TASK, taskId, action), + redirectTask: (taskId, instruction) => + invoke(AGENT_HOST_CHANNELS.REDIRECT_TASK, taskId, instruction), cancel: (jobId) => invoke(AGENT_HOST_CHANNELS.CANCEL, jobId), respond: (response) => invoke(AGENT_HOST_CHANNELS.RESPOND, response), onRequest: (callback) => { diff --git a/packages/app/src/electro-bridge/ipc/agent-host-context.test.ts b/packages/app/src/electro-bridge/ipc/agent-host-context.test.ts index 21c23a56..26d627b3 100644 --- a/packages/app/src/electro-bridge/ipc/agent-host-context.test.ts +++ b/packages/app/src/electro-bridge/ipc/agent-host-context.test.ts @@ -45,6 +45,7 @@ function mainIPC() { const dispatch: AgentHostDispatch = { channelId: "channel", + channelKind: "channel", conversationId: "conversation", triggerMessageId: "message", contextMessageIds: ["message"], @@ -129,4 +130,47 @@ describe("Agent Host IPC", () => { expect(host.enqueue).not.toHaveBeenCalled(); expect(bridge.respond).not.toHaveBeenCalled(); }); + + it("routes task controls through validated IPC methods", async () => { + const sender = new FakeWebContents(); + const host = { + listTasks: vi.fn(async () => [{ id: "task-1" }]), + pauseTask: vi.fn(async () => true), + resumeTask: vi.fn(async () => true), + cancelTask: vi.fn(async () => true), + redirectTask: vi.fn(async () => ({ id: "job-2" })), + } as unknown as AgentHost; + const { handlers, ipc } = mainIPC(); + setupAgentHostIPC( + { host, getAllowedWebContents: () => sender as never }, + ipc as never, + ); + + expect( + await handlers.get(AGENT_HOST_CHANNELS.LIST_TASKS)?.( + event(sender), + "agent:fizz" as never, + ), + ).toEqual({ success: true, tasks: [{ id: "task-1" }] }); + expect( + await handlers.get(AGENT_HOST_CHANNELS.CONTROL_TASK)?.( + event(sender), + "task-1" as never, + "pause" as never, + ), + ).toEqual({ success: true, changed: true }); + expect( + await handlers.get(AGENT_HOST_CHANNELS.REDIRECT_TASK)?.( + event(sender), + "task-1" as never, + "Show the diff first" as never, + ), + ).toEqual({ success: true, job: { id: "job-2" } }); + expect(host.listTasks).toHaveBeenCalledWith("agent:fizz"); + expect(host.pauseTask).toHaveBeenCalledWith("task-1"); + expect(host.redirectTask).toHaveBeenCalledWith( + "task-1", + "Show the diff first", + ); + }); }); diff --git a/packages/app/src/electro-bridge/ipc/agent-host-context.ts b/packages/app/src/electro-bridge/ipc/agent-host-context.ts index 48da84e6..6c141e8f 100644 --- a/packages/app/src/electro-bridge/ipc/agent-host-context.ts +++ b/packages/app/src/electro-bridge/ipc/agent-host-context.ts @@ -1,6 +1,7 @@ import type { AgentHostDispatch, AgentHostRendererResponse, + AgentHostTaskAction, } from "@/shared/types/agent-host"; import type { AgentHost } from "@/electron/agent-host/host"; import type { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge"; @@ -38,6 +39,9 @@ export function setupAgentHostIPC( AGENT_HOST_CHANNELS.ENQUEUE, AGENT_HOST_CHANNELS.READY, AGENT_HOST_CHANNELS.LIST_JOBS, + AGENT_HOST_CHANNELS.LIST_TASKS, + AGENT_HOST_CHANNELS.CONTROL_TASK, + AGENT_HOST_CHANNELS.REDIRECT_TASK, AGENT_HOST_CHANNELS.CANCEL, AGENT_HOST_CHANNELS.RESPOND, ]) { @@ -89,6 +93,96 @@ export function setupAgentHostIPC( } }); + mainIPC.handle( + AGENT_HOST_CHANNELS.LIST_TASKS, + async (event, agentMemberId: unknown) => { + if (!allowed(event, options)) { + return { + success: false, + error: "Agent Host IPC sender is not allowed.", + }; + } + if (!options.host) { + return { success: false, error: "Agent Host is unavailable." }; + } + if (agentMemberId !== undefined && typeof agentMemberId !== "string") { + return { success: false, error: "Agent member id must be a string." }; + } + try { + return { + success: true, + tasks: await options.host.listTasks(agentMemberId), + }; + } catch (error) { + return { success: false, error: errorMessage(error) }; + } + }, + ); + + mainIPC.handle( + AGENT_HOST_CHANNELS.CONTROL_TASK, + async (event, taskId: unknown, action: unknown) => { + if (!allowed(event, options)) { + return { + success: false, + error: "Agent Host IPC sender is not allowed.", + }; + } + if (!options.host) { + return { success: false, error: "Agent Host is unavailable." }; + } + if (typeof taskId !== "string" || !taskId) { + return { success: false, error: "A task id is required." }; + } + if (!isTaskAction(action)) { + return { + success: false, + error: "Task action must be pause, resume, or cancel.", + }; + } + try { + const changed = + action === "pause" + ? await options.host.pauseTask(taskId) + : action === "resume" + ? await options.host.resumeTask(taskId) + : await options.host.cancelTask(taskId); + return { success: true, changed }; + } catch (error) { + return { success: false, error: errorMessage(error) }; + } + }, + ); + + mainIPC.handle( + AGENT_HOST_CHANNELS.REDIRECT_TASK, + async (event, taskId: unknown, instruction: unknown) => { + if (!allowed(event, options)) { + return { + success: false, + error: "Agent Host IPC sender is not allowed.", + }; + } + if (!options.host) { + return { success: false, error: "Agent Host is unavailable." }; + } + if (typeof taskId !== "string" || !taskId) { + return { success: false, error: "A task id is required." }; + } + if (typeof instruction !== "string") { + return { success: false, error: "Task guidance must be a string." }; + } + try { + return { + success: true, + job: await options.host.redirectTask(taskId, instruction), + }; + } catch (error) { + return { success: false, error: errorMessage(error) }; + } + }, + ); + mainIPC.handle(AGENT_HOST_CHANNELS.CANCEL, async (event, jobId: unknown) => { if (!allowed(event, options)) { return { success: false, error: "Agent Host IPC sender is not allowed." }; @@ -128,3 +222,7 @@ export function setupAgentHostIPC( }, ); } + +function isTaskAction(value: unknown): value is AgentHostTaskAction { + return value === "pause" || value === "resume" || value === "cancel"; +} diff --git a/packages/app/src/electron/agent-host/executor.test.ts b/packages/app/src/electron/agent-host/executor.test.ts index d1fb94dd..d9a0dbdb 100644 --- a/packages/app/src/electron/agent-host/executor.test.ts +++ b/packages/app/src/electron/agent-host/executor.test.ts @@ -9,7 +9,9 @@ import { LocalAiAgentHostExecutor } from "./executor"; const job: AgentHostJob = { id: "job", + taskId: "job", channelId: "channel", + channelKind: "channel", conversationId: "trusted-conversation", triggerMessageId: "message", contextMessageIds: ["message"], @@ -18,6 +20,7 @@ const job: AgentHostJob = { agentId: "trusted", agentMemberId: "agent:trusted", chain: { hops: 0, invoked: ["agent:trusted"] }, + controlInstructions: [], status: "running", attempts: 1, createdAt: new Date().toISOString(), @@ -66,6 +69,11 @@ describe("LocalAiAgentHostExecutor", () => { expect.objectContaining({ conversationId: "trusted-conversation", concurrent: true, + agentHost: { + jobId: "job", + taskId: "job", + channelKind: "channel", + }, agent: { id: "trusted", memberId: "agent:trusted" }, }), expect.any(Function), diff --git a/packages/app/src/electron/agent-host/executor.ts b/packages/app/src/electron/agent-host/executor.ts index 4a410f5f..fb9f117e 100644 --- a/packages/app/src/electron/agent-host/executor.ts +++ b/packages/app/src/electron/agent-host/executor.ts @@ -31,6 +31,11 @@ export class LocalAiAgentHostExecutor implements AgentHostExecutor { ...prepared.request, conversationId: job.conversationId, concurrent: true, + agentHost: { + jobId: job.id, + taskId: job.taskId, + channelKind: job.channelKind, + }, agent: { ...prepared.request.agent, id: job.agentId, diff --git a/packages/app/src/electron/agent-host/host.test.ts b/packages/app/src/electron/agent-host/host.test.ts index 9cca3354..afc31bf8 100644 --- a/packages/app/src/electron/agent-host/host.test.ts +++ b/packages/app/src/electron/agent-host/host.test.ts @@ -23,6 +23,7 @@ function dispatch( ): AgentHostDispatch { return { channelId: `channel:${conversationId}`, + channelKind: "channel", conversationId, triggerMessageId, contextMessageIds: [triggerMessageId], @@ -39,7 +40,9 @@ function dispatch( function storedJob(status: AgentHostJob["status"]): AgentHostJob { return { id: "stored", + taskId: "stored", channelId: "channel:c1", + channelKind: "channel", conversationId: "c1", triggerMessageId: "message:c1", contextMessageIds: ["message:c1"], @@ -48,6 +51,7 @@ function storedJob(status: AgentHostJob["status"]): AgentHostJob { agentId: "a", agentMemberId: "agent:a", chain: { hops: 0, invoked: ["agent:a"] }, + controlInstructions: [], status, attempts: 1, createdAt: new Date().toISOString(), @@ -167,4 +171,130 @@ describe("AgentHost", () => { ); expect((await host.listJobs())[0].status).toBe("cancelled"); }); + + it("pauses and resumes one task without changing its stable identity", async () => { + const gates = [deferred(), deferred()]; + const execute = vi.fn(() => gates[execute.mock.calls.length - 1].promise); + const cancel = vi.fn(async () => { + gates[0].resolve(); + return true; + }); + const host = new AgentHost({ + repository: new InMemoryAgentHostJobRepository(), + executor: { execute, cancel }, + createId: () => "task-1", + }); + const [job] = await host.enqueue(dispatch("c1")); + await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce()); + + expect(await host.pauseTask(job.taskId)).toBe(true); + expect((await host.listTasks())[0]).toMatchObject({ + id: "task-1", + status: "paused", + runCount: 1, + }); + expect(await host.resumeTask(job.taskId)).toBe(true); + await vi.waitFor(() => expect(execute).toHaveBeenCalledTimes(2)); + expect((await host.listTasks())[0].id).toBe("task-1"); + gates[1].resolve(); + }); + + it("keeps a running task live when its provider refuses to pause", async () => { + const gate = deferred(); + const host = new AgentHost({ + repository: new InMemoryAgentHostJobRepository(), + executor: { + execute: () => gate.promise, + cancel: async () => false, + }, + createId: () => "task-1", + }); + const [job] = await host.enqueue(dispatch("c1")); + await vi.waitFor(async () => + expect((await host.listTasks())[0].status).toBe("running"), + ); + + expect(await host.pauseTask(job.taskId)).toBe(false); + expect((await host.listTasks())[0].status).toBe("running"); + gate.resolve(); + await vi.waitFor(async () => + expect((await host.listTasks())[0].status).toBe("completed"), + ); + }); + + it("redirects a task into a replacement run with private guidance", async () => { + const ids = ["z-original", "a-replacement"]; + const host = new AgentHost({ + repository: new InMemoryAgentHostJobRepository(), + executor: { execute: async () => undefined }, + startPaused: true, + createId: () => ids.shift() as string, + }); + const [original] = await host.enqueue(dispatch("c1")); + const replacement = await host.redirectTask( + original.taskId, + "Show me the diff before opening a PR.", + ); + + expect(replacement).toMatchObject({ + id: "a-replacement", + taskId: "z-original", + parentJobId: "z-original", + status: "queued", + controlInstructions: ["Show me the diff before opening a PR."], + }); + expect(await host.listTasks()).toEqual([ + expect.objectContaining({ + id: "z-original", + currentJobId: "a-replacement", + runCount: 2, + status: "queued", + }), + ]); + expect( + (await host.listJobs()).find((entry) => entry.id === "z-original"), + ).toMatchObject({ status: "cancelled" }); + }); + + it("serializes simultaneous redirects into one replacement chain", async () => { + const ids = ["original", "replacement-1", "replacement-2"]; + const host = new AgentHost({ + repository: new InMemoryAgentHostJobRepository(), + executor: { execute: async () => undefined }, + startPaused: true, + createId: () => ids.shift() as string, + }); + const [original] = await host.enqueue(dispatch("c1")); + + await Promise.all([ + host.redirectTask(original.taskId, "First guidance"), + host.redirectTask(original.taskId, "Second guidance"), + ]); + + expect(await host.listTasks()).toEqual([ + expect.objectContaining({ + currentJobId: "replacement-2", + runCount: 3, + controlInstructions: ["First guidance", "Second guidance"], + }), + ]); + expect( + (await host.listJobs()).find((job) => job.id === "replacement-2"), + ).toMatchObject({ parentJobId: "replacement-1" }); + }); + + it("does not let one agent control another agent's task", async () => { + const host = new AgentHost({ + repository: new InMemoryAgentHostJobRepository(), + executor: { execute: async () => undefined }, + startPaused: true, + createId: () => "task-a", + }); + const [job] = await host.enqueue(dispatch("c1", ["agent:a"])); + + expect(await host.pauseTask(job.taskId, "agent:b")).toBe(false); + await expect( + host.redirectTask(job.taskId, "Change direction", "agent:b"), + ).rejects.toThrow("not found for this agent"); + }); }); diff --git a/packages/app/src/electron/agent-host/host.ts b/packages/app/src/electron/agent-host/host.ts index 284020a2..23f58b80 100644 --- a/packages/app/src/electron/agent-host/host.ts +++ b/packages/app/src/electron/agent-host/host.ts @@ -3,6 +3,7 @@ import type { AgentHostDispatch, AgentHostEvent, AgentHostJob, + AgentHostTaskSummary, } from "@/shared/types/agent-host"; import type { AgentHostJobRepository } from "./repository"; @@ -33,6 +34,8 @@ const TERMINAL = new Set([ ]); const IDENTIFIER = /^[A-Za-z0-9._:-]{1,256}$/; const MAX_CONTEXT_MESSAGES = 500; +const MAX_CONTROL_INSTRUCTIONS = 100; +const MAX_CONTROL_INSTRUCTION_LENGTH = 4_000; function actorKey(job: Pick) { return `${job.conversationId}\0${job.agentMemberId}`; @@ -41,6 +44,7 @@ function actorKey(job: Pick) { function validateDispatch(dispatch: AgentHostDispatch): void { if ( !IDENTIFIER.test(dispatch.channelId) || + !["channel", "dm"].includes(dispatch.channelKind) || !IDENTIFIER.test(dispatch.conversationId) || !IDENTIFIER.test(dispatch.triggerMessageId) || !Array.isArray(dispatch.contextMessageIds) || @@ -76,6 +80,64 @@ function validateDispatch(dispatch: AgentHostDispatch): void { } } +function byCreation(left: AgentHostJob, right: AgentHostJob): number { + return ( + left.createdAt.localeCompare(right.createdAt) || + left.id.localeCompare(right.id) + ); +} + +function currentJob(jobs: AgentHostJob[]): AgentHostJob { + const replacedJobIds = new Set( + jobs.flatMap((job) => (job.parentJobId ? [job.parentJobId] : [])), + ); + return ( + jobs.find((job) => !replacedJobIds.has(job.id)) ?? + ([...jobs].sort(byCreation).at(-1) as AgentHostJob) + ); +} + +export function summarizeAgentHostTasks( + jobs: AgentHostJob[], + agentMemberId?: string, +): AgentHostTaskSummary[] { + const grouped = new Map(); + for (const job of jobs) { + if (agentMemberId && job.agentMemberId !== agentMemberId) continue; + const runs = grouped.get(job.taskId); + if (runs) runs.push(job); + else grouped.set(job.taskId, [job]); + } + return [...grouped.entries()] + .map(([taskId, runs]) => { + const current = currentJob(runs); + const first = [...runs].sort(byCreation)[0]; + return { + id: taskId, + channelId: current.channelId, + channelKind: current.channelKind, + conversationId: current.conversationId, + triggerMessageId: current.triggerMessageId, + agentId: current.agentId, + agentMemberId: current.agentMemberId, + currentJobId: current.id, + status: current.status, + runCount: runs.length, + controlInstructions: [...current.controlInstructions], + createdAt: first.createdAt, + updatedAt: current.updatedAt, + startedAt: current.startedAt, + completedAt: current.completedAt, + error: current.error, + }; + }) + .sort( + (left, right) => + right.updatedAt.localeCompare(left.updatedAt) || + right.id.localeCompare(left.id), + ); +} + export class AgentHost { private readonly repository: AgentHostJobRepository; private readonly executor: AgentHostExecutor; @@ -85,7 +147,9 @@ export class AgentHost { private readonly jobs = new Map(); private readonly listeners = new Set(); private readonly running = new Set(); + private readonly pausing = new Set(); private readonly activeActors = new Set(); + private readonly taskControlQueues = new Map>(); private ready: Promise; private accepting = true; private drainQueued = false; @@ -143,6 +207,11 @@ export class AgentHost { .map((job) => structuredClone(job)); } + async listTasks(agentMemberId?: string): Promise { + await this.ready; + return summarizeAgentHostTasks([...this.jobs.values()], agentMemberId); + } + async enqueue(dispatch: AgentHostDispatch): Promise { await this.ready; if (!this.accepting) throw new Error("Agent Host is stopping."); @@ -167,9 +236,12 @@ export class AgentHost { created.push(structuredClone(duplicate)); continue; } + const id = this.createId(); const job: AgentHostJob = { - id: this.createId(), + id, + taskId: id, channelId: dispatch.channelId, + channelKind: dispatch.channelKind, conversationId: dispatch.conversationId, triggerMessageId: dispatch.triggerMessageId, contextMessageIds: [...dispatch.contextMessageIds], @@ -178,6 +250,7 @@ export class AgentHost { agentId: target.agentId, agentMemberId: target.memberId, chain: structuredClone(dispatch.chain), + controlInstructions: [], status: "queued", attempts: 0, createdAt: now, @@ -204,6 +277,145 @@ export class AgentHost { return true; } + pauseTask(taskId: string, agentMemberId?: string): Promise { + return this.serializeTaskControl(taskId, () => + this.pauseTaskNow(taskId, agentMemberId), + ); + } + + private async pauseTaskNow( + taskId: string, + agentMemberId?: string, + ): Promise { + await this.ready; + const job = this.taskCurrentJob(taskId, agentMemberId); + if (!job || TERMINAL.has(job.status) || job.status === "paused") { + return false; + } + if (job.status === "running") { + this.pausing.add(job.id); + const cancelled = await this.executor.cancel?.(structuredClone(job)); + if (cancelled === false) { + this.pausing.delete(job.id); + return false; + } + } + job.status = "paused"; + job.error = undefined; + job.completedAt = undefined; + job.updatedAt = this.now().toISOString(); + await this.repository.put(job); + this.emit({ type: "job", job }); + this.pausing.delete(job.id); + return true; + } + + resumeTask(taskId: string, agentMemberId?: string): Promise { + return this.serializeTaskControl(taskId, () => + this.resumeTaskNow(taskId, agentMemberId), + ); + } + + private async resumeTaskNow( + taskId: string, + agentMemberId?: string, + ): Promise { + await this.ready; + const job = this.taskCurrentJob(taskId, agentMemberId); + if (!job || job.status !== "paused") return false; + job.status = "queued"; + job.error = undefined; + job.completedAt = undefined; + job.updatedAt = this.now().toISOString(); + await this.repository.put(job); + this.emit({ type: "job", job }); + this.scheduleDrain(); + return true; + } + + cancelTask(taskId: string, agentMemberId?: string): Promise { + return this.serializeTaskControl(taskId, () => + this.cancelTaskNow(taskId, agentMemberId), + ); + } + + private async cancelTaskNow( + taskId: string, + agentMemberId?: string, + ): Promise { + await this.ready; + const job = this.taskCurrentJob(taskId, agentMemberId); + return job ? this.cancel(job.id) : false; + } + + redirectTask( + taskId: string, + instruction: string, + agentMemberId?: string, + ): Promise { + return this.serializeTaskControl(taskId, () => + this.redirectTaskNow(taskId, instruction, agentMemberId), + ); + } + + private async redirectTaskNow( + taskId: string, + instruction: string, + agentMemberId?: string, + ): Promise { + await this.ready; + const normalized = instruction.trim(); + if (!normalized || normalized.length > MAX_CONTROL_INSTRUCTION_LENGTH) { + throw new Error( + `Task guidance must contain 1-${MAX_CONTROL_INSTRUCTION_LENGTH} characters.`, + ); + } + const source = this.taskCurrentJob(taskId, agentMemberId); + if (!source) + throw new Error(`Task ${taskId} was not found for this agent.`); + if (source.controlInstructions.length >= MAX_CONTROL_INSTRUCTIONS) { + throw new Error( + `Task ${taskId} already has the maximum ${MAX_CONTROL_INSTRUCTIONS} guidance entries.`, + ); + } + if (!TERMINAL.has(source.status)) { + const cancelled = await this.cancel(source.id); + if (!cancelled) { + throw new Error( + `Task ${taskId} could not stop its current run. Retry after it reaches a safe boundary.`, + ); + } + } + + const now = this.now().toISOString(); + const id = this.createId(); + const successor: AgentHostJob = { + id, + taskId: source.taskId, + parentJobId: source.id, + channelId: source.channelId, + channelKind: source.channelKind, + conversationId: source.conversationId, + triggerMessageId: source.triggerMessageId, + contextMessageIds: [...source.contextMessageIds], + mode: source.mode, + offeredAgentMemberIds: [...source.offeredAgentMemberIds], + agentId: source.agentId, + agentMemberId: source.agentMemberId, + chain: structuredClone(source.chain), + controlInstructions: [...source.controlInstructions, normalized], + status: "queued", + attempts: 0, + createdAt: now, + updatedAt: now, + }; + this.jobs.set(id, successor); + await this.repository.put(successor); + this.emit({ type: "job", job: successor }); + this.scheduleDrain(); + return structuredClone(successor); + } + async dispose(): Promise { await this.ready; this.accepting = false; @@ -221,6 +433,37 @@ export class AgentHost { }); } + private taskCurrentJob( + taskId: string, + agentMemberId?: string, + ): AgentHostJob | undefined { + const jobs = [...this.jobs.values()].filter( + (job) => + job.taskId === taskId && + (!agentMemberId || job.agentMemberId === agentMemberId), + ); + return jobs.length > 0 ? currentJob(jobs) : undefined; + } + + private async serializeTaskControl( + taskId: string, + operation: () => Promise, + ): Promise { + const previous = this.taskControlQueues.get(taskId) ?? Promise.resolve(); + const result = previous.then(operation, operation); + const settled = result.then( + () => undefined, + () => undefined, + ); + this.taskControlQueues.set(taskId, settled); + void settled.then(() => { + if (this.taskControlQueues.get(taskId) === settled) { + this.taskControlQueues.delete(taskId); + } + }); + return result; + } + private async drain(): Promise { await this.ready; while (this.running.size < this.maxConcurrency) { @@ -254,10 +497,10 @@ export class AgentHost { await this.executor.execute(structuredClone(job), (event) => this.emit(event), ); - if (job.status !== "running") return; + if (job.status !== "running" || this.pausing.has(job.id)) return; await this.finish(job, "completed"); } catch (error) { - if (job.status === "running") { + if (job.status === "running" && !this.pausing.has(job.id)) { await this.finish( job, "failed", diff --git a/packages/app/src/electron/agent-host/repository.test.ts b/packages/app/src/electron/agent-host/repository.test.ts index 97d3b456..44bbb17f 100644 --- a/packages/app/src/electron/agent-host/repository.test.ts +++ b/packages/app/src/electron/agent-host/repository.test.ts @@ -13,7 +13,9 @@ function job(id: string, status: AgentHostJob["status"]): AgentHostJob { ).toISOString(); return { id, + taskId: id, channelId: "channel", + channelKind: "channel", conversationId: "conversation", triggerMessageId: `message-${id}`, contextMessageIds: [`message-${id}`], @@ -22,6 +24,7 @@ function job(id: string, status: AgentHostJob["status"]): AgentHostJob { agentId: "a", agentMemberId: "agent:a", chain: { hops: 0, invoked: ["agent:a"] }, + controlInstructions: [], status, attempts: status === "queued" ? 0 : 1, createdAt: timestamp, @@ -100,4 +103,29 @@ describe("JsonAgentHostJobRepository", () => { }), ]); }); + + it("migrates version two jobs into stable tasks", async () => { + const directory = await mkdtemp(join(tmpdir(), "convera-agent-host-")); + temporaryDirectories.push(directory); + const path = join(directory, "jobs.json"); + const current = job("1", "completed"); + const versionTwoJob: Partial = { ...current }; + delete versionTwoJob.taskId; + delete versionTwoJob.channelKind; + delete versionTwoJob.controlInstructions; + await writeFile( + path, + JSON.stringify({ schemaVersion: 2, jobs: [versionTwoJob] }), + "utf8", + ); + + expect(await new JsonAgentHostJobRepository({ path }).list()).toEqual([ + expect.objectContaining({ + id: "1", + taskId: "1", + channelKind: "channel", + controlInstructions: [], + }), + ]); + }); }); diff --git a/packages/app/src/electron/agent-host/repository.ts b/packages/app/src/electron/agent-host/repository.ts index b185e7e1..c7b7d8c3 100644 --- a/packages/app/src/electron/agent-host/repository.ts +++ b/packages/app/src/electron/agent-host/repository.ts @@ -18,6 +18,7 @@ const chainSchema = z.object({ const statusSchema = z.enum([ "queued", "running", + "paused", "completed", "failed", "cancelled", @@ -42,13 +43,20 @@ const legacyJobSchema = z.object({ completedAt: z.string().datetime().optional(), }); -const jobSchema = legacyJobSchema.extend({ +const versionTwoJobSchema = legacyJobSchema.extend({ contextMessageIds: z.array(z.string().min(1)).min(1).max(500), mode: z.enum(["open-floor", "direct"]), offeredAgentMemberIds: z.array(z.string().min(1)).max(16), agentId: z.string().min(1), }); +const jobSchema = versionTwoJobSchema.extend({ + taskId: z.string().min(1), + parentJobId: z.string().min(1).optional(), + channelKind: z.enum(["channel", "dm"]), + controlInstructions: z.array(z.string().min(1).max(4_000)).max(100), +}); + const legacyStateSchema = z.object({ schemaVersion: z.literal(1), jobs: z.array(legacyJobSchema), @@ -56,6 +64,11 @@ const legacyStateSchema = z.object({ const stateSchema = z.object({ schemaVersion: z.literal(2), + jobs: z.array(versionTwoJobSchema), +}); + +const currentStateSchema = z.object({ + schemaVersion: z.literal(3), jobs: z.array(jobSchema), }); @@ -65,12 +78,15 @@ function migrateLegacyJob(job: LegacyJob): AgentHostJob { const incompatible = job.status === "queued" || job.status === "running"; return { ...job, + taskId: job.id, + channelKind: "channel", agentId: job.agentMemberId.startsWith("agent:") ? job.agentMemberId.slice("agent:".length) : job.agentMemberId, contextMessageIds: [job.triggerMessageId], mode: "direct", offeredAgentMemberIds: [job.agentMemberId], + controlInstructions: [], status: incompatible ? "interrupted" : job.status, error: incompatible ? "This queued Agent Host job predates frozen offer context and was not replayed." @@ -81,6 +97,17 @@ function migrateLegacyJob(job: LegacyJob): AgentHostJob { }; } +function migrateVersionTwoJob( + job: z.infer, +): AgentHostJob { + return { + ...job, + taskId: job.id, + channelKind: "channel", + controlInstructions: [], + }; +} + function prune(jobs: AgentHostJob[], limit: number): AgentHostJob[] { const terminal = jobs .filter((job) => @@ -129,29 +156,39 @@ export class JsonAgentHostJobRepository implements AgentHostJobRepository { private async readState(): Promise<{ state: { - schemaVersion: 2; + schemaVersion: 3; jobs: AgentHostJob[]; }; migrated: boolean; }> { const value = await this.file.read(); if (value === undefined) { - return { state: { schemaVersion: 2, jobs: [] }, migrated: false }; + return { state: { schemaVersion: 3, jobs: [] }, migrated: false }; } const version = z.object({ schemaVersion: z.number() }).parse(value); if (version.schemaVersion === 1) { const legacy = legacyStateSchema.parse(value); return { state: { - schemaVersion: 2, + schemaVersion: 3, jobs: legacy.jobs.map(migrateLegacyJob), }, migrated: true, }; } + if (version.schemaVersion === 2) { + const prior = stateSchema.parse(value); + return { + state: { + schemaVersion: 3, + jobs: prior.jobs.map(migrateVersionTwoJob), + }, + migrated: true, + }; + } return { - state: stateSchema.parse(value) as { - schemaVersion: 2; + state: currentStateSchema.parse(value) as { + schemaVersion: 3; jobs: AgentHostJob[]; }, migrated: false, @@ -159,10 +196,10 @@ export class JsonAgentHostJobRepository implements AgentHostJobRepository { } private async writeState(state: { - schemaVersion: 2; + schemaVersion: 3; jobs: AgentHostJob[]; }): Promise { - await this.file.write(stateSchema.parse(state)); + await this.file.write(currentStateSchema.parse(state)); } async list(): Promise { @@ -170,7 +207,7 @@ export class JsonAgentHostJobRepository implements AgentHostJobRepository { const { state, migrated } = await this.readState(); const jobs = prune(state.jobs, this.maxTerminalJobs); if (migrated || jobs.length !== state.jobs.length) { - await this.writeState({ schemaVersion: 2, jobs }); + await this.writeState({ schemaVersion: 3, jobs }); } return structuredClone(jobs); }); diff --git a/packages/app/src/electron/ai/__tests__/agent-host-tools.test.ts b/packages/app/src/electron/ai/__tests__/agent-host-tools.test.ts new file mode 100644 index 00000000..3b7fba34 --- /dev/null +++ b/packages/app/src/electron/ai/__tests__/agent-host-tools.test.ts @@ -0,0 +1,116 @@ +import { describe, expect, it, vi } from "vitest"; +import type { AgentHost } from "@/electron/agent-host/host"; +import type { + AgentHostJob, + AgentHostTaskSummary, +} from "@/shared/types/agent-host"; +import type { LocalAiTurnHookInput } from "../runtime"; +import { withAgentHostTools } from "../agent-host-tools"; + +const task: AgentHostTaskSummary = { + id: "task-1", + channelId: "channel-1", + channelKind: "channel", + conversationId: "conversation-1", + triggerMessageId: "message-1", + agentId: "fizz", + agentMemberId: "agent:fizz", + currentJobId: "job-1", + status: "running", + runCount: 1, + controlInstructions: [], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), +}; + +function input(channelKind: "channel" | "dm"): LocalAiTurnHookInput { + return { + request: { + requestId: "request", + conversationId: "conversation", + turnId: "turn", + providerId: "codex-cli", + operation: { + kind: "append", + message: { role: "user", content: "Control the task" }, + }, + agent: { id: "fizz", memberId: "agent:fizz" }, + agentHost: { jobId: "dm-job", taskId: "dm-task", channelKind }, + }, + prepared: {} as LocalAiTurnHookInput["prepared"], + requestInteraction: async () => ({}), + }; +} + +describe("Agent Host task tools", () => { + it("injects one private task tool only into agent DMs", async () => { + const host = { + listTasks: vi.fn(async () => [task]), + } as unknown as AgentHost; + const hooks = withAgentHostTools({}, () => host); + + const dm = await hooks.prepareTurnContext?.(input("dm")); + const channel = await hooks.prepareTurnContext?.(input("channel")); + + expect(dm?.additionalTools?.map((tool) => tool.qualifiedName)).toEqual([ + "task:manage_task", + ]); + expect(dm?.systemContext).toContain("private direct conversation"); + expect(channel).toBeUndefined(); + }); + + it("lists only non-DM tasks owned by the speaking agent", async () => { + const host = { + listTasks: vi.fn(async () => [ + task, + { ...task, id: "dm-task", channelKind: "dm" }, + ]), + } as unknown as AgentHost; + const prepared = await withAgentHostTools( + {}, + () => host, + ).prepareTurnContext?.(input("dm")); + const tool = prepared?.additionalTools?.[0]; + + expect(await tool?.execute({ action: "list" })).toMatchObject({ + ok: true, + tasks: [{ task_id: "task-1", channel_id: "channel-1" }], + }); + expect(host.listTasks).toHaveBeenCalledWith("agent:fizz"); + }); + + it("passes redirect guidance through the ownership fence", async () => { + const successor = { + id: "job-2", + taskId: "task-1", + status: "queued", + } as AgentHostJob; + const host = { + listTasks: vi.fn(async () => [task]), + redirectTask: vi.fn(async () => successor), + } as unknown as AgentHost; + const prepared = await withAgentHostTools( + {}, + () => host, + ).prepareTurnContext?.(input("dm")); + const tool = prepared?.additionalTools?.[0]; + + expect( + await tool?.execute({ + action: "redirect", + task_id: "task-1", + instruction: "Do not open the PR until I see the diff.", + }), + ).toEqual({ + ok: true, + task_id: "task-1", + job_id: "job-2", + status: "queued", + }); + expect(host.redirectTask).toHaveBeenCalledWith( + "task-1", + "Do not open the PR until I see the diff.", + "agent:fizz", + ); + }); +}); diff --git a/packages/app/src/electron/ai/agent-host-tools.ts b/packages/app/src/electron/ai/agent-host-tools.ts new file mode 100644 index 00000000..8d7cf992 --- /dev/null +++ b/packages/app/src/electron/ai/agent-host-tools.ts @@ -0,0 +1,190 @@ +import type { AgentHost } from "@/electron/agent-host/host"; +import type { AgentTool } from "./agent-tools"; +import type { LocalAiTurnHooks, PreparedLocalAiTurnContext } from "./runtime"; +import { z } from "zod"; + +const actionSchema = z.enum([ + "list", + "inspect", + "pause", + "resume", + "cancel", + "redirect", +]); + +const inputSchema = z.object({ + action: actionSchema.describe( + "Operation to perform. Use list before acting when the user did not provide an exact task id.", + ), + task_id: z + .string() + .min(1) + .optional() + .describe("Stable task id returned by list. Required except for list."), + instruction: z + .string() + .min(1) + .max(4_000) + .optional() + .describe( + "Replacement guidance for redirect. The new run keeps the original task identity and applies this instruction after older guidance.", + ), +}); + +type Input = z.infer; + +function jsonSchema(): Record { + return { + type: "object", + properties: { + action: { + type: "string", + enum: actionSchema.options, + description: + "Use list to discover tasks, inspect for detail, pause/resume/cancel for lifecycle control, or redirect to restart with new guidance.", + }, + task_id: { + type: "string", + description: + "Stable task id returned by list. Required except for list.", + }, + instruction: { + type: "string", + minLength: 1, + maxLength: 4_000, + description: "Required only for redirect.", + }, + }, + required: ["action"], + additionalProperties: false, + }; +} + +function failure(code: string, message: string, recovery: string) { + return { ok: false, error: { code, message, recovery } }; +} + +function taskTool(host: AgentHost, agentMemberId: string): AgentTool { + return { + name: "manage_task", + qualifiedName: "task:manage_task", + description: + "Inspect or control this agent's background tasks while speaking privately with the user. Use list when the user refers to a task by channel, topic, or relative time. Pause, resume, cancel, and redirect change real Agent Host state; do not claim success unless the returned ok field is true. Redirect stops the current run and starts a replacement run with the supplied private guidance.", + inputSchema: jsonSchema(), + inputShape: inputSchema.shape, + inputValidator: inputSchema, + execute: async (raw) => { + const input = inputSchema.parse(raw) as Input; + const visible = (await host.listTasks(agentMemberId)).filter( + (task) => task.channelKind !== "dm", + ); + if (input.action === "list") { + return { + ok: true, + truncated: visible.length > 20, + tasks: visible.slice(0, 20).map((task) => ({ + task_id: task.id, + channel_id: task.channelId, + status: task.status, + runs: task.runCount, + updated_at: task.updatedAt, + latest_guidance: task.controlInstructions.at(-1), + })), + }; + } + + if (!input.task_id) { + return failure( + "TASK_ID_REQUIRED", + `The ${input.action} action requires task_id.`, + "Call task:manage_task with action=list, choose one returned task_id, then retry.", + ); + } + const task = visible.find((candidate) => candidate.id === input.task_id); + if (!task) { + return failure( + "TASK_NOT_FOUND", + `Task ${input.task_id} is not a controllable task owned by this agent.`, + "Call task:manage_task with action=list and use an exact task_id from the result.", + ); + } + if (input.action === "inspect") return { ok: true, task }; + + if (input.action === "redirect") { + if (!input.instruction) { + return failure( + "INSTRUCTION_REQUIRED", + "Redirect requires a non-empty instruction.", + "Retry with instruction describing what should change in the replacement run.", + ); + } + try { + const job = await host.redirectTask( + task.id, + input.instruction, + agentMemberId, + ); + return { + ok: true, + task_id: job.taskId, + job_id: job.id, + status: job.status, + }; + } catch (error) { + return failure( + "TASK_REDIRECT_FAILED", + error instanceof Error ? error.message : String(error), + "Inspect the task and retry after the current run reaches a safe boundary.", + ); + } + } + + const changed = + input.action === "pause" + ? await host.pauseTask(task.id, agentMemberId) + : input.action === "resume" + ? await host.resumeTask(task.id, agentMemberId) + : await host.cancelTask(task.id, agentMemberId); + if (!changed) { + return failure( + "TASK_STATE_CONFLICT", + `Task ${task.id} cannot ${input.action} from status ${task.status}.`, + "Inspect the task, then choose an action valid for its current status.", + ); + } + return { ok: true, task_id: task.id, action: input.action }; + }, + }; +} + +export function withAgentHostTools( + hooks: LocalAiTurnHooks, + getHost: () => AgentHost | undefined, +): LocalAiTurnHooks { + return { + ...hooks, + prepareTurnContext: async ( + input, + ): Promise => { + const prepared = await hooks.prepareTurnContext?.(input); + const memberId = input.request.agent?.memberId?.trim(); + const host = getHost(); + if (!memberId || !host || input.request.agentHost?.channelKind !== "dm") { + return prepared; + } + return { + ...prepared, + systemContext: [ + prepared?.systemContext, + "This is your private direct conversation with the user. You may inspect and control your background channel tasks with task:manage_task. Work state comes from that tool; do not guess from chat history.", + ] + .filter(Boolean) + .join("\n\n"), + additionalTools: [ + ...(prepared?.additionalTools ?? []), + taskTool(host, memberId), + ], + }; + }, + }; +} diff --git a/packages/app/src/electron/main.ts b/packages/app/src/electron/main.ts index 0dfaed32..d0fb08a3 100644 --- a/packages/app/src/electron/main.ts +++ b/packages/app/src/electron/main.ts @@ -22,6 +22,7 @@ import { AgentHost } from "@/electron/agent-host/host"; import { JsonAgentHostJobRepository } from "@/electron/agent-host/repository"; import { AgentHostRendererBridge } from "@/electron/agent-host/renderer-bridge"; import { LocalAiAgentHostExecutor } from "@/electron/agent-host/executor"; +import { withAgentHostTools } from "@/electron/ai/agent-host-tools"; import { getCurrentShortcut } from "@/electro-bridge/ipc/ipc-handlers"; @@ -154,7 +155,10 @@ app.whenReady().then(async () => { }); localAIRuntime = new LocalAiRuntime({ sessionRepository, - turnHooks: withWorkspacePerception(memoryCoordinator), + turnHooks: withAgentHostTools( + withWorkspacePerception(memoryCoordinator), + () => agentHost, + ), memoryService: memoryCoordinator, resolveSandbox: async (request) => { const agentId = request.agent?.id?.trim(); diff --git a/packages/app/src/renderer/components/chat/AgentContextPanel.tsx b/packages/app/src/renderer/components/chat/AgentContextPanel.tsx index 586d961a..3c1c941b 100644 --- a/packages/app/src/renderer/components/chat/AgentContextPanel.tsx +++ b/packages/app/src/renderer/components/chat/AgentContextPanel.tsx @@ -3,6 +3,9 @@ import { type AgentContextInspection, } from "@/renderer/libs/agent-context-inspector"; import { cn } from "@/renderer/libs/utils/tailwind"; +import { useAgentHostTasks } from "@/renderer/libs/hooks/use-agent-host-jobs"; +import { memberIdForAgent } from "@/renderer/libs/db"; +import { useChannels } from "@/renderer/libs/stores/channel-store"; import { motion } from "framer-motion"; import { Activity, @@ -10,10 +13,15 @@ import { Eye, Loader2, LockKeyhole, + Pause, + Play, + RotateCcw, + Send, + Square, Wrench, X, } from "lucide-react"; -import React, { useEffect, useState } from "react"; +import React, { useEffect, useMemo, useState } from "react"; function Section({ icon: Icon, @@ -41,12 +49,201 @@ function Empty({ children }: { children: React.ReactNode }) { ); } -/** - * Private, out-of-band inspection of the context Convera can account for. - * Nothing rendered here is appended to the DM transcript or exposed as an - * agent tool. Home owns whether the panel is mounted; the loader itself still - * rejects non-DM channels so a public header cannot accidentally reuse it. - */ +function taskStatusLabel(status: string): string { + return status === "running" + ? "Working" + : status === "queued" + ? "Queued" + : status.charAt(0).toUpperCase() + status.slice(1); +} + +function AgentWorkSection({ agentId }: { agentId: string }) { + const agentMemberId = memberIdForAgent(agentId); + const { tasks, error, control, redirect } = useAgentHostTasks(agentMemberId); + const channels = useChannels(); + const [guidingTaskId, setGuidingTaskId] = useState(); + const [instruction, setInstruction] = useState(""); + const [busy, setBusy] = useState(); + const channelNames = useMemo( + () => + new Map((channels ?? []).map((channel) => [channel.id, channel.name])), + [channels], + ); + const visibleTasks = tasks + .filter((task) => task.channelKind !== "dm") + .slice(0, 20); + + async function runControl( + taskId: string, + action: "pause" | "resume" | "cancel", + ) { + setBusy(`${taskId}:${action}`); + try { + await control(taskId, action); + } finally { + setBusy(undefined); + } + } + + async function submitGuidance(taskId: string) { + const value = instruction.trim(); + if (!value) return; + setBusy(`${taskId}:redirect`); + try { + if (await redirect(taskId, value)) { + setInstruction(""); + setGuidingTaskId(undefined); + } + } finally { + setBusy(undefined); + } + } + + return ( +
+

+ Chat here to let this agent inspect and control its own work. Buttons + below act directly on Agent Host and do not wait for the model. +

+ {error && ( +

{error}

+ )} + {visibleTasks.length === 0 ? ( + No channel tasks are recorded for this agent yet. + ) : ( +
+ {visibleTasks.map((task) => { + const isActive = + task.status === "running" || task.status === "queued"; + const isPaused = task.status === "paused"; + const channelName = + channelNames.get(task.channelId) ?? task.channelId; + const latestGuidance = task.controlInstructions.at(-1); + const guiding = guidingTaskId === task.id; + return ( +
+
+
+

+ #{channelName} +

+

+ {taskStatusLabel(task.status)} · {task.runCount} run + {task.runCount === 1 ? "" : "s"} · {task.id.slice(-8)} +

+ {latestGuidance && ( +

+ Guidance: {latestGuidance} +

+ )} + {task.error && ( +

+ {task.error} +

+ )} +
+
+ {isActive && ( + + )} + {isPaused && ( + + )} + + {![ + "completed", + "failed", + "cancelled", + "interrupted", + ].includes(task.status) && ( + + )} +
+
+ {guiding && ( +
+