Merge v3-t47: subagents as a world extension (T47, P3-h)
This commit is contained in:
@@ -0,0 +1,27 @@
|
||||
# Subagents
|
||||
|
||||
The `subagent` tool starts named definitions as independent child conversations.
|
||||
The tool supports `start`, `message`, `wait`, `stop`, and `list`.
|
||||
Definitions come from world files and project overlays.
|
||||
Only `isolate: none` works until P3-k supplies workspaces.
|
||||
A worktree request returns a clear error before it creates a child.
|
||||
|
||||
An anchor task in the parent owns each child.
|
||||
The conversation chain therefore includes the child and its parent.
|
||||
A parent stop does not stop the child.
|
||||
A child stop cancels captured reporter inputs and current work, but permits later reuse.
|
||||
|
||||
The reporter uses two durable phases: `deliver` and `report`.
|
||||
The delivery phase submits an instruction with a stable request ID.
|
||||
The report phase steers the parent with the stored answer and its returned model identity.
|
||||
A stable request ID deduplicates each answer across replays and restarts.
|
||||
The `afterTools` hook acknowledges committed foreground results and withdraws duplicate reports.
|
||||
A foreground wait yields when the parent receives steering. The child continues independently.
|
||||
|
||||
The recovery observer and startup scan deliver answers to one-way parent messages and parent replies.
|
||||
They also recover answers after interrupted or failed delivery.
|
||||
They do not report ordinary child inputs or messages from other conversations.
|
||||
|
||||
The service creates standing conversations for resident definitions.
|
||||
The sections list available definitions and child work that remains active.
|
||||
The definitions editor, workspaces, `land`, and `discard` remain outside this extension's current scope.
|
||||
@@ -0,0 +1,99 @@
|
||||
/**
|
||||
* Subagents: conversations delegated to agent definitions, and the standing agents. Other extensions create
|
||||
* configured child conversations and open standing agents here; definitions themselves are read with `lib/agents.ts`.
|
||||
*/
|
||||
import type { Context } from '@earendil-works/chord'
|
||||
import { defineDoc } from '@earendil-works/pi-durable'
|
||||
import type { ConversationId, EntryId, TaskId, Tx } from '@earendil-works/pi-durable'
|
||||
const defineApi = <S>() => (name: string): {name: string; service?: S} => ({name})
|
||||
import { RunsDoc, StandingDoc } from './refs.ts'
|
||||
import type { Run } from './refs.ts'
|
||||
import type { createChild } from './children.ts'
|
||||
|
||||
export { Started, StandingDoc } from './refs.ts'
|
||||
|
||||
/** Each conversation's subagents and pending subagent actions, on the conversation that started them. */
|
||||
export const Subagents = defineDoc<{
|
||||
agents: Record<
|
||||
string,
|
||||
{
|
||||
conversationId: ConversationId
|
||||
agent: string
|
||||
background: boolean
|
||||
reported: EntryId[]
|
||||
}
|
||||
>
|
||||
actions: Record<
|
||||
string,
|
||||
{
|
||||
conversationId: ConversationId
|
||||
reporter: TaskId
|
||||
answer?: EntryId
|
||||
/**
|
||||
* Unset while delivering; then `done`, `unanswered` (stopped by the parent), `interrupted` (stopped in the
|
||||
* child), `failed` (a provider error the watchdog holds, or a delivery error) or `given-up` (the watchdog
|
||||
* reported that provider error instead of retrying it: final).
|
||||
*/
|
||||
status?: string
|
||||
}
|
||||
>
|
||||
}>({
|
||||
kind: 'ultimate.subagents',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({ agents: {}, actions: {} }),
|
||||
})
|
||||
|
||||
/** `{ [name]: conversationId }` of the standing agents, on the root conversation. */
|
||||
export const Standing = defineDoc<Record<string, ConversationId>>({
|
||||
kind: StandingDoc.kind,
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({}),
|
||||
})
|
||||
|
||||
/** `{ [conversationId]: Run }` of the children `subagent start` made, on the root, for the sidebar and call cards. */
|
||||
export const Runs = defineDoc<Record<string, Run>>({
|
||||
kind: RunsDoc.kind,
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({}),
|
||||
})
|
||||
|
||||
/**
|
||||
* The definition a conversation was configured from: children (including automation runs) and standing agents. An agent
|
||||
* made on the fly has none: `adhoc`, so nothing looks a definition up by its label.
|
||||
*/
|
||||
export const Profile = defineDoc<{ definition?: string; adhoc?: true }>({
|
||||
kind: 'ultimate.agent-profile',
|
||||
checkpointWhen: () => false,
|
||||
version: 1,
|
||||
scope: 'conversation',
|
||||
history: 'latest',
|
||||
fork: 'initial',
|
||||
initial: () => ({}),
|
||||
})
|
||||
|
||||
export type SubagentsService = {
|
||||
/** The standing agents that run in this home, by name and description; use it to list valid `@name` addresses. */
|
||||
residents(ctx: Context): Promise<{ name: string; description: string }[]>
|
||||
/** The conversation of standing agent `name`, created or refreshed from its definition; throws for unknown names. */
|
||||
standing(name: string, ctx: Context): Promise<ConversationId>
|
||||
/** Create, inside `tx`, a child conversation configured from a definition or as a copy of the parent's agent. */
|
||||
createChild: (
|
||||
tx: Tx,
|
||||
options: Parameters<typeof createChild>[2],
|
||||
) => ReturnType<typeof createChild>
|
||||
}
|
||||
/** Child conversations and standing agents; `need` it where delegation is required, else `want` it. */
|
||||
export const SubagentsApi = defineApi<SubagentsService>()('subagents')
|
||||
export const tier = 'core'
|
||||
@@ -0,0 +1,37 @@
|
||||
/** Child configuration and start events share the transaction that creates ownership. */
|
||||
import { AgentDoc, configure } from '@earendil-works/pi-durable'
|
||||
import { toolsFor } from '../core/agents.ts'
|
||||
import { Started } from './refs.ts'
|
||||
import { Profile } from './api.ts'
|
||||
export async function started(host: any, tx: any, conversation: any, agent: string, kind: 'child'|'standing'|'branch') {
|
||||
const {model, thinkingLevel} = await tx.doc(AgentDoc, conversation.id)
|
||||
const parent = conversation.owner?.conversationId ?? conversation.parent?.conversationId
|
||||
await host.emit(tx, Started, {id:conversation.id, agent, kind,
|
||||
...(model ? {model:`${model.provider}/${model.modelId}`} : {}),
|
||||
...(thinkingLevel ? {thinking:thinkingLevel} : {}), ...(parent === undefined ? {} : {parent})})
|
||||
}
|
||||
export async function createChild(host: any, tx: any, options: any) {
|
||||
const {parentAgent:parent, definition} = options
|
||||
const child = options.fork
|
||||
? await tx.forkConversation(options.fork.conversationId, options.fork.at, {ownership:options.owner})
|
||||
: await tx.createConversation({ownership:options.owner})
|
||||
await configure(tx, child.id, childConfiguration(options))
|
||||
Object.assign(await tx.doc(host.metadata, child.id), {label:options.label, title:options.title})
|
||||
if (definition) (await tx.doc(Profile, child.id)).definition = definition.name
|
||||
if (definition?.summaries === false) await host.want({name:'conversations'})?.setAutomaticSummaries(tx, child.id, false)
|
||||
return child
|
||||
}
|
||||
/** Standing refreshes and new children apply the same definition and tool denials. */
|
||||
export function childConfiguration(options: any) {
|
||||
const {parentAgent:parent, definition} = options
|
||||
const inherited = parent.tools.map((tool: any) => tool.name)
|
||||
const names = (definition ? toolsFor(definition, inherited) : inherited).filter((name: string) => !options.excludeTools?.includes(name))
|
||||
return {
|
||||
tools:parent.tools.filter((tool: any) => names.includes(tool.name)),
|
||||
extensions:parent.extensions.filter((extension: any) => !options.excludeExtensions?.includes(extension.name)),
|
||||
instructions:definition?.instructions ?? parent.instructions ?? null,
|
||||
model:options.model ?? definition?.runs?.model ?? parent.model ?? null,
|
||||
thinkingLevel:options.thinking ?? definition?.runs?.thinking ?? definition?.thinking ?? parent.thinkingLevel ?? null,
|
||||
cwd:options.cwd ?? parent.cwd ?? '/src/sausage',
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
/** Temporary world-file resolver until the models service supplies definitions. */
|
||||
import { discoverAgents, parseFamilies } from '../core/agents.ts'
|
||||
export async function discover(host: any, cwd = '/src/sausage') {
|
||||
const root = '/src/sausage'
|
||||
const families = parseFamilies(await host.fs.read(`${root}/agents/models.json`))
|
||||
const models = host.models().getModels().map((model: any) => `${model.provider}/${model.id}`)
|
||||
const files: {path:string; text:string}[] = []
|
||||
for (const directory of [`${root}/agents`, `${cwd}/.sausage/agents`]) {
|
||||
let entries
|
||||
try { entries = await host.fs.list(directory) } catch (error) { if ((error as any).code === 'not_found') continue; throw error }
|
||||
for (const entry of entries) if (entry.kind === 'file' && entry.name.endsWith('.md')) {
|
||||
const path = `${directory}/${entry.name}`
|
||||
files.push({path, text:await host.fs.read(path)})
|
||||
}
|
||||
}
|
||||
return discoverAgents(files, root, cwd, families, models)
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
/**
|
||||
* Where a page opens when no conversation is chosen (a first load, a new chapter, a remembered one that is gone):
|
||||
* @ultimate's conversation, once the shared state and the standing agents are known; 'create' when @ultimate has no
|
||||
* conversation yet, as on a fresh preview. The root is only its history.
|
||||
*/
|
||||
export const landing = (
|
||||
loaded: boolean,
|
||||
chosen: unknown,
|
||||
standing: Readonly<Record<string, number>> | null | undefined,
|
||||
): number | 'create' | undefined => {
|
||||
if (!loaded || chosen !== undefined || standing === undefined) return undefined
|
||||
return standing?.ultimate ?? 'create'
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
/** Stored provenance distinguishes child instructions and reports from ordinary user text. */
|
||||
export type AgentOrigin = {
|
||||
body: string
|
||||
agent: string
|
||||
failed?: boolean
|
||||
sender: number
|
||||
entry?: number
|
||||
kind: 'instruction' | 'report'
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
/** Browser-safe identities for child starts and durable indices. */
|
||||
import { defineEvent } from '@earendil-works/sausage-events'
|
||||
export const Started = defineEvent<{id:number; agent:string; model?:string; thinking?:string; parent?:number; kind:'child'|'standing'|'branch'}>('subagents.started')
|
||||
export type Run = {model?:string; adhoc?:true}
|
||||
export const RunsDoc = {kind:'subagents.runs'}
|
||||
export const StandingDoc = {kind:'ultimate.standing-agents'}
|
||||
@@ -0,0 +1,841 @@
|
||||
/** Durable child coordination. Reporter delivery and report are separate, replay-safe phases. */
|
||||
import { BACKGROUND_CONTEXT, withCancel } from '@earendil-works/chord/context'
|
||||
import type { Context } from '@earendil-works/chord'
|
||||
import { bool, int, obj, oneOf, opt, str } from '@earendil-works/pi-schema'
|
||||
import type { AssistantMessage } from '@earendil-works/pi-ai'
|
||||
import { AgentDoc, AssistantEntry, configure, ToolResultEntry, defineExtension, defineTask, defineTool, LiveDoc, InboxDoc, GenerationTask, hook, ROOT_CONVERSATION_ID, section } from '@earendil-works/pi-durable'
|
||||
import type { ConversationId, Cursor, EntryId, EntryRecord, TaskId, ToolExecutionApi, SubmissionRecord } from '@earendil-works/pi-durable'
|
||||
import { Subagents, Standing, Profile, Runs, SubagentsApi } from './api.ts'
|
||||
import { createChild, childConfiguration, started } from './children.ts'
|
||||
import { discover } from './definitions.ts'
|
||||
export const docs = [Subagents, Standing, Profile, Runs]
|
||||
export const runsLabel = (runs: any) => runs ? `${runs.model.provider}/${runs.model.modelId}${runs.thinking ? ` · ${runs.thinking}` : ''}` : ''
|
||||
const onCommit = (harness: any, callback: any) => {
|
||||
const observation = harness.observeCommits({advance: callback, end: () => {}})
|
||||
return () => observation.detach()
|
||||
}
|
||||
const Anchor = defineTask<null, { phase: 'done' }, null>({
|
||||
name: 'ultimate.subagent-anchor',
|
||||
version: 1,
|
||||
initial: () => ({ phase: 'done' }),
|
||||
phases: {
|
||||
done: (_task, runtime, context) =>
|
||||
runtime.commit(
|
||||
() => ({ status: 'terminal', outcome: { status: 'completed', result: null } }),
|
||||
context,
|
||||
),
|
||||
},
|
||||
abort: (_task, runtime, context) =>
|
||||
runtime.commit(() => ({ status: 'terminal', outcome: { status: 'aborted' } }), context),
|
||||
})
|
||||
const textOf = (message: AssistantMessage | undefined) =>
|
||||
(message?.content ?? []).flatMap((part) => (part.type === 'text' ? [part.text] : [])).join('')
|
||||
// A report is keyed by its answer, independently of the tool call that may return it.
|
||||
const answerRequest = (answer: EntryId) => `subagent-answer:${answer}`
|
||||
/** An aborted or failed turn keeps its partial message, often only an empty thinking block: not an answer. */
|
||||
const answered = (message: AssistantMessage | undefined) =>
|
||||
message !== undefined &&
|
||||
message.stopReason !== 'aborted' &&
|
||||
message.stopReason !== 'error' &&
|
||||
textOf(message).trim() !== ''
|
||||
const unanswered = (message: AssistantMessage | undefined) =>
|
||||
`ended its turn without an answer${message?.stopReason === 'aborted' || message?.stopReason === 'error' ? ` (${message.stopReason})` : ''}`
|
||||
const answerText = (message: AssistantMessage | undefined) =>
|
||||
`Returned model: ${message?.provider && message?.model ? `${message.provider}/${message.model}` : 'unknown'}\n${textOf(message)}`
|
||||
const reportText = (id: ConversationId, agent: string, message: AssistantMessage | undefined) =>
|
||||
answered(message)
|
||||
? `[subagent ${id} (${agent}) answered] ${answerText(message)}`
|
||||
: `[subagent ${id} failed: ${unanswered(message)}]`
|
||||
/** What a foreground start, message or wait returns for a child's answer entry. */
|
||||
const replyText = (id: ConversationId, message: AssistantMessage | undefined) =>
|
||||
answered(message)
|
||||
? answerText(message)
|
||||
: `Subagent ${id} ${unanswered(message)}. Message it to continue.`
|
||||
function createReporter() {
|
||||
return defineTask<
|
||||
{
|
||||
conversationId: ConversationId
|
||||
text: string
|
||||
steer: boolean
|
||||
sourceEntry?: EntryId
|
||||
answer?: EntryId
|
||||
},
|
||||
{ phase: 'deliver' } | { phase: 'report'; report?: string; answer?: EntryId },
|
||||
null
|
||||
>({
|
||||
name: 'ultimate.subagent-reporter',
|
||||
version: 1,
|
||||
initial: (input) =>
|
||||
input.answer === undefined
|
||||
? { phase: 'deliver' }
|
||||
: { phase: 'report', report: input.text, answer: input.answer },
|
||||
phases: {
|
||||
deliver: async (task, runtime, context) => {
|
||||
try {
|
||||
const child = (await runtime.conversation(task.input.conversationId, context))!
|
||||
const content = task.input.text
|
||||
const settled = await (
|
||||
await child.submit(
|
||||
{
|
||||
type: 'input',
|
||||
content,
|
||||
entry: {
|
||||
kind: 'subagents.instruction',
|
||||
data: {
|
||||
sender: runtime.conversationId,
|
||||
kind: 'instruction',
|
||||
agent: (await runtime.snapshot(Subagents, runtime.conversationId, context))!
|
||||
.agents[String(child.id)].agent,
|
||||
body: content,
|
||||
...(task.input.sourceEntry ? { entry: task.input.sourceEntry } : {}),
|
||||
},
|
||||
},
|
||||
whenBusy: task.input.steer ? 'steer' : 'followUp',
|
||||
requestId: `subagent:${task.id}`,
|
||||
},
|
||||
context,
|
||||
)
|
||||
).wait(context)
|
||||
await runtime.commit(async (tx) => {
|
||||
const state = await tx.doc(Subagents, runtime.conversationId)
|
||||
const action = Object.values(state.actions).find(
|
||||
(candidate) => candidate.reporter === task.id,
|
||||
)!
|
||||
action.status = settled.status
|
||||
const agent = state.agents[String(child.id)]
|
||||
if (settled.status === 'done' && settled.type === 'input') {
|
||||
action.answer = settled.answer
|
||||
if (agent.reported.includes(settled.answer))
|
||||
return { status: 'running', checkpoint: { phase: 'report' } }
|
||||
agent.reported.push(settled.answer)
|
||||
const answer = (await tx.entry(AssistantEntry, settled.answer))?.model?.[0] as
|
||||
| AssistantMessage
|
||||
| undefined
|
||||
return {
|
||||
status: 'running',
|
||||
checkpoint: {
|
||||
phase: 'report',
|
||||
report: reportText(child.id, agent.agent, answer),
|
||||
answer: settled.answer,
|
||||
},
|
||||
}
|
||||
}
|
||||
// Provider failures belong to the watchdog, which marks the action `given-up` once it reports
|
||||
// instead of retrying. Either way recover() delivers the answer of a later watchdog retry (or
|
||||
// manual continuation).
|
||||
if (
|
||||
settled.status === 'unanswered' &&
|
||||
(settled.reason === 'model_error' || settled.reason === 'no_model')
|
||||
) {
|
||||
action.status = 'failed'
|
||||
return { status: 'running', checkpoint: { phase: 'report' } }
|
||||
}
|
||||
if (settled.status === 'unanswered' && settled.reason !== 'aborted') {
|
||||
const failure = settled.reason
|
||||
return { status: 'running', checkpoint: { phase: 'report', report: failure } }
|
||||
}
|
||||
// A parent stop aborts this task before the input, so this abort came from the child's
|
||||
// own conversation: whatever answers its next input there still belongs to the parent.
|
||||
if (settled.status === 'unanswered') action.status = 'interrupted'
|
||||
return { status: 'running', checkpoint: { phase: 'report' } }
|
||||
}, context)
|
||||
} catch (error) {
|
||||
if (runtime.signal.aborted) throw error
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
await runtime.commit(async (tx) => {
|
||||
const state = await tx.doc(Subagents, runtime.conversationId)
|
||||
const action = Object.values(state.actions).find(
|
||||
(candidate) => candidate.reporter === task.id,
|
||||
)!
|
||||
action.status = 'failed'
|
||||
return {
|
||||
status: 'running',
|
||||
checkpoint: {
|
||||
phase: 'report',
|
||||
report: message,
|
||||
},
|
||||
}
|
||||
}, context)
|
||||
}
|
||||
},
|
||||
report: async (task, runtime, context) => {
|
||||
const { report, answer } = task.state.checkpoint
|
||||
if (report !== undefined) {
|
||||
const parent = (await runtime.conversation(runtime.conversationId, context))!
|
||||
const agent = (await runtime.snapshot(Subagents, runtime.conversationId, context))!
|
||||
.agents[String(task.input.conversationId)].agent
|
||||
let message: AssistantMessage | undefined
|
||||
if (answer !== undefined)
|
||||
await runtime.commit(async (tx) => {
|
||||
message = (await tx.entry(answer))?.model?.[0] as AssistantMessage | undefined
|
||||
}, context)
|
||||
const failed = !answered(message)
|
||||
const body =
|
||||
answer === undefined
|
||||
? `Failed: ${report}`
|
||||
: failed
|
||||
? `Failed: ${unanswered(message)}`
|
||||
: answerText(message)
|
||||
await parent.submit(
|
||||
{
|
||||
type: 'input',
|
||||
content:
|
||||
answer === undefined
|
||||
? `[subagent ${task.input.conversationId} failed: ${report}]`
|
||||
: report,
|
||||
entry: {
|
||||
kind: 'subagents.report',
|
||||
data: {
|
||||
sender: task.input.conversationId,
|
||||
kind: 'report',
|
||||
agent,
|
||||
failed,
|
||||
body,
|
||||
...(answer ? { entry: answer } : {}),
|
||||
},
|
||||
},
|
||||
whenBusy: 'steer',
|
||||
requestId:
|
||||
answer === undefined ? `subagent-report:${task.id}` : answerRequest(answer),
|
||||
},
|
||||
context,
|
||||
)
|
||||
}
|
||||
await runtime.commit(
|
||||
() => ({ status: 'terminal', outcome: { status: 'completed', result: null } }),
|
||||
context,
|
||||
)
|
||||
},
|
||||
},
|
||||
abort: (task, runtime, context) =>
|
||||
runtime.commit(async (tx) => {
|
||||
const state = await tx.doc(Subagents, runtime.conversationId)
|
||||
const action = Object.values(state.actions).find(
|
||||
(candidate) => candidate.reporter === task.id,
|
||||
)!
|
||||
action.status = 'unanswered'
|
||||
return { status: 'terminal', outcome: { status: 'aborted' } }
|
||||
}, context),
|
||||
})
|
||||
}
|
||||
|
||||
/** Details name the child and whether it answers in the background, for the call's renderer. */
|
||||
const details = (
|
||||
conversationId: ConversationId,
|
||||
background: boolean,
|
||||
delivery?: { yielded?: true; answerId?: EntryId },
|
||||
) => ({
|
||||
conversationId,
|
||||
...(background ? { background } : {}),
|
||||
...delivery,
|
||||
})
|
||||
const reply = (text: string, info?: ReturnType<typeof details>) => ({
|
||||
output: [{ type: 'text' as const, text }],
|
||||
...(info === undefined ? {} : { details: info }),
|
||||
})
|
||||
|
||||
/** Stop only these local waits; child and reporter ownership never changes. */
|
||||
async function waitInterruptibly(
|
||||
api: ToolExecutionApi,
|
||||
id: ConversationId,
|
||||
reporters: TaskId[],
|
||||
idle: boolean,
|
||||
context: Context,
|
||||
) {
|
||||
await api.commit(async (tx) => {
|
||||
await tx.doc(InboxDoc, api.conversationId)
|
||||
}, context)
|
||||
const local = withCancel(context)
|
||||
const inbox = (await api.watchDoc(InboxDoc, api.conversationId, local.context))!
|
||||
let wake!: () => void
|
||||
const steering = new Promise<void>((resolve) => {
|
||||
wake = resolve
|
||||
})
|
||||
const inspect = (value: typeof inbox.value) => {
|
||||
if (value?.items.some((item) => item.mode === 'steer')) wake()
|
||||
}
|
||||
inbox.start(async (call) => {
|
||||
if (call.kind === 'value') inspect(call.value)
|
||||
})
|
||||
inspect(inbox.value)
|
||||
const child = (await api.conversation(id, context))!
|
||||
const ready = async () => {
|
||||
if (idle && (await api.snapshot(LiveDoc, id, context))?.run !== undefined) return false
|
||||
for (const reporter of reporters) {
|
||||
const task = await api.getTask(reporter, context)
|
||||
if (
|
||||
task?.state.status !== 'terminal' &&
|
||||
!(
|
||||
task?.state.status === 'running' &&
|
||||
(task.state.checkpoint as { phase?: string } | null)?.phase === 'report'
|
||||
)
|
||||
)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
const completion = async () => {
|
||||
for (const reporter of reporters) await api.waitForTask(reporter, local.context)
|
||||
if (idle) await child.waitForIdle(local.context)
|
||||
}
|
||||
try {
|
||||
const result = await Promise.race([
|
||||
completion().then(() => true),
|
||||
steering.then(async () => {
|
||||
// Our reporter may have just queued the steer. Prefer its completed answer.
|
||||
if (await ready()) {
|
||||
await completion()
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}),
|
||||
])
|
||||
return result
|
||||
} finally {
|
||||
local.cancel()
|
||||
await inbox.stop()
|
||||
}
|
||||
}
|
||||
|
||||
export default function subagents(host: any) {
|
||||
host.implement(SubagentsApi, {
|
||||
createChild: async (tx: any, options: any) => {
|
||||
const child = await createChild(host, tx, options)
|
||||
await started(host, tx, child, options.definition?.name ?? options.label, 'child')
|
||||
return child
|
||||
},
|
||||
residents: async () => (await discover(host)).definitions.filter((d: any) => d.resident).map(({name, description}: any) => ({name, description})),
|
||||
standing: async (name: string, ctx: Context) => {
|
||||
const found = await discover(host)
|
||||
const definition = found.definitions.find((d: any) => d.name === name && d.resident)
|
||||
if (!definition) throw new Error(`Unknown standing agent @${name}`)
|
||||
const root = (await host.harness().conversation(ROOT_CONVERSATION_ID, ctx))!
|
||||
const parentAgent = await root.agent(ctx)
|
||||
return host.harness().commit(async (tx: any) => {
|
||||
const index = await tx.doc(Standing, ROOT_CONVERSATION_ID)
|
||||
const existing = index[name] && await tx.conversation(index[name])
|
||||
const lifecycle = host.want({name:'conversations'})
|
||||
if (existing) {
|
||||
await lifecycle?.setStanding(tx, existing.id, true)
|
||||
await lifecycle?.setAutomaticSummaries(tx, existing.id, false)
|
||||
const own = await tx.doc(AgentDoc, existing.id)
|
||||
await configure(tx, existing.id, childConfiguration({definition, parentAgent:{...parentAgent, model:own.model ?? parentAgent.model, thinkingLevel:own.thinkingLevel ?? parentAgent.thinkingLevel}}))
|
||||
return existing.id
|
||||
}
|
||||
const child = await createChild(host, tx, {owner: {kind:'ownerless'}, definition, parentAgent, metadata:host.metadata, label:name, title:`@${name}`})
|
||||
index[name] = child.id
|
||||
await lifecycle?.setStanding(tx, child.id, true)
|
||||
await lifecycle?.setAutomaticSummaries(tx, child.id, false)
|
||||
await started(host, tx, child, name, 'standing')
|
||||
return child.id
|
||||
}, ctx)
|
||||
},
|
||||
})
|
||||
const spawnDefinition = async (args: any, definitions: any[]) => {
|
||||
const definition = definitions.find(d => d.name === args.agent)
|
||||
if (!definition) throw new Error(`Unknown agent: ${args.agent}. Available definitions: ${definitions.map(d => d.name).join(', ')}`)
|
||||
if ((args.isolate ?? definition.isolate ?? 'none') !== 'none') throw new Error('worktree isolation requires the workspaces extension (P3-k)')
|
||||
return definition
|
||||
}
|
||||
const Reporter = createReporter()
|
||||
let unsubscribe: (() => void) | undefined
|
||||
let recovery = Promise.resolve()
|
||||
const recover = async (submission: SubmissionRecord) => {
|
||||
if (
|
||||
submission.type !== 'input' ||
|
||||
submission.status !== 'done' ||
|
||||
submission.requestId?.startsWith('subagent:')
|
||||
)
|
||||
return
|
||||
await host.harness().commit(async (tx) => {
|
||||
const child = await tx.conversation(submission.conversationId)
|
||||
const parent = child?.owner?.conversationId
|
||||
if (parent === undefined) return
|
||||
const state = await tx.doc(Subagents, parent)
|
||||
const agent = state.agents[String(submission.conversationId)]
|
||||
if (!agent || agent.reported.includes(submission.answer)) return
|
||||
const childActions = Object.values(state.actions).filter(
|
||||
(candidate) => candidate.conversationId === submission.conversationId,
|
||||
)
|
||||
// A one-way message from the parent, or its reply to the child's own question, has no
|
||||
// reporter and no reply: without this, the answer it starts never reaches the parent,
|
||||
// which stays idle. A pending reporter whose input was answered in the same run delivers
|
||||
// that answer itself.
|
||||
const input = await tx.entry(submission.entry)
|
||||
const message = input?.data as { from?: unknown; reply?: unknown } | undefined
|
||||
let messaged =
|
||||
message?.from === parent &&
|
||||
((input?.kind === 'board.message' && message.reply === false) ||
|
||||
input?.kind === 'board.reply')
|
||||
for (const action of childActions) {
|
||||
if (!messaged || action.status !== undefined) continue
|
||||
const sibling = await tx.submissionByRequest(
|
||||
submission.conversationId,
|
||||
`subagent:${action.reporter}`,
|
||||
)
|
||||
if (sibling?.status === 'done' && sibling.answer === submission.answer) messaged = false
|
||||
}
|
||||
// Watchdog continuations also answer failed one-way messages, which have no reporter action.
|
||||
if (!messaged && !submission.requestId?.startsWith('watchdog-retry:')) {
|
||||
const action = childActions.at(-1)
|
||||
const interrupted = action?.status === 'interrupted'
|
||||
if (
|
||||
!action ||
|
||||
(!['unanswered', 'failed', 'given-up'].includes(action.status ?? '') && !interrupted)
|
||||
)
|
||||
return
|
||||
const failed = await tx.submissionByRequest(
|
||||
submission.conversationId,
|
||||
`subagent:${action.reporter}`,
|
||||
)
|
||||
// IDs share one allocation order; an input withdrawn before placement has no entry.
|
||||
if (
|
||||
failed?.status !== 'unanswered' ||
|
||||
(failed.reason === 'aborted') !== interrupted ||
|
||||
(failed.entry === undefined && !interrupted) ||
|
||||
submission.entry <= (failed.entry ?? failed.id)
|
||||
)
|
||||
return
|
||||
}
|
||||
const answer = (await tx.entry(AssistantEntry, submission.answer))?.model?.[0] as
|
||||
| AssistantMessage
|
||||
| undefined
|
||||
const reporter = await tx.createTask(
|
||||
Reporter,
|
||||
{
|
||||
conversationId: submission.conversationId,
|
||||
text: reportText(submission.conversationId, agent.agent, answer),
|
||||
steer: false,
|
||||
answer: submission.answer,
|
||||
},
|
||||
{ conversationId: parent, ownership: { kind: 'conversation' }, background: true },
|
||||
)
|
||||
agent.reported.push(submission.answer)
|
||||
state.actions[`subagent-answer:${submission.answer}`] = {
|
||||
conversationId: submission.conversationId,
|
||||
reporter,
|
||||
answer: submission.answer,
|
||||
status: 'done',
|
||||
}
|
||||
}, BACKGROUND_CONTEXT)
|
||||
}
|
||||
host.onClose(async () => {
|
||||
unsubscribe?.()
|
||||
unsubscribe = undefined
|
||||
await recovery
|
||||
})
|
||||
host.onOpen(async () => {
|
||||
const found = await discover(host)
|
||||
if (!found.errors.length) await host.harness().commit(async (tx: any) => {
|
||||
const index = await tx.doc(Standing, ROOT_CONVERSATION_ID)
|
||||
for (const name of Object.keys(index)) if (!found.definitions.some((d: any) => d.name === name && d.resident)) {
|
||||
const id = index[name]
|
||||
delete index[name]
|
||||
if (await tx.conversation(id)) {
|
||||
delete (await tx.doc(host.metadata, id)).label
|
||||
await host.want({name:'conversations'})?.setStanding(tx, id, false)
|
||||
}
|
||||
}
|
||||
}, BACKGROUND_CONTEXT)
|
||||
unsubscribe?.()
|
||||
// Not a reactor: subagent answers settle as Pi submission records, not domain events.
|
||||
unsubscribe = onCommit(host.harness(), (publication) => {
|
||||
for (const change of publication.changes) {
|
||||
if (change.type !== 'submission') continue
|
||||
recovery = recovery
|
||||
.then(() => recover(change.value))
|
||||
.catch((error: unknown) => {
|
||||
host.notice(`Subagent answer recovery failed: ${String(error)}`)
|
||||
})
|
||||
}
|
||||
})
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await host.storage.scanSubmissions({ status: 'done' }, 500, cursor, BACKGROUND_CONTEXT)
|
||||
for (const submission of page.items) await recover(submission)
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
})
|
||||
return defineExtension({
|
||||
name: 'subagents',
|
||||
tasks: [Anchor, Reporter],
|
||||
hooks: [
|
||||
hook(GenerationTask, {
|
||||
afterTools: async (_assistant, results, api, context) => {
|
||||
// Tool settlement is not a receipt. Only committed outer results may acknowledge.
|
||||
await host.harness().commit(async (tx) => {
|
||||
const answers = new Set<EntryId>()
|
||||
for (const id of results) {
|
||||
const message = (await tx.entry(ToolResultEntry, id))?.model?.[0]
|
||||
if (message?.role !== 'toolResult' || message.isError) continue
|
||||
const info = message.details as
|
||||
| { answerId?: EntryId; subagentAnswers?: EntryId[] }
|
||||
| undefined
|
||||
if (message.toolName === 'subagent' && info?.answerId !== undefined)
|
||||
answers.add(info.answerId)
|
||||
if (message.toolName === 'codemode')
|
||||
for (const answer of info?.subagentAnswers ?? []) answers.add(answer)
|
||||
}
|
||||
if (!answers.size) return
|
||||
const submissions = []
|
||||
for (const answer of answers)
|
||||
submissions.push({
|
||||
answer,
|
||||
submission: await tx.submissionByRequest(api.conversationId, answerRequest(answer)),
|
||||
})
|
||||
const queued = submissions.flatMap(({ submission }) =>
|
||||
submission?.status === 'queued' ? [submission.id] : [],
|
||||
)
|
||||
const inbox = await tx.doc(InboxDoc, api.conversationId)
|
||||
inbox.items = inbox.items.filter((item) => !queued.includes(item.id))
|
||||
for (const { answer, submission } of submissions) {
|
||||
// A reporter not yet submitted must also see this receipt through request deduplication.
|
||||
if (!submission)
|
||||
await tx.createSubmission({
|
||||
conversationId: api.conversationId,
|
||||
requestId: answerRequest(answer),
|
||||
type: 'input',
|
||||
status: 'unanswered',
|
||||
reason: 'subagent-tool-delivered',
|
||||
})
|
||||
else if (submission.status === 'queued')
|
||||
tx.settleSubmission(submission.id, {
|
||||
status: 'unanswered',
|
||||
reason: 'subagent-tool-delivered',
|
||||
})
|
||||
}
|
||||
}, context)
|
||||
},
|
||||
}),
|
||||
],
|
||||
sections: [
|
||||
section('subagent_state', async (input, context) => {
|
||||
const state = await input.read.snapshot(Subagents, input.conversationId, context)
|
||||
const running: string[] = []
|
||||
const pending = new Set<ConversationId>()
|
||||
for (const action of Object.values(state?.actions ?? {}))
|
||||
if (
|
||||
action.status === undefined &&
|
||||
(await host.harness().getTask(action.reporter, context))?.state.status !== 'terminal'
|
||||
)
|
||||
pending.add(action.conversationId)
|
||||
const queued: string[] = []
|
||||
for (const agent of Object.values(state?.agents ?? {})) {
|
||||
const label = `${agent.conversationId} (${agent.agent})`
|
||||
if (
|
||||
(await input.read.snapshot(LiveDoc, agent.conversationId, context))?.run !== undefined
|
||||
)
|
||||
running.push(label)
|
||||
else if (pending.has(agent.conversationId)) queued.push(label)
|
||||
}
|
||||
if (!running.length && !queued.length) return undefined
|
||||
return [
|
||||
...(running.length ? [`Still running subagents: ${running.join(', ')}.`] : []),
|
||||
...(queued.length ? [`Pending subagent inputs: ${queued.join(', ')}.`] : []),
|
||||
'A yielded wait is not a child answer. Use message, stop, or wait with these IDs; parent stop does not stop them. Queued inputs are delivered only at the normal boundary; do not act on them early.',
|
||||
].join(' ')
|
||||
}),
|
||||
section('subagents', async (input) => {
|
||||
const usesDefinitions = input.agent.tools.some((tool) =>
|
||||
['subagent', 'branch', 'automation'].includes(tool.name),
|
||||
)
|
||||
if (!usesDefinitions) return undefined
|
||||
const discovered = await discover(host, input.agent.cwd)
|
||||
const definitions = discovered.definitions
|
||||
.map((agent) => {
|
||||
const runs = runsLabel(agent.runs)
|
||||
return `${agent.name}${runs ? ` (${runs})` : ''}: ${agent.description}`
|
||||
})
|
||||
.join('\n')
|
||||
return `Available definitions:\n${definitions}`
|
||||
}),
|
||||
],
|
||||
tools: [
|
||||
defineTool({
|
||||
name: 'subagent',
|
||||
description:
|
||||
'Delegate to a named definition. start returns an answer or a background child ID. message steers a busy child unless steer is false. wait returns its latest answer. Foreground waits yield on parent steering. stop ends current child work. list shows your children. Parent stop leaves children independent. fork inherits history before this call. Only isolate none is available until workspaces arrive. Background answers steer the parent exactly once.',
|
||||
// Call-id receipts deduplicate starts/messages; memoized targets and guards protect stop/workspace replays.
|
||||
replay: 'safe',
|
||||
parameters: obj({
|
||||
action: oneOf(['start', 'message', 'wait', 'stop', 'list']),
|
||||
agent: opt(
|
||||
str({
|
||||
description:
|
||||
'For start: an Available definitions name.',
|
||||
}),
|
||||
),
|
||||
label: opt(
|
||||
str({
|
||||
description:
|
||||
'For start: an optional label for the child.',
|
||||
}),
|
||||
),
|
||||
task: opt(str()),
|
||||
background: opt(bool()),
|
||||
fork: opt(bool()),
|
||||
id: opt(int({ minimum: 1 })),
|
||||
text: opt(str()),
|
||||
steer: opt(bool()),
|
||||
isolate: opt(oneOf(['none', 'worktree'])),
|
||||
}),
|
||||
execute: async (args, api, context) => {
|
||||
const cached = await api.memo<ReturnType<typeof reply>>('result', context)
|
||||
if (cached) return cached
|
||||
const finish = (text: string, info?: ReturnType<typeof details>) =>
|
||||
api.memo('result', reply(text, info), context)
|
||||
const index = (await api.snapshot(Subagents, api.conversationId, context)) ?? {
|
||||
agents: {},
|
||||
actions: {},
|
||||
}
|
||||
if (args.action === 'list') {
|
||||
const rows = await Promise.all(
|
||||
Object.values(index.agents).map(async (agent) => ({
|
||||
...agent,
|
||||
busy:
|
||||
(await api.snapshot(LiveDoc, agent.conversationId, context))?.run !== undefined,
|
||||
})),
|
||||
)
|
||||
return finish(JSON.stringify(rows.map(({ reported: _reported, ...row }) => row)))
|
||||
}
|
||||
if (
|
||||
args.action !== 'start' &&
|
||||
(!args.id || !Object.hasOwn(index.agents, String(args.id)))
|
||||
)
|
||||
throw new Error('id must identify a child of this conversation')
|
||||
if (args.action === 'start' || args.action === 'message') {
|
||||
const text = args.action === 'start' ? args.task : args.text
|
||||
if (!text?.trim())
|
||||
throw new Error(
|
||||
`${args.action} requires ${args.action === 'start' ? 'task' : 'text'}`,
|
||||
)
|
||||
const parent = await api.agent(context)
|
||||
const existingAction = index.actions[api.callId]
|
||||
const definitions =
|
||||
args.action === 'start' && !existingAction
|
||||
? (
|
||||
await discover(host, parent.cwd)
|
||||
).definitions
|
||||
: undefined
|
||||
const definition = definitions && (await spawnDefinition(args, definitions))
|
||||
if (definition?.tools?.some((name) => !parent.tools.some((tool) => tool.name === name)))
|
||||
throw new Error('Definition tools must be a subset of parent tools')
|
||||
const label = definition && (args.label ?? definition.name)
|
||||
const task = await api.getTask(api.taskId, context)
|
||||
// Pi upstream: @earendil-works/pi-durable@1.5.0 src/harness/tool.ts — internal ToolTaskInput.assistant locates the calling assistant. Drop when ToolExecutionApi exposes the calling assistant entry ID.
|
||||
const assistant = (task?.input as { assistant: EntryId } | undefined)?.assistant
|
||||
let forkAt: EntryId | undefined
|
||||
if (definition && args.fork) {
|
||||
const conversation = (await host.harness().conversation(api.conversationId, context))!
|
||||
const { entries } = await conversation.context(context)
|
||||
forkAt = entries[entries.findIndex((entry) => entry.id === assistant) - 1]?.id
|
||||
if (forkAt === undefined) throw new Error('No parent history before this call')
|
||||
}
|
||||
|
||||
const action = await api.commit(async (tx) => {
|
||||
const state = await tx.doc(Subagents, api.conversationId)
|
||||
if (state.actions[api.callId]) return { ...state.actions[api.callId] }
|
||||
let conversationId = args.id as ConversationId
|
||||
if (definition) {
|
||||
const owner = await tx.createTask(Anchor, null, {
|
||||
ownership: { kind: 'conversation' },
|
||||
background: true,
|
||||
})
|
||||
const child = await createChild(host, tx, {
|
||||
owner: { kind: 'task', taskId: owner },
|
||||
definition,
|
||||
parentAgent: parent,
|
||||
metadata: host.metadata,
|
||||
label: label!,
|
||||
title: text.trim().slice(0, 80),
|
||||
excludeTools: definition.tools?.includes('subagent') ? [] : ['subagent'],
|
||||
excludeExtensions: definition.tools?.includes('subagent') ? [] : ['subagents'],
|
||||
...(forkAt === undefined
|
||||
? {}
|
||||
: { fork: { conversationId: api.conversationId, at: forkAt } }),
|
||||
cwd: parent.cwd ?? '/src/sausage',
|
||||
})
|
||||
conversationId = child.id
|
||||
await started(host, tx, child, label!, 'child')
|
||||
const runs = await tx.doc(Runs, ROOT_CONVERSATION_ID)
|
||||
const { model: ran } = await tx.doc(AgentDoc, child.id)
|
||||
runs[String(child.id)] = {
|
||||
...(ran ? { model: ran.modelId } : {}),
|
||||
}
|
||||
state.agents[String(child.id)] = {
|
||||
conversationId,
|
||||
agent: label!,
|
||||
background: args.background === true,
|
||||
reported: [],
|
||||
}
|
||||
}
|
||||
const reporter = await tx.createTask(
|
||||
Reporter,
|
||||
{
|
||||
conversationId,
|
||||
text,
|
||||
steer: args.steer !== false,
|
||||
...(assistant ? { sourceEntry: assistant } : {}),
|
||||
},
|
||||
{ ownership: { kind: 'conversation' }, background: true },
|
||||
)
|
||||
const createdAction = { conversationId, reporter }
|
||||
state.actions[api.callId] = createdAction
|
||||
return createdAction
|
||||
}, context)
|
||||
const isolateInfo = ''
|
||||
const background =
|
||||
index.agents[String(action.conversationId)]?.background ?? args.background === true
|
||||
const info = details(action.conversationId, background)
|
||||
await api.details(info, context)
|
||||
if (background)
|
||||
return finish(
|
||||
`Started ${args.action === 'message' ? 'message to' : 'subagent'} ${action.conversationId}.${isolateInfo}`,
|
||||
info,
|
||||
)
|
||||
if (
|
||||
!(await waitInterruptibly(
|
||||
api,
|
||||
action.conversationId,
|
||||
[action.reporter],
|
||||
false,
|
||||
context,
|
||||
))
|
||||
)
|
||||
return finish(
|
||||
`Wait yielded for subagent ${action.conversationId}; work continues.${isolateInfo}`,
|
||||
details(action.conversationId, background, { yielded: true }),
|
||||
)
|
||||
const settled = (await api.snapshot(Subagents, api.conversationId, context))!.actions[
|
||||
api.callId
|
||||
]
|
||||
const answer =
|
||||
settled.answer === undefined
|
||||
? undefined
|
||||
: await api.commit((tx) => tx.entry(AssistantEntry, settled.answer!), context)
|
||||
return finish(
|
||||
(answer
|
||||
? replyText(
|
||||
action.conversationId,
|
||||
answer.model?.[0] as AssistantMessage | undefined,
|
||||
)
|
||||
: `Subagent ${action.conversationId}: ${settled.status}`) + isolateInfo,
|
||||
details(
|
||||
action.conversationId,
|
||||
background,
|
||||
settled.answer === undefined ? undefined : { answerId: settled.answer },
|
||||
),
|
||||
)
|
||||
}
|
||||
const id = args.id as ConversationId
|
||||
const info = details(id, index.agents[String(id)].background)
|
||||
await api.details(info, context)
|
||||
if (args.action === 'stop') {
|
||||
// Capture identities before cancellation; replay must leave later reuse alone.
|
||||
const live = await api.snapshot(LiveDoc, id, context)
|
||||
const state = await api.snapshot(Subagents, api.conversationId, context)
|
||||
const watchdogs = await api.commit(async (tx) => {
|
||||
const tasks = []
|
||||
let cursor: Cursor | undefined
|
||||
do {
|
||||
const page = await tx.scanTasks(
|
||||
{ conversationId: api.conversationId, kind: 'watchdog.failure' },
|
||||
500,
|
||||
cursor,
|
||||
)
|
||||
tasks.push(
|
||||
...page.items.filter(
|
||||
(task) => (task.input as { conversation?: number }).conversation === id,
|
||||
),
|
||||
)
|
||||
cursor = page.next
|
||||
} while (cursor)
|
||||
return tasks.map((task) => ({
|
||||
task: task.id,
|
||||
request: `watchdog-retry:${(task.input as { failedTask: number }).failedTask}`,
|
||||
}))
|
||||
}, context)
|
||||
const target = await api.memo(
|
||||
'stopTarget',
|
||||
{
|
||||
run: live?.run?.taskId ?? null,
|
||||
inputs: live?.run?.inputs ?? [],
|
||||
watchdogs,
|
||||
reporters: Object.values(state?.actions ?? {}).flatMap((action) =>
|
||||
action.conversationId === id ? [action.reporter] : [],
|
||||
),
|
||||
},
|
||||
context,
|
||||
)
|
||||
// Join delivery invocations before inspecting admissions that may have raced stop.
|
||||
// Capture watchdog identities too: Stop must not restart the child after its delay.
|
||||
for (const reporter of [
|
||||
...target.reporters,
|
||||
...target.watchdogs.map((task) => task.task),
|
||||
]) {
|
||||
await host.harness().abortTask(reporter, context)
|
||||
await api.waitForTask(reporter, context)
|
||||
}
|
||||
const submissions = await api.commit(async (tx) => {
|
||||
return Promise.all(
|
||||
[
|
||||
...target.reporters.map((reporter) => `subagent:${reporter}`),
|
||||
...target.watchdogs.map((task) => task.request),
|
||||
].map((request) => tx.submissionByRequest(id, request)),
|
||||
)
|
||||
}, context)
|
||||
for (const submission of submissions)
|
||||
if (submission?.status === 'queued')
|
||||
await host.harness().abortSubmission(submission.id, context, id)
|
||||
const inputs = new Set([
|
||||
...target.inputs,
|
||||
...submissions.flatMap((submission) => (submission ? [submission.id] : [])),
|
||||
])
|
||||
// A tool round can hand off between the snapshot and abort; follow only captured inputs.
|
||||
while (true) {
|
||||
const run = (await api.snapshot(LiveDoc, id, context))?.run
|
||||
if (
|
||||
!run ||
|
||||
(run.taskId !== target.run && !run.inputs.some((input) => inputs.has(input)))
|
||||
)
|
||||
break
|
||||
await host.harness().abortTask(run.taskId, context)
|
||||
await api.waitForTask(run.taskId, context)
|
||||
}
|
||||
return finish(`Stopped ${id}.`, info)
|
||||
}
|
||||
const reporters = Object.values(index.actions).flatMap((action) =>
|
||||
action.conversationId === id ? [action.reporter] : [],
|
||||
)
|
||||
if (!(await waitInterruptibly(api, id, reporters, true, context)))
|
||||
return finish(
|
||||
`Wait yielded for subagent ${id}; work continues.`,
|
||||
details(id, index.agents[String(id)].background, { yielded: true }),
|
||||
)
|
||||
const answer = await api.commit(async (tx) => {
|
||||
let cursor: Cursor | undefined
|
||||
let latest: EntryRecord | undefined
|
||||
do {
|
||||
const page = await tx.scanEntries({ conversationId: id }, 256, cursor)
|
||||
for (const entry of page.items)
|
||||
if (AssistantEntry.is(entry) && latest === undefined) latest = entry
|
||||
cursor = page.next
|
||||
} while (cursor !== undefined)
|
||||
if (!latest) return undefined
|
||||
const agent = (await tx.doc(Subagents, api.conversationId)).agents[String(id)]
|
||||
if (!agent.reported.includes(latest.id)) agent.reported.push(latest.id)
|
||||
return latest
|
||||
}, context)
|
||||
return finish(
|
||||
replyText(id, answer?.model?.[0] as AssistantMessage | undefined),
|
||||
details(
|
||||
id,
|
||||
index.agents[String(id)].background,
|
||||
answer === undefined ? undefined : { answerId: answer.id },
|
||||
),
|
||||
)
|
||||
},
|
||||
}),
|
||||
],
|
||||
})
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js" },
|
||||
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js && esbuild extensions-test.ts --bundle --format=esm --platform=browser --outfile=extensions-test-bundle.js && esbuild events-test.ts --bundle --format=esm --platform=browser --outfile=events-test-bundle.js && esbuild subagents-test.ts --bundle --format=esm --platform=browser --outfile=subagents-test-bundle.js" },
|
||||
"dependencies": {
|
||||
"@earendil-works/pi-durable": "1.5.0",
|
||||
"@earendil-works/pi-ai": "1.5.0",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,110 @@
|
||||
/** Scripted world-loaded coordination, steering, ownership, and crash recovery. */
|
||||
import { BACKGROUND_CONTEXT as context, awaitWithContext } from '@earendil-works/chord/context';
|
||||
import { createModels } from '@earendil-works/pi-ai/models';
|
||||
import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
|
||||
import { Harness, createRegistry, defineExtension, defineTool, LiveDoc, InboxDoc } from '@earendil-works/pi-durable';
|
||||
import { obj } from '@earendil-works/pi-schema';
|
||||
import { loadExtensions, fs } from './extensions';
|
||||
import { conversationChains } from './environment';
|
||||
import { openStorage } from './sqlite';
|
||||
const assert = (value:any,message:string)=>{if(!value)throw new Error(message);};
|
||||
const until = async(test:()=>Promise<any>|any)=>{const deadline=Date.now()+15000;while(!await test()){if(Date.now()>deadline)throw new Error('Subagent deadline');await new Promise(r=>setTimeout(r,0));}};
|
||||
const models=createModels(), faux=fauxProvider({tokensPerSecond:100000}); models.setProvider(faux.provider);
|
||||
const registry=createRegistry(), storage=await openStorage(globalThis.storagePath);
|
||||
const harness=await Harness.open(storage,{models,registry,settings:{get extensions(){return registry.snapshot().installed();}}},context);
|
||||
const root=await harness.root(context,{agent:{model:{provider:'faux',modelId:'faux-1'},cwd:'/src/sausage'}});
|
||||
const gate=Promise.withResolvers<void>(), parentGate=Promise.withResolvers<void>(), holding=new Set<number>();
|
||||
const held=defineExtension({name:'hold',tools:[defineTool({name:'hold',description:'Wait for test gate',parameters:obj({}),replay:'safe',execute:async(_args,api,ctx)=>{holding.add(api.conversationId);await awaitWithContext(api.conversationId===root.id?parentGate.promise:gate.promise,ctx);return {output:[{type:'text',text:'released'}]};}})]});
|
||||
if(globalThis.subagentsMode==='kill') {
|
||||
const install=registry.install.bind(registry);
|
||||
registry.install=(extension:any)=>install(extension.name!=='subagents'?extension:{...extension,tasks:extension.tasks.map((task:any)=>task.definition.name!=='ultimate.subagent-reporter'?task:{definition:{...task.definition,phases:{...task.definition.phases,report:async(_task:any,_runtime:any,ctx:any)=>{await fs.write('/report-ready','yes');await awaitWithContext(new Promise(()=>{}),ctx);}}}})});
|
||||
}
|
||||
const route=(transcript:any):any=>{
|
||||
faux.appendResponses([route]);
|
||||
const last=transcript.messages.findLast((message:any)=>message.role!=='system');
|
||||
const text=typeof last.content==='string'?last.content:last.content.filter((part:any)=>part.type==='text').map((part:any)=>part.text).join('');
|
||||
if(last.role==='user'&&text.startsWith('CALL ')) return fauxAssistantMessage([fauxToolCall('subagent',JSON.parse(text.slice(5)))],{stopReason:'toolUse'});
|
||||
if(last.role==='user'&&text==='HOLD') return fauxAssistantMessage([fauxToolCall('hold',{})],{stopReason:'toolUse'});
|
||||
return fauxAssistantMessage(last.role==='user'&&text.startsWith('[subagent')?'Report received':`Child: ${text}`);
|
||||
};
|
||||
faux.setResponses([route]);
|
||||
const loader=await loadExtensions(harness,registry,storage,models,context,[held]);
|
||||
assert(!loader.state().error,`Extension load: ${loader.state().error}`);
|
||||
const snapshot=await fs.read('/src/sausage/.git/sausage-last-good');
|
||||
const {Subagents}=await import(`file:///run/sausage/extensions/${snapshot}/extensions/subagents/api.ts`);
|
||||
const chain=conversationChains(harness,context);
|
||||
const children=async()=>Object.values((await harness.snapshot(Subagents,root.id,context))?.agents??{}) as any[];
|
||||
const reports=async()=>(await root.context(context)).entries.filter((entry:any)=>entry.kind==='subagents.report');
|
||||
const start=async(args:any)=>root.submit({type:'input',content:`CALL ${JSON.stringify(args)}`},context);
|
||||
const call=async(args:any)=>{await(await start(args)).wait(context);return (await root.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;};
|
||||
const text=(message:any)=>message?.content.filter((part:any)=>part.type==='text').map((part:any)=>part.text).join('');
|
||||
harness.resume();
|
||||
try {
|
||||
if(globalThis.subagentsMode==='kill') {
|
||||
await call({action:'start',agent:'scout',task:'crash answer',background:true});
|
||||
await until(async()=>{try{return await fs.read('/report-ready')}catch{return false}});
|
||||
await new Promise(()=>{});
|
||||
} else if(globalThis.subagentsMode==='resume') {
|
||||
await until(async()=>(await reports()).length===1);
|
||||
await harness.waitForIdle(context);
|
||||
assert((await children()).length===1,'Restart must retain one child');
|
||||
assert((await reports()).length===1,'Restart must report exactly once');
|
||||
assert(JSON.stringify(await reports()).includes('crash answer'),'Restart lost answer');
|
||||
} else {
|
||||
const service=globalThis.subagentService;
|
||||
assert((await service.residents(context))[0].name==='scout','Resident definitions');
|
||||
const standing=await service.standing('scout',context);
|
||||
assert(await service.standing('scout',context)===standing,'Standing refresh must reuse its conversation');
|
||||
const standingRecord=await harness.commit((tx:any)=>tx.conversation(standing),context);
|
||||
assert(!standingRecord.owner,'Standing must be ownerless');
|
||||
const scoutFile=await fs.read('/src/sausage/agents/scout.md');
|
||||
await fs.write('/src/sausage/agents/scout.md',scoutFile.replace('Scout instructions','Updated instructions'));
|
||||
await service.standing('scout',context);
|
||||
assert((await (await harness.conversation(standing,context))!.agent(context)).instructions==='Updated instructions','Standing refresh must reread instructions');
|
||||
const answer=await call({action:'start',agent:'scout',task:'facts'});
|
||||
assert(text(answer)==='Returned model: faux/faux-1\nChild: facts',`Foreground answer: ${JSON.stringify(answer)}`);
|
||||
const [first]=await children();
|
||||
assert(JSON.stringify(await chain(first.conversationId))===JSON.stringify([String(first.conversationId),String(root.id)]),'Anchor ownership chain');
|
||||
const record=await harness.commit((tx:any)=>tx.conversation(first.conversationId),context);
|
||||
const anchor=await harness.getTask(record!.owner!.taskId,context);
|
||||
assert(anchor?.kind==='ultimate.subagent-anchor','Child owner must be anchor');
|
||||
const child=(await harness.conversation(first.conversationId,context))!;
|
||||
assert(!(await child.agent(context)).tools.some((tool:any)=>tool.name==='subagent'),'Tool denials');
|
||||
assert((await root.agent(context)).tools.some((tool:any)=>tool.name==='subagent'),'Parent tools remain unchanged');
|
||||
assert((await reports()).length===0,'Foreground receipt must suppress duplicate report');
|
||||
assert(text(await call({action:'message',id:first.conversationId,text:'next'})).includes('Child: next'),'Foreground message');
|
||||
assert(text(await call({action:'wait',id:first.conversationId})).includes('Child: next'),'Wait latest answer');
|
||||
await (await child.submit({type:'input',content:'one way',entry:{kind:'board.message',data:{from:root.id,reply:false}}},context)).wait(context);
|
||||
await until(async()=>(await reports()).length===1);await harness.waitForIdle(context);
|
||||
assert(JSON.stringify(await reports()).includes('one way'),'One-way recover lost answer');
|
||||
await call({action:'start',agent:'scout',task:'HOLD',background:true});
|
||||
const second=(await children()).at(-1)!;
|
||||
await until(()=>holding.has(second.conversationId));
|
||||
const wait=await start({action:'wait',id:second.conversationId});
|
||||
await until(async()=>(await harness.snapshot(LiveDoc,root.id,context))?.run);
|
||||
await root.submit({type:'input',content:'change direction',whenBusy:'steer'},context);
|
||||
await wait.wait(context);
|
||||
const yielded=(await root.context(context)).entries.filter((entry:any)=>entry.kind==='pi.tool-result').at(-1)?.model?.[0] as any;
|
||||
assert(yielded.details?.yielded,'Wait must yield on steering');
|
||||
assert((await harness.snapshot(LiveDoc,second.conversationId,context))?.run,'Steering must leave child alive');
|
||||
await root.submit({type:'input',content:'HOLD'},context);
|
||||
await until(()=>holding.has(root.id));
|
||||
gate.resolve();
|
||||
await until(async()=>(await harness.snapshot(InboxDoc,root.id,context))?.items.some((item:any)=>item.mode==='steer'&&item.entry?.kind==='subagents.report'));
|
||||
parentGate.resolve();
|
||||
await until(async()=>(await reports()).length===2);await harness.waitForIdle(context);
|
||||
assert((await reports()).length===2,'Background answer must report exactly once');
|
||||
const listed=JSON.parse(text(await call({action:'list'})));assert(listed.length===2,'List children');
|
||||
// Use a fresh unresolved gate for the stop case through a child input that never resolves.
|
||||
const stopGate=Promise.withResolvers<void>();
|
||||
registry.install(defineExtension({name:'hold',tools:[defineTool({name:'hold',description:'Stop gate',parameters:obj({}),replay:'safe',execute:async(_a,api,ctx)=>{holding.add(api.conversationId);await awaitWithContext(stopGate.promise,ctx);return {output:[]};}})]}));
|
||||
holding.clear();
|
||||
await call({action:'start',agent:'scout',task:'HOLD',background:true});
|
||||
const third=(await children()).at(-1)!;await until(()=>holding.has(third.conversationId));
|
||||
assert((await harness.snapshot(LiveDoc,third.conversationId,context))?.run,'Stop fixture must hold an active child');
|
||||
assert(text(await call({action:'stop',id:third.conversationId}))===`Stopped ${third.conversationId}.`,'Stop result');
|
||||
assert(!(await harness.snapshot(LiveDoc,third.conversationId,context))?.run,'Stop must end child');
|
||||
assert((await reports()).length===2,'Stop must suppress aborted report');
|
||||
assert(text(await call({action:'start',agent:'scout',task:'no',isolate:'worktree'})).includes('requires the workspaces extension'),'Worktree seam');
|
||||
}
|
||||
} finally {gate.resolve();parentGate.resolve();await loader.close();await harness.close(context);}
|
||||
@@ -0,0 +1,162 @@
|
||||
//! World-loaded subagents and a SIGKILL at the reporter phase boundary.
|
||||
use sausage::{harness::run_module, world::World};
|
||||
use std::{
|
||||
path::Path,
|
||||
process::Command,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
const BUNDLE: &str = include_str!("../js/subagents-test-bundle.js");
|
||||
fn prepare(dir: &Path) -> anyhow::Result<()> {
|
||||
let disk = dir.join("world");
|
||||
let repo = disk.join("src/sausage");
|
||||
if repo.join(".git").exists() {
|
||||
return Ok(());
|
||||
}
|
||||
let source = Path::new(env!("CARGO_MANIFEST_DIR")).join("..");
|
||||
let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
|
||||
let userland = source.join("userland");
|
||||
if !userland.join("cwd-init/init.o").exists() {
|
||||
assert!(
|
||||
Command::new("nu")
|
||||
.args(["--no-config-file", "cwd-init/build.nu"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
Command::new("cargo")
|
||||
.args(["build", "--release", "-p", "git", "--manifest-path"])
|
||||
.arg(userland.join("Cargo.toml"))
|
||||
.args(["--target", "wasm32-wasip2", "--target-dir"])
|
||||
.arg(target)
|
||||
.args(["-j", "2", "--locked"])
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
std::fs::create_dir_all(disk.join("bin"))?;
|
||||
std::fs::copy(
|
||||
target.join("wasm32-wasip2/release/git.wasm"),
|
||||
disk.join("bin/git.wasm"),
|
||||
)?;
|
||||
for directory in ["extensions/subagents", "extensions/core", "agents"] {
|
||||
std::fs::create_dir_all(repo.join(directory))?;
|
||||
}
|
||||
for entry in std::fs::read_dir(source.join("extensions/subagents"))? {
|
||||
let entry = entry?;
|
||||
if entry.path().extension().is_some_and(|ext| ext == "ts") {
|
||||
std::fs::copy(
|
||||
entry.path(),
|
||||
repo.join("extensions/subagents").join(entry.file_name()),
|
||||
)?;
|
||||
}
|
||||
}
|
||||
std::fs::create_dir_all(repo.join("extensions/probe"))?;
|
||||
std::fs::write(
|
||||
repo.join("extensions/probe/server.ts"),
|
||||
"export default host => { globalThis.subagentService = host.need({name:'subagents'}); return {name:'probe'}; }",
|
||||
)?;
|
||||
std::fs::copy(
|
||||
source.join("extensions/core/agents.ts"),
|
||||
repo.join("extensions/core/agents.ts"),
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/models.json"),
|
||||
r#"{"default":{"model":"faux/faux-1"}}"#,
|
||||
)?;
|
||||
std::fs::write(
|
||||
repo.join("agents/scout.md"),
|
||||
"---\nname: scout\ndescription: Search\nresident: true\ndenyTools: [subagent]\nisolate: none\n---\nScout instructions\n",
|
||||
)?;
|
||||
for args in [
|
||||
vec!["init", "-b", "main"],
|
||||
vec!["add", "."],
|
||||
vec!["commit", "--no-gpg-sign", "-m", "subagents fixture"],
|
||||
] {
|
||||
assert!(
|
||||
Command::new("git")
|
||||
.args(args)
|
||||
.current_dir(&repo)
|
||||
.env("GIT_AUTHOR_NAME", "Test")
|
||||
.env("GIT_AUTHOR_EMAIL", "test@example.com")
|
||||
.env("GIT_COMMITTER_NAME", "Test")
|
||||
.env("GIT_COMMITTER_EMAIL", "test@example.com")
|
||||
.status()?
|
||||
.success()
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
async fn run(dir: &Path, mode: &str) -> anyhow::Result<()> {
|
||||
let world = World::open(dir)?;
|
||||
prepare(dir)?;
|
||||
let exec = sausage::process::exec(World {
|
||||
root: world.root.clone(),
|
||||
host_dir: world.host_dir.clone(),
|
||||
processes: world.processes.clone(),
|
||||
})?;
|
||||
run_module(
|
||||
world,
|
||||
exec,
|
||||
&format!("globalThis.subagentsMode={mode:?};\n{BUNDLE}"),
|
||||
)
|
||||
.await
|
||||
}
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn coordination() -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
run(dir.path(), "normal").await
|
||||
}
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn restart_child() -> anyhow::Result<()> {
|
||||
if let Ok(dir) = std::env::var("SUBAGENTS_DIR") {
|
||||
run(Path::new(&dir), &std::env::var("SUBAGENTS_MODE")?).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
#[test]
|
||||
fn sigkill_between_deliver_and_report() -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
World::open(dir.path())?;
|
||||
prepare(dir.path())?;
|
||||
let spawn = |mode: &str| {
|
||||
Command::new(std::env::current_exe().unwrap())
|
||||
.args(["--exact", "restart_child", "--nocapture"])
|
||||
.env("SUBAGENTS_DIR", dir.path())
|
||||
.env("SUBAGENTS_MODE", mode)
|
||||
.spawn()
|
||||
};
|
||||
let mut child = spawn("kill")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
while !dir.path().join("world/report-ready").exists() {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
anyhow::bail!("Child exited before report: {status}");
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Reporter did not reach report");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
for _ in 0..2 {
|
||||
let mut child = spawn("resume")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
loop {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
assert!(status.success());
|
||||
break;
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Report recovery timed out");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user