diff --git a/packages/core/src/conversations/events.ts b/packages/core/src/conversations/events.ts index da713a39..bd07bbe0 100644 --- a/packages/core/src/conversations/events.ts +++ b/packages/core/src/conversations/events.ts @@ -146,16 +146,12 @@ export type ConversationEvent = Data.TaggedEnum<{ /** The asking turn stopped waiting; the answer will resume it instead. */ CollaborationStoppedWaiting: CollaborationChange; /** The collaborator's reply was recorded as the answer. */ - CollaborationAnswered: CollaborationChange & { - readonly askingAgentId: string; - /** The asking turn had stopped waiting, so nothing is reading the answer yet. */ - readonly askerMovedOn: boolean; - }; + CollaborationAnswered: CollaborationChange & AskerResume; /** * The collaboration ended unanswered: the routine run it worked for ended, * or the collaborator's turn ended without an answer. */ - CollaborationFailed: CollaborationChange; + CollaborationFailed: CollaborationChange & AskerResume; /** `person` read further into the thread, at `readAt`: every message created up to `readThrough`. */ ThreadRead: { readonly threadId: string; @@ -248,3 +244,14 @@ export interface CollaborationChange { readonly parentMessageId: string; readonly collaboration: CollaborationPart; } + +/** Whether a collaboration's end is owed to the agent that asked, in a turn of its own. */ +export interface AskerResume { + /** + * The asking agent, when its turn stopped waiting and so nothing is reading + * how the collaboration ended. Null while the asking tool still waits, and + * when the routine run the collaboration worked for ended, which takes no + * more turns. + */ + readonly askerToResume: string | null; +} diff --git a/packages/core/src/conversations/floor/floor-control.ts b/packages/core/src/conversations/floor/floor-control.ts index f6cddd5d..50214f57 100644 --- a/packages/core/src/conversations/floor/floor-control.ts +++ b/packages/core/src/conversations/floor/floor-control.ts @@ -19,8 +19,8 @@ export interface Interface { readonly giveFloor: (committed: FloorMessage) => Effect.Effect; /** * Asks for the collaborator's turn when a collaboration opens, for the - * asking agent's when an answer arrives after it stopped waiting, and for - * the bot's in a thread another bot started and briefed it in. + * asking agent's when a collaboration ends after it stopped waiting, and + * for the bot's in a thread another bot started and briefed it in. */ readonly handler: DomainEvents.Handler; } @@ -79,11 +79,11 @@ export const make = Effect.gen(function* () { reason: "brief", }); case "CollaborationAnswered": - // While it was still waiting, the asking tool reads the answer itself. - return event.askerMovedOn + case "CollaborationFailed": + return event.askerToResume ? turns.ask({ threadId: event.parentThreadId, - agentId: event.askingAgentId, + agentId: event.askerToResume, triggerMessageId: event.parentMessageId, reason: "resume", }) diff --git a/packages/core/src/conversations/sql.ts b/packages/core/src/conversations/sql.ts index 4500fa50..ccf399de 100644 --- a/packages/core/src/conversations/sql.ts +++ b/packages/core/src/conversations/sql.ts @@ -384,8 +384,8 @@ export const threadParticipant = pgTable( /** * Why an agent was given a turn: a person mentioned it or replied to it, it was the default, - * it was asked to collaborate, a routine ran, or a collaboration answer - * resumed it, it opened its setup interview, or a bot started its thread + * it was asked to collaborate, a routine ran, a collaboration it stopped + * waiting on ended, it opened its setup interview, or a bot started its thread * for a person and briefed it. */ export type TurnReason = diff --git a/packages/core/src/conversations/tools/collaborate/collaborations.test.ts b/packages/core/src/conversations/tools/collaborate/collaborations.test.ts index 86380fd6..338afb29 100644 --- a/packages/core/src/conversations/tools/collaborate/collaborations.test.ts +++ b/packages/core/src/conversations/tools/collaborate/collaborations.test.ts @@ -458,6 +458,10 @@ describe.skipIf(!process.env.DATABASE_URL)("collaboration, against Postgres", as expect(willRetry).toBe(false); expect(await statusOf(opened.collaboration.id)).toBe("failed"); + // It told the person it would follow up, so it gets a turn to say what happened. + expect( + await runOnPostgres(waitingTurns({ threadId: rootThreadId, agentId: host.id })), + ).toMatchObject([{ agentId: host.id, triggerMessageId: reply.messageId, reason: "resume" }]); }); it("keeps the collaboration open while the failed turn runs again", async () => { @@ -500,6 +504,17 @@ describe.skipIf(!process.env.DATABASE_URL)("collaboration, against Postgres", as }); }); + it("gives the asker no turn when the routine run its collaboration worked for ends", async () => { + const opened = await collaborations.open({ from: from(), to: helper.name, brief: "Look" }); + await collaborations.collectAnswer(opened.collaboration.id); + + await collaborations.failUnder([rootThreadId]); + + expect(await runOnPostgres(waitingTurns({ threadId: rootThreadId, agentId: host.id }))).toEqual( + [], + ); + }); + it("tells the asking model a collaboration failed while it waited, rather than that an answer is coming", async () => { const tool = collaborateTool({ from: { diff --git a/packages/core/src/conversations/tools/collaborate/collaborations.ts b/packages/core/src/conversations/tools/collaborate/collaborations.ts index 89a46ee0..3652e4ac 100644 --- a/packages/core/src/conversations/tools/collaborate/collaborations.ts +++ b/packages/core/src/conversations/tools/collaborate/collaborations.ts @@ -61,8 +61,8 @@ export interface Interface { /** * Fails a collaboration once its collaborator's turn in its thread ends for * good without answering: failed with no retry, cancelled, or given up - * on. Otherwise the asking agent, told the answer would follow, waits for - * one that never comes. + * on. An asking agent that stopped waiting gets a turn to say so, as it + * told the person it would follow up. */ readonly handler: DomainEvents.Handler; /** @@ -148,7 +148,7 @@ export const make = Effect.gen(function* () { operation("collectAnswer", repository.stopWaiting(collaborationId)), answer: ({ threadId, answer }) => - operation("answer", transaction(repository.answer(threadId, answer)).pipe(Effect.asVoid)), + operation("answer", transaction(repository.answer(threadId, answer))), handler: (events) => Effect.forEach(events.flatMap(unansweredBy), (ended) => repository.failUnanswered(ended), { diff --git a/packages/core/src/conversations/tools/collaborate/repository.ts b/packages/core/src/conversations/tools/collaborate/repository.ts index 34f12f54..1d5d34cd 100644 --- a/packages/core/src/conversations/tools/collaborate/repository.ts +++ b/packages/core/src/conversations/tools/collaborate/repository.ts @@ -51,9 +51,9 @@ export interface Records { readonly stopWaiting: (collaborationId: string) => Effect.Effect; /** * Records `answer` as the reply to the collaboration whose thread is - * `childThreadId`. `undefined` when there is none outstanding. + * `childThreadId`, if one is outstanding. */ - readonly answer: (childThreadId: string, answer: string) => Effect.Effect; + readonly answer: (childThreadId: string, answer: string) => Effect.Effect; /** * Fails the collaborations asked for in these threads that are still * waiting or pending, because the routine run they work for ended. One @@ -165,21 +165,15 @@ export const makeRecords = Effect.gen(function* () { Effect.gen(function* () { const current = yield* locked(eq(collaboration.childThreadId, childThreadId)); if (!current || current.row.status === "answered" || current.row.status === "failed") { - return undefined; + return; } const updated = yield* writeStatus(current.row.id, { status: "answered", answer }); yield* emit([ ConversationEvent.CollaborationAnswered({ ...collaborationChange(updated, current.collaboratorName), - askingAgentId: current.askingAgentId, - askerMovedOn: current.row.status === "pending", + askerToResume: askerToResume(current), }), ]); - return { - collaboration: updated, - askingAgentId: current.askingAgentId, - askerMovedOn: current.row.status === "pending", - }; }), ), ), @@ -206,9 +200,10 @@ export const makeRecords = Effect.gen(function* () { ); const failed = yield* Effect.forEach(unanswered, (row) => Effect.map(writeStatus(row.collaboration.id, { status: "failed" }), (updated) => - ConversationEvent.CollaborationFailed( - collaborationChange(updated, row.collaboratorName), - ), + ConversationEvent.CollaborationFailed({ + ...collaborationChange(updated, row.collaboratorName), + askerToResume: null, + }), ), ); yield* emit(failed); @@ -245,9 +240,10 @@ export const makeRecords = Effect.gen(function* () { if (stillAnswering) return; const updated = yield* writeStatus(current.row.id, { status: "failed" }); yield* emit([ - ConversationEvent.CollaborationFailed( - collaborationChange(updated, current.collaboratorName), - ), + ConversationEvent.CollaborationFailed({ + ...collaborationChange(updated, current.collaboratorName), + askerToResume: askerToResume(current), + }), ]); }), ), @@ -265,14 +261,6 @@ export type WaitOutcome = | { readonly _tag: "MovedOn" } | { readonly _tag: "Failed" }; -/** A collaboration answered, and who asked for it. */ -export interface Answered { - readonly collaboration: schema.CollaborationRow; - readonly askingAgentId: string; - /** The asking turn had stopped waiting, so nothing is reading the answer yet. */ - readonly askerMovedOn: boolean; -} - /** * The collaboration, locked for update so a transition is serialised, with * the names its announcement and the asker's resume need. @@ -303,6 +291,14 @@ const locked = (by: SQL | undefined) => : undefined, ); +/** The asking agent, if it stopped waiting for the collaboration about to end. */ +function askerToResume(current: { + row: schema.CollaborationRow; + askingAgentId: string; +}): string | null { + return current.row.status === "pending" ? current.askingAgentId : null; +} + const writeStatus = ( collaborationId: string, change: Pick & diff --git a/packages/core/src/conversations/tools/collaborate/tool.ts b/packages/core/src/conversations/tools/collaborate/tool.ts index 57db27b6..8928bb86 100644 --- a/packages/core/src/conversations/tools/collaborate/tool.ts +++ b/packages/core/src/conversations/tools/collaborate/tool.ts @@ -104,7 +104,7 @@ export function collaborateTool({ return { status: "pending", threadId: collaboration.threadId, - note: `${collaborator.name} has not answered yet. Tell the person you have asked and will follow up when the answer arrives.`, + note: `${collaborator.name} is still working on it. When they answer, or it turns out they cannot, you get a new turn to tell the person. For now, tell the person you have asked ${collaborator.name} and will follow up.`, }; }, }); diff --git a/packages/core/src/conversations/turns/context.ts b/packages/core/src/conversations/turns/context.ts index ce558d99..eb3b1713 100644 --- a/packages/core/src/conversations/turns/context.ts +++ b/packages/core/src/conversations/turns/context.ts @@ -441,7 +441,7 @@ function whyYou(context: TurnContext): string { case "collaboration": return "It is your turn because another agent asked you to collaborate."; case "resume": - return "It is your turn because an answer you were waiting on has arrived."; + return "It is your turn because a collaboration you were waiting on has ended. Tell the person what came of it."; case "brief": // With no other bot named, the bot briefs itself, so the brief reads as its own reply. return context.messages[0] && isOwnMessage(context.messages[0], context.agent.id) diff --git a/packages/demo/src/showcase/stage.ts b/packages/demo/src/showcase/stage.ts index 4dd09611..f6a5edad 100644 --- a/packages/demo/src/showcase/stage.ts +++ b/packages/demo/src/showcase/stage.ts @@ -730,8 +730,7 @@ const collaborate = ( parentThreadId: reply.thread.id, parentMessageId: reply.messageId, collaboration: toCollaborationPart(row, collaborator.name), - askingAgentId: reply.bot.id, - askerMovedOn: false, + askerToResume: null, }), ]); }),