From 84e7f71bd831c6cb61591b3ff0703053b7a7f1be Mon Sep 17 00:00:00 2001 From: dex Date: Tue, 13 Oct 2026 04:20:48 +0000 Subject: [PATCH 1/8] feat(core): add task lifecycle rows and synthetic user-message origins Additive v1 schema changes for background subagents: - task_started / task_finished log entries (status, summary, error, usage, duration, last tool, run count) - optional user-message origin { _tag: 'task-notification', taskIds } so runtime-written turns are distinguishable; old logs decode unchanged - task-progress ephemeral delta part SessionControls queues now carry an origin, adjacent queued task notifications drain as one message, and runningAgentIds exposes the running registry. AgentRuntime and Session.send pass origins through. Co-Authored-By: Claude Opus 5.5 HumanLayer-Session: https://app.dev.codelayer.gg/sessions/01a13295-e128-724d-bb90-666e8fbf2d01 --- .../src/AgentEvents/AgentEventsService.ts | 14 +++ .../src/AgentRuntime/AgentRuntimeLayer.ts | 40 ++++--- .../src/AgentRuntime/AgentRuntimeService.ts | 3 + packages/fold-core/src/EventLog/Schemas.ts | 109 ++++++++++++++++++ .../fold-core/src/Session/SessionControls.ts | 103 ++++++++++++++--- .../fold-core/src/Session/SessionLayer.ts | 4 +- .../fold-core/src/Session/SessionService.ts | 16 ++- .../EventLog/StoredLogEntryDecoder.vi.test.ts | 76 ++++++++++++ .../Session/SessionControlsQueue.vi.test.ts | 47 ++++++++ 9 files changed, 376 insertions(+), 36 deletions(-) create mode 100644 packages/fold-core/test/Session/SessionControlsQueue.vi.test.ts diff --git a/packages/fold-core/src/AgentEvents/AgentEventsService.ts b/packages/fold-core/src/AgentEvents/AgentEventsService.ts index 0db942d..1a85ce8 100644 --- a/packages/fold-core/src/AgentEvents/AgentEventsService.ts +++ b/packages/fold-core/src/AgentEvents/AgentEventsService.ts @@ -17,6 +17,20 @@ export type DeltaPart = | { readonly type: 'text-delta'; readonly id: string; readonly delta: string } | { readonly type: 'reasoning-delta'; readonly id: string; readonly delta: string } | { readonly type: 'tool-progress'; readonly toolName: string; readonly payload: Schema.Json } + /** + * Live progress of one background task run, published on the task agent's envelope after each of + * its turns. Latest wins; the durable `task_finished` row carries the final numbers. + */ + | { + readonly type: 'task-progress' + readonly taskId: AgentId + readonly turns: number + readonly toolUses: number + readonly totalTokens: number | null + readonly lastToolName: string | null + /** The task agent's latest assistant text, truncated: what it is doing right now. */ + readonly progressSummary: string | null + } /** One event on the merged session stream: a durable log row or an ephemeral delta. */ export type FoldEvent = diff --git a/packages/fold-core/src/AgentRuntime/AgentRuntimeLayer.ts b/packages/fold-core/src/AgentRuntime/AgentRuntimeLayer.ts index 961f4f1..aa17aa6 100644 --- a/packages/fold-core/src/AgentRuntime/AgentRuntimeLayer.ts +++ b/packages/fold-core/src/AgentRuntime/AgentRuntimeLayer.ts @@ -26,6 +26,7 @@ import { type CompactionLogEntry, type LogEntry, type LogEntryInput, + type UserMessageOrigin, } from '../EventLog/Schemas' import { usageFromResponseUsage } from '../EventLog/Usage' import { HookRunner } from '../HookRunner/HookRunnerService' @@ -137,19 +138,24 @@ export const liveAgentRuntimeLayer: Layer.Layer< Effect.map((entries): ReadonlyArray => entries), ) - const appendUserMessage = (input: RunAgentInput, text: string): Effect.Effect => + const appendUserMessage = ( + input: RunAgentInput, + text: string, + origin: UserMessageOrigin | null = null, + ): Effect.Effect => ids.makeMessageId.pipe( - Effect.flatMap((messageId) => - appendToEventLog( - LogEntryInputs['user-message']({ - agentId: input.agentId, - parentAgentId: input.parentAgentId, - toolCallId: input.toolCallId, - messageId, - message: encodeUserMessage(Prompt.userMessage({ content: [Prompt.textPart({ text })] })), - }), - ), - ), + Effect.flatMap((messageId) => { + const envelope = { + agentId: input.agentId, + parentAgentId: input.parentAgentId, + toolCallId: input.toolCallId, + messageId, + message: encodeUserMessage(Prompt.userMessage({ content: [Prompt.textPart({ text })] })), + } + return appendToEventLog( + LogEntryInputs['user-message'](origin === null ? envelope : { ...envelope, origin }), + ) + }), Effect.asVoid, ) @@ -323,8 +329,8 @@ export const liveAgentRuntimeLayer: Layer.Layer< // Drain queued steering before this turn's model call (D8): each steered message becomes an // ordinary user-message right here, so the log records it exactly where the model saw it. const steered = yield* sessionControls.drainSteering(input.agentId) - for (const text of steered) { - yield* appendUserMessage(input, text) + for (const message of steered) { + yield* appendUserMessage(input, message.text, message.origin) } const entriesBeforeCompaction = yield* collectEntries @@ -506,8 +512,8 @@ export const liveAgentRuntimeLayer: Layer.Layer< // complete naturally - each becomes an ordinary user-message and the run continues. const followUps = yield* sessionControls.drainFollowUps(input.agentId) if (Arr.isReadonlyArrayNonEmpty(followUps)) { - for (const text of followUps) { - yield* appendUserMessage(input, text) + for (const message of followUps) { + yield* appendUserMessage(input, message.text, message.origin) } return TurnResult.continue() } @@ -647,7 +653,7 @@ export const liveAgentRuntimeLayer: Layer.Layer< const doomLoopRef = yield* Ref.make(initialDoomLoopState) for (const text of input.messages) { - yield* appendUserMessage(input, text) + yield* appendUserMessage(input, text, input.origin ?? null) } while (true) { diff --git a/packages/fold-core/src/AgentRuntime/AgentRuntimeService.ts b/packages/fold-core/src/AgentRuntime/AgentRuntimeService.ts index 02661df..5097969 100644 --- a/packages/fold-core/src/AgentRuntime/AgentRuntimeService.ts +++ b/packages/fold-core/src/AgentRuntime/AgentRuntimeService.ts @@ -15,6 +15,7 @@ import type { AgentLaunchMode, AgentStartedLogEntry, CompactionLogEntry, + UserMessageOrigin, } from '../EventLog/Schemas' import type { AgentId, ToolCallId } from '../Ids' @@ -47,6 +48,8 @@ export type RunAgentInput = { * messages carry a dispatch prompt followed by injected context such as a preloaded skill (D21). */ readonly messages: Arr.NonEmptyReadonlyArray + /** Origin recorded on the initial user messages when the runtime, not a person, wrote them. */ + readonly origin?: UserMessageOrigin | undefined } /** Input for switching one started agent to a different model (a new epoch - D17). */ diff --git a/packages/fold-core/src/EventLog/Schemas.ts b/packages/fold-core/src/EventLog/Schemas.ts index fd30085..1b229e9 100644 --- a/packages/fold-core/src/EventLog/Schemas.ts +++ b/packages/fold-core/src/EventLog/Schemas.ts @@ -339,6 +339,26 @@ export const SystemMessageLogEntry = Schema.TaggedStruct('system-message', { .annotate({ identifier: 'SystemMessageLogEntry' }) export type SystemMessageLogEntry = typeof SystemMessageLogEntry.Type +/** + * Origin of a user message the runtime wrote on its own: a task notification delivered to the agent + * that launched one or more background tasks. The ids are the finished tasks' agent ids. + */ +export const TaskNotificationOrigin = Schema.TaggedStruct('task-notification', { + taskIds: Schema.Array(AgentId), +}).annotate({ identifier: 'TaskNotificationOrigin' }) +export type TaskNotificationOrigin = typeof TaskNotificationOrigin.Type + +/** + * Why a user message exists when a person did not type it. Absent on ordinary user turns (and on every + * entry written before origins existed), so v1 logs decode unchanged. UIs use it to render synthetic + * messages as system rows; the model still sees the message text. + */ +export const UserMessageOrigin = Schema.Union([TaskNotificationOrigin]).annotate({ + identifier: 'UserMessageOrigin', + discriminator: '_tag', +}) +export type UserMessageOrigin = typeof UserMessageOrigin.Type + /** Input for a user message log entry. */ export const UserMessageLogEntryInput = Schema.TaggedStruct('user-message', { agentId: AgentId, @@ -346,6 +366,7 @@ export const UserMessageLogEntryInput = Schema.TaggedStruct('user-message', { toolCallId: Schema.NullOr(ToolCallId), messageId: MessageId, message: UserMessageEncoded, + origin: Schema.optionalKey(UserMessageOrigin), }) .check(AgentRunContextFilter) .annotate({ identifier: 'UserMessageLogEntryInput' }) @@ -359,6 +380,7 @@ export const UserMessageLogEntry = Schema.TaggedStruct('user-message', { toolCallId: Schema.NullOr(ToolCallId), messageId: MessageId, message: UserMessageEncoded, + origin: Schema.optionalKey(UserMessageOrigin), }) .check(AgentRunContextFilter) .annotate({ identifier: 'UserMessageLogEntry' }) @@ -638,6 +660,89 @@ export const ErrorLogEntry = Schema.TaggedStruct('error', { }).annotate({ identifier: 'ErrorLogEntry' }) export type ErrorLogEntry = typeof ErrorLogEntry.Type +/** What kind of work a background task runs. Only subagents today; the vocabulary may grow. */ +export const BackgroundTaskType = Schema.Literals(['subagent']).annotate({ identifier: 'BackgroundTaskType' }) +export type BackgroundTaskType = typeof BackgroundTaskType.Type + +/** How one run of a background task ended. A later run of the same task starts with a new task_started. */ +export const BackgroundTaskFinalStatus = Schema.Literals(['completed', 'failed', 'stopped']).annotate({ + identifier: 'BackgroundTaskFinalStatus', +}) +export type BackgroundTaskFinalStatus = typeof BackgroundTaskFinalStatus.Type + +const NonNegativeInt = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)) + +/** Work done by one background task run, folded from the task agent's own assistant messages. */ +export const BackgroundTaskUsage = Schema.Struct({ + /** Assistant turns (model calls) in this run. */ + turns: NonNegativeInt, + /** Tool calls the task agent made in this run. */ + toolUses: NonNegativeInt, + /** Context size at the latest turn (input + output tokens), or null when the provider reported none. */ + totalTokens: Schema.NullOr(NonNegativeInt), +}).annotate({ identifier: 'BackgroundTaskUsage' }) +export type BackgroundTaskUsage = typeof BackgroundTaskUsage.Type + +const TaskStartedFields = { + /** The task id: the background agent's own id. */ + agentId: AgentId, + /** The agent whose tool call launched (or resumed) this run, and is notified when it ends. */ + parentAgentId: AgentId, + /** The launching tool call: the subagent call, or a send_message call that resumed the task. */ + toolCallId: ToolCallId, + taskType: BackgroundTaskType, + /** Registry type name of the agent; null for forks. */ + agentType: Schema.NullOr(Schema.String), + /** Short label for the task (the subagent call's description, or the agent type). */ + description: Schema.String, + /** The message this run started with. */ + prompt: Schema.String, + /** True when the run does not block its launcher. Foreground runs are not recorded as tasks today. */ + isBackgrounded: Schema.Boolean, + /** 1 for the first run; each resume of the same task id increments it. */ + runCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(1)), +} + +/** Input for the durable start of one background task run. */ +export const TaskStartedLogEntryInput = Schema.TaggedStruct('task_started', TaskStartedFields).annotate({ + identifier: 'TaskStartedLogEntryInput', +}) +export type TaskStartedLogEntryInput = typeof TaskStartedLogEntryInput.Type + +/** Durable start of one background task run. Additive to v1: older runtimes never wrote it. */ +export const TaskStartedLogEntry = Schema.TaggedStruct('task_started', { + ...CurrentStoredLogEntryEnvelope, + ...TaskStartedFields, +}).annotate({ identifier: 'TaskStartedLogEntry' }) +export type TaskStartedLogEntry = typeof TaskStartedLogEntry.Type + +const TaskFinishedFields = { + agentId: AgentId, + parentAgentId: AgentId, + toolCallId: ToolCallId, + status: BackgroundTaskFinalStatus, + /** The task agent's final assistant text for this run, or null when it produced none. */ + summary: Schema.NullOr(Schema.String), + /** Model-visible error description; set when status is failed. */ + error: Schema.NullOr(Schema.String), + usage: BackgroundTaskUsage, + durationMs: Schema.Finite.check(Schema.isGreaterThanOrEqualTo(0)), + lastToolName: Schema.NullOr(Schema.String), +} + +/** Input for the durable end of one background task run. */ +export const TaskFinishedLogEntryInput = Schema.TaggedStruct('task_finished', TaskFinishedFields).annotate({ + identifier: 'TaskFinishedLogEntryInput', +}) +export type TaskFinishedLogEntryInput = typeof TaskFinishedLogEntryInput.Type + +/** Durable end of one background task run. Additive to v1: older runtimes never wrote it. */ +export const TaskFinishedLogEntry = Schema.TaggedStruct('task_finished', { + ...CurrentStoredLogEntryEnvelope, + ...TaskFinishedFields, +}).annotate({ identifier: 'TaskFinishedLogEntry' }) +export type TaskFinishedLogEntry = typeof TaskFinishedLogEntry.Type + /** Append-time log entry schema. `EventLog.append` assigns `seq` and `ts`. */ export const LogEntryInput = Schema.Union([ SessionStartedLogEntryInput, @@ -654,6 +759,8 @@ export const LogEntryInput = Schema.Union([ AgentFinishedLogEntryInput, SessionTitleLogEntryInput, ErrorLogEntryInput, + TaskStartedLogEntryInput, + TaskFinishedLogEntryInput, ]).annotate({ identifier: 'LogEntryInput', discriminator: '_tag' }) export type LogEntryInput = typeof LogEntryInput.Type export const LogEntryInputs = Data.taggedEnum() @@ -674,6 +781,8 @@ export const LogEntryV1 = Schema.Union([ AgentFinishedLogEntry, SessionTitleLogEntry, ErrorLogEntry, + TaskStartedLogEntry, + TaskFinishedLogEntry, ]).annotate({ identifier: 'LogEntryV1', discriminator: '_tag' }) export type LogEntryV1 = typeof LogEntryV1.Type diff --git a/packages/fold-core/src/Session/SessionControls.ts b/packages/fold-core/src/Session/SessionControls.ts index 926da44..160fbf0 100644 --- a/packages/fold-core/src/Session/SessionControls.ts +++ b/packages/fold-core/src/Session/SessionControls.ts @@ -14,10 +14,14 @@ * - The session-wide graceful-stop signal (D9): first-write-wins, observed by every agent's loop at * its batch boundaries (root and subagents alike - the tree stops together), cleared by the facade * when the next send begins. + * + * Queued messages may carry a user-message origin (background task notifications). Adjacent queued + * task notifications drain as ONE message, so completions that land close together reach the model as + * a single user turn. */ import { Array as Arr, Context, Deferred, Effect, type Exit, Fiber, Layer, Ref, SynchronizedRef } from 'effect' -import type { AgentFinishedLogEntry } from '../EventLog/Schemas' +import { TaskNotificationOrigin, type AgentFinishedLogEntry, type UserMessageOrigin } from '../EventLog/Schemas' import type { AgentId } from '../Ids' import { AgentNotRunningError } from './Errors' @@ -33,11 +37,54 @@ export type FollowUpTicket = { readonly consumed: Effect.Effect } -type FollowUpEntry = { +/** One queued message as the agent loop appends it: its text and, for synthetic messages, its origin. */ +export type QueuedUserMessage = { readonly text: string + readonly origin: UserMessageOrigin | null +} + +type FollowUpEntry = QueuedUserMessage & { readonly consumed: Deferred.Deferred } +type QueuedTaskNotification = QueuedUserMessage & { readonly origin: TaskNotificationOrigin } + +const isTaskNotification = (message: QueuedUserMessage): message is QueuedTaskNotification => + message.origin?._tag === 'task-notification' + +/** + * Merge runs of adjacent task-notification messages into one message (texts joined, task ids + * concatenated); every other message passes through in order. + */ +export const coalesceQueuedMessages = ( + messages: ReadonlyArray, +): ReadonlyArray => { + const merged: Array = [] + for (const message of messages) { + const previous = merged.at(-1) + if (previous !== undefined && isTaskNotification(previous) && isTaskNotification(message)) { + merged[merged.length - 1] = { + text: `${previous.text}\n\n${message.text}`, + origin: TaskNotificationOrigin.make({ + taskIds: [...previous.origin.taskIds, ...message.origin.taskIds], + }), + } + continue + } + merged.push(message) + } + return merged +} + +/** The leading drain unit of a queue: one message, or a run of adjacent task notifications. */ +const leadingGroupLength = (queued: ReadonlyArray): number => { + const first = queued[0] + if (first === undefined) return 0 + if (!isTaskNotification(first)) return 1 + const end = queued.findIndex((message) => !isTaskNotification(message)) + return end === -1 ? queued.length : end +} + type RunningAgentRecord = { /** The run fiber; null between the claim and the fork. */ readonly fiber: Fiber.Fiber | null @@ -52,6 +99,8 @@ export type SessionControlsService = { /** Release a claim; abandons the agent's queued steering and follow-ups (undrained = never seen). */ readonly releaseRunning: (agentId: AgentId) => Effect.Effect readonly isRunning: (agentId: AgentId) => Effect.Effect + /** Snapshot of every agent currently claimed as running. */ + readonly runningAgentIds: Effect.Effect> /** Await the current run's exit; null when the agent is not running (or its fiber is not attached yet). */ readonly awaitRunning: (agentId: AgentId) => Effect.Effect | null> /** Interrupt one running agent's fiber and await its termination; false when it was not running. */ @@ -59,13 +108,24 @@ export type SessionControlsService = { /** Interrupt every running agent (D10 session interrupt). */ readonly interruptAllRunning: Effect.Effect /** Queue a steering message for a RUNNING agent; drained between its turns (D8). */ - readonly steer: (agentId: AgentId, text: string) => Effect.Effect - /** Pop queued steering per the session's steering mode; the loop appends each as a user message. */ - readonly drainSteering: (agentId: AgentId) => Effect.Effect> + readonly steer: ( + agentId: AgentId, + text: string, + origin?: UserMessageOrigin, + ) => Effect.Effect + /** + * Pop queued steering per the session's steering mode; the loop appends each as a user message. In + * one-at-a-time mode a run of adjacent task notifications counts as one message. + */ + readonly drainSteering: (agentId: AgentId) => Effect.Effect> /** Queue a follow-up for a RUNNING agent; drained when its run would complete naturally (D8). */ - readonly pushFollowUp: (agentId: AgentId, text: string) => Effect.Effect + readonly pushFollowUp: ( + agentId: AgentId, + text: string, + origin?: UserMessageOrigin, + ) => Effect.Effect /** Pop ALL queued follow-ups (resolving their tickets true); the loop appends them and continues. */ - readonly drainFollowUps: (agentId: AgentId) => Effect.Effect> + readonly drainFollowUps: (agentId: AgentId) => Effect.Effect> /** Request a session-wide graceful stop (D9); first reason wins until cleared. */ readonly requestSessionStop: (reason: string) => Effect.Effect readonly sessionStopReason: Effect.Effect @@ -96,7 +156,7 @@ export const layerSessionControls = (options?: { const steeringMode = options?.steeringMode ?? 'one-at-a-time' const running = yield* SynchronizedRef.make>(new Map()) - const steering = yield* Ref.make>>(new Map()) + const steering = yield* Ref.make>>(new Map()) const followUps = yield* Ref.make>>(new Map()) const stopReason = yield* Ref.make(null) @@ -147,6 +207,10 @@ export const layerSessionControls = (options?: { const isRunning = (agentId: AgentId): Effect.Effect => SynchronizedRef.get(running).pipe(Effect.map((current) => current.has(agentId))) + const runningAgentIds: Effect.Effect> = SynchronizedRef.get(running).pipe( + Effect.map((current) => [...current.keys()]), + ) + const awaitRunning = (agentId: AgentId): Effect.Effect | null> => SynchronizedRef.get(running).pipe( Effect.flatMap((current) => { @@ -169,22 +233,27 @@ export const layerSessionControls = (options?: { ), ) - const steer = (agentId: AgentId, text: string): Effect.Effect => + const steer = ( + agentId: AgentId, + text: string, + origin?: UserMessageOrigin, + ): Effect.Effect => Effect.gen(function* () { if (!(yield* isRunning(agentId))) return yield* notRunning(agentId) yield* Ref.update(steering, (current) => { const queued = current.get(agentId) ?? [] - return new Map(current).set(agentId, [...queued, text]) + return new Map(current).set(agentId, [...queued, { text, origin: origin ?? null }]) }) }) - const drainSteering = (agentId: AgentId): Effect.Effect> => + const drainSteering = (agentId: AgentId): Effect.Effect> => Ref.modify(steering, (current) => { const queued = current.get(agentId) ?? [] if (Arr.isReadonlyArrayEmpty(queued)) return [[], current] as const - const drained = steeringMode === 'all' ? queued : queued.slice(0, 1) - const remaining = steeringMode === 'all' ? [] : queued.slice(1) + const take = steeringMode === 'all' ? queued.length : leadingGroupLength(queued) + const drained = coalesceQueuedMessages(queued.slice(0, take)) + const remaining = queued.slice(take) const next = new Map(current) if (Arr.isReadonlyArrayEmpty(remaining)) next.delete(agentId) else next.set(agentId, remaining) @@ -194,18 +263,19 @@ export const layerSessionControls = (options?: { const pushFollowUp = ( agentId: AgentId, text: string, + origin?: UserMessageOrigin, ): Effect.Effect => Effect.gen(function* () { if (!(yield* isRunning(agentId))) return yield* notRunning(agentId) const consumed = yield* Deferred.make() yield* Ref.update(followUps, (current) => { const queued = current.get(agentId) ?? [] - return new Map(current).set(agentId, [...queued, { text, consumed }]) + return new Map(current).set(agentId, [...queued, { text, origin: origin ?? null, consumed }]) }) return { consumed: Deferred.await(consumed) } }) - const drainFollowUps = (agentId: AgentId): Effect.Effect> => + const drainFollowUps = (agentId: AgentId): Effect.Effect> => Effect.gen(function* () { const queued = yield* Ref.modify(followUps, (current) => { const entries = current.get(agentId) ?? [] @@ -214,7 +284,7 @@ export const layerSessionControls = (options?: { return [entries, next] as const }) yield* Effect.forEach(queued, (entry) => Deferred.succeed(entry.consumed, true), { discard: true }) - return queued.map((entry) => entry.text) + return coalesceQueuedMessages(queued.map((entry) => ({ text: entry.text, origin: entry.origin }))) }) const requestSessionStop = (reason: string): Effect.Effect => @@ -225,6 +295,7 @@ export const layerSessionControls = (options?: { setRunningFiber, releaseRunning, isRunning, + runningAgentIds, awaitRunning, interruptRunning, interruptAllRunning, diff --git a/packages/fold-core/src/Session/SessionLayer.ts b/packages/fold-core/src/Session/SessionLayer.ts index 4e7b257..efa43cb 100644 --- a/packages/fold-core/src/Session/SessionLayer.ts +++ b/packages/fold-core/src/Session/SessionLayer.ts @@ -15,6 +15,7 @@ import { Ids } from '../Ids' import { SessionAlreadyStartedError, SessionNotStartedError } from './Errors' import { Session, + type SendSessionInput, type SessionService, type StartedSession, type StartSessionInput, @@ -108,7 +109,7 @@ export const liveSessionLayer: Layer.Layer + const send: SessionService['send'] = Effect.fn('fold.session.send')((input: SendSessionInput) => Effect.gen(function* () { const started = yield* Ref.get(startedRef) @@ -121,6 +122,7 @@ export const liveSessionLayer: Layer.Layer Effect.Effect readonly adopt: (input: StartedSession) => Effect.Effect - readonly send: (input: { readonly text: string }) => Effect.Effect + readonly send: (input: SendSessionInput) => Effect.Effect readonly switchModel: (input: SwitchSessionModelInput) => Effect.Effect readonly compact: ( options?: CompactSessionOptions, diff --git a/packages/fold-core/test/EventLog/StoredLogEntryDecoder.vi.test.ts b/packages/fold-core/test/EventLog/StoredLogEntryDecoder.vi.test.ts index 0c98976..bedc1e3 100644 --- a/packages/fold-core/test/EventLog/StoredLogEntryDecoder.vi.test.ts +++ b/packages/fold-core/test/EventLog/StoredLogEntryDecoder.vi.test.ts @@ -6,8 +6,10 @@ import { CURRENT_LOG_ENTRY_VERSION, EventId, EventLogUnsupportedVersionError, + MessageId, SessionId, SUPPORTED_LOG_ENTRY_VERSIONS, + ToolCallId, decodeStoredLogEntry, } from '../../src/index' @@ -122,3 +124,77 @@ it.effect('rejects an unsupported event format without guessing its schema', () } }), ) + +const legacyUserMessageEntry = () => ({ + _tag: 'user-message', + seq: 2, + eventId: EventId.create(), + ts: 3, + agentId: AgentId.create(), + parentAgentId: null, + toolCallId: null, + messageId: MessageId.create(), + message: { role: 'user', content: [{ type: 'text', text: 'hello' }] }, +}) + +it.effect('decodes v1 user messages written before message origins existed', () => + Effect.gen(function* () { + const entry = yield* decodeStoredLogEntry(legacyUserMessageEntry()) + + expect(entry._tag).toBe('user-message') + if (!Predicate.isTagged(entry, 'user-message')) throw new Error('expected a user-message') + expect(entry.origin).toBeUndefined() + }), +) + +it.effect('decodes task-notification origins and background task lifecycle rows', () => + Effect.gen(function* () { + const taskId = AgentId.create() + const parentAgentId = AgentId.create() + const toolCallId = ToolCallId.create() + + const notification = yield* decodeStoredLogEntry({ + ...legacyUserMessageEntry(), + version: CURRENT_LOG_ENTRY_VERSION, + origin: { _tag: 'task-notification', taskIds: [taskId] }, + }) + if (!Predicate.isTagged(notification, 'user-message')) throw new Error('expected a user-message') + expect(notification.origin).toEqual({ _tag: 'task-notification', taskIds: [taskId] }) + + const started = yield* decodeStoredLogEntry({ + _tag: 'task_started', + seq: 3, + eventId: EventId.create(), + ts: 4, + version: CURRENT_LOG_ENTRY_VERSION, + agentId: taskId, + parentAgentId, + toolCallId, + taskType: 'subagent', + agentType: 'researcher', + description: 'find it', + prompt: 'look', + isBackgrounded: true, + runCount: 1, + }) + expect(started._tag).toBe('task_started') + + const finished = yield* decodeStoredLogEntry({ + _tag: 'task_finished', + seq: 4, + eventId: EventId.create(), + ts: 5, + agentId: taskId, + parentAgentId, + toolCallId, + status: 'completed', + summary: 'found it', + error: null, + usage: { turns: 1, toolUses: 2, totalTokens: null }, + durationMs: 10, + lastToolName: 'read', + }) + expect(finished._tag).toBe('task_finished') + expect(finished.version).toBe(1) + }), +) diff --git a/packages/fold-core/test/Session/SessionControlsQueue.vi.test.ts b/packages/fold-core/test/Session/SessionControlsQueue.vi.test.ts new file mode 100644 index 0000000..87e81ea --- /dev/null +++ b/packages/fold-core/test/Session/SessionControlsQueue.vi.test.ts @@ -0,0 +1,47 @@ +/** + * SessionControls queue semantics for runtime-written messages: queued messages keep their origin, and + * adjacent task notifications drain as one message (one-at-a-time steering counts the run as one unit). + */ +import { expect, it } from '@effect/vitest' +import { Effect } from 'effect' + +import { AgentId, coalesceQueuedMessages, layerSessionControls, SessionControls } from '../../src/index' + +const first = AgentId.create() +const second = AgentId.create() +const notification = (taskId: AgentId, text: string) => ({ + text, + origin: { _tag: 'task-notification' as const, taskIds: [taskId] }, +}) + +it('coalesces adjacent task notifications and keeps other messages in order', () => { + const merged = coalesceQueuedMessages([ + { text: 'plain', origin: null }, + notification(first, 'one'), + notification(second, 'two'), + { text: 'after', origin: null }, + ]) + + expect(merged).toEqual([ + { text: 'plain', origin: null }, + { text: 'one\n\ntwo', origin: { _tag: 'task-notification', taskIds: [first, second] } }, + { text: 'after', origin: null }, + ]) +}) + +it.effect('one-at-a-time steering drains a run of notifications as one message', () => + Effect.gen(function* () { + const controls = yield* SessionControls + const agent = AgentId.create() + yield* controls.claimRunning(agent) + yield* controls.steer(agent, 'one', { _tag: 'task-notification', taskIds: [first] }) + yield* controls.steer(agent, 'two', { _tag: 'task-notification', taskIds: [second] }) + yield* controls.steer(agent, 'person') + + expect(yield* controls.drainSteering(agent)).toEqual([ + { text: 'one\n\ntwo', origin: { _tag: 'task-notification', taskIds: [first, second] } }, + ]) + expect(yield* controls.drainSteering(agent)).toEqual([{ text: 'person', origin: null }]) + expect(yield* controls.runningAgentIds).toEqual([agent]) + }).pipe(Effect.provide(layerSessionControls())), +) From d165b072558bfc7009fd359754d634fcf41ebcfe Mon Sep 17 00:00:00 2001 From: dex Date: Tue, 13 Oct 2026 04:20:48 +0000 Subject: [PATCH 2/8] feat(core): run subagents in the background and notify their launcher - subagent tool: run_in_background for dispatch, fork, and resume; the call returns a receipt once the agent has started - BackgroundTasks service: per-run task scopes forked from the session scope, durable task rows, live progress, and notification delivery (follow-up to a running launcher, otherwise the root: follow-up when busy, a synthetic task-notification wake when idle; coalesced) - session API: backgroundTasks, backgroundTaskChanges, stopTask, awaitBackgroundIdle; interrupt() leaves background tasks running; stop() stops them with deferred notices; close records them stopped; reopening rebuilds the registry and closes orphaned runs - list_agents, stop_agent, send_message model tools (agentControlTools) Co-Authored-By: Claude Opus 5.5 HumanLayer-Session: https://app.dev.codelayer.gg/sessions/01a13295-e128-724d-bb90-666e8fbf2d01 --- packages/fold-core/src/Api/OpenSession.ts | 224 +++-- packages/fold-core/src/Api/ToolDefinition.ts | 2 + .../src/Subagents/AgentControlTools.ts | 184 ++++ .../src/Subagents/BackgroundTasks.ts | 795 ++++++++++++++++++ packages/fold-core/src/Subagents/Schemas.ts | 40 +- .../fold-core/src/Subagents/SubagentEngine.ts | 213 +++-- .../fold-core/src/Subagents/SubagentTool.ts | 106 ++- packages/fold-core/src/Tools/Contracts.ts | 67 ++ packages/fold-core/src/index.ts | 2 + .../test/Subagents/BackgroundHarness.ts | 118 +++ .../Subagents/BackgroundSubagents.vi.test.ts | 505 +++++++++++ 11 files changed, 2086 insertions(+), 170 deletions(-) create mode 100644 packages/fold-core/src/Subagents/AgentControlTools.ts create mode 100644 packages/fold-core/src/Subagents/BackgroundTasks.ts create mode 100644 packages/fold-core/test/Subagents/BackgroundHarness.ts create mode 100644 packages/fold-core/test/Subagents/BackgroundSubagents.vi.test.ts diff --git a/packages/fold-core/src/Api/OpenSession.ts b/packages/fold-core/src/Api/OpenSession.ts index ccd21db..17832cd 100644 --- a/packages/fold-core/src/Api/OpenSession.ts +++ b/packages/fold-core/src/Api/OpenSession.ts @@ -43,6 +43,7 @@ import { type LogEntry, type LogSeq, type ToolResultLogEntry, + type UserMessageOrigin, } from '../EventLog/Schemas' import { Ids, layerLiveIdFactory, type AgentId, type IdsService, type SessionId } from '../Ids' import { ModelCatalog, modelCatalogFromEntries, type ModelCatalogEntry } from '../Model/ModelCatalog' @@ -74,6 +75,13 @@ import { collectAgentDefinitions, type CollectedAgentDefinitions, } from '../Subagents/AgentRegistry' +import { + BackgroundTasks, + isOwnedByBackgroundTask, + layerBackgroundTasks, + type BackgroundTask, + type BackgroundTasksService, +} from '../Subagents/BackgroundTasks' import { type SubagentBusyError, SubagentNotFoundError } from '../Subagents/Errors' import { SessionAgents, type RealizedAgentTools, type RootAgentSnapshot } from '../Subagents/SessionAgents' import { continueSubagent, type ContinueSubagentInput } from '../Subagents/SubagentEngine' @@ -184,18 +192,38 @@ export type FoldSession = { /** * Request a session-wide graceful stop (D9): every running agent finishes its in-flight tool batch, * appends the results, and ends its run with `agent-finished{stopped}` - no further model calls. - * The signal clears when the next send begins. + * Running background tasks are stopped too (interrupted, recorded `stopped`); their notifications + * do not wake the root, and are delivered at the start of the root's next run instead. The signal + * clears when the next send begins. */ readonly stop: (reason?: string) => Effect.Effect /** - * Hard interrupt (D10). With no target, cancels every running agent's fiber tree (in-flight - * inference, tools, hooks, and subagents); uninterruptible finalizers write the durable markers - - * partial assistant text, interrupted tool results, subagent markers, and the terminal - * `agent-finished{interrupted}` - so resume sees coherent, honest history. With a target agentId, - * interrupts just that agent's run (a dispatched subagent's interruption folds into its - * dispatcher's tool result; the dispatcher keeps running). + * Hard interrupt (D10). With no target, cancels the foreground work: the root's run and every + * running agent that is not a background task or beneath one (in-flight inference, tools, hooks, + * and foreground subagents). Background tasks keep running and still notify the root when they + * finish; use {@link stopTask} or {@link stop} to end them. Uninterruptible finalizers write the + * durable markers - partial assistant text, interrupted tool results, subagent markers, and the + * terminal `agent-finished{interrupted}` - so resume sees coherent, honest history. With a target + * agentId, interrupts just that agent's run (a dispatched subagent's interruption folds into its + * dispatcher's tool result; the dispatcher keeps running; a background task ends `stopped`). */ readonly interrupt: (options?: AgentTargetOptions) => Effect.Effect + /** Every background task this session has run (running and finished), in first-launch order. */ + readonly backgroundTasks: Effect.Effect> + /** The background task list now, then after every change (status, progress, notification). */ + readonly backgroundTaskChanges: Stream.Stream> + /** + * Stop one running background task by id (full or unique short reference): its run is interrupted, + * recorded `task_finished{stopped}`, and its launcher is notified like any completion. Resolves + * false when the task is not running. + */ + readonly stopTask: (taskId: AgentId | string) => Effect.Effect + /** + * Resolve once background work has settled: no task is running, no task-notification wake is + * scheduled or running, and no notification is waiting for delivery. Headless hosts await this + * after `send` before exiting. + */ + readonly awaitBackgroundIdle: Effect.Effect /** * Switch the root agent to a different provider/model, optionally replacing its prompt blocks and * installed tools in the same transition. Durably records the epoch transition - `model-change`, the @@ -227,6 +255,9 @@ export type FoldSession = { ) => Effect.Effect } +/** One wake's message: the coalesced notification text and its origin. */ +type WakeMessage = { readonly text: string; readonly origin: UserMessageOrigin | null } + /** The switchable slice of a session's configuration, tracked so omitted switch options carry forward. */ type SessionAgentConfig = { readonly model: FoldModel @@ -261,6 +292,8 @@ type SessionGraph = { readonly eventLog: EventLogService readonly ids: IdsService readonly controls: SessionControlsService + readonly backgroundTasks: BackgroundTasksService + readonly sessionScope: Scope.Scope readonly systemPromptService: SystemPromptService readonly continueSubagent: ( input: ContinueSubagentInput, @@ -430,6 +463,9 @@ const assembleSessionGraph = (options: { const idsLayer = layerLiveIdFactory const eventLog = yield* EventLog const infraLayer = Layer.mergeAll(Layer.succeed(EventLog, eventLog), idsLayer, liveAgentEventsLayer) + const controlsLayer = layerSessionControls( + options.steering === undefined ? {} : { steeringMode: options.steering }, + ) const servicesLayer = Layer.mergeAll( infraLayer, Layer.succeedContext(hostServices), @@ -450,7 +486,10 @@ const assembleSessionGraph = (options: { // entries build the empty catalog, which behaves exactly like the Reference default. Layer.succeed(ModelCatalog, modelCatalogFromEntries(options.catalog ?? [])), Layer.succeed(StopConditions, agent.stopConditions ?? {}), - layerSessionControls(options.steering === undefined ? {} : { steeringMode: options.steering }), + controlsLayer, + // Background subagent runs: task scopes forked from this session scope, their durable task rows, + // and the delivery of their completions. Shares the session's one controls instance (memoized). + layerBackgroundTasks.pipe(Layer.provide(Layer.mergeAll(infraLayer, controlsLayer))), // Session-wide role->model bindings (profiles slice): one mutable map shared by the facade's // setProfile and the subagent operations' per-dispatch/resume resolution. layerProfiles(initialProfiles), @@ -504,6 +543,8 @@ const assembleSessionGraph = (options: { eventLog: Context.get(sessionServices, EventLog), ids: Context.get(sessionServices, Ids), controls: Context.get(sessionServices, SessionControls), + backgroundTasks: Context.get(sessionServices, BackgroundTasks), + sessionScope, systemPromptService: Context.get(sessionServices, SystemPrompt), continueSubagent: (input) => continueSubagent(input).pipe(Effect.provideContext(sessionServices)), profiles: Context.get(sessionServices, Profiles), @@ -536,9 +577,27 @@ const assembleSessionGraph = (options: { } }) -/** Build the public handle over one assembled, started-or-adopted session. */ -const makeSessionHandle = (graph: SessionGraph, identity: StartedSession): FoldSession => { - const { session, eventLog, ids, controls, continueSubagent, configRef, profiles } = graph +/** + * Build the public handle over one assembled, started-or-adopted session, and start delivering + * background task notifications for it in the session scope. + */ +const makeSessionHandle = (graph: SessionGraph, identity: StartedSession): Effect.Effect> => + Effect.gen(function* () { + const handle = sessionHandle(graph, identity) + yield* graph.backgroundTasks + .serveNotifications({ rootAgentId: identity.rootAgentId, wake: handle.wake }) + .pipe(Effect.forkIn(graph.sessionScope)) + return handle.session + }) + +/** The public handle plus the root wake it gives the notification server. */ +type SessionHandleParts = { + readonly session: FoldSession + readonly wake: (take: Effect.Effect) => Effect.Effect +} + +const sessionHandle = (graph: SessionGraph, identity: StartedSession): SessionHandleParts => { + const { session, eventLog, ids, controls, backgroundTasks, continueSubagent, configRef, profiles } = graph const rootAgentId = identity.rootAgentId const collectEntries: Effect.Effect> = Stream.runCollect(eventLog.entries()).pipe( @@ -617,43 +676,59 @@ const makeSessionHandle = (graph: SessionGraph, identity: StartedSession): // is mid-flight, and the durable epoch entries must land before the next send projects them. const gate = Semaphore.makeUnsafe(1) - /** One root run on a registered fiber: interruptible externally, honest markers on teardown. */ - const runRootSend = (text: string): Effect.Effect => - gate.withPermit( - Effect.gen(function* () { - // A new send clears a previous graceful-stop request (D9): stop targets current work. - yield* controls.clearSessionStop + /** + * One root run on a registered fiber: interruptible externally, honest markers on teardown. Callers + * hold the gate. `origin` marks a runtime-written turn (a task-notification wake). + */ + const runRootUnderGate = (text: string, origin?: UserMessageOrigin): Effect.Effect => + Effect.gen(function* () { + // A new send clears a previous graceful-stop request (D9): stop targets current work. + yield* controls.clearSessionStop - const claimed = yield* controls.claimRunning(rootAgentId) - if (!claimed) { - return yield* Effect.die(new Error('root agent already running while the send gate was held')) - } + const claimed = yield* controls.claimRunning(rootAgentId) + if (!claimed) { + return yield* Effect.die(new Error('root agent already running while the send gate was held')) + } - const entries = yield* collectEntries - const baselineSeq = entries.at(-1)?.seq - if (baselineSeq === undefined) { - return yield* Effect.die(new Error('send on an empty session log')) - } + const entries = yield* collectEntries + const baselineSeq = entries.at(-1)?.seq + if (baselineSeq === undefined) { + return yield* Effect.die(new Error('send on an empty session log')) + } - const runFiber = yield* Effect.forkChild( - session - .send({ text }) - .pipe( - Effect.orDie, - Effect.onExit(writeRootInterruptMarker(baselineSeq)), - Effect.ensuring(controls.releaseRunning(rootAgentId)), - ), - ) - yield* controls.setRunningFiber(rootAgentId, runFiber) + const runFiber = yield* Effect.forkChild( + session + .send({ text, origin }) + .pipe( + Effect.orDie, + Effect.onExit(writeRootInterruptMarker(baselineSeq)), + Effect.ensuring(controls.releaseRunning(rootAgentId)), + ), + ) + yield* controls.setRunningFiber(rootAgentId, runFiber) + // The root is running now: queued task notifications become follow-ups of this run. + yield* backgroundTasks.nudge + + const exit = yield* Fiber.await(runFiber) + if (Exit.isSuccess(exit)) return exit.value + if (Cause.hasInterrupts(exit.cause)) { + // The marker finalizer already wrote the durable interrupted outcome; return it. + return yield* lastFinishedFor(rootAgentId) + } + return yield* Effect.failCause(exit.cause) + }) - const exit = yield* Fiber.await(runFiber) - if (Exit.isSuccess(exit)) return exit.value - if (Cause.hasInterrupts(exit.cause)) { - // The marker finalizer already wrote the durable interrupted outcome; return it. - return yield* lastFinishedFor(rootAgentId) - } - return yield* Effect.failCause(exit.cause) - }), + const runRootSend = (text: string): Effect.Effect => gate.withPermit(runRootUnderGate(text)) + + /** A task-notification wake: take the queued notifications at the gate and run the root on them. */ + const wake = (take: Effect.Effect): Effect.Effect => + gate.withPermit( + take.pipe( + Effect.flatMap((message) => + message === null ? Effect.void : runRootUnderGate(message.text, message.origin ?? undefined), + ), + Effect.asVoid, + ), ) const send = ( @@ -778,12 +853,27 @@ const makeSessionHandle = (graph: SessionGraph, identity: StartedSession): return { call, result } }) + // Background tasks stop first, with deferred notices, so none of them can end gracefully on the stop + // signal and wake the root with an ordinary notification. const stop = (reason?: string): Effect.Effect => - controls.requestSessionStop(reason ?? 'the user requested a stop') + backgroundTasks + .stopAll('defer') + .pipe(Effect.andThen(controls.requestSessionStop(reason ?? 'the user requested a stop'))) + + /** Interrupt the foreground: every running agent that is not a background task or beneath one. */ + const interruptForeground: Effect.Effect = Effect.gen(function* () { + const running = yield* controls.runningAgentIds + const liveTasks = new Set( + (yield* backgroundTasks.list).filter((task) => task.status === 'running').map((task) => task.taskId), + ) + const entries = liveTasks.size === 0 ? [] : yield* collectEntries + const foreground = running.filter((agentId) => !isOwnedByBackgroundTask(entries, agentId, liveTasks)) + yield* Effect.forEach(foreground, (agentId) => controls.interruptRunning(agentId), { discard: true }) + }) const interrupt = (options?: AgentTargetOptions): Effect.Effect => options?.agentId === undefined - ? controls.interruptAllRunning + ? interruptForeground : resolveTarget(options.agentId).pipe( Effect.flatMap((target) => controls.interruptRunning(target)), // An unresolvable target is not running by definition - same no-op as an idle full id. @@ -791,6 +881,9 @@ const makeSessionHandle = (graph: SessionGraph, identity: StartedSession): Effect.asVoid, ) + const stopTask = (taskId: AgentId | string): Effect.Effect => + resolveTarget(taskId).pipe(Effect.flatMap((target) => backgroundTasks.stop(target, 'notify'))) + const switchModel = (model: FoldModel, switchOptions?: SwitchModelOptions): Effect.Effect => gate.withPermit( Effect.gen(function* () { @@ -875,19 +968,26 @@ const makeSessionHandle = (graph: SessionGraph, identity: StartedSession): .pipe(Effect.asVoid, Effect.orDie) return { - sessionId: identity.sessionId, - rootAgentId, - send, - steer, - injectSkill, - stop, - interrupt, - switchModel, - compact, - setProfile, - events: (fromSeq?: LogSeq) => session.events(fromSeq), - entries: collectEntries, - setTitle, + session: { + sessionId: identity.sessionId, + rootAgentId, + send, + steer, + injectSkill, + stop, + interrupt, + backgroundTasks: backgroundTasks.list, + backgroundTaskChanges: backgroundTasks.changes, + stopTask, + awaitBackgroundIdle: backgroundTasks.awaitIdle, + switchModel, + compact, + setProfile, + events: (fromSeq?: LogSeq) => session.events(fromSeq), + entries: collectEntries, + setTitle, + }, + wake, } } @@ -964,7 +1064,7 @@ const openWithLog = ( }) .pipe(Effect.orDie) - return makeSessionHandle(graph, started) + return yield* makeSessionHandle(graph, started) } if (!Predicate.isTagged(sessionStarted, 'session_started')) { return yield* Effect.die(new Error('cannot open: invalid session history')) @@ -974,6 +1074,8 @@ const openWithLog = ( rootAgentId: sessionStarted.rootAgentId, } yield* graph.session.adopt(identity).pipe(Effect.orDie) + // Rebuild the background task registry; runs the previous process left open are recorded stopped. + yield* graph.backgroundTasks.hydrate(entries) // D17 resume ruling + D20 resume roster rule: one epoch transition iff the configuration no // longer matches the log's projected root state - the model binding, or the leading block set @@ -1009,5 +1111,5 @@ const openWithLog = ( .pipe(Effect.orDie) } - return makeSessionHandle(graph, identity) + return yield* makeSessionHandle(graph, identity) }) diff --git a/packages/fold-core/src/Api/ToolDefinition.ts b/packages/fold-core/src/Api/ToolDefinition.ts index 7cf33ab..96a433c 100644 --- a/packages/fold-core/src/Api/ToolDefinition.ts +++ b/packages/fold-core/src/Api/ToolDefinition.ts @@ -21,6 +21,7 @@ import type { ModelRequestSettings } from '../Model/ModelRequestSettings' import type { Profiles } from '../Session/Profiles' import type { SessionControls } from '../Session/SessionControls' import type { SkillSourceService } from '../Skills/SkillSource' +import type { BackgroundTasks } from '../Subagents/BackgroundTasks' import type { SessionAgents } from '../Subagents/SessionAgents' import type { SubagentToolCapabilities } from '../Subagents/SubagentTool' import type { SystemPrompt } from '../SystemPrompt/SystemPromptService' @@ -61,6 +62,7 @@ export type ToolHandlerServices = | SessionControls | Profiles | SessionAgents + | BackgroundTasks type ToolDependency = | typeof ToolState diff --git a/packages/fold-core/src/Subagents/AgentControlTools.ts b/packages/fold-core/src/Subagents/AgentControlTools.ts new file mode 100644 index 0000000..8e699e8 --- /dev/null +++ b/packages/fold-core/src/Subagents/AgentControlTools.ts @@ -0,0 +1,184 @@ +/** + * This file builds the model-facing controls over background agents: `list_agents`, `stop_agent`, and + * `send_message`. Install them next to a `subagentTool` value (`agentControlTools()` returns all + * three). They are thin adapters over the session's BackgroundTasks registry, its run controls, and + * the subagent operations: listing reads the registry, stopping interrupts one run (silently - the + * caller already knows), and messaging steers a running agent or resumes a finished one in the + * background. Every expected failure comes back to the model as an instructive tool failure. + */ +import { Array as Arr, Effect, Match, Predicate, Stream } from 'effect' + +import { defineTool, type FoldTool } from '../Api/ToolDefinition' +import { EventLog } from '../EventLog/EventLogService' +import type { LogEntry } from '../EventLog/Schemas' +import type { AgentId } from '../Ids' +import { SessionControls } from '../Session/SessionControls' +import { CurrentAgent } from '../ToolRuntime/ToolContextServices' +import { listAgentsToolContract, sendMessageToolContract, stopAgentToolContract } from '../Tools/Contracts' +import { ToolResultFailure, ToolResultText } from '../Tools/ToolResultContent' +import { agentIdsFromEntries, resolveAgentIdRef, shortAgentId } from './AgentIdRef' +import { BackgroundTasks, type BackgroundTask } from './BackgroundTasks' +import { resumeSubagentInBackground } from './SubagentEngine' +import { renderBackgroundLaunch } from './SubagentTool' + +const RESULT_PREVIEW_LENGTH = 300 + +const preview = (text: string): string => { + const flat = text.replace(/\s+/g, ' ').trim() + return flat.length <= RESULT_PREVIEW_LENGTH ? flat : `${flat.slice(0, RESULT_PREVIEW_LENGTH)}...` +} + +/** One list_agents line (plus an indented detail line) for a task. */ +const renderTaskLine = (task: BackgroundTask): string => { + const label = task.agentType ?? 'fork' + const work = `${task.usage.turns} turn${task.usage.turns === 1 ? '' : 's'}, ${task.usage.toolUses} tool use${task.usage.toolUses === 1 ? '' : 's'}` + const head = `- ${shortAgentId(task.taskId)} [${task.status}] ${label}: "${task.description}" (${work})` + const detail = Match.value(task.status).pipe( + Match.when('running', () => (task.lastToolName === null ? null : ` last tool: ${task.lastToolName}`)), + Match.when('failed', () => (task.error === null ? null : ` error: ${preview(task.error)}`)), + Match.orElse(() => (task.summary === null ? null : ` result: ${preview(task.summary)}`)), + ) + return detail === null ? head : `${head}\n${detail}` +} + +/** Render the whole registry for the model: running tasks first, then finished ones. */ +export const renderAgentList = (tasks: ReadonlyArray): string => { + if (Arr.isReadonlyArrayEmpty(tasks)) return 'No background agents in this session.' + const running = tasks.filter((task) => task.status === 'running') + const finished = tasks.filter((task) => task.status !== 'running') + return [ + `Background agents: ${running.length} running, ${finished.length} finished.`, + ...running.map(renderTaskLine), + ...finished.map(renderTaskLine), + ].join('\n') +} + +const collectEntries: Effect.Effect, never, EventLog> = EventLog.use((eventLog) => + Stream.runCollect(eventLog.entries()), +).pipe( + Effect.orDie, + Effect.map((entries): ReadonlyArray => entries), +) + +const unknownAgent = (ref: string, candidates?: ReadonlyArray): ToolResultFailure => + ToolResultFailure.make({ + text: + candidates === undefined + ? `No agent with agent_id "${ref}" exists in this session. Use list_agents to see the agent ids.` + : `agent_id "${ref}" is ambiguous: it matches ${candidates.length} agents ` + + `(${[...new Set(candidates)].join(', ')}). Provide more characters of the agent_id.`, + }) + +/** Resolve a model-supplied reference against a set of known agent ids, as a tool failure when it does not match. */ +const resolveRef = (known: Iterable, ref: string): Effect.Effect => + Match.valueTags(resolveAgentIdRef(known, ref), { + resolved: ({ agentId }) => Effect.succeed(agentId), + 'not-found': () => Effect.fail(unknownAgent(ref)), + ambiguous: ({ candidates }) => Effect.fail(unknownAgent(ref, candidates)), + }) + +/** True when `target` is the caller itself or one of the caller's running ancestors. */ +const isSelfOrAncestor = (entries: ReadonlyArray, caller: AgentId, target: AgentId): boolean => { + let current: AgentId | null = caller + const seen = new Set() + while (current !== null && !seen.has(current)) { + if (current === target) return true + seen.add(current) + const id: AgentId = current + const started = entries.find((entry) => Predicate.isTagged(entry, 'agent_started') && entry.agentId === id) + current = started === undefined ? null : started.parentAgentId + } + return false +} + +/** list_agents: the registry, rendered. */ +export const listAgentsTool = (): FoldTool => + defineTool({ + ...listAgentsToolContract, + handler: () => + BackgroundTasks.use((tasks) => tasks.list).pipe( + Effect.map((tasks) => ToolResultText.make({ text: renderAgentList(tasks) })), + ), + }) + +/** stop_agent: interrupt one running background task. The caller asked, so no notification follows. */ +export const stopAgentTool = (): FoldTool => + defineTool({ + ...stopAgentToolContract, + handler: ({ agent_id }) => + Effect.gen(function* () { + const tasks = yield* BackgroundTasks + const known = (yield* tasks.list).map((task) => task.taskId) + const taskId = yield* resolveRef(known, agent_id) + const stopped = yield* tasks.stop(taskId, 'silent') + if (!stopped) { + const task = yield* tasks.get(taskId) + return yield* Effect.fail( + ToolResultFailure.make({ + text: `Agent ${shortAgentId(taskId)} is not running (status: ${task?.status ?? 'unknown'}).`, + }), + ) + } + return ToolResultText.make({ + text: + `Stopped background agent ${shortAgentId(taskId)}. Its work so far is kept; send_message ` + + `with this agent_id continues it.`, + }) + }), + }) + +/** send_message: steer a running subagent, or continue a finished one as a background task. */ +export const sendMessageTool = (): FoldTool => + defineTool({ + ...sendMessageToolContract, + handler: ({ agent_id, message }) => + Effect.gen(function* () { + const tasks = yield* BackgroundTasks + const controls = yield* SessionControls + const caller = yield* CurrentAgent + const entries = yield* collectEntries + const target = yield* resolveRef(agentIdsFromEntries(entries), agent_id) + + if (isSelfOrAncestor(entries, caller.agentId, target)) { + return yield* Effect.fail( + ToolResultFailure.make({ + text: `Agent ${shortAgentId(target)} is you or an agent waiting on you; it cannot be messaged.`, + }), + ) + } + + const steerRunning = controls.steer(target, message).pipe( + Effect.as( + ToolResultText.make({ + text: + `Message queued for running agent ${shortAgentId(target)}; it reads it after its ` + + `current step.`, + }), + ), + ) + const steered = yield* steerRunning.pipe( + Effect.catchTag('AgentNotRunningError', () => Effect.succeed(null)), + ) + if (steered !== null) return steered + + // A run that just ended may still be releasing the agent; wait for that before resuming it. + yield* tasks.awaitSettled(target) + return yield* resumeSubagentInBackground({ agentId: target, prompt: message, skill: null }).pipe( + Effect.map((task) => ToolResultText.make({ text: renderBackgroundLaunch(task) })), + Effect.catchTags({ + SubagentNotFoundError: () => Effect.fail(unknownAgent(agent_id)), + SubagentBusyError: () => + Effect.fail( + ToolResultFailure.make({ + text: `Agent ${shortAgentId(target)} is busy right now. Try again after its current run ends.`, + }), + ), + SkillNotFoundError: (error) => + Effect.fail(ToolResultFailure.make({ text: `Skill "${error.name}" not found.` })), + }), + ) + }), + }) + +/** The three background-agent controls, for installing next to a `subagentTool` value. */ +export const agentControlTools = (): ReadonlyArray => [listAgentsTool(), stopAgentTool(), sendMessageTool()] diff --git a/packages/fold-core/src/Subagents/BackgroundTasks.ts b/packages/fold-core/src/Subagents/BackgroundTasks.ts new file mode 100644 index 0000000..55156ea --- /dev/null +++ b/packages/fold-core/src/Subagents/BackgroundTasks.ts @@ -0,0 +1,795 @@ +/** + * This file implements BackgroundTasks - one session's registry of background subagent runs and the + * delivery of their completions. It is the deep module behind `subagent { run_in_background: true }`: + * + * - Lifetime. Each run gets its own scope forked from the session scope. The subagent's claim, its + * provisioned runtime, and its run fiber live there, so the launching tool call returns at once and + * the run outlives it. Closing the session closes every task scope, which interrupts the run. + * - Durable record. `task_started` lands after the agent's own start rows and before its first message; + * `task_finished` is written by an uninterruptible exit finalizer on the run fiber, so a run that + * completes, fails, is stopped, or is torn down by session close always gets one. + * - Live record. A SubscriptionRef holds every task (running and finished) for snapshots and change + * streams; a watcher folds the task agent's assistant messages into turns, tool uses, tokens, and the + * last tool name, and publishes them as ephemeral `task-progress` deltas. + * - Notification. When a run ends, its `` is queued for the agent that launched it. + * `serveNotifications` delivers queued notifications: to a running launcher as a follow-up (at its + * natural completion), otherwise to the root agent - as a follow-up when the root is running, or by + * waking it with a new run when it is idle. Everything pending when a wake takes the root's gate goes + * into that one wake, and adjacent queued notifications coalesce into one user message. + * - Stop. A stop interrupts one run's fiber. The stop notice decides how its notification travels: + * `notify` wakes the launcher like any completion, `defer` holds it for the root's next run (session + * stop), and `silent` drops it (the launcher stopped the task itself and already knows). + */ +import { + Array as Arr, + Cause, + Clock, + Context, + Deferred, + Effect, + Exit, + Fiber, + Layer, + Match, + Predicate, + Queue, + Ref, + Scope, + Stream, + SubscriptionRef, +} from 'effect' + +import { AgentEvents } from '../AgentEvents/AgentEventsService' +import { EventLog } from '../EventLog/EventLogService' +import { encodedContentParts, encodedContentText, isPartOfType } from '../EventLog/MessageContent' +import { + LogEntryInputs, + type AgentFinishedLogEntry, + type AssistantMessageLogEntry, + type BackgroundTaskFinalStatus, + type BackgroundTaskUsage, + type LogEntry, + type LogSeq, + type TaskFinishedLogEntry, + type TaskStartedLogEntry, + TaskNotificationOrigin, +} from '../EventLog/Schemas' +import { usageInputTotal, usageOutputTotal } from '../EventLog/Usage' +import type { AgentId, ToolCallId } from '../Ids' +import { SessionControls, type QueuedUserMessage } from '../Session/SessionControls' +import { modelVisibleErrorDetailsFromCause } from '../ToolRuntime/ModelVisibleErrors' +import { shortAgentId } from './AgentIdRef' + +/** Status of a background task: running, or how its latest run ended. */ +export type BackgroundTaskStatus = 'running' | BackgroundTaskFinalStatus + +/** + * One background task as hosts and tools see it. The task id is the background agent's id; a task + * keeps its id across runs (a resume starts a new run and bumps `runCount`). + */ +export type BackgroundTask = { + readonly taskId: AgentId + /** The tool call that launched the latest run (a subagent call, or a send_message resume). */ + readonly toolCallId: ToolCallId + /** The agent that launched the latest run; it is notified when the run ends. */ + readonly parentAgentId: AgentId + readonly taskType: 'subagent' + /** Registry type name; null for forks. */ + readonly agentType: string | null + readonly description: string + /** The message the latest run started with. */ + readonly prompt: string + readonly status: BackgroundTaskStatus + readonly isBackgrounded: boolean + /** Epoch millis when the latest run started. */ + readonly startedAt: number + /** Epoch millis when the latest run ended; null while running. */ + readonly endedAt: number | null + /** Epoch millis when the latest run's notification was handed to its launcher; null until then. */ + readonly notifiedAt: number | null + readonly usage: BackgroundTaskUsage + readonly lastToolName: string | null + /** The task agent's latest assistant text (truncated) while it runs. */ + readonly progressSummary: string | null + /** The latest run's final assistant text. */ + readonly summary: string | null + readonly error: string | null + readonly runCount: number +} + +/** Who launched a background run, and what it was asked to do. */ +export type LaunchBackgroundTaskInput = { + readonly taskId: AgentId + readonly parentAgentId: AgentId + readonly toolCallId: ToolCallId + readonly agentType: string | null + readonly description: string + readonly prompt: string +} + +/** A prepared run: the agent is claimed and started; forking `run` executes it. */ +export type PreparedRun = { readonly run: Effect.Effect } + +/** How a stopped run's notification travels: wake the launcher, hold it for the next root run, or drop it. */ +export type TaskStopNotice = 'notify' | 'defer' | 'silent' + +/** Hands one wake's message (taken at the root's gate) to a fresh root run; null means nothing to deliver. */ +export type RootWake = (take: Effect.Effect) => Effect.Effect + +export type BackgroundTasksService = { + /** + * Start one background run. `prepare` runs in the new task scope and returns the agent's run + * effect: it claims the agent, provisions its runtime, and writes its start rows. A failure of + * `prepare` (a busy agent) closes the task scope and propagates; nothing is recorded. + */ + readonly launch: ( + input: LaunchBackgroundTaskInput, + prepare: Effect.Effect, + ) => Effect.Effect> + /** Every task this session has seen (running and finished), in first-launch order. */ + readonly list: Effect.Effect> + /** The task list now, then after every change. */ + readonly changes: Stream.Stream> + readonly get: (taskId: AgentId) => Effect.Effect + /** Interrupt one running task and wait for its durable end; false when it was not running. */ + readonly stop: (taskId: AgentId, notice: TaskStopNotice) => Effect.Effect + /** Stop every running task. */ + readonly stopAll: (notice: TaskStopNotice) => Effect.Effect + /** Wait until the task's latest run has fully released its agent (no-op when not running). */ + readonly awaitSettled: (taskId: AgentId) => Effect.Effect + /** + * Adopt a resumed log: rebuild the registry from its task rows, and close any run the previous + * process left open with a `stopped` task_finished whose notification waits for the next root run. + */ + readonly hydrate: (entries: ReadonlyArray) => Effect.Effect + /** Deliver queued notifications forever. The session handle forks this into the session scope. */ + readonly serveNotifications: (options: { + readonly rootAgentId: AgentId + readonly wake: RootWake + }) => Effect.Effect + /** Re-check queued notifications now (the root just started a run, or a stop request was cleared). */ + readonly nudge: Effect.Effect + /** + * Resolve once no task is running, no wake run is scheduled or running, and every queued + * notification has been delivered (deferred ones excepted: they wait for the next root run). + */ + readonly awaitIdle: Effect.Effect +} + +/** BackgroundTasks service tag; one instance per session, shared by the engine, tools, and handle. */ +export class BackgroundTasks extends Context.Service()( + 'fold/BackgroundTasks', +) {} + +/** One finished run waiting for delivery. */ +type TaskNotification = { + readonly taskId: AgentId + readonly parentAgentId: AgentId + readonly text: string + /** Deliver only alongside a root run someone else started; never wakes the root by itself. */ + readonly deferred: boolean +} + +type State = { + readonly tasks: ReadonlyArray + readonly pending: ReadonlyArray + /** Notifications handed to a running agent whose consumption is not yet confirmed. */ + readonly inFlight: number + /** A wake run is waiting for the root's gate or running. */ + readonly waking: boolean +} + +/** Per-run control state, kept only while the run's scope is open. */ +type ActiveRun = { + readonly stopNotice: Ref.Ref + readonly settled: Deferred.Deferred +} + +const PROGRESS_SUMMARY_LENGTH = 200 +const NOTIFICATION_RESULT_LENGTH = 20_000 + +const emptyUsage: BackgroundTaskUsage = { turns: 0, toolUses: 0, totalTokens: null } + +/** Progress of one run so far: work done, last tool, and latest assistant text. */ +type RunProgress = { + readonly usage: BackgroundTaskUsage + readonly lastToolName: string | null + readonly text: string | null +} + +const emptyProgress: RunProgress = { usage: emptyUsage, lastToolName: null, text: null } + +const truncate = (text: string, length: number): string => + text.length <= length ? text : `${text.slice(0, length)}... [truncated]` + +/** Fold one of the task agent's assistant messages into its run progress. */ +const foldAssistantTurn = (progress: RunProgress, entry: AssistantMessageLogEntry): RunProgress => { + const toolCalls = encodedContentParts(entry.message.content).filter(isPartOfType('tool-call')) + const text = encodedContentText(entry.message.content).trim() + const usage = entry.finish?.usage + return { + usage: { + turns: progress.usage.turns + 1, + toolUses: progress.usage.toolUses + toolCalls.length, + totalTokens: + usage === undefined + ? progress.usage.totalTokens + : Math.round(usageInputTotal(usage) + usageOutputTotal(usage)), + }, + lastToolName: toolCalls.at(-1)?.name ?? progress.lastToolName, + text: text.length > 0 ? text : progress.text, + } +} + +const isRunTurn = + (taskId: AgentId, afterSeq: LogSeq) => + (entry: LogEntry): entry is AssistantMessageLogEntry => + Predicate.isTagged(entry, 'assistant-message') && entry.agentId === taskId && entry.seq > afterSeq + +/** Progress of one run, folded from the log: the task agent's own turns after its task_started row. */ +const progressForRun = (entries: ReadonlyArray, taskId: AgentId, afterSeq: LogSeq) => + entries.filter(isRunTurn(taskId, afterSeq)).reduce(foldAssistantTurn, emptyProgress) + +/** How a run ended, from its fiber exit: the durable terminal entry, or the cause that tore it down. */ +const finalStatusFor = ( + exit: Exit.Exit, +): { + readonly status: BackgroundTaskFinalStatus + readonly error: string | null + readonly resultText: string | null +} => { + if (Exit.isFailure(exit)) { + return Cause.hasInterrupts(exit.cause) + ? { status: 'stopped', error: null, resultText: null } + : { status: 'failed', error: modelVisibleErrorDetailsFromCause(exit.cause), resultText: null } + } + const finished = exit.value + return Match.value(finished.outcome).pipe( + Match.when('completed', () => ({ status: 'completed' as const, error: null, resultText: finished.resultText })), + Match.when('error', () => ({ + status: 'failed' as const, + error: finished.reason, + resultText: finished.resultText, + })), + Match.whenOr('stopped', 'interrupted', () => ({ + status: 'stopped' as const, + error: null, + resultText: finished.resultText, + })), + Match.exhaustive, + ) +} + +const statusPhrase = (status: BackgroundTaskFinalStatus): string => + Match.value(status).pipe( + Match.when('completed', () => 'completed'), + Match.when('failed', () => 'failed'), + Match.when('stopped', () => 'was stopped'), + Match.exhaustive, + ) + +/** The `` block the launcher reads when a run ends. */ +export const renderTaskNotification = (task: BackgroundTask): string => { + const status = task.status === 'running' ? 'completed' : task.status + const id = shortAgentId(task.taskId) + const lines = [ + '', + `${id}`, + `${status}`, + `Background agent "${task.description}" ${statusPhrase(status)}`, + ] + if (task.summary !== null) lines.push(`\n${truncate(task.summary, NOTIFICATION_RESULT_LENGTH)}\n`) + if (task.error !== null) lines.push(`${task.error}`) + lines.push( + `turns: ${task.usage.turns}, tool uses: ${task.usage.toolUses}`, + `The agent keeps its context. Send it a message with agent_id ${id} to continue it.`, + '', + ) + return lines.join('\n') +} + +const notificationMessage = (notifications: ReadonlyArray): QueuedUserMessage => ({ + text: notifications.map((notification) => notification.text).join('\n\n'), + origin: TaskNotificationOrigin.make({ taskIds: notifications.map((notification) => notification.taskId) }), +}) + +const replaceTask = ( + tasks: ReadonlyArray, + taskId: AgentId, + update: (task: BackgroundTask) => BackgroundTask, +): ReadonlyArray => tasks.map((task) => (task.taskId === taskId ? update(task) : task)) + +const upsertTask = (tasks: ReadonlyArray, next: BackgroundTask): ReadonlyArray => + tasks.some((task) => task.taskId === next.taskId) ? replaceTask(tasks, next.taskId, () => next) : [...tasks, next] + +const taskFromStarted = (started: TaskStartedLogEntry, previous: BackgroundTask | undefined): BackgroundTask => ({ + taskId: started.agentId, + toolCallId: started.toolCallId, + parentAgentId: started.parentAgentId, + taskType: started.taskType, + agentType: started.agentType, + description: started.description, + prompt: started.prompt, + status: 'running', + isBackgrounded: started.isBackgrounded, + startedAt: started.ts, + endedAt: null, + notifiedAt: null, + usage: emptyUsage, + lastToolName: null, + progressSummary: null, + summary: previous?.summary ?? null, + error: null, + runCount: started.runCount, +}) + +const applyFinished = (task: BackgroundTask, finished: TaskFinishedLogEntry): BackgroundTask => ({ + ...task, + status: finished.status, + endedAt: finished.ts, + usage: finished.usage, + lastToolName: finished.lastToolName, + progressSummary: null, + summary: finished.summary, + error: finished.error, +}) + +/** + * True when an agent is a running background task or runs somewhere beneath one (a foreground + * subagent of a background task). A root-only interrupt leaves these alone. + */ +export const isOwnedByBackgroundTask = ( + entries: ReadonlyArray, + agentId: AgentId, + runningTaskIds: ReadonlySet, +): boolean => { + const seen = new Set() + let current: AgentId | null = agentId + while (current !== null && !seen.has(current)) { + if (runningTaskIds.has(current)) return true + seen.add(current) + const id: AgentId = current + const started = entries.find((entry) => Predicate.isTagged(entry, 'agent_started') && entry.agentId === id) + current = started === undefined ? null : started.parentAgentId + } + return false +} + +/** The live layer. Built into the session scope, which owns every task scope and delivery fiber. */ +export const layerBackgroundTasks: Layer.Layer = + Layer.effect( + BackgroundTasks, + Effect.gen(function* () { + const sessionScope = yield* Effect.scope + const eventLog = yield* EventLog + const controls = yield* SessionControls + const agentEvents = yield* AgentEvents + + const state = yield* SubscriptionRef.make({ tasks: [], pending: [], inFlight: 0, waking: false }) + const activeRuns = yield* Ref.make>(new Map()) + const kicks = yield* Queue.sliding(1) + + const collectEntries: Effect.Effect> = Stream.runCollect(eventLog.entries()).pipe( + Effect.orDie, + Effect.map((entries): ReadonlyArray => entries), + ) + const kick: Effect.Effect = Queue.offer(kicks, undefined).pipe(Effect.asVoid) + const updateState = (f: (current: State) => State): Effect.Effect => SubscriptionRef.update(state, f) + const getTask = (taskId: AgentId): Effect.Effect => + SubscriptionRef.get(state).pipe( + Effect.map((current) => current.tasks.find((task) => task.taskId === taskId) ?? null), + ) + + const enqueue = (notification: TaskNotification): Effect.Effect => + updateState((current) => ({ ...current, pending: [...current.pending, notification] })).pipe( + Effect.andThen(kick), + ) + + /** Fold the task agent's turns into live progress as they land, publishing each as a delta. */ + const watchProgress = (task: BackgroundTask, startedSeq: LogSeq): Effect.Effect => + eventLog.subscribe(startedSeq + 1).pipe( + Stream.orDie, + Stream.filter(isRunTurn(task.taskId, startedSeq)), + Stream.runFoldEffect( + () => emptyProgress, + (progress, entry) => { + const next = foldAssistantTurn(progress, entry) + const progressSummary = + next.text === null ? null : truncate(next.text, PROGRESS_SUMMARY_LENGTH) + return updateState((current) => ({ + ...current, + tasks: replaceTask(current.tasks, task.taskId, (existing) => + existing.status === 'running' + ? { + ...existing, + usage: next.usage, + lastToolName: next.lastToolName, + progressSummary, + } + : existing, + ), + })).pipe( + Effect.andThen( + agentEvents.publish({ + kind: 'delta', + agentId: task.taskId, + parentAgentId: task.parentAgentId, + toolCallId: task.toolCallId, + part: { + type: 'task-progress', + taskId: task.taskId, + turns: next.usage.turns, + toolUses: next.usage.toolUses, + totalTokens: next.usage.totalTokens, + lastToolName: next.lastToolName, + progressSummary, + }, + }), + ), + Effect.as(next), + ) + }, + ), + Effect.asVoid, + ) + + /** Uninterruptible run-exit finalizer: the durable task_finished row and the live record. */ + const finishRun = + (task: BackgroundTask, startedSeq: LogSeq) => + (exit: Exit.Exit): Effect.Effect => + Effect.gen(function* () { + const entries = yield* collectEntries + const progress = progressForRun(entries, task.taskId, startedSeq) + const ending = finalStatusFor(exit) + const endedAt = yield* Clock.currentTimeMillis + const finished = yield* eventLog + .append( + LogEntryInputs.task_finished({ + agentId: task.taskId, + parentAgentId: task.parentAgentId, + toolCallId: task.toolCallId, + status: ending.status, + summary: ending.resultText ?? progress.text, + error: ending.error, + usage: progress.usage, + durationMs: Math.max(0, endedAt - task.startedAt), + lastToolName: progress.lastToolName, + }), + ) + .pipe(Effect.orDie) + if (!Predicate.isTagged(finished, 'task_finished')) { + return yield* Effect.die(new Error(`EventLog returned ${finished._tag} for task_finished`)) + } + yield* updateState((current) => ({ + ...current, + tasks: replaceTask(current.tasks, task.taskId, (existing) => + applyFinished(existing, finished), + ), + })) + }) + + const launch = ( + input: LaunchBackgroundTaskInput, + prepare: Effect.Effect, + ): Effect.Effect> => + Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const taskScope = yield* Scope.fork(sessionScope) + const { run } = yield* restore(prepare).pipe( + Scope.provide(taskScope), + Effect.onExit((exit) => + Exit.isSuccess(exit) ? Effect.void : Scope.close(taskScope, exit), + ), + ) + + const previous = yield* getTask(input.taskId) + const started = yield* eventLog + .append( + LogEntryInputs.task_started({ + agentId: input.taskId, + parentAgentId: input.parentAgentId, + toolCallId: input.toolCallId, + taskType: 'subagent', + agentType: input.agentType, + description: input.description, + prompt: input.prompt, + isBackgrounded: true, + runCount: previous === null ? 1 : previous.runCount + 1, + }), + ) + .pipe(Effect.orDie) + if (!Predicate.isTagged(started, 'task_started')) { + return yield* Effect.die(new Error(`EventLog returned ${started._tag} for task_started`)) + } + const task = taskFromStarted(started, previous ?? undefined) + yield* updateState((current) => ({ ...current, tasks: upsertTask(current.tasks, task) })) + + const activeRun: ActiveRun = { + stopNotice: yield* Ref.make('notify'), + settled: yield* Deferred.make(), + } + yield* Ref.update(activeRuns, (current) => new Map(current).set(task.taskId, activeRun)) + + yield* watchProgress(task, started.seq).pipe(Effect.forkIn(taskScope)) + const runFiber = yield* run.pipe( + Effect.onExit(finishRun(task, started.seq)), + Effect.forkIn(taskScope), + ) + // Closing the task scope (session close) interrupts the run and waits for its finalizers, + // so the durable markers land before the session's services go away. + yield* Scope.addFinalizer(taskScope, Fiber.interrupt(runFiber)) + yield* controls.setRunningFiber(task.taskId, runFiber) + + // Settle the run: release its scope (the agent's claim and runtime) BEFORE the notification + // is queued, so a launcher woken by it can immediately resume the same agent. + yield* Effect.gen(function* () { + yield* Fiber.await(runFiber) + yield* Scope.close(taskScope, Exit.void) + const notice = yield* Ref.get(activeRun.stopNotice) + yield* Ref.update(activeRuns, (current) => { + const next = new Map(current) + next.delete(task.taskId) + return next + }) + const ended = yield* getTask(task.taskId) + if (ended !== null && notice !== 'silent') { + yield* enqueue({ + taskId: ended.taskId, + parentAgentId: ended.parentAgentId, + text: renderTaskNotification(ended), + deferred: notice === 'defer', + }) + } + yield* Deferred.succeed(activeRun.settled, undefined) + }).pipe(Effect.forkIn(sessionScope)) + + return task + }), + ) + + const stop = (taskId: AgentId, notice: TaskStopNotice): Effect.Effect => + Effect.gen(function* () { + const run = (yield* Ref.get(activeRuns)).get(taskId) + if (run === undefined) return false + yield* Ref.set(run.stopNotice, notice) + return yield* controls.interruptRunning(taskId) + }) + + const stopAll = (notice: TaskStopNotice): Effect.Effect => + Ref.get(activeRuns).pipe( + Effect.flatMap((runs) => + Effect.forEach([...runs.keys()], (taskId) => stop(taskId, notice), { + concurrency: 'unbounded', + discard: true, + }), + ), + ) + + const awaitSettled = (taskId: AgentId): Effect.Effect => + Ref.get(activeRuns).pipe( + Effect.flatMap((runs) => { + const run = runs.get(taskId) + return run === undefined ? Effect.void : Deferred.await(run.settled) + }), + ) + + const hydrate = (entries: ReadonlyArray): Effect.Effect => + Effect.gen(function* () { + let tasks: ReadonlyArray = [] + const openRuns = new Map() + for (const entry of entries) { + if (Predicate.isTagged(entry, 'task_started')) { + const previous = tasks.find((task) => task.taskId === entry.agentId) + tasks = upsertTask(tasks, taskFromStarted(entry, previous)) + openRuns.set(entry.agentId, entry) + } else if (Predicate.isTagged(entry, 'task_finished')) { + tasks = replaceTask(tasks, entry.agentId, (task) => applyFinished(task, entry)) + openRuns.delete(entry.agentId) + } + } + + // A run the previous process never finished: record it stopped, and tell the launcher on + // the root's next run. + const orphaned: Array = [] + for (const started of openRuns.values()) { + const progress = progressForRun(entries, started.agentId, started.seq) + const finished = yield* eventLog + .append( + LogEntryInputs.task_finished({ + agentId: started.agentId, + parentAgentId: started.parentAgentId, + toolCallId: started.toolCallId, + status: 'stopped', + summary: progress.text, + error: 'The session ended before this background agent finished.', + usage: progress.usage, + durationMs: 0, + lastToolName: progress.lastToolName, + }), + ) + .pipe(Effect.orDie) + if (!Predicate.isTagged(finished, 'task_finished')) continue + tasks = replaceTask(tasks, started.agentId, (task) => applyFinished(task, finished)) + const task = tasks.find((candidate) => candidate.taskId === started.agentId) + if (task !== undefined) { + orphaned.push({ + taskId: task.taskId, + parentAgentId: task.parentAgentId, + text: renderTaskNotification(task), + deferred: true, + }) + } + } + + yield* updateState((current) => ({ ...current, tasks, pending: [...current.pending, ...orphaned] })) + }) + + /** Hand notifications to a running agent as one follow-up; requeue them if its run ends first. */ + const deliverAsFollowUp = ( + agentId: AgentId, + notifications: ReadonlyArray, + requeueTo: AgentId, + ): Effect.Effect => + Effect.gen(function* () { + const message = notificationMessage(notifications) + const ticket = yield* controls + .pushFollowUp(agentId, message.text, message.origin ?? undefined) + .pipe(Effect.catchTag('AgentNotRunningError', () => Effect.succeed(null))) + if (ticket === null) { + yield* updateState((current) => ({ + ...current, + pending: [ + ...notifications.map((notification) => ({ ...notification, parentAgentId: requeueTo })), + ...current.pending, + ], + })) + return yield* kick + } + yield* markNotified(notifications) + yield* updateState((current) => ({ ...current, inFlight: current.inFlight + 1 })) + yield* ticket.consumed.pipe( + Effect.flatMap((consumed) => + updateState((current) => ({ + ...current, + inFlight: current.inFlight - 1, + pending: consumed + ? current.pending + : [ + ...notifications.map((notification) => ({ + ...notification, + parentAgentId: requeueTo, + })), + ...current.pending, + ], + })).pipe(Effect.andThen(consumed ? Effect.void : kick)), + ), + Effect.forkIn(sessionScope), + ) + }) + + const markNotified = (notifications: ReadonlyArray): Effect.Effect => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => + updateState((current) => ({ + ...current, + tasks: current.tasks.map((task) => + notifications.some((notification) => notification.taskId === task.taskId) + ? { ...task, notifiedAt: now } + : task, + ), + })), + ), + ) + + /** Take the pending notifications matching a predicate, leaving the rest queued. */ + const takePending = ( + predicate: (notification: TaskNotification) => boolean, + ): Effect.Effect> => + SubscriptionRef.modify(state, (current) => { + const [taken, kept] = [ + current.pending.filter(predicate), + current.pending.filter((notification) => !predicate(notification)), + ] + return [taken, { ...current, pending: kept }] as const + }) + + const deliverPending = (rootAgentId: AgentId, wake: RootWake): Effect.Effect => + Effect.gen(function* () { + const current = yield* SubscriptionRef.get(state) + if (Arr.isReadonlyArrayEmpty(current.pending)) return + + // A launcher that is still running hears about its own tasks directly. + const running = new Set(yield* controls.runningAgentIds) + const launchers = [ + ...new Set( + current.pending + .filter( + (notification) => + !notification.deferred && + notification.parentAgentId !== rootAgentId && + running.has(notification.parentAgentId), + ) + .map((notification) => notification.parentAgentId), + ), + ] + for (const launcher of launchers) { + const taken = yield* takePending( + (notification) => !notification.deferred && notification.parentAgentId === launcher, + ) + if (Arr.isReadonlyArrayNonEmpty(taken)) yield* deliverAsFollowUp(launcher, taken, rootAgentId) + } + + // Everything else goes to the root. + if (running.has(rootAgentId)) { + const deferred = yield* takePending((notification) => notification.deferred) + if (Arr.isReadonlyArrayNonEmpty(deferred)) { + const message = notificationMessage(deferred) + const steered = yield* controls + .steer(rootAgentId, message.text, message.origin ?? undefined) + .pipe( + Effect.as(true), + Effect.catchTag('AgentNotRunningError', () => Effect.succeed(false)), + ) + if (steered) yield* markNotified(deferred) + else yield* updateState((state) => ({ ...state, pending: [...deferred, ...state.pending] })) + } + const live = yield* takePending(() => true) + if (Arr.isReadonlyArrayNonEmpty(live)) yield* deliverAsFollowUp(rootAgentId, live, rootAgentId) + return + } + + const after = yield* SubscriptionRef.get(state) + if (after.waking || !after.pending.some((notification) => !notification.deferred)) return + + yield* updateState((state) => ({ ...state, waking: true })) + const take: Effect.Effect = takePending(() => true).pipe( + Effect.tap(markNotified), + Effect.map((taken) => (Arr.isReadonlyArrayEmpty(taken) ? null : notificationMessage(taken))), + ) + yield* wake(take).pipe( + Effect.ensuring( + updateState((state) => ({ ...state, waking: false })).pipe(Effect.andThen(kick)), + ), + Effect.forkIn(sessionScope), + ) + }) + + const serveNotifications = (options: { + readonly rootAgentId: AgentId + readonly wake: RootWake + }): Effect.Effect => + Queue.take(kicks).pipe( + Effect.andThen(deliverPending(options.rootAgentId, options.wake)), + Effect.forever, + ) + + const isIdle = (current: State): boolean => + !current.waking && + current.inFlight === 0 && + current.tasks.every((task) => task.status !== 'running') && + current.pending.every((notification) => notification.deferred) + + const awaitIdle: Effect.Effect = SubscriptionRef.changes(state).pipe( + Stream.filter(isIdle), + Stream.take(1), + Stream.runDrain, + ) + + return { + launch, + list: SubscriptionRef.get(state).pipe(Effect.map((current) => current.tasks)), + changes: SubscriptionRef.changes(state).pipe( + Stream.map((current) => current.tasks), + Stream.changes, + ), + get: getTask, + stop, + stopAll, + awaitSettled, + hydrate, + serveNotifications, + nudge: kick, + awaitIdle, + } + }), + ) diff --git a/packages/fold-core/src/Subagents/Schemas.ts b/packages/fold-core/src/Subagents/Schemas.ts index 1cee448..fd518a1 100644 --- a/packages/fold-core/src/Subagents/Schemas.ts +++ b/packages/fold-core/src/Subagents/Schemas.ts @@ -49,14 +49,19 @@ export const DispatchSubagentCommand = Schema.TaggedStruct('dispatch', { }), prompt: Schema.String, skill: Schema.NullOr(Schema.String), + /** Launch as a background task: the call returns once the agent has started. */ + background: Schema.Boolean, }).annotate({ identifier: 'DispatchSubagentCommand' }) export type DispatchSubagentCommand = typeof DispatchSubagentCommand.Type /** Resume a previously dispatched subagent by reference: its full id or a unique short prefix. */ export const ResumeSubagentCommand = Schema.TaggedStruct('resume', { agentId: AgentIdRef, + description: Schema.optionalKey(Schema.String), prompt: Schema.String, skill: Schema.NullOr(Schema.String), + /** Resume as a background task: the call returns once the agent has restarted. */ + background: Schema.Boolean, }).annotate({ identifier: 'ResumeSubagentCommand' }) export type ResumeSubagentCommand = typeof ResumeSubagentCommand.Type @@ -67,6 +72,8 @@ export const ForkSubagentCommand = Schema.TaggedStruct('fork', { }), prompt: Schema.String, skill: Schema.NullOr(Schema.String), + /** Launch the fork as a background task. */ + background: Schema.Boolean, }).annotate({ identifier: 'ForkSubagentCommand' }) export type ForkSubagentCommand = typeof ForkSubagentCommand.Type @@ -99,6 +106,7 @@ export type SubagentToolWireParameters = { readonly agent?: string readonly agent_id?: string readonly fork?: boolean + readonly run_in_background?: boolean } const nonEmptyOptionalString = (value: string | undefined): string | undefined => @@ -133,33 +141,15 @@ export const parseSubagentCommand = ( }) } + // Optional keys stay absent (not undefined) on the parsed command. + const label = description === undefined ? {} : { description } + const background = params.run_in_background === true + if (agent !== undefined) { - if (description === undefined) { - return DispatchSubagentCommand.make({ - agent, - prompt: params.prompt, - skill, - }) - } - return DispatchSubagentCommand.make({ - agent, - description, - prompt: params.prompt, - skill, - }) + return DispatchSubagentCommand.make({ agent, ...label, prompt: params.prompt, skill, background }) } if (params.fork === true) { - if (description === undefined) { - return ForkSubagentCommand.make({ - prompt: params.prompt, - skill, - }) - } - return ForkSubagentCommand.make({ - description, - prompt: params.prompt, - skill, - }) + return ForkSubagentCommand.make({ ...label, prompt: params.prompt, skill, background }) } const agentId = yield* decodeAgentIdRef(agentIdRef).pipe( @@ -173,5 +163,5 @@ export const parseSubagentCommand = ( ), ) - return ResumeSubagentCommand.make({ agentId, prompt: params.prompt, skill }) + return ResumeSubagentCommand.make({ agentId, ...label, prompt: params.prompt, skill, background }) }) diff --git a/packages/fold-core/src/Subagents/SubagentEngine.ts b/packages/fold-core/src/Subagents/SubagentEngine.ts index a47b016..2e649af 100644 --- a/packages/fold-core/src/Subagents/SubagentEngine.ts +++ b/packages/fold-core/src/Subagents/SubagentEngine.ts @@ -55,6 +55,7 @@ import { } from '../ToolRuntime/ToolContextServices' import { agentIdsFromEntries, resolveAgentIdRef, shortAgentId } from './AgentIdRef' import type { AgentRegistry, RegisteredAgentType } from './AgentRegistry' +import { BackgroundTasks, type BackgroundTask } from './BackgroundTasks' import { SubagentBusyError, SubagentNotFoundError, SubagentTypeNotInRosterError } from './Errors' import type { ForkSubagentInput, SubagentResult } from './Schemas' import { SessionAgents, type RealizedAgentTools } from './SessionAgents' @@ -71,6 +72,8 @@ export type DispatchSubagentInput = { readonly skill: string | null /** The executing subagentTool value's roster (from its closure) - the dispatch authority (§1a). */ readonly allowedAgents: ReadonlyArray + /** Short label recorded on a background task; defaults to the agent type name. */ + readonly description?: string | undefined } /** @@ -82,6 +85,14 @@ export type ResumeSubagentInput = { readonly agentId: AgentId | string readonly prompt: string readonly skill: string | null + /** Short label recorded on a background task; defaults to the task's previous label. */ + readonly description?: string | undefined +} + +/** Input for forking the executing agent into the background (fork input plus a task label). */ +export type ForkSubagentInBackgroundInput = ForkSubagentInput & { + /** Short label recorded on the background task; defaults to "fork of ". */ + readonly description?: string | undefined } /** Input for continuing a finished subagent directly from the SDK (D8 `send(message, { agentId })`). */ @@ -468,57 +479,69 @@ const watchSubagentTurns = Effect.fnUntraced(function* (params: LaunchSubagentPa }) /** - * Run one subagent launch/resume to its result: claim the id, provision this agent's runtime into the - * dispatch scope, write the start (or resume transition), fork the run fiber, await its Exit, and fold. - * The dispatch scope closing (dispatcher interrupted) interrupts the run fiber; the onExit finalizer - * then writes the durable markers before the interruption propagates. + * Prepare one subagent launch/resume inside the ambient scope: claim the id (Busy when it is already + * running), provision this agent's runtime into the scope, and write the start (or resume transition). + * Returns the run effect, carrying the uninterruptible exit markers. Foreground calls fork it into the + * dispatch scope; background launches fork it into the task's own scope. + */ +const prepareSubagentRun = (params: LaunchSubagentParams) => + Effect.gen(function* () { + yield* holdRunningSubagent(params.subagentId) + + const agentRuntimeForSubagent = yield* provisionAgentRuntime({ + model: params.model, + tools: params.tools, + hooks: params.hooks, + }) + + if (SubagentLaunch.$is('start')(params.launch)) { + yield* agentRuntimeForSubagent.start({ + agentId: params.subagentId, + parentAgentId: params.parentAgentId, + toolCallId: params.toolCallId, + mode: params.mode, + fork: params.fork, + skill: params.skillParam, + agentType: params.agentTypeName, + model: params.model.activeModel, + promptCacheKey: params.promptCacheKey, + systemPrompt: params.systemPrompt, + }) + } else if (params.launch.modelTransition !== null) { + yield* agentRuntimeForSubagent.switchModel({ + agentId: params.subagentId, + parentAgentId: params.parentAgentId, + toolCallId: params.toolCallId, + model: params.model.activeModel, + systemPrompt: params.launch.modelTransition.systemPrompt, + reason: 'resume: the configured model for this agent changed since it last ran', + }) + } + + const markerServices = yield* Effect.context() + const run: Effect.Effect = agentRuntimeForSubagent + .run({ + agentId: params.subagentId, + parentAgentId: params.parentAgentId, + toolCallId: params.toolCallId, + messages: params.messages, + }) + .pipe(Effect.onExit(writeSubagentExitMarkers(params)), Effect.provideContext(markerServices)) + return { run } + }) + +/** + * Run one subagent launch/resume to its result: prepare it in the dispatch scope, fork the run fiber, + * await its Exit, and fold. The dispatch scope closing (dispatcher interrupted) interrupts the run + * fiber; the onExit finalizer then writes the durable markers before the interruption propagates. */ const runSubagentToResult = (params: LaunchSubagentParams) => Effect.scoped( Effect.gen(function* () { const controls = yield* SessionControls - yield* holdRunningSubagent(params.subagentId) - - const agentRuntimeForSubagent = yield* provisionAgentRuntime({ - model: params.model, - tools: params.tools, - hooks: params.hooks, - }) + const { run } = yield* prepareSubagentRun(params) - if (SubagentLaunch.$is('start')(params.launch)) { - yield* agentRuntimeForSubagent.start({ - agentId: params.subagentId, - parentAgentId: params.parentAgentId, - toolCallId: params.toolCallId, - mode: params.mode, - fork: params.fork, - skill: params.skillParam, - agentType: params.agentTypeName, - model: params.model.activeModel, - promptCacheKey: params.promptCacheKey, - systemPrompt: params.systemPrompt, - }) - } else if (params.launch.modelTransition !== null) { - yield* agentRuntimeForSubagent.switchModel({ - agentId: params.subagentId, - parentAgentId: params.parentAgentId, - toolCallId: params.toolCallId, - model: params.model.activeModel, - systemPrompt: params.launch.modelTransition.systemPrompt, - reason: 'resume: the configured model for this agent changed since it last ran', - }) - } - - const subagentRunFiber = yield* Effect.forkScoped( - agentRuntimeForSubagent - .run({ - agentId: params.subagentId, - parentAgentId: params.parentAgentId, - toolCallId: params.toolCallId, - messages: params.messages, - }) - .pipe(Effect.onExit(writeSubagentExitMarkers(params))), - ) + const subagentRunFiber = yield* Effect.forkScoped(run) // Closing the dispatch scope interrupts the subagent and awaits its exit finalizers, so the // durable interrupt/error markers land before the dispatch call returns. yield* Effect.addFinalizer(() => Fiber.interrupt(subagentRunFiber)) @@ -532,6 +555,28 @@ const runSubagentToResult = (params: LaunchSubagentParams) => }), ) +/** + * Launch one subagent launch/resume as a background task: the run lives in a task scope owned by the + * session, so the dispatching tool call returns as soon as the agent has started. + */ +const launchSubagentInBackground = ( + params: LaunchSubagentParams, + task: { readonly description: string; readonly prompt: string }, +) => + BackgroundTasks.use((tasks) => + tasks.launch( + { + taskId: params.subagentId, + parentAgentId: params.parentAgentId, + toolCallId: params.toolCallId, + agentType: params.agentTypeName, + description: task.description, + prompt: task.prompt, + }, + prepareSubagentRun(params), + ), + ) + /** A fresh id can never be running; a Busy claim failure on one is an engine bug. */ const dieOnBusy = (error: SubagentBusyError): Effect.Effect => Effect.die(new Error(`freshly minted subagent id ${error.agentId} was already running`)) @@ -654,8 +699,8 @@ export const continueSubagent = Effect.fn('fold.subagents.continue')(function* ( ) }) -/** Dispatch one fresh subagent of a registered type from inside the executing subagent tool call. */ -export const dispatchSubagent = Effect.fn('fold.subagents.dispatch')(function* (input: DispatchSubagentInput) { +/** Resolve one fresh dispatch of a registered type into its launch parameters. */ +const resolveDispatch = Effect.fnUntraced(function* (input: DispatchSubagentInput) { if (!input.allowedAgents.includes(input.agent)) { return yield* new SubagentTypeNotInRosterError({ requested: input.agent, @@ -694,7 +739,7 @@ export const dispatchSubagent = Effect.fn('fold.subagents.dispatch')(function* ( const realized = agents.realizeTools(entry.tools) - return yield* runSubagentToResult({ + const params: LaunchSubagentParams = { subagentId, agentTypeName: entry.name, agentLabel: entry.name, @@ -712,11 +757,29 @@ export const dispatchSubagent = Effect.fn('fold.subagents.dispatch')(function* ( messages: preloaded === null ? [input.prompt] : [input.prompt, preloaded], launch: SubagentLaunch.start(), interruptNote, + } + return params +}) + +/** Dispatch one fresh subagent of a registered type from inside the executing subagent tool call. */ +export const dispatchSubagent = Effect.fn('fold.subagents.dispatch')(function* (input: DispatchSubagentInput) { + const params = yield* resolveDispatch(input) + return yield* runSubagentToResult(params).pipe(Effect.catchTag('SubagentBusyError', dieOnBusy)) +}) + +/** Dispatch one fresh subagent as a background task; returns once it has started. */ +export const dispatchSubagentInBackground = Effect.fn('fold.subagents.dispatch_background')(function* ( + input: DispatchSubagentInput, +) { + const params = yield* resolveDispatch(input) + return yield* launchSubagentInBackground(params, { + description: input.description ?? input.agent, + prompt: input.prompt, }).pipe(Effect.catchTag('SubagentBusyError', dieOnBusy)) }) -/** Fork the executing agent: a new subagent over a copy of the caller's context. */ -export const forkSubagent = Effect.fn('fold.subagents.fork')(function* (input: ForkSubagentInput) { +/** Resolve one fork of the executing agent into its launch parameters. */ +const resolveFork = Effect.fnUntraced(function* (input: ForkSubagentInput) { const agents = yield* SessionAgents const ids = yield* Ids const dispatcher = yield* CurrentAgent @@ -759,7 +822,7 @@ export const forkSubagent = Effect.fn('fold.subagents.fork')(function* (input: F const fork = input.forkAgentDefinitionId === null ? forkPoint : { ...forkPoint, definitionId: input.forkAgentDefinitionId } - return yield* runSubagentToResult({ + const params: LaunchSubagentParams = { subagentId, agentTypeName: null, agentLabel, @@ -777,11 +840,29 @@ export const forkSubagent = Effect.fn('fold.subagents.fork')(function* (input: F messages: preloaded === null ? [input.prompt] : [input.prompt, preloaded], launch: SubagentLaunch.start(), interruptNote, + } + return params +}) + +/** Fork the executing agent: a new subagent over a copy of the caller's context. */ +export const forkSubagent = Effect.fn('fold.subagents.fork')(function* (input: ForkSubagentInput) { + const params = yield* resolveFork(input) + return yield* runSubagentToResult(params).pipe(Effect.catchTag('SubagentBusyError', dieOnBusy)) +}) + +/** Fork the executing agent into a background task; returns once the fork has started. */ +export const forkSubagentInBackground = Effect.fn('fold.subagents.fork_background')(function* ( + input: ForkSubagentInBackgroundInput, +) { + const params = yield* resolveFork(input) + return yield* launchSubagentInBackground(params, { + description: input.description ?? params.agentLabel, + prompt: input.prompt, }).pipe(Effect.catchTag('SubagentBusyError', dieOnBusy)) }) -/** Resume a previously launched subagent by reference, from inside the executing subagent tool call. */ -export const resumeSubagent = Effect.fn('fold.subagents.resume')(function* (input: ResumeSubagentInput) { +/** Resolve one resume by reference into its launch parameters. */ +const resolveResume = Effect.fnUntraced(function* (input: ResumeSubagentInput) { const agents = yield* SessionAgents const dispatcher = yield* CurrentAgent const currentCall = yield* CurrentToolCall @@ -831,7 +912,7 @@ export const resumeSubagent = Effect.fn('fold.subagents.resume')(function* (inpu const agentLabel = started.agentType ?? `subagent ${shortAgentId(agentId)}` yield* interruptNote.set(interruptedSubagentNote(agentLabel, agentId, 0)) - return yield* runSubagentToResult({ + const params: LaunchSubagentParams = { subagentId: agentId, agentTypeName: started.agentType, agentLabel, @@ -848,5 +929,29 @@ export const resumeSubagent = Effect.fn('fold.subagents.resume')(function* (inpu messages: preloaded === null ? [input.prompt] : [input.prompt, preloaded], launch: SubagentLaunch.resume({ modelTransition }), interruptNote, + } + return params +}) + +/** Resume a previously launched subagent by reference, from inside the executing subagent tool call. */ +export const resumeSubagent = Effect.fn('fold.subagents.resume')(function* (input: ResumeSubagentInput) { + const params = yield* resolveResume(input) + return yield* runSubagentToResult(params) +}) + +/** + * Resume a previously launched subagent by reference as a background task; returns once it has + * started. Busy when the agent is running (including as a background task). + */ +export const resumeSubagentInBackground = Effect.fn('fold.subagents.resume_background')(function* ( + input: ResumeSubagentInput, +) { + const params = yield* resolveResume(input) + const previous = yield* BackgroundTasks.use((tasks) => tasks.get(params.subagentId)) + return yield* launchSubagentInBackground(params, { + description: input.description ?? previous?.description ?? params.agentLabel, + prompt: input.prompt, }) }) + +export type { BackgroundTask } diff --git a/packages/fold-core/src/Subagents/SubagentTool.ts b/packages/fold-core/src/Subagents/SubagentTool.ts index 417f7cb..5b38cbc 100644 --- a/packages/fold-core/src/Subagents/SubagentTool.ts +++ b/packages/fold-core/src/Subagents/SubagentTool.ts @@ -16,11 +16,19 @@ import type { SkillNotFoundError } from '../Skills/SkillSource' import { subagentToolContract } from '../Tools/Contracts' import { ToolResultFailure, ToolResultText } from '../Tools/ToolResultContent' import { shortAgentId } from './AgentIdRef' +import type { BackgroundTask } from './BackgroundTasks' import type { SubagentBusyError, SubagentNotFoundError, SubagentTypeNotInRosterError } from './Errors' import type { ForkAgentDefinition } from './ForkAgentDefinition' import { parseSubagentCommand, type SubagentResult } from './Schemas' import { subagentsNeedingAll, type SubagentDefinition, type SubagentDefinitionServices } from './SubagentDefinition' -import { dispatchSubagent, forkSubagent, resumeSubagent } from './SubagentEngine' +import { + dispatchSubagent, + dispatchSubagentInBackground, + forkSubagent, + forkSubagentInBackground, + resumeSubagent, + resumeSubagentInBackground, +} from './SubagentEngine' /** Runtime capabilities attached to a model-visible delegation tool. */ export type SubagentToolCapabilities = { @@ -56,6 +64,16 @@ export const renderSubagentResult = (result: SubagentResult): string => { return note === null ? `${header}\n\n${body}` : `${header}\n\n${body}\n${note}` } +/** + * Render the receipt for a background launch: the agent id and how its result will arrive. The run + * keeps going after this tool call returns; its notification arrives as a later user message. + */ +export const renderBackgroundLaunch = (task: BackgroundTask): string => + `Background agent ${task.runCount > 1 ? 'resumed' : 'started'}. agent_id: ${shortAgentId(task.taskId)}\n` + + `It is working in the background. You will be notified with a message when it ` + + `finishes, so do not poll for it. You know nothing about its results until then: do not report, assume, ` + + `or predict them. If the user asks before it arrives, say it is still running.` + /** The system-information note appended for non-completed outcomes; null for clean completions. */ const outcomeNoteFor = (result: SubagentResult): string | null => Match.value(result.outcome).pipe( @@ -129,7 +147,8 @@ const rosterDescriptionSuffix = (agents: ReadonlyArray, RF = never>( ), ) + const renderResult = (result: SubagentResult) => + ToolResultText.make({ text: renderSubagentResult(result) }) + const renderLaunch = (task: BackgroundTask) => + ToolResultText.make({ text: renderBackgroundLaunch(task) }) + return yield* Match.value(command).pipe( Match.tagsExhaustive({ - dispatch: (dispatchCommand) => - dispatchSubagent({ + dispatch: (dispatchCommand) => { + const input = { agent: dispatchCommand.agent, prompt: dispatchCommand.prompt, skill: dispatchCommand.skill, allowedAgents, - }).pipe( - Effect.catchTags({ - SubagentTypeNotInRosterError: (error) => Effect.fail(rosterFailure(error)), - SkillNotFoundError: (error) => Effect.fail(skillFailure(error, allowedAgents)), - }), - Effect.map((result) => ToolResultText.make({ text: renderSubagentResult(result) })), - ), - fork: (forkCommand) => - forkSubagent({ + description: dispatchCommand.description, + } + const failures = { + SubagentTypeNotInRosterError: (error: SubagentTypeNotInRosterError) => + Effect.fail(rosterFailure(error)), + SkillNotFoundError: (error: SkillNotFoundError) => + Effect.fail(skillFailure(error, allowedAgents)), + } + return dispatchCommand.background + ? dispatchSubagentInBackground(input).pipe( + Effect.catchTags(failures), + Effect.map(renderLaunch), + ) + : dispatchSubagent(input).pipe(Effect.catchTags(failures), Effect.map(renderResult)) + }, + fork: (forkCommand) => { + const input = { prompt: forkCommand.prompt, skill: forkCommand.skill, forkAgentDefinitionId: options?.forkAgent?.id ?? null, - history: 'all', - }).pipe( - Effect.catchTag('SkillNotFoundError', (error) => + history: 'all' as const, + } + const failures = { + SkillNotFoundError: (error: SkillNotFoundError) => Effect.fail(skillFailure(error, allowedAgents)), - ), - Effect.map((result) => ToolResultText.make({ text: renderSubagentResult(result) })), - ), - resume: (resumeCommand) => - resumeSubagent({ + } + return forkCommand.background + ? forkSubagentInBackground({ ...input, description: forkCommand.description }).pipe( + Effect.catchTags(failures), + Effect.map(renderLaunch), + ) + : forkSubagent(input).pipe(Effect.catchTags(failures), Effect.map(renderResult)) + }, + resume: (resumeCommand) => { + const input = { agentId: resumeCommand.agentId, prompt: resumeCommand.prompt, skill: resumeCommand.skill, - }).pipe( - Effect.catchTags({ - SubagentNotFoundError: (error) => - Effect.fail(notFoundFailure(error, allowedAgents)), - SubagentBusyError: (error) => Effect.fail(busyFailure(error, allowedAgents)), - SkillNotFoundError: (error) => Effect.fail(skillFailure(error, allowedAgents)), - }), - Effect.map((result) => ToolResultText.make({ text: renderSubagentResult(result) })), - ), + description: resumeCommand.description, + } + const failures = { + SubagentNotFoundError: (error: SubagentNotFoundError) => + Effect.fail(notFoundFailure(error, allowedAgents)), + SubagentBusyError: (error: SubagentBusyError) => + Effect.fail(busyFailure(error, allowedAgents)), + SkillNotFoundError: (error: SkillNotFoundError) => + Effect.fail(skillFailure(error, allowedAgents)), + } + return resumeCommand.background + ? resumeSubagentInBackground(input).pipe( + Effect.catchTags(failures), + Effect.map(renderLaunch), + ) + : resumeSubagent(input).pipe(Effect.catchTags(failures), Effect.map(renderResult)) + }, }), ) }), diff --git a/packages/fold-core/src/Tools/Contracts.ts b/packages/fold-core/src/Tools/Contracts.ts index c48e9d8..47e03cb 100644 --- a/packages/fold-core/src/Tools/Contracts.ts +++ b/packages/fold-core/src/Tools/Contracts.ts @@ -273,6 +273,14 @@ const SubagentParameters = Schema.Struct({ skill: Schema.optionalKey(Schema.String).annotate({ description: 'Optional skill name (from your available skills) to preload into the subagent after the prompt.', }), + run_in_background: Schema.optionalKey(Schema.Boolean).annotate({ + description: + 'Run the subagent in the background. The call returns at once with its agent_id, and you are ' + + 'notified with a message when it finishes - do not poll for it, and do not ' + + 'guess its results before then. Use it when other work can usefully proceed meanwhile (for ' + + 'example several independent investigations launched together); leave it unset when your very ' + + 'next step depends on the result. Works with agent, agent_id, and fork.', + }), }) const SubagentSuccess = ToolResultText @@ -294,6 +302,62 @@ export const subagentToolContract = { failure: SubagentFailure, } satisfies ToolContract +// --- background agent controls ----------------------------------------------------------------------- + +const ListAgentsParameters = Schema.Struct({}) + +/** Contract for list_agents: the session's background agents with their status and latest result. */ +export const listAgentsToolContract = { + name: 'list_agents', + description: + "List the background agents in this session - running and finished - with each one's agent_id, " + + 'status, description, progress, and the start of its result. Use it when you need to know what is ' + + 'still running; you do not need it to learn about completions, which arrive as ' + + 'messages on their own.', + parameters: ListAgentsParameters, + success: ToolResultText, + failure: ToolResultFailure, +} satisfies ToolContract + +const StopAgentParameters = Schema.Struct({ + agent_id: Schema.String.annotate({ + description: 'The agent_id of the running background agent to stop (from its launch receipt or list_agents).', + }), +}) + +/** Contract for stop_agent: stop one running background agent; its context is kept. */ +export const stopAgentToolContract = { + name: 'stop_agent', + description: + 'Stop a running background agent. Its work so far is kept, and send_message can continue it later. ' + + 'Use it when the agent is no longer needed or is going in the wrong direction.', + parameters: StopAgentParameters, + success: ToolResultText, + failure: ToolResultFailure, +} satisfies ToolContract + +const SendMessageParameters = Schema.Struct({ + agent_id: Schema.String.annotate({ + description: 'The agent_id of the subagent to message (from its launch receipt, result, or list_agents).', + }), + message: Schema.String.annotate({ + description: 'The message the agent sees next.', + }), +}) + +/** Contract for send_message: steer a running subagent, or continue a finished one in the background. */ +export const sendMessageToolContract = { + name: 'send_message', + description: + 'Send a message to a subagent by agent_id. If it is running, it reads the message after its current ' + + 'step. If it has finished (or was stopped), it continues in the background with its full context ' + + 'and the message as its next input, and you are notified with a when it ' + + 'finishes again.', + parameters: SendMessageParameters, + success: ToolResultText, + failure: ToolResultFailure, +} satisfies ToolContract + /** * The isomorphic built-in toolkit: every core tool contract, without handlers (D18). Platform packages * bind handlers with `defineTool({ ...contract, handler })`; hosts on other substrates (browser, @@ -308,4 +372,7 @@ export const builtinToolContracts = { web_search: webSearchToolContract, skill: skillToolContract, subagent: subagentToolContract, + list_agents: listAgentsToolContract, + stop_agent: stopAgentToolContract, + send_message: sendMessageToolContract, } as const diff --git a/packages/fold-core/src/index.ts b/packages/fold-core/src/index.ts index 7595ff7..85f10cc 100644 --- a/packages/fold-core/src/index.ts +++ b/packages/fold-core/src/index.ts @@ -13,8 +13,10 @@ export { type InjectedSkillEntries, } from './Api/OpenSession' export * from './Api/ToolDefinition' +export * from './Subagents/AgentControlTools' export * from './Subagents/AgentIdRef' export * from './Subagents/AgentRegistry' +export * from './Subagents/BackgroundTasks' export * from './Subagents/Errors' export * from './Subagents/ForkAgentDefinition' export * from './Subagents/Schemas' diff --git a/packages/fold-core/test/Subagents/BackgroundHarness.ts b/packages/fold-core/test/Subagents/BackgroundHarness.ts new file mode 100644 index 0000000..bf5fe54 --- /dev/null +++ b/packages/fold-core/test/Subagents/BackgroundHarness.ts @@ -0,0 +1,118 @@ +/** + * Harness for background-subagent tests: a scripted model whose turns can each wait on a gate, so a + * test decides exactly when a root or background agent finishes a turn; plus small readers over the + * session log. Turns that never get their gate released hang forever - the interruption target. + */ +import { Deferred, Effect, Predicate, Ref, Stream } from 'effect' +import { AiError, LanguageModel } from 'effect/ai' + +import { + customModel, + encodedContentText, + type ActiveModel, + type BackgroundTask, + type FoldModel, + type FoldSession, + type LogEntry, + type TaskFinishedLogEntry, + type TaskStartedLogEntry, + type UserMessageLogEntry, +} from '../../src/index' +import type { ScriptedTurn } from '../TestLayers/ScriptedLanguageModel' + +/** One scripted turn held until `gate` opens; `started` is signaled when it is requested. */ +export type GatedTurn = { + readonly _tag: 'gated' + readonly turn: ScriptedTurn + readonly gate: Deferred.Deferred | null + readonly started: Deferred.Deferred | null +} + +/** Hold a scripted turn until `gate` opens (never, if it never does); signal `started` on request. */ +export const gatedTurn = ( + turn: ScriptedTurn, + options: { readonly gate?: Deferred.Deferred; readonly started?: Deferred.Deferred }, +): GatedTurn => ({ _tag: 'gated', turn, gate: options.gate ?? null, started: options.started ?? null }) + +export type GatedModel = { + readonly model: FoldModel + /** Append turns between sends (for params only known at runtime, like agent ids). */ + readonly push: (turns: ReadonlyArray) => Effect.Effect + readonly requests: Effect.Effect +} + +/** A model serving its turns in order across every runtime provisioned from it; gated turns wait. */ +export const makeGatedModel = ( + activeModel: ActiveModel, + turns: ReadonlyArray, +): Effect.Effect => + Effect.gen(function* () { + const normalize = (input: ReadonlyArray): ReadonlyArray => + input.map((turn) => (Predicate.isTagged(turn, 'gated') ? turn : gatedTurn(turn, {}))) + const turnsRef = yield* Ref.make(normalize(turns)) + const requests = yield* Ref.make(0) + + const make = LanguageModel.make({ + generateText: () => Effect.die(new Error('gated model supports streamText only')), + streamText: () => + Stream.unwrap( + Effect.gen(function* () { + yield* Ref.update(requests, (count) => count + 1) + const remaining = yield* Ref.get(turnsRef) + const next = remaining[0] + if (next === undefined) return yield* Effect.die(new Error('gated model: script exhausted')) + yield* Ref.set(turnsRef, remaining.slice(1)) + if (next.started !== null) yield* Deferred.succeed(next.started, undefined) + if (next.gate !== null) yield* Deferred.await(next.gate) + return Predicate.isTagged(next.turn, 'failure') + ? Stream.fail( + AiError.make({ + module: 'GatedModel', + method: 'streamText', + reason: new AiError.UnknownError({ description: next.turn.message }), + }), + ) + : Stream.fromIterable(next.turn.parts) + }), + ), + }) + + return { + model: customModel({ activeModel, make }), + push: (extra) => Ref.update(turnsRef, (remaining) => [...remaining, ...normalize(extra)]), + requests: Ref.get(requests), + } + }) + +export const taskStartedRows = (entries: ReadonlyArray): ReadonlyArray => + entries.filter((entry): entry is TaskStartedLogEntry => Predicate.isTagged(entry, 'task_started')) + +export const taskFinishedRows = (entries: ReadonlyArray): ReadonlyArray => + entries.filter((entry): entry is TaskFinishedLogEntry => Predicate.isTagged(entry, 'task_finished')) + +/** User messages the runtime wrote to deliver task notifications. */ +export const notificationMessages = (entries: ReadonlyArray): ReadonlyArray => + entries.filter( + (entry): entry is UserMessageLogEntry => + Predicate.isTagged(entry, 'user-message') && entry.origin?._tag === 'task-notification', + ) + +export const messageText = (entry: UserMessageLogEntry): string => encodedContentText(entry.message.content) + +/** The root's terminal rows, in log order. */ +export const rootFinishedOutcomes = (entries: ReadonlyArray, rootAgentId: string) => + entries.flatMap((entry) => + Predicate.isTagged(entry, 'agent-finished') && entry.agentId === rootAgentId ? [entry] : [], + ) + +/** The text of every tool result in log order (JSON of its first content part). */ +export const toolResultTexts = (entries: ReadonlyArray): ReadonlyArray => + entries.flatMap((entry) => + Predicate.isTagged(entry, 'tool-result') ? [JSON.stringify(entry.message.content[0])] : [], + ) + +/** Wait until the session's task list satisfies a predicate. */ +export const awaitTasks = ( + session: FoldSession, + predicate: (tasks: ReadonlyArray) => boolean, +): Effect.Effect => session.backgroundTaskChanges.pipe(Stream.filter(predicate), Stream.take(1), Stream.runDrain) diff --git a/packages/fold-core/test/Subagents/BackgroundSubagents.vi.test.ts b/packages/fold-core/test/Subagents/BackgroundSubagents.vi.test.ts new file mode 100644 index 0000000..1034986 --- /dev/null +++ b/packages/fold-core/test/Subagents/BackgroundSubagents.vi.test.ts @@ -0,0 +1,505 @@ +/** + * Background subagents end to end through the public facade: the real `subagent` tool with + * `run_in_background`, the model-facing controls, and the session handle. Scripted models with gated + * turns decide when each agent finishes, so every ordering below is deterministic: the launch returns + * at once and the root's turn completes while the child runs; an idle root is woken by a synthetic + * task-notification turn; a busy root gets completions as one coalesced follow-up; stopTask, interrupt, + * stop, and session close each end background work the documented way. + */ +import * as NodeFileSystem from '@effect/platform-node/NodeFileSystem' +import { expect, it } from '@effect/vitest' +import { Deferred, Effect, Fiber, Layer, Predicate, Stream } from 'effect' + +import { + agentControlTools, + defineAgent, + defineSubagent, + EventLog, + layerInMemoryEventLogWithIds, + layerLiveIdFactory, + LogEntryInputs, + Session, + shortAgentId, + subagentTool, + type BackgroundTask, + type FoldModel, + type LogEntry, +} from '../../src/index' +import { claudeActiveModel, gptActiveModel } from '../Api/ApiTestHelpers' +import { textTurn, toolCallTurn } from '../TestLayers/ScriptedLanguageModel' +import { + awaitTasks, + gatedTurn, + makeGatedModel, + messageText, + notificationMessages, + rootFinishedOutcomes, + taskFinishedRows, + taskStartedRows, + toolResultTexts, +} from './BackgroundHarness' + +const backgroundCall = (id: string, description: string, prompt = 'dig into it') => + ({ + id, + name: 'subagent', + params: { agent: 'researcher', description, prompt, run_in_background: true }, + }) as const + +const openBackgroundSession = (rootModel: FoldModel, childModel: FoldModel) => { + const researcher = defineSubagent({ + name: 'researcher', + description: 'explores', + systemPrompt: 'You are a researcher.', + model: childModel, + }) + return Session.open({ + agent: defineAgent({ + model: rootModel, + systemPrompt: 'root', + tools: [subagentTool([researcher]), ...agentControlTools()], + }), + }) +} + +/** The outcome of an agent's latest agent-finished row, or null when it has none. */ +const lastOutcome = (entries: ReadonlyArray, agentId: string): string | null => + entries + .flatMap((entry) => + Predicate.isTagged(entry, 'agent-finished') && entry.agentId === agentId ? [entry.outcome] : [], + ) + .at(-1) ?? null + +const onlyTask = (tasks: ReadonlyArray): BackgroundTask => { + const task = tasks[0] + if (task === undefined || tasks.length !== 1) throw new Error(`expected exactly one task, got ${tasks.length}`) + return task +} + +it.effect('a background launch returns at once, and the idle root is woken by a task notification', () => + Effect.gen(function* () { + const release = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [ + gatedTurn(textTurn('found the config loader'), { gate: release }), + ]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([backgroundCall('provider-call-1', 'find the config loader')]), + textTurn('I started a researcher.'), + textTurn('The researcher found the config loader.'), + ]) + const session = yield* openBackgroundSession(root.model, child.model) + + // The root's turn completes while the child is still blocked on its first model call. + const finished = yield* session.send('where is the config loaded?') + expect(finished.outcome).toBe('completed') + expect(finished.resultText).toBe('I started a researcher.') + + const running = onlyTask(yield* session.backgroundTasks) + expect(running.status).toBe('running') + expect(running.description).toBe('find the config loader') + expect(running.agentType).toBe('researcher') + expect(running.prompt).toBe('dig into it') + expect(running.runCount).toBe(1) + + const launched = yield* session.entries + expect(toolResultTexts(launched)[0]).toContain( + `Background agent started. agent_id: ${shortAgentId(running.taskId)}`, + ) + const started = taskStartedRows(launched) + expect(started).toHaveLength(1) + expect(started[0]?.isBackgrounded).toBe(true) + expect(started[0]?.parentAgentId).toBe(session.rootAgentId) + // task_started lands after the child's own start rows and before its first message. + const childTags = launched.filter((entry) => entry.agentId === running.taskId).map((entry) => entry._tag) + expect(childTags).toEqual(['agent_started', 'system-message', 'task_started', 'user-message']) + + yield* Deferred.succeed(release, undefined) + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + const taskFinished = taskFinishedRows(entries) + expect(taskFinished).toHaveLength(1) + expect(taskFinished[0]?.status).toBe('completed') + expect(taskFinished[0]?.summary).toBe('found the config loader') + expect(taskFinished[0]?.usage).toEqual({ turns: 1, toolUses: 0, totalTokens: 15 }) + + // The wake: a synthetic root user message carrying the notification, then a root run on it. + const notifications = notificationMessages(entries) + expect(notifications).toHaveLength(1) + const notification = notifications[0] + if (notification === undefined) throw new Error('expected a notification') + expect(notification.agentId).toBe(session.rootAgentId) + expect(notification.origin).toEqual({ _tag: 'task-notification', taskIds: [running.taskId] }) + expect(messageText(notification)).toContain('') + expect(messageText(notification)).toContain('completed') + expect(messageText(notification)).toContain('found the config loader') + + const rootRuns = rootFinishedOutcomes(entries, session.rootAgentId) + expect(rootRuns.map((run) => run.resultText)).toEqual([ + 'I started a researcher.', + 'The researcher found the config loader.', + ]) + + const done = onlyTask(yield* session.backgroundTasks) + expect(done.status).toBe('completed') + expect(done.summary).toBe('found the config loader') + expect(done.notifiedAt).not.toBeNull() + expect(done.endedAt).not.toBeNull() + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +it.effect('completions during a busy root arrive as one coalesced follow-up, without a separate wake', () => + Effect.gen(function* () { + const rootGate = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [textTurn('first result'), textTurn('second result')]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([ + backgroundCall('provider-call-1', 'task one'), + backgroundCall('provider-call-2', 'task two'), + ]), + gatedTurn(textTurn('both launched'), { gate: rootGate }), + textTurn('handled both results'), + ]) + const session = yield* openBackgroundSession(root.model, child.model) + + const sendFiber = yield* Effect.forkScoped(session.send('do two things')) + // Both children finish (and are handed to the running root) while the root's turn is held. + yield* awaitTasks( + session, + (tasks) => + tasks.length === 2 && tasks.every((task) => task.status === 'completed' && task.notifiedAt !== null), + ) + yield* Deferred.succeed(rootGate, undefined) + + const finished = yield* Fiber.join(sendFiber) + expect(finished.resultText).toBe('handled both results') + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + const notifications = notificationMessages(entries) + expect(notifications).toHaveLength(1) + const taskIds = (yield* session.backgroundTasks).map((task) => task.taskId) + expect(notifications[0]?.origin?.taskIds).toHaveLength(2) + expect([...(notifications[0]?.origin?.taskIds ?? [])].sort()).toEqual([...taskIds].sort()) + const text = notifications[0] === undefined ? '' : messageText(notifications[0]) + expect(text.match(//g)).toHaveLength(2) + // One root run consumed everything: no extra wake. + expect(rootFinishedOutcomes(entries, session.rootAgentId)).toHaveLength(1) + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +it.effect('stopTask ends one background run as stopped and notifies the root', () => + Effect.gen(function* () { + const hang = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [gatedTurn(textTurn('never'), { gate: hang })]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([backgroundCall('provider-call-1', 'long job')]), + textTurn('launched'), + textTurn('noted that it was stopped'), + ]) + const session = yield* openBackgroundSession(root.model, child.model) + yield* session.send('start a long job') + const task = onlyTask(yield* session.backgroundTasks) + + expect(yield* session.stopTask(shortAgentId(task.taskId))).toBe(true) + expect(yield* session.stopTask(task.taskId)).toBe(false) + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + expect(taskFinishedRows(entries).map((row) => row.status)).toEqual(['stopped']) + expect(lastOutcome(entries, task.taskId)).toBe('interrupted') + const notifications = notificationMessages(entries) + expect(notifications).toHaveLength(1) + expect(notifications[0] === undefined ? '' : messageText(notifications[0])).toContain( + 'stopped', + ) + expect(rootFinishedOutcomes(entries, session.rootAgentId).at(-1)?.resultText).toBe('noted that it was stopped') + expect(onlyTask(yield* session.backgroundTasks).status).toBe('stopped') + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +it.effect('interrupt() stops only the root turn; background tasks keep running and still notify', () => + Effect.gen(function* () { + const childGate = yield* Deferred.make() + const rootHang = yield* Deferred.make() + const rootHangStarted = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [ + gatedTurn(textTurn('child result'), { gate: childGate }), + ]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([backgroundCall('provider-call-1', 'survivor')]), + gatedTurn(textTurn('never'), { gate: rootHang, started: rootHangStarted }), + textTurn('relayed after interrupt'), + ]) + const session = yield* openBackgroundSession(root.model, child.model) + + const sendFiber = yield* Effect.forkScoped(session.send('go')) + yield* Deferred.await(rootHangStarted) + yield* session.interrupt() + const interrupted = yield* Fiber.join(sendFiber) + expect(interrupted.outcome).toBe('interrupted') + expect(onlyTask(yield* session.backgroundTasks).status).toBe('running') + + yield* Deferred.succeed(childGate, undefined) + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + expect(taskFinishedRows(entries).map((row) => row.status)).toEqual(['completed']) + expect(rootFinishedOutcomes(entries, session.rootAgentId).map((run) => run.outcome)).toEqual([ + 'interrupted', + 'completed', + ]) + expect(notificationMessages(entries)).toHaveLength(1) + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +it.effect('stop() stops background tasks without waking the root; the next run hears about them first', () => + Effect.gen(function* () { + const hang = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [gatedTurn(textTurn('never'), { gate: hang })]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([backgroundCall('provider-call-1', 'stoppable')]), + textTurn('launched'), + textTurn('next reply'), + ]) + const session = yield* openBackgroundSession(root.model, child.model) + yield* session.send('go') + + yield* session.stop() + yield* session.awaitBackgroundIdle + const afterStop = yield* session.entries + expect(taskFinishedRows(afterStop).map((row) => row.status)).toEqual(['stopped']) + expect(notificationMessages(afterStop)).toHaveLength(0) + expect(rootFinishedOutcomes(afterStop, session.rootAgentId)).toHaveLength(1) + + // The held notification is steered into the next root run, ahead of its first model call. + const next = yield* session.send('what happened?') + expect(next.resultText).toBe('next reply') + const entries = yield* session.entries + const notifications = notificationMessages(entries) + expect(notifications).toHaveLength(1) + const userTexts = entries.flatMap((entry) => + Predicate.isTagged(entry, 'user-message') && entry.agentId === session.rootAgentId + ? [messageText(entry)] + : [], + ) + expect(userTexts.at(-2)).toBe('what happened?') + expect(userTexts.at(-1)).toContain('stopped') + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +it.effect('closing the session stops background tasks durably; reopening rebuilds the registry', () => + Effect.gen(function* () { + const hang = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [gatedTurn(textTurn('never'), { gate: hang })]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([backgroundCall('provider-call-1', 'outlived')]), + textTurn('launched'), + ]) + const eventLog = yield* EventLog + + const taskId = yield* Effect.scoped( + Effect.gen(function* () { + const session = yield* openBackgroundSession(root.model, child.model) + yield* session.send('go') + return onlyTask(yield* session.backgroundTasks).taskId + }), + ) + + const closed = yield* Stream.runCollect(eventLog.entries()).pipe(Effect.orDie) + expect(taskFinishedRows(closed).map((row) => [row.agentId, row.status])).toEqual([[taskId, 'stopped']]) + expect(lastOutcome(closed, taskId)).toBe('interrupted') + + // A run a crashed process left open: task_started with no task_finished. + const firstRun = taskStartedRows(closed)[0] + if (firstRun === undefined) throw new Error('expected the first task_started row') + yield* eventLog + .append( + LogEntryInputs.task_started({ + agentId: taskId, + parentAgentId: firstRun.parentAgentId, + toolCallId: firstRun.toolCallId, + taskType: 'subagent', + agentType: 'researcher', + description: 'orphaned run', + prompt: 'again', + isBackgrounded: true, + runCount: 2, + }), + ) + .pipe(Effect.orDie) + + const reopened = yield* Effect.scoped( + Effect.gen(function* () { + const session = yield* openBackgroundSession(root.model, child.model) + return yield* session.backgroundTasks + }), + ) + const task = onlyTask(reopened) + expect(task.taskId).toBe(taskId) + expect(task.status).toBe('stopped') + expect(task.runCount).toBe(2) + expect(task.error).toContain('session ended') + const after = yield* Stream.runCollect(eventLog.entries()).pipe(Effect.orDie) + expect(taskFinishedRows(after)).toHaveLength(2) + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(NodeFileSystem.layer, layerInMemoryEventLogWithIds.pipe(Layer.provide(layerLiveIdFactory))), + ), + ), +) + +it.effect('list_agents, send_message, and stop_agent drive background agents from the model', () => + Effect.gen(function* () { + const firstRequestStarted = yield* Deferred.make() + const hang = yield* Deferred.make() + const child = yield* makeGatedModel(claudeActiveModel, [ + gatedTurn(textTurn('never'), { gate: hang, started: firstRequestStarted }), + textTurn('resumed answer'), + ]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([backgroundCall('provider-call-1', 'controlled')]), + textTurn('launched'), + ]) + const session = yield* openBackgroundSession(root.model, child.model) + yield* session.send('go') + yield* Deferred.await(firstRequestStarted) + const task = onlyTask(yield* session.backgroundTasks) + const ref = shortAgentId(task.taskId) + + yield* root.push([ + toolCallTurn([{ id: 'provider-call-2', name: 'list_agents', params: {} }]), + toolCallTurn([ + { id: 'provider-call-3', name: 'send_message', params: { agent_id: ref, message: 'also check X' } }, + ]), + toolCallTurn([{ id: 'provider-call-4', name: 'stop_agent', params: { agent_id: ref } }]), + toolCallTurn([{ id: 'provider-call-5', name: 'stop_agent', params: { agent_id: ref } }]), + textTurn('managed'), + ]) + const managed = yield* session.send('manage it') + expect(managed.resultText).toBe('managed') + yield* session.awaitBackgroundIdle + + const afterManage = yield* session.entries + const results = toolResultTexts(afterManage) + expect(results[1]).toContain('1 running, 0 finished') + expect(results[1]).toContain(`${ref} [running] researcher: \\"controlled\\"`) + expect(results[2]).toContain('Message queued for running agent') + expect(results[3]).toContain(`Stopped background agent ${ref}`) + expect(results[4]).toContain('is not running (status: stopped)') + // The model stopped it itself: no notification, no wake. + expect(notificationMessages(afterManage)).toHaveLength(0) + expect(rootFinishedOutcomes(afterManage, session.rootAgentId)).toHaveLength(2) + + // Messaging a stopped agent continues it in the background; its completion wakes the root. + yield* root.push([ + toolCallTurn([ + { id: 'provider-call-6', name: 'send_message', params: { agent_id: ref, message: 'resume please' } }, + ]), + textTurn('resumed it'), + textTurn('got the resumed answer'), + ]) + yield* session.send('continue it') + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + expect(toolResultTexts(entries)[5]).toContain(`Background agent resumed. agent_id: ${ref}`) + const runs = taskStartedRows(entries) + expect(runs.map((row) => row.runCount)).toEqual([1, 2]) + expect(runs[1]?.prompt).toBe('resume please') + expect(runs[1]?.description).toBe('controlled') + expect(taskFinishedRows(entries).map((row) => row.status)).toEqual(['stopped', 'completed']) + const resumed = onlyTask(yield* session.backgroundTasks) + expect(resumed.status).toBe('completed') + expect(resumed.summary).toBe('resumed answer') + expect(resumed.runCount).toBe(2) + expect(rootFinishedOutcomes(entries, session.rootAgentId).at(-1)?.resultText).toBe('got the resumed answer') + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +const openNestedSession = (rootModel: FoldModel, coordinatorModel: FoldModel, researcherModel: FoldModel) => { + const researcher = defineSubagent({ + name: 'researcher', + description: 'explores', + systemPrompt: 'You are a researcher.', + model: researcherModel, + }) + const coordinator = defineSubagent({ + name: 'coordinator', + description: 'delegates', + systemPrompt: 'You coordinate.', + model: coordinatorModel, + tools: [subagentTool([researcher])], + }) + return Session.open({ + agent: defineAgent({ model: rootModel, systemPrompt: 'root', tools: [subagentTool([coordinator])] }), + }) +} + +const coordinatorCall = { + id: 'provider-call-root', + name: 'subagent', + params: { agent: 'coordinator', prompt: 'coordinate' }, +} as const + +it.effect('a background task launched by a running subagent notifies that subagent', () => + Effect.gen(function* () { + const coordinatorGate = yield* Deferred.make() + const researcher = yield* makeGatedModel(claudeActiveModel, [textTurn('grandchild result')]) + const coordinator = yield* makeGatedModel(claudeActiveModel, [ + toolCallTurn([backgroundCall('provider-call-c1', 'nested task')]), + gatedTurn(textTurn('waiting on it'), { gate: coordinatorGate }), + textTurn('coordinator saw the result'), + ]) + const root = yield* makeGatedModel(gptActiveModel, [toolCallTurn([coordinatorCall]), textTurn('root done')]) + const session = yield* openNestedSession(root.model, coordinator.model, researcher.model) + + const sendFiber = yield* Effect.forkScoped(session.send('go')) + yield* awaitTasks(session, (tasks) => tasks.length === 1 && tasks.every((task) => task.notifiedAt !== null)) + yield* Deferred.succeed(coordinatorGate, undefined) + const finished = yield* Fiber.join(sendFiber) + expect(finished.resultText).toBe('root done') + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + const task = onlyTask(yield* session.backgroundTasks) + const notifications = notificationMessages(entries) + expect(notifications).toHaveLength(1) + expect(notifications[0]?.agentId).toBe(task.parentAgentId) + expect(notifications[0]?.agentId).not.toBe(session.rootAgentId) + expect(rootFinishedOutcomes(entries, session.rootAgentId)).toHaveLength(1) + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) + +it.effect('a background task whose launcher already finished notifies the root', () => + Effect.gen(function* () { + const release = yield* Deferred.make() + const researcher = yield* makeGatedModel(claudeActiveModel, [ + gatedTurn(textTurn('late result'), { gate: release }), + ]) + const coordinator = yield* makeGatedModel(claudeActiveModel, [ + toolCallTurn([backgroundCall('provider-call-c1', 'outlives its launcher')]), + textTurn('coordinator done'), + ]) + const root = yield* makeGatedModel(gptActiveModel, [ + toolCallTurn([coordinatorCall]), + textTurn('root done'), + textTurn('root relayed the late result'), + ]) + const session = yield* openNestedSession(root.model, coordinator.model, researcher.model) + + yield* session.send('go') + yield* Deferred.succeed(release, undefined) + yield* session.awaitBackgroundIdle + + const entries = yield* session.entries + const notifications = notificationMessages(entries) + expect(notifications).toHaveLength(1) + expect(notifications[0]?.agentId).toBe(session.rootAgentId) + expect(rootFinishedOutcomes(entries, session.rootAgentId).map((run) => run.resultText)).toEqual([ + 'root done', + 'root relayed the late result', + ]) + }).pipe(Effect.scoped, Effect.provide(NodeFileSystem.layer)), +) From 1d013e5fc8c61f782423a919fa8828cb79003675 Mon Sep 17 00:00:00 2001 From: dex Date: Tue, 13 Oct 2026 04:21:26 +0000 Subject: [PATCH 3/8] feat(agent): give the coding and RLM modes the background-agent controls Install list_agents, stop_agent, and send_message next to the subagent tool, and tell the coding agent when to launch subagents in the background. Co-Authored-By: Claude Opus 5.5 HumanLayer-Session: https://app.dev.codelayer.gg/sessions/01a13295-e128-724d-bb90-666e8fbf2d01 --- packages/fold-agent/src/Mode/Mode.ts | 9 ++++++--- packages/fold-agent/src/Mode/Rlm.ts | 3 ++- packages/fold-agent/test/Mode/Launch.vi.test.ts | 1 + packages/fold-agent/test/Mode/Rlm.vi.test.ts | 1 + 4 files changed, 10 insertions(+), 4 deletions(-) diff --git a/packages/fold-agent/src/Mode/Mode.ts b/packages/fold-agent/src/Mode/Mode.ts index ff96662..b16e729 100644 --- a/packages/fold-agent/src/Mode/Mode.ts +++ b/packages/fold-agent/src/Mode/Mode.ts @@ -6,12 +6,12 @@ * * `defaultCodingMode` is the batteries-included local coding agent: the full filesystem toolset (read, * write, edit, apply_patch, bash - the family policy advertises the right editing subset per model), - * the disk skill tool, and the default subagent roster (general-purpose, bash, researcher - see - * Mode/Subagents). Other modes are just other `FoldMode` values (Mode/Rlm); Mode/ModeName maps the + * the disk skill tool, the default subagent roster (general-purpose, bash, researcher - see + * Mode/Subagents), and the background-agent controls (list_agents, stop_agent, send_message). Other modes are just other `FoldMode` values (Mode/Rlm); Mode/ModeName maps the * selectable names to them. RPI is not a mode: the `rpi` context flag appends the RPI specialist * roster to ANY mode's subagents (Mode/Rpi). */ -import { skillTool, subagentTool, type FoldTool } from '@humanlayer/fold-core' +import { agentControlTools, skillTool, subagentTool, type FoldTool } from '@humanlayer/fold-core' import type { FileSystem, Path } from 'effect' import type { HttpClient } from 'effect/http' import type { ChildProcessSpawner } from 'effect/process' @@ -74,6 +74,8 @@ export const DEFAULT_CODING_PROMPT: string = 'Delegate to a subagent when a step would burn context you need for the main task: researcher for ' + 'locating and explaining code, web-search-researcher for current web research, bash for running a ' + 'command and summarizing its output, and general-purpose for a self-contained multi-step task. Do simple work yourself. ' + + 'Launch independent subagents with run_in_background when you can keep working meanwhile; each result ' + + 'arrives later as a message, and list_agents, send_message, and stop_agent manage them. ' + 'Keep responses concise; let the tools do the work.' /** The batteries-included local coding agent: full filesystem toolset, disk skills, default subagents. */ @@ -85,5 +87,6 @@ export const defaultCodingMode: FoldMode = { ...codingTools({ cwd }), skillTool(skillsFromDisk({ cwd })), subagentTool(modeSubagents({ cwd, rpi })), + ...agentControlTools(), ], } diff --git a/packages/fold-agent/src/Mode/Rlm.ts b/packages/fold-agent/src/Mode/Rlm.ts index 70b8056..67f6e6e 100644 --- a/packages/fold-agent/src/Mode/Rlm.ts +++ b/packages/fold-agent/src/Mode/Rlm.ts @@ -8,7 +8,7 @@ * The mode runs on the `orchestrator` config role, which falls back to `smart` when the config * declares no orchestrator binding (D25). */ -import { skillTool, subagentTool } from '@humanlayer/fold-core' +import { agentControlTools, skillTool, subagentTool } from '@humanlayer/fold-core' import { skillsFromDisk } from '../Skills/DiskSkills' import { applyPatchTool } from '../Tools/ApplyPatchTool' @@ -64,5 +64,6 @@ export const rlmMode: FoldMode = { applyPatchTool({ cwd }), skillTool(skillsFromDisk({ cwd })), subagentTool(modeSubagents({ cwd, rpi })), + ...agentControlTools(), ], } diff --git a/packages/fold-agent/test/Mode/Launch.vi.test.ts b/packages/fold-agent/test/Mode/Launch.vi.test.ts index 9087e22..db5936b 100644 --- a/packages/fold-agent/test/Mode/Launch.vi.test.ts +++ b/packages/fold-agent/test/Mode/Launch.vi.test.ts @@ -127,6 +127,7 @@ it.effect('launchSession composes the model, agentfiles, and mode tools over Ses expect(tools).toContain('bash') expect(tools).toContain('skill') expect(tools).toContain('subagent') + expect(tools).toEqual(expect.arrayContaining(['list_agents', 'stop_agent', 'send_message'])) }), ) }).pipe(Effect.provide(launchTestServices)), diff --git a/packages/fold-agent/test/Mode/Rlm.vi.test.ts b/packages/fold-agent/test/Mode/Rlm.vi.test.ts index d400428..e6ec081 100644 --- a/packages/fold-agent/test/Mode/Rlm.vi.test.ts +++ b/packages/fold-agent/test/Mode/Rlm.vi.test.ts @@ -68,6 +68,7 @@ it('holds the file tools, skills, and subagents - and deliberately no bash', () expect(names).not.toContain('bash') expect(names.filter((name) => name === 'subagent')).toHaveLength(1) + expect(names).toEqual(expect.arrayContaining(['list_agents', 'stop_agent', 'send_message'])) expect(names.filter((name) => name === 'skill')).toHaveLength(1) expect(names).toContain('read') expect(names).toContain('write') From 11dadb3b783fb7cd319e072a812aa6b1e883eda0 Mon Sep 17 00:00:00 2001 From: dex Date: Tue, 13 Oct 2026 04:26:53 +0000 Subject: [PATCH 4/8] feat(cli): keep --prompt runs open until background agents settle runPrompt now waits after the prompt's own turn until background agents and their task-notification turns are done, and reports the root's latest outcome. Agents still running after --background-timeout seconds (default 600) are stopped; a signal while waiting stops them too. Co-Authored-By: Claude Opus 5.5 HumanLayer-Session: https://app.dev.codelayer.gg/sessions/01a13295-e128-724d-bb90-666e8fbf2d01 --- packages/fold-cli/src/Commands.ts | 14 +- packages/fold-cli/src/Run.ts | 63 +++++++- .../fold-cli/test/BackgroundRun.vi.test.ts | 141 ++++++++++++++++++ .../fold-cli/test/SessionFlags.vi.test.ts | 1 + 4 files changed, 212 insertions(+), 7 deletions(-) create mode 100644 packages/fold-cli/test/BackgroundRun.vi.test.ts diff --git a/packages/fold-cli/src/Commands.ts b/packages/fold-cli/src/Commands.ts index feb415d..52c6910 100644 --- a/packages/fold-cli/src/Commands.ts +++ b/packages/fold-cli/src/Commands.ts @@ -45,11 +45,11 @@ import { type XaiAuthError, type XaiAuthOptions, } from '@humanlayer/fold-xai' -import { Array as Arr, Clock, Console, Effect, Layer, Match, Option, Schema } from 'effect' +import { Array as Arr, Clock, Console, Duration, Effect, Layer, Match, Option, Schema } from 'effect' import { type CliError, Command, Flag } from 'effect/cli' import { jsonOutputRenderer, promptOutputRenderer, type JsonOutputMode } from './Renderer' -import { ResumeTarget, runPrompt, type CliSessionOptions } from './Run' +import { ResumeTarget, runPrompt, type CliSessionOptions, type PromptRunOptions } from './Run' declare const FOLD_VERSION: string /** The release version the native build defines as a global; source runs have no such global. */ @@ -186,6 +186,10 @@ const commonFlags = { 'Keep this many recent tokens verbatim', ), compactionPrompt: optionalString('compaction-prompt', 'Override the compaction summarization prompt'), + backgroundTimeout: optionalInteger( + 'background-timeout', + 'Seconds a --prompt run waits for background agents before stopping them (default 600)', + ), } /** Decoded values of {@link commonFlags}, the root `fold` command's flag set. */ @@ -211,6 +215,7 @@ export type CommonFlagValues = { readonly compactionReserveTokens: Option.Option readonly compactionKeepRecentTokens: Option.Option readonly compactionPrompt: Option.Option + readonly backgroundTimeout: Option.Option } const optionValue = (option: Option.Option): A | undefined => Option.getOrUndefined(option) @@ -399,7 +404,10 @@ const run = Command.make('foldcode', commonFlags, (input) => ? promptOutputRenderer({ colors: !input.noColor, verbose: input.verbose, catalog }) : jsonOutputRenderer({ mode: outputMode }) - const finished = yield* runPrompt({ ...sessionOptions, prompt }, renderer) + const promptOptions: Mutable = { ...sessionOptions, prompt } + const backgroundTimeout = optionValue(input.backgroundTimeout) + if (backgroundTimeout !== undefined) promptOptions.backgroundTimeout = Duration.seconds(backgroundTimeout) + const finished = yield* runPrompt(promptOptions, renderer) if (finished.outcome === 'completed') return process.exitCode = finished.outcome === 'interrupted' ? 130 : 1 diff --git a/packages/fold-cli/src/Run.ts b/packages/fold-cli/src/Run.ts index 0dadc68..fb24319 100644 --- a/packages/fold-cli/src/Run.ts +++ b/packages/fold-cli/src/Run.ts @@ -36,6 +36,7 @@ import { Predicate, Cause, Clock, + Duration, Effect, Exit, Fiber, @@ -80,8 +81,16 @@ export type CliSessionOptions = { /** Options for one non-interactive `--prompt` run. */ export type PromptRunOptions = CliSessionOptions & { readonly prompt: string + /** + * How long the run waits for background agents (and the notification turns they trigger) after + * the prompt's own turn ends, before stopping them. Defaults to {@link DEFAULT_BACKGROUND_TIMEOUT}. + */ + readonly backgroundTimeout?: Duration.Input } +/** Default cap on a headless run's wait for background agents. */ +export const DEFAULT_BACKGROUND_TIMEOUT: Duration.Input = Duration.minutes(10) + type OpenedSession = { readonly session: FoldSession< | FileSystem.FileSystem @@ -315,6 +324,7 @@ const withProcessSignals = ( >, renderer: OutputRenderer, effect: Effect.Effect, + onSignal: Effect.Effect = session.interrupt(), ): Effect.Effect => Effect.acquireUseRelease( Effect.sync(() => { @@ -323,9 +333,7 @@ const withProcessSignals = ( if (fired) return fired = true Effect.runFork( - renderer - .renderNote('interrupt requested; saving session state') - .pipe(Effect.andThen(session.interrupt())), + renderer.renderNote('interrupt requested; saving session state').pipe(Effect.andThen(onSignal)), ) } process.on('SIGINT', handler) @@ -376,6 +384,45 @@ const forkStartupEnsures = ( }), ).pipe(Effect.asVoid) +/** + * After a headless prompt's own turn: wait until background agents have finished and every + * notification turn they triggered has run, then return the root's latest outcome. Agents still + * running after `timeout` are stopped (recorded `stopped`; no further turn). An interrupted prompt + * stops background agents at once instead of waiting. + */ +export const settleBackgroundWork = ( + session: FoldSession, + renderer: Pick, + sent: AgentFinishedLogEntry, + timeout: Duration.Input = DEFAULT_BACKGROUND_TIMEOUT, +): Effect.Effect => + Effect.gen(function* () { + if (sent.outcome === 'interrupted') { + yield* session.stop('the user interrupted the run') + return sent + } + + const running = (yield* session.backgroundTasks).filter((task) => task.status === 'running').length + if (running > 0) { + yield* renderer.renderNote(`waiting for ${running} background agent${running === 1 ? '' : 's'} to finish`) + } + const settled = yield* session.awaitBackgroundIdle.pipe(Effect.timeoutOption(timeout)) + if (Option.isNone(settled)) { + yield* renderer.renderNote( + `background agents still running after ${Duration.format(Duration.fromInputUnsafe(timeout))}; stopping them`, + ) + yield* session.stop('background agents did not finish before the run timed out') + yield* session.awaitBackgroundIdle + } + + const entries = yield* session.entries + const latest = entries.findLast( + (entry): entry is AgentFinishedLogEntry => + Predicate.isTagged(entry, 'agent-finished') && entry.agentId === session.rootAgentId, + ) + return latest ?? sent + }) + /** Open a session, print its header, and run one CI-friendly prompt. */ export const runPrompt = ( options: PromptRunOptions, @@ -398,7 +445,7 @@ export const runPrompt = ( yield* renderer.renderHeader(yield* sessionHeader(opened, options)) const renderFiber = yield* renderLiveEvents(opened.session, renderer) yield* forkStartupEnsures(options, renderer) - const finished = yield* withProcessSignals( + const sent = yield* withProcessSignals( opened.session, renderer, opened.session.send(options.prompt).pipe( @@ -408,6 +455,14 @@ export const runPrompt = ( ), ), ) + // Background agents keep the run open: their notification turns are part of this prompt's answer. + // A signal while waiting stops them (an interrupt alone would leave them running). + const finished = yield* withProcessSignals( + opened.session, + renderer, + settleBackgroundWork(opened.session, renderer, sent, options.backgroundTimeout), + opened.session.stop('the user interrupted the run'), + ) yield* Effect.yieldNow yield* renderer.renderFinish(finished) yield* Fiber.interrupt(renderFiber) diff --git a/packages/fold-cli/test/BackgroundRun.vi.test.ts b/packages/fold-cli/test/BackgroundRun.vi.test.ts new file mode 100644 index 0000000..2da0de9 --- /dev/null +++ b/packages/fold-cli/test/BackgroundRun.vi.test.ts @@ -0,0 +1,141 @@ +/** + * Headless runs and background agents: `settleBackgroundWork` keeps a `--prompt` run open until + * background agents and the notification turns they trigger are done, returns the root's latest + * outcome, and stops agents that outlive the timeout. Real fold-core sessions over a scripted model. + */ +import { expect, it } from '@effect/vitest' +import { + customModel, + defineAgent, + defineSubagent, + Session, + subagentTool, + type ActiveModel, + type FoldModel, +} from '@humanlayer/fold-core' +import { Deferred, Duration, Effect, Fiber, Ref, Stream } from 'effect' +import { LanguageModel, type Response } from 'effect/ai' +import { TestClock } from 'effect/testing' + +import { settleBackgroundWork } from '../src/Run' + +const activeModel: ActiveModel = { + providerId: 'scripted', + providerKind: 'openai-compatible', + modelId: 'scripted', + role: null, + requestedReasoningLevel: 'off', + reasoning: { _tag: 'disabled' }, +} + +const finish = (reason: Response.FinishReason): Response.StreamPartEncoded => ({ + type: 'finish', + reason, + response: undefined, + usage: { + inputTokens: { uncached: undefined, total: 10, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 5, text: undefined, reasoning: undefined }, + }, +}) + +type Turn = { readonly parts: ReadonlyArray; readonly gate: Deferred.Deferred | null } + +const text = (value: string, gate: Deferred.Deferred | null = null): Turn => ({ + parts: [ + { type: 'text-start', id: 't' }, + { type: 'text-delta', id: 't', delta: value }, + { type: 'text-end', id: 't' }, + finish('stop'), + ], + gate, +}) + +const launchInBackground: Turn = { + parts: [ + { + type: 'tool-call', + id: 'call-1', + name: 'subagent', + params: { agent: 'worker', description: 'slow work', prompt: 'work', run_in_background: true }, + providerExecuted: false, + }, + finish('tool-calls'), + ], + gate: null, +} + +const scripted = (turns: ReadonlyArray): Effect.Effect => + Effect.gen(function* () { + const remaining = yield* Ref.make(turns) + const make = LanguageModel.make({ + generateText: () => Effect.die(new Error('streamText only')), + streamText: () => + Stream.unwrap( + Effect.gen(function* () { + const [next, ...rest] = yield* Ref.get(remaining) + if (next === undefined) return yield* Effect.die(new Error('script exhausted')) + yield* Ref.set(remaining, rest) + if (next.gate !== null) yield* Deferred.await(next.gate) + return Stream.fromIterable(next.parts) + }), + ), + }) + return customModel({ activeModel, make }) + }) + +const openSession = (root: FoldModel, child: FoldModel) => + Session.open({ + agent: defineAgent({ + model: root, + tools: [subagentTool([defineSubagent({ name: 'worker', description: 'works', model: child })])], + }), + }) + +const silent = { renderNote: () => Effect.void } + +it.effect('waits for background agents and their notification turn, then returns the latest root outcome', () => + Effect.gen(function* () { + const release = yield* Deferred.make() + const session = yield* openSession( + yield* scripted([launchInBackground, text('started'), text('final answer with the result')]), + yield* scripted([text('worker result', release)]), + ) + const sent = yield* session.send('go') + expect(sent.resultText).toBe('started') + + const settling = yield* Effect.forkScoped(settleBackgroundWork(session, silent, sent)) + yield* Deferred.succeed(release, undefined) + const settled = yield* Fiber.join(settling) + + expect(settled.resultText).toBe('final answer with the result') + expect((yield* session.backgroundTasks).map((task) => task.status)).toEqual(['completed']) + }).pipe(Effect.scoped), +) + +it.effect('stops background agents that outlive the timeout, without another turn', () => + Effect.gen(function* () { + const never = yield* Deferred.make() + const notes: Array = [] + const session = yield* openSession( + yield* scripted([launchInBackground, text('started')]), + yield* scripted([text('never', never)]), + ) + const sent = yield* session.send('go') + + const settling = yield* Effect.forkScoped( + settleBackgroundWork( + session, + { renderNote: (note) => Effect.sync(() => notes.push(note)) }, + sent, + Duration.minutes(1), + ), + ) + yield* TestClock.adjust(Duration.minutes(1)) + const settled = yield* Fiber.join(settling) + + expect(settled.resultText).toBe('started') + expect((yield* session.backgroundTasks).map((task) => task.status)).toEqual(['stopped']) + expect(notes[0]).toBe('waiting for 1 background agent to finish') + expect(notes[1]).toContain('stopping them') + }).pipe(Effect.scoped), +) diff --git a/packages/fold-cli/test/SessionFlags.vi.test.ts b/packages/fold-cli/test/SessionFlags.vi.test.ts index 356afd6..8b73a44 100644 --- a/packages/fold-cli/test/SessionFlags.vi.test.ts +++ b/packages/fold-cli/test/SessionFlags.vi.test.ts @@ -29,6 +29,7 @@ const baseFlags: CommonFlagValues = { compactionReserveTokens: Option.none(), compactionKeepRecentTokens: Option.none(), compactionPrompt: Option.none(), + backgroundTimeout: Option.none(), } it.effect('lowers --mode into the session options', () => From 451459190cf6ad3d665fb546211df4fea791fec9 Mon Sep 17 00:00:00 2001 From: dex Date: Tue, 13 Oct 2026 04:26:53 +0000 Subject: [PATCH 5/8] feat(cli): show background agents in the TUI and line renderer - SUBAGENTS rail: BG badge, status from the task rows, X STOP hint; the rail header and activity label count running background agents - x on a selected running background agent stops it (session.stopTask) - a task-notification wake renders as an AGENTS notification row with the per-agent summaries instead of a USER row - line renderer prints [background] start/finish lines and a [notification] line for wakes; JSON verbose mode encodes task-progress Co-Authored-By: Claude Opus 5.5 HumanLayer-Session: https://app.dev.codelayer.gg/sessions/01a13295-e128-724d-bb90-666e8fbf2d01 --- packages/fold-cli/src/Renderer.ts | 46 +++++++++ packages/fold-cli/src/tui/App.tsx | 61 +++++++++++- packages/fold-cli/src/tui/EventViews.tsx | 2 + packages/fold-cli/src/tui/HostedTuiSession.ts | 22 +++++ packages/fold-cli/src/tui/SessionState.ts | 46 ++++++++- packages/fold-cli/src/tui/Shell.tsx | 1 + packages/fold-cli/src/tui/Subagents.ts | 45 +++++++-- packages/fold-cli/test/Renderer.vi.test.ts | 78 +++++++++++++++ .../fold-cli/test/TuiSubagents.vi.test.ts | 78 ++++++++++++++- .../fold-cli/test/tui/SessionState.vi.test.ts | 95 +++++++++++++++++++ 10 files changed, 458 insertions(+), 16 deletions(-) diff --git a/packages/fold-cli/src/Renderer.ts b/packages/fold-cli/src/Renderer.ts index 950174c..dc59cc2 100644 --- a/packages/fold-cli/src/Renderer.ts +++ b/packages/fold-cli/src/Renderer.ts @@ -15,6 +15,7 @@ import { usageOutputTotal, type ActiveModel, type AgentFinishedLogEntry, + type TaskFinishedLogEntry, type AgentStartedLogEntry, type AssistantMessageLogEntry, type ModelChangeLogEntry, @@ -104,6 +105,15 @@ const DeltaPart = Schema.Union([ Schema.Struct({ type: Schema.Literal('text-delta'), id: Schema.String, delta: Schema.String }), Schema.Struct({ type: Schema.Literal('reasoning-delta'), id: Schema.String, delta: Schema.String }), Schema.Struct({ type: Schema.Literal('tool-progress'), toolName: Schema.String, payload: Schema.Json }), + Schema.Struct({ + type: Schema.Literal('task-progress'), + taskId: AgentId, + turns: Schema.Finite, + toolUses: Schema.Finite, + totalTokens: Schema.NullOr(Schema.Finite), + lastToolName: Schema.NullOr(Schema.String), + progressSummary: Schema.NullOr(Schema.String), + }), ]) const FoldEventJson = Schema.fromJsonString( @@ -300,6 +310,16 @@ const resumeCommand = ( return `foldcode ${flags.join(' ')}` } +const taskStatusColor = (ansi: AnsiPalette, status: TaskFinishedLogEntry['status']): ((text: string) => string) => + Match.value(status).pipe( + Match.when('completed', () => ansi.green), + Match.when('stopped', () => ansi.yellow), + Match.when('failed', () => ansi.red), + Match.exhaustive, + ) + +const plural = (count: number, noun: string): string => `${count} ${noun}${count === 1 ? '' : 's'}` + const outcomeColor = (ansi: AnsiPalette, outcome: AgentFinishedLogEntry['outcome']): ((text: string) => string) => Match.value(outcome).pipe( Match.when('completed', () => ansi.green), @@ -448,9 +468,32 @@ export const humanOutputRenderer = (options?: RendererOptions): OutputRenderer = ) }, 'user-message': (message) => { + // A task-notification wake is runtime-written: show which agents it reports, not its XML. + if (message.origin !== undefined) { + const ids = message.origin.taskIds.map(displayAgentId).join(', ') + return renderAgentLine( + message.agentId, + `${label(ansi, 'notification')} background agent ${ids} finished`, + ) + } const text = encodedContentText(message.message.content) return text.length === 0 ? Effect.void : renderAgentLine(message.agentId, `${ansi.cyan('>')} ${text}`) }, + task_started: (started) => + renderAgentLine( + started.agentId, + `${label(ansi, 'background')} ${displayAgentId(started.agentId)} ` + + `${started.runCount > 1 ? 'resumed' : 'started'}: ${started.description}`, + ), + task_finished: (finished) => { + const color = taskStatusColor(ansi, finished.status) + const work = `${plural(finished.usage.turns, 'turn')}, ${plural(finished.usage.toolUses, 'tool use')}` + const error = finished.error === null ? '' : ` - ${finished.error}` + return renderAgentLine( + finished.agentId, + `${label(ansi, 'background')} ${displayAgentId(finished.agentId)} ${color(finished.status)} (${work})${error}`, + ) + }, 'assistant-message': (message) => { if (message.finish !== null) { latestUsage.set(message.agentId, { @@ -578,6 +621,9 @@ export const humanOutputRenderer = (options?: RendererOptions): OutputRenderer = `${label(ansi, 'tool')} ${ansi.cyan(toolName)} ${truncate(jsonText(payload), verbose ? 2000 : 300)}`, ) }, + + // Background progress is visible through the agent's own lines; the counts land on task_finished. + 'task-progress': () => Effect.void, }), ) } diff --git a/packages/fold-cli/src/tui/App.tsx b/packages/fold-cli/src/tui/App.tsx index 6aab21b..1a6071b 100644 --- a/packages/fold-cli/src/tui/App.tsx +++ b/packages/fold-cli/src/tui/App.tsx @@ -63,6 +63,8 @@ export type TuiAppProps = { readonly onStop?: () => void readonly onTargetSubmit?: (agentId: string, text: string, verb: RootInputVerb) => void readonly onTargetInterrupt?: (agentId: string) => void + /** Stop a running background agent (`x` on the selected subagent). */ + readonly onStopTask?: (agentId: string) => void readonly onInjectSkill?: (skill: string, agentId: string | null) => void readonly initialSelectedAgentId?: string readonly gitSnapshot?: Accessor @@ -222,9 +224,16 @@ export const TuiApp = (props: TuiAppProps) => { ) const runningCount = createMemo(() => meta().running + (props.state().status === 'RUNNING' ? 1 : 0)) const sessionActivityLabel = createMemo(() => { - if (sessionActivity() === 'running') return runningCount() > 1 ? `${runningCount()} RUNNING` : 'RUNNING' + const background = meta().background > 0 ? ` · ${meta().background} BG` : '' + if (sessionActivity() === 'running') + return `${runningCount() > 1 ? `${runningCount()} RUNNING` : 'RUNNING'}${background}` return sessionActivity().toUpperCase() }) + /** The focused subagent when it is a running background agent: the target of `x`. */ + const stoppableAgent = createMemo(() => { + const agent = focusedAgent() + return agent !== undefined && agent.background && agent.status === 'running' ? agent : undefined + }) const reasoningLabel = createMemo(() => { const level = props.state().reasoningLevel return level === null ? '' : ` · ${level.toUpperCase()}` @@ -696,6 +705,14 @@ export const TuiApp = (props: TuiAppProps) => { return case 'd': toggleChanges() + return + case 'x': { + const agent = stoppableAgent() + if (agent === undefined) return + key.preventDefault() + props.onStopTask?.(agent.agentId) + return + } } }) @@ -1089,6 +1106,18 @@ export const TuiApp = (props: TuiAppProps) => { ^C INTERRUPT + + { + const agent = stoppableAgent() + if (agent !== undefined) props.onStopTask?.(agent.agentId) + }} + > + {' · X STOP'} + +