From d126d46f534cf31c4b6193f759bd07c42c7e46b5 Mon Sep 17 00:00:00 2001 From: Dnyaneshvn Date: Mon, 13 Jul 2026 10:57:28 +0530 Subject: [PATCH] Link spin-off threads back to the conversation that spawned them When the agent messages someone new mid-conversation ("go ask Alex..."), nothing recorded why the new thread existed, so the reply came back to a fresh session with no link home. - record the active conversation per turn, the same seam as channel-hint.ts - on a send to a new recipient, store a spawn link (parent session, reply target, and the outbound message as the why) - on the reply, inject a short context block and set modelParentSessionKey to the parent so it inherits context and can relay the answer back Fixes #10 --- src/inbound/session.ts | 85 ++++++--- src/spawn-context.ts | 143 ++++++++++++++ src/tools/send-email.ts | 6 + src/tools/send-imessage.ts | 6 + src/tools/send-sms.ts | 6 + tests/inbound/spinoff-thread.test.ts | 272 +++++++++++++++++++++++++++ 6 files changed, 492 insertions(+), 26 deletions(-) create mode 100644 src/spawn-context.ts create mode 100644 tests/inbound/spinoff-thread.test.ts diff --git a/src/inbound/session.ts b/src/inbound/session.ts index b9020a6..ed6455a 100644 --- a/src/inbound/session.ts +++ b/src/inbound/session.ts @@ -26,6 +26,12 @@ import { import type { InkboxRuntime, PluginLogger } from "../client.js"; import type { ResolvedInkboxAccount } from "../accounts.js"; import { recordInboundChannelHint } from "../channel-hint.js"; +import { + buildSpawnContextBlock, + consumeSpawnLink, + normalizeRecipientKey, + setActiveConversation, +} from "../spawn-context.js"; import { classifySendRejection, claimDeliveryFailure, @@ -1940,11 +1946,21 @@ async function dispatchInboundTurn( const baseSessionKey = route.sessionKey; const effectiveSessionKey = opts.turn.mode === "voice" ? voiceSessionKey(route.agentId, opts.turn) : baseSessionKey; + // A spun-off thread: this reply is from someone the agent messaged during + // another conversation. Inherit that parent so the turn knows why it exists + // and can relay the answer back — see spawn-context.ts. + const isMessagingMode = + opts.turn.mode === "sms" || opts.turn.mode === "imessage" || opts.turn.mode === "email"; + const spawnLink = isMessagingMode ? consumeSpawnLink(opts.turn.remoteAddress) : undefined; + const spawnContextBlock = spawnLink ? buildSpawnContextBlock(spawnLink) : undefined; + const agentBody = spawnContextBlock ? `${spawnContextBlock}\n${opts.turn.body}` : opts.turn.body; + const modelParentSessionKey = + opts.turn.mode === "voice" ? baseSessionKey : spawnLink?.parentSessionKey; const { storePath, body } = buildEnvelope({ channel: "Inkbox", from: opts.turn.fromLabel, timestamp, - body: opts.turn.body, + body: agentBody, }); const conversationPrefix = opts.turn.mode === "imessage" ? "imessage" : "sms"; const smsReplyTarget = opts.turn.conversationId @@ -1975,7 +1991,7 @@ async function dispatchInboundTurn( agentId: route.agentId, accountId: routeAccountId, routeSessionKey: effectiveSessionKey, - ...(opts.turn.mode === "voice" ? { modelParentSessionKey: baseSessionKey } : {}), + ...(modelParentSessionKey ? { modelParentSessionKey } : {}), }, reply: { to: @@ -1999,9 +2015,9 @@ async function dispatchInboundTurn( }, message: { body, - bodyForAgent: opts.turn.body, - rawBody: opts.turn.body, - commandBody: opts.turn.body, + bodyForAgent: agentBody, + rawBody: agentBody, + commandBody: agentBody, envelopeFrom: opts.turn.fromLabel, }, extra: { @@ -2087,28 +2103,45 @@ async function dispatchInboundTurn( ); }, }; - await core.inbound.dispatchReply({ - cfg: opts.cfg as any, - channel: "inkbox", - accountId: opts.account.accountId, - agentId: route.agentId, - routeSessionKey: effectiveSessionKey, - storePath, - ctxPayload, - recordInboundSession: core.session.recordInboundSession, - dispatchReplyWithBufferedBlockDispatcher: - core.reply.dispatchReplyWithBufferedBlockDispatcher, - ...(replyOptions ? { replyOptions } : {}), - delivery, - replyPipeline: {}, - record: { - onRecordError: (error: unknown) => { - opts.logger?.warn?.( - `Inkbox session record failed: ${error instanceof Error ? error.message : String(error)}`, - ); + // Mark this conversation active while the agent runs, so a send tool it + // invokes to a new recipient parents the spin-off back to here. Only real + // messaging turns can be relayed back to; voice/warmup/external cannot. + if (isMessagingMode) { + setActiveConversation({ + sessionKey: baseSessionKey, + replyTarget: smsReplyTarget, + label: opts.turn.conversationLabel ?? opts.turn.fromLabel, + party: normalizeRecipientKey(opts.turn.remoteAddress), + }); + } + try { + await core.inbound.dispatchReply({ + cfg: opts.cfg as any, + channel: "inkbox", + accountId: opts.account.accountId, + agentId: route.agentId, + routeSessionKey: effectiveSessionKey, + storePath, + ctxPayload, + recordInboundSession: core.session.recordInboundSession, + dispatchReplyWithBufferedBlockDispatcher: + core.reply.dispatchReplyWithBufferedBlockDispatcher, + ...(replyOptions ? { replyOptions } : {}), + delivery, + replyPipeline: {}, + record: { + onRecordError: (error: unknown) => { + opts.logger?.warn?.( + `Inkbox session record failed: ${error instanceof Error ? error.message : String(error)}`, + ); + }, }, - }, - }); + }); + } finally { + if (isMessagingMode) { + setActiveConversation(undefined); + } + } } const REALTIME_CONSULT_TIMED_OUT = Symbol("realtime-consult-timed-out"); diff --git a/src/spawn-context.ts b/src/spawn-context.ts new file mode 100644 index 0000000..97bb037 --- /dev/null +++ b/src/spawn-context.ts @@ -0,0 +1,143 @@ +// Spin-off thread context (issue #10). +// +// When the agent messages someone new mid-conversation ("go ask Alex about +// X"), the send goes out through a send tool and Alex's eventual reply routes +// to a fresh session with no link back to the conversation that spawned it. +// +// This is the same in-process seam as channel-hint.ts: the inbound bridge +// records the conversation it is currently processing, the send tools read +// that to parent a spin-off, and the inbound bridge consumes the link when +// the new recipient replies — so the spawned turn knows why it exists, holds +// a reference to its parent thread, and can relay the answer back. + +// The conversation currently being processed. Set by the inbound bridge for +// the duration of a turn's agent run (same global-latest tradeoff as +// channel-hint.ts): a send tool invoked during the run reads this to know +// which conversation it is spinning off from. +export interface ActiveConversation { + // Session key of the conversation the agent is currently in. + sessionKey: string; + // Where a relayed answer goes back (conversation id or address). + replyTarget: string; + // Human label of the parent conversation, for the injected context block. + label: string; + // The parent's remote party, normalized — used to tell a spin-off (a new + // recipient) apart from a reply to the same person. + party?: string; +} + +let active: ActiveConversation | undefined; + +export function setActiveConversation(ctx: ActiveConversation | undefined): void { + active = ctx; +} + +export function getActiveConversation(): ActiveConversation | undefined { + return active; +} + +// A recorded spin-off: the new recipient's reply should inherit this parent. +export interface SpawnLink { + parentSessionKey: string; + parentReplyTarget: string; + parentLabel: string; + // The message the agent sent out, as the "why" the thread exists. + why: string; +} + +type StoredSpawnLink = SpawnLink & { recordedAt: number }; + +// A spin-off waiting on a reply is short-lived; bound by TTL and count so an +// abandoned outreach can't linger and mis-parent a much later, unrelated +// conversation with the same person. +const SPAWN_LINK_TTL_MS = 6 * 60 * 60 * 1000; +const SPAWN_LINK_MAX_ENTRIES = 500; +const WHY_SNIPPET_MAX_CHARS = 500; +const spawnLinks = new Map(); + +// Normalize a recipient into the same key an inbound reply will resolve to: +// email addresses lowercased, phone numbers reduced to their +digits. +export function normalizeRecipientKey(value: string | undefined): string | undefined { + const trimmed = value?.trim(); + if (!trimmed) { + return undefined; + } + if (trimmed.includes("@")) { + return trimmed.toLowerCase(); + } + const digits = trimmed.replace(/[^+\d]/g, ""); + return digits || trimmed; +} + +function prune(now = Date.now()): void { + for (const [key, link] of spawnLinks) { + if (now - link.recordedAt > SPAWN_LINK_TTL_MS) { + spawnLinks.delete(key); + } + } + while (spawnLinks.size > SPAWN_LINK_MAX_ENTRIES) { + const oldest = spawnLinks.keys().next().value as string | undefined; + if (oldest === undefined) break; + spawnLinks.delete(oldest); + } +} + +// Record that the currently-active conversation spun off a message to +// `recipient`. No-op when there is no active conversation, or when the +// recipient is the active conversation's own party (a reply, not a spin-off). +export function recordSpawnFromActive(params: { + recipient: string | undefined; + body: string | undefined; +}): void { + const parent = active; + if (!parent) { + return; + } + const key = normalizeRecipientKey(params.recipient); + if (!key || key === parent.party) { + return; + } + prune(); + spawnLinks.set(key, { + parentSessionKey: parent.sessionKey, + parentReplyTarget: parent.replyTarget, + parentLabel: parent.label, + why: (params.body ?? "").slice(0, WHY_SNIPPET_MAX_CHARS), + recordedAt: Date.now(), + }); +} + +// Consume the spawn link for a replying recipient, if any. One-shot: the +// first reply inherits the parent (context block + parent session), after +// which the spawned session carries the history itself. +export function consumeSpawnLink(recipient: string | undefined): SpawnLink | undefined { + prune(); + const key = normalizeRecipientKey(recipient); + if (!key) { + return undefined; + } + const link = spawnLinks.get(key); + if (!link) { + return undefined; + } + spawnLinks.delete(key); + const { recordedAt: _recordedAt, ...rest } = link; + return rest; +} + +// The context block prepended to a spun-off reply's turn: what the thread is +// for, and how to relay the answer back to the parent conversation. +export function buildSpawnContextBlock(link: SpawnLink): string { + return [ + `[inkbox:spinoff_thread parent=${JSON.stringify(link.parentLabel)} relay_to=${link.parentReplyTarget}]`, + `You started this conversation from another thread (${link.parentLabel}).`, + `You reached out here to: ${JSON.stringify(link.why)}`, + "When you have what you need, relay the answer back to that original conversation using your messaging tools; do not leave it waiting.", + ].join("\n"); +} + +// Test hook — the module-level stores persist across vitest cases otherwise. +export function resetSpawnContextForTest(): void { + active = undefined; + spawnLinks.clear(); +} diff --git a/src/tools/send-email.ts b/src/tools/send-email.ts index c352f2f..fefae6f 100644 --- a/src/tools/send-email.ts +++ b/src/tools/send-email.ts @@ -2,6 +2,7 @@ import { Type } from "typebox"; import type { InkboxRuntime } from "../client.js"; import { runTool, toolText, toolError } from "../errors.js"; import { checkOutboundRecipients } from "../allowlist.js"; +import { recordSpawnFromActive } from "../spawn-context.js"; // Outbound email — the primary write path for the email channel. export function registerSendEmail( @@ -51,6 +52,11 @@ export function registerSendEmail( bcc: params.bcc, inReplyToMessageId: params.inReplyToMessageId, }); + // A single-recipient email to a new address mid-conversation is a + // spin-off; link it back so the reply inherits the parent thread. + if (Array.isArray(params.to) && params.to.length === 1) { + recordSpawnFromActive({ recipient: params.to[0], body: params.bodyText ?? params.bodyHtml }); + } return toolText( `Sent email id=${msg.id} to=${params.to.join(",")} subject="${params.subject}"`, ); diff --git a/src/tools/send-imessage.ts b/src/tools/send-imessage.ts index 2ab5ad1..53a360a 100644 --- a/src/tools/send-imessage.ts +++ b/src/tools/send-imessage.ts @@ -2,6 +2,7 @@ import { Type } from "typebox"; import type { InkboxRuntime } from "../client.js"; import { runTool, toolText, toolError } from "../errors.js"; import { checkOutboundRecipient } from "../allowlist.js"; +import { recordSpawnFromActive } from "../spawn-context.js"; import { IMESSAGE_MAX_TEXT_CHARS, imessageTextTooLongMessage } from "../message-limits.js"; // Outbound iMessage — recipient-first channel: a person must have connected @@ -97,6 +98,11 @@ export function registerSendIMessage( ...(mediaUrls?.length ? { mediaUrls } : {}), ...(params.sendStyle ? { sendStyle: params.sendStyle } : {}), }); + // A send addressed to a new number (not an existing conversation) is a + // spin-off; link it back so the reply inherits the parent thread. + if (to) { + recordSpawnFromActive({ recipient: to, body: text || undefined }); + } const target = conversationId ? `conversation=${conversationId}` : `to=${to}`; return toolText( `Sent iMessage id=${msg.id} ${target} conversation_id=${msg.conversationId} status=${msg.status ?? "unknown"}`, diff --git a/src/tools/send-sms.ts b/src/tools/send-sms.ts index bb8d777..b18ac25 100644 --- a/src/tools/send-sms.ts +++ b/src/tools/send-sms.ts @@ -2,6 +2,7 @@ import { Type } from "typebox"; import type { InkboxRuntime } from "../client.js"; import { runTool, toolText, toolError } from "../errors.js"; import { checkOutboundRecipient } from "../allowlist.js"; +import { recordSpawnFromActive } from "../spawn-context.js"; import { SMS_MAX_TEXT_CHARS, smsTextTooLongMessage } from "../message-limits.js"; function normalizeRecipients(value: unknown): string[] | undefined { @@ -121,6 +122,11 @@ export function registerSendSms( : { to: toList!.length === 1 ? toList![0] : toList }), }; const msg = await identity.sendText(payload); + // A 1:1 send to a new number mid-conversation is a spin-off; link it + // back so the recipient's reply inherits the parent thread. + if (!hasConversation && toList && toList.length === 1) { + recordSpawnFromActive({ recipient: toList[0], body: params.text }); + } const target = formatTargetSummary(msg, params); const status = msg.deliveryStatus ?? "unknown"; return toolText( diff --git a/tests/inbound/spinoff-thread.test.ts b/tests/inbound/spinoff-thread.test.ts new file mode 100644 index 0000000..d1b3aaf --- /dev/null +++ b/tests/inbound/spinoff-thread.test.ts @@ -0,0 +1,272 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@inkbox/sdk", () => ({ verifyWebhook: vi.fn(() => true) })); + +vi.mock("openclaw/plugin-sdk/inbound-envelope", () => ({ + resolveInboundRouteEnvelopeBuilderWithRuntime: vi.fn(() => ({ + route: { + agentId: "main", + accountId: "default", + sessionKey: "agent:main:inkbox:direct:spawned", + }, + buildEnvelope: ({ body }: { body: string }) => ({ + storePath: "memory://inkbox/test", + body, + }), + })), +})); + +vi.mock("openclaw/plugin-sdk/realtime-voice", () => ({ + REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME: "consult_agent", + REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ: { + encoding: "g711_ulaw", + sampleRateHz: 8000, + channels: 1, + }, + buildRealtimeVoiceAgentConsultChatMessage: vi.fn(), + buildRealtimeVoiceAgentConsultPolicyInstructions: vi.fn(() => "Consult policy."), + buildRealtimeVoiceAgentConsultWorkingResponse: vi.fn(), + resolveRealtimeVoiceAgentConsultToolPolicy: vi.fn((v: any, f: any) => v ?? f), + resolveRealtimeVoiceAgentConsultTools: vi.fn(() => []), + resolveConfiguredRealtimeVoiceProvider: vi.fn(() => { + throw new Error("realtime voice not configured in this test"); + }), + createRealtimeVoiceBridgeSession: vi.fn(), +})); + +import { createInkboxSessionBridge } from "../../src/inbound/session.js"; +import { registerSendSms } from "../../src/tools/send-sms.js"; +import { + consumeSpawnLink, + getActiveConversation, + normalizeRecipientKey, + recordSpawnFromActive, + resetSpawnContextForTest, + setActiveConversation, +} from "../../src/spawn-context.js"; + +const PARENT = { + sessionKey: "agent:main:inkbox:direct:parent", + replyTarget: "sms:parent-conv", + label: "Dima", + party: normalizeRecipientKey("+15551110000"), +}; + +function seedSpawnLink(recipient: string, why: string): void { + setActiveConversation(PARENT); + recordSpawnFromActive({ recipient, body: why }); + setActiveConversation(undefined); +} + +function createRuntime() { + return { + getIdentity: vi.fn(async () => ({ + agentHandle: "smoke-agent", + id: "identity-1", + emailAddress: "smoke-agent@inkboxmail.com", + mailbox: { emailAddress: "smoke-agent@inkboxmail.com" }, + sendText: vi.fn(async () => ({ id: "txt-reply" })), + sendIMessage: vi.fn(async () => ({ id: "im-reply", conversationId: "imconv-1" })), + sendEmail: vi.fn(async () => ({ id: "mail-reply" })), + sendIMessageTyping: vi.fn(async () => undefined), + listTextConversations: vi.fn(async () => []), + })), + getClient: vi.fn(async () => ({ + contacts: { lookup: vi.fn(async () => []) }, + })), + }; +} + +function createChannelRuntime() { + const dispatchReply = vi.fn(async () => undefined); + return { + inbound: { buildContext: vi.fn((input: any) => input), dispatchReply }, + session: { recordInboundSession: vi.fn() }, + reply: { dispatchReplyWithBufferedBlockDispatcher: vi.fn() }, + }; +} + +function createBridge(runtime: any, channelRuntime: any) { + return createInkboxSessionBridge({ + cfg: {}, + account: { accountId: "default", config: { identity: "smoke-agent" } } as any, + runtime, + channelRuntime, + logger: { info: vi.fn(), warn: vi.fn() }, + }); +} + +function inboundText(remote: string): any { + return { + id: "evt-txt-in", + event_type: "text.received", + timestamp: "2026-07-13T00:00:00Z", + data: { + contacts: [], + agent_identities: [], + recipient_phone_number: null, + text_message: { + id: "txt-in-1", + direction: "inbound", + local_phone_number: "+16282028580", + remote_phone_number: remote, + sender_phone_number: remote, + conversation_id: "conv-alex", + text: "Sure, the launch is at 4pm.", + type: "sms", + media: null, + is_read: false, + delivery_status: null, + origin: "user_initiated", + error_code: null, + error_detail: null, + sent_at: null, + delivered_at: null, + failed_at: null, + recipients: null, + created_at: "2026-07-13T00:00:00Z", + updated_at: "2026-07-13T00:00:00Z", + }, + }, + }; +} + +function runOf(channelRuntime: any) { + return channelRuntime.inbound.dispatchReply.mock.calls[0][0]; +} + +describe("spin-off thread context (module)", () => { + beforeEach(() => resetSpawnContextForTest()); + + it("normalizes phone and email recipient keys", () => { + expect(normalizeRecipientKey("+1 (555) 111-0000")).toBe("+15551110000"); + expect(normalizeRecipientKey("Alex@Example.com")).toBe("alex@example.com"); + expect(normalizeRecipientKey(" ")).toBeUndefined(); + }); + + it("records a spawn link to a new recipient and consumes it once", () => { + setActiveConversation(PARENT); + recordSpawnFromActive({ recipient: "+15559999999", body: "Can you confirm the time?" }); + + const link = consumeSpawnLink("+15559999999"); + expect(link).toMatchObject({ + parentSessionKey: PARENT.sessionKey, + parentReplyTarget: PARENT.replyTarget, + parentLabel: "Dima", + why: "Can you confirm the time?", + }); + // One-shot: a second consume finds nothing. + expect(consumeSpawnLink("+15559999999")).toBeUndefined(); + }); + + it("does not record a link when the recipient is the active conversation's own party", () => { + setActiveConversation(PARENT); + recordSpawnFromActive({ recipient: "+1 555 111 0000", body: "replying to you" }); + expect(consumeSpawnLink("+15551110000")).toBeUndefined(); + }); + + it("does not record a link when there is no active conversation", () => { + setActiveConversation(undefined); + recordSpawnFromActive({ recipient: "+15559999999", body: "no parent" }); + expect(consumeSpawnLink("+15559999999")).toBeUndefined(); + }); +}); + +describe("spin-off thread context (inbound reply)", () => { + let runtime: any; + let channelRuntime: any; + + beforeEach(() => { + resetSpawnContextForTest(); + runtime = createRuntime(); + channelRuntime = createChannelRuntime(); + }); + + it("inherits the parent when a spun-off recipient replies", async () => { + seedSpawnLink("+15559999999", "Hey Alex, can you confirm the launch time?"); + const bridge = createBridge(runtime, channelRuntime); + + await bridge.handlers.onText?.(inboundText("+15559999999")); + + const run = runOf(channelRuntime); + const body = run.ctxPayload.message.bodyForAgent; + expect(body).toContain("[inkbox:spinoff_thread"); + expect(body).toContain('parent="Dima"'); + expect(body).toContain("relay_to=sms:parent-conv"); + expect(body).toContain("Hey Alex, can you confirm the launch time?"); + // The reply's session inherits the parent's context. + expect(run.ctxPayload.route.modelParentSessionKey).toBe(PARENT.sessionKey); + }); + + it("leaves an ordinary reply untouched when there is no spawn link", async () => { + const bridge = createBridge(runtime, channelRuntime); + + await bridge.handlers.onText?.(inboundText("+15558888888")); + + const run = runOf(channelRuntime); + expect(run.ctxPayload.message.bodyForAgent).not.toContain("spinoff_thread"); + expect(run.ctxPayload.route.modelParentSessionKey).toBeUndefined(); + }); + + it("consumes the link so a later reply from the same person is ordinary", async () => { + seedSpawnLink("+15559999999", "one-shot question"); + const bridge = createBridge(runtime, channelRuntime); + + await bridge.handlers.onText?.(inboundText("+15559999999")); + await bridge.handlers.onText?.(inboundText("+15559999999")); + + const second = channelRuntime.inbound.dispatchReply.mock.calls[1][0]; + expect(second.ctxPayload.message.bodyForAgent).not.toContain("spinoff_thread"); + expect(second.ctxPayload.route.modelParentSessionKey).toBeUndefined(); + }); +}); + +describe("spin-off thread context (send tool records the link)", () => { + beforeEach(() => resetSpawnContextForTest()); + + function registerTool() { + let execute: ((id: string, params: any) => Promise) | undefined; + const api = { registerTool: (t: any) => { execute = t.execute; } }; + const runtime = createRuntime(); + registerSendSms(api as any, runtime as any); + return (params: any) => execute!("call-1", params); + } + + it("records a spawn link when the agent texts a new number mid-conversation", async () => { + const run = registerTool(); + setActiveConversation(PARENT); + await run({ to: "+15559999999", text: "Hey Alex, quick question." }); + setActiveConversation(undefined); + + const link = consumeSpawnLink("+15559999999"); + expect(link?.parentSessionKey).toBe(PARENT.sessionKey); + expect(link?.why).toBe("Hey Alex, quick question."); + }); + + it("does not record a spawn link when texting the active party back", async () => { + const run = registerTool(); + setActiveConversation(PARENT); + await run({ to: "+15551110000", text: "replying to you directly" }); + setActiveConversation(undefined); + + expect(consumeSpawnLink("+15551110000")).toBeUndefined(); + }); + + it("does not parent a group send (ambiguous which recipient is the spin-off)", async () => { + const run = registerTool(); + setActiveConversation(PARENT); + await run({ to: ["+15559999999", "+15557777777"], text: "group ping" }); + setActiveConversation(undefined); + + expect(consumeSpawnLink("+15559999999")).toBeUndefined(); + expect(consumeSpawnLink("+15557777777")).toBeUndefined(); + }); + + it("clears the active conversation after a turn so idle sends are unparented", async () => { + // Sanity: getActiveConversation reflects set/clear. + setActiveConversation(PARENT); + expect(getActiveConversation()).toEqual(PARENT); + setActiveConversation(undefined); + expect(getActiveConversation()).toBeUndefined(); + }); +});