From 2c3072d43e0aac6a0c3226f1c7cc16f4e2c2a3c9 Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 11:34:39 -0400 Subject: [PATCH 01/13] feat(slack): stream native agent progress --- src/api/slack-core-client.ts | 2 + src/runs/turn-stream.ts | 2 + src/slack/core-bridge.ts | 2 + src/slack/lib.ts | 2 + src/slack/messaging.ts | 27 ++++++ src/slack/presenters.ts | 131 +++++++++++++++++++++++++++ src/slack/turn-handler.ts | 52 ++++++++++- test/slack-index.integration.test.ts | 66 +++++++++++++- test/slack-messaging.test.ts | 30 ++++++ test/slack-presenters.test.ts | 81 ++++++++++++++++- test/turn-stream.test.ts | 16 ++++ 11 files changed, 404 insertions(+), 7 deletions(-) create mode 100644 test/slack-messaging.test.ts diff --git a/src/api/slack-core-client.ts b/src/api/slack-core-client.ts index 67f88aa9..988b98b1 100644 --- a/src/api/slack-core-client.ts +++ b/src/api/slack-core-client.ts @@ -28,6 +28,7 @@ import { resolveRuntimeChoiceDurable, type RuntimeChoice } from "../harness/harn import { modelDisplayName } from "../model/pi-models.ts"; interface SlackRunHooks { + onDelta?(delta: string): void | Promise; onFirstBlock?(text: string): void; onSurfacePosted?(): void; onTasks?(tasks: Array<{ id: string; title: string; status: TaskStatus }>): void | Promise; @@ -187,6 +188,7 @@ export function createSlackCoreClient(deps: SlackCoreClientDeps): SlackCoreClien const waiters = terminalWaiters.get(runId) ?? new Set(); terminalWaiters.set(runId, waiters); const unsubscribe = deps.turnStream.subscribe(runId, { + onDelta: (delta) => hooks.onDelta?.(delta), onFirstBlock: signalFirstBlock, onSurfacePosted: signalSurface, }); diff --git a/src/runs/turn-stream.ts b/src/runs/turn-stream.ts index 26373579..722bcef8 100644 --- a/src/runs/turn-stream.ts +++ b/src/runs/turn-stream.ts @@ -16,6 +16,7 @@ export interface TurnStream { } interface TurnStreamListener { + onDelta?(delta: string): void | Promise; onFirstBlock?(text: string): void; onSurfacePosted?(): void; } @@ -101,6 +102,7 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream { if (entry.firstBlockOpen && entry.firstBlock.length < FIRST_BLOCK_MAX_CHARS) entry.firstBlock = (entry.firstBlock + delta).slice(0, FIRST_BLOCK_MAX_CHARS); if (entry.text.length < maxChars) entry.text = (entry.text + delta).slice(0, maxChars); + for (const l of listeners.get(runId) ?? []) l.onDelta?.(delta); }, publishBlockStart(runId) { diff --git a/src/slack/core-bridge.ts b/src/slack/core-bridge.ts index 5946cb8b..811a0e8b 100644 --- a/src/slack/core-bridge.ts +++ b/src/slack/core-bridge.ts @@ -10,6 +10,7 @@ interface CoreCallHooks { /** The turn was folded into a run that was ALREADY live (a mid-turn steer), so this handler * owns nothing: the envelope is durably accepted, but the reply belongs to the run's owner. */ onSteered?: (runId: string) => void; + onDelta?: (delta: string) => void | Promise; onFirstBlock?: (text: string) => void; onSurfacePosted?: () => void; onTasks?: (tasks: RunTaskView[]) => void; @@ -140,6 +141,7 @@ export function createCoreBridge(core: SlackCoreClient): CoreBridge { let result: TurnResult | null; try { result = await core.waitRun(runId, { + ...(hooks.onDelta ? { onDelta: hooks.onDelta } : {}), ...(hooks.onFirstBlock ? { onFirstBlock: hooks.onFirstBlock } : {}), ...(hooks.onSurfacePosted ? { onSurfacePosted: hooks.onSurfacePosted } : {}), ...(hooks.onTasks ? { onTasks: hooks.onTasks } : {}), diff --git a/src/slack/lib.ts b/src/slack/lib.ts index d0d33980..831f24fe 100644 --- a/src/slack/lib.ts +++ b/src/slack/lib.ts @@ -139,4 +139,6 @@ export { renderTaskList, type TaskListPresenter, createTaskListPresenter, + type NativeAgentPresenter, + createNativeAgentPresenter, } from "./presenters.ts"; diff --git a/src/slack/messaging.ts b/src/slack/messaging.ts index 3e7321b7..e056b6b9 100644 --- a/src/slack/messaging.ts +++ b/src/slack/messaging.ts @@ -74,6 +74,33 @@ export function stripSlackDirectives(text: string): string { return stripAgentRequestDirectives(stripReactionDirectives(text)); } +export interface StreamingReplyFilter { + push(delta: string): string; + flush(): string; +} + +export function createStreamingReplyFilter(): StreamingReplyFilter { + let raw = ""; + let emitted = ""; + const take = (flush: boolean): string => { + const cleaned = stripSlackDirectives(raw); + const limit = flush ? cleaned.length : Math.max(0, cleaned.length - 16); + if (limit <= emitted.length) return ""; + const next = cleaned.slice(emitted.length, limit); + emitted += next; + return next; + }; + return { + push(delta) { + raw += delta; + return take(false); + }, + flush() { + return take(true); + }, + }; +} + export async function applyAndLogReactions( client: any, channel: string, diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index 25150f86..ac206861 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -175,6 +175,137 @@ export interface TaskListPresenter { settle(): Promise; } +type NativeAgentChunk = + | { type: "markdown_text"; text: string } + | { + type: "task_update"; + id: string; + title: string; + status: "pending" | "in_progress" | "complete" | "error"; + }; + +export interface NativeAgentPresenter { + begin(): Promise; + onDelta(delta: string): Promise; + onTasks(tasks: RunTaskView[]): Promise; + finalize(): Promise; + settle(): Promise; + ownsSurface(): boolean; +} + +export function createNativeAgentPresenter(deps: { + setStatus(status: string): Promise; + start(chunks: NativeAgentChunk[]): Promise; + append(ts: string, chunks: NativeAgentChunk[]): Promise; + stop(ts: string): Promise; + checkpoint(ts: string): Promise; + onSurfacePosted(): void; + onError?(error: unknown): void; +}): NativeAgentPresenter { + let state: "idle" | "starting" | "active" | "disabled" | "stopped" = "idle"; + let messageTs: string | undefined; + let chain = Promise.resolve(); + let statusStarted = false; + let statusCleared = false; + const nativeTaskStatus = (status: RunTaskStatus): "pending" | "in_progress" | "complete" | "error" => { + if (status === "completed" || status === "skipped") return "complete"; + if (status === "failed") return "error"; + return status; + }; + const clearStatus = async (): Promise => { + if (!statusStarted || statusCleared) return; + statusCleared = true; + await deps.setStatus("").catch((error) => deps.onError?.(error)); + }; + const enqueue = (operation: () => Promise): Promise => { + chain = chain.then(operation); + return chain; + }; + const send = async (chunks: NativeAgentChunk[]): Promise => { + if (state === "disabled" || state === "stopped") return false; + if (!messageTs && state === "idle") state = "starting"; + await enqueue(async () => { + if (state === "disabled" || state === "stopped") return; + try { + if (!messageTs) { + const ts = await deps.start(chunks); + if (!ts) throw new Error("chat.startStream returned no message timestamp"); + await deps.checkpoint(ts); + messageTs = ts; + state = "active"; + deps.onSurfacePosted(); + return; + } + await deps.append(messageTs, chunks); + } catch (error) { + state = "disabled"; + deps.onError?.(error); + } + }); + return state === "active"; + }; + return { + async begin() { + try { + await deps.setStatus("Thinking…"); + statusStarted = true; + } catch (error) { + deps.onError?.(error); + } + }, + onDelta(delta) { + if (!delta) return Promise.resolve(state === "active"); + return send([{ type: "markdown_text", text: delta }]); + }, + onTasks(tasks) { + if (!tasks.length) return Promise.resolve(state === "active"); + const chunks: NativeAgentChunk[] = tasks.slice(0, 20).map((task) => ({ + type: "task_update", + id: task.id, + title: task.title + .replace(/[\r\n\t]+/g, " ") + .replace(/\s+/g, " ") + .trim() + .slice(0, 120), + status: nativeTaskStatus(task.status), + })); + return send(chunks); + }, + async finalize() { + await chain; + if (state !== "active" || !messageTs) { + await clearStatus(); + return false; + } + try { + await deps.stop(messageTs); + state = "stopped"; + await clearStatus(); + return true; + } catch (error) { + deps.onError?.(error); + await clearStatus(); + return true; + } + }, + async settle() { + await chain; + if (state === "active" && messageTs) { + try { + await deps.stop(messageTs); + state = "stopped"; + } catch (error) { + deps.onError?.(error); + } + } + await clearStatus(); + }, + ownsSurface() { + return state === "starting" || state === "active" || state === "stopped"; + }, + }; +} + export function createTaskListPresenter(deps: { post(text: string, blocks: Array>): Promise; update(ts: string, text: string, blocks: Array>): Promise; diff --git a/src/slack/turn-handler.ts b/src/slack/turn-handler.ts index 0214c17a..c7eb087b 100644 --- a/src/slack/turn-handler.ts +++ b/src/slack/turn-handler.ts @@ -7,6 +7,7 @@ import { type OverheardMessage, type ReactionTally, type RunTaskView, + type NativeAgentPresenter, type SlackFile, type TaskListPresenter, DEFAULT_ACK_REACTIONS, @@ -16,6 +17,7 @@ import { buildReactionTurnText, createAckPresenter, createDeduper, + createNativeAgentPresenter, createTaskListPresenter, createThreadTracker, decodeSlackEntities, @@ -59,6 +61,7 @@ import { type SlackConversationKind, applyAndLogReactions, cleanAgentReplyForSlack, + createStreamingReplyFilter, conversationPlaceLabel, slackSurfaceInstructions, } from "./messaging.ts"; @@ -257,6 +260,7 @@ export function createTurnHandler(deps: { let queuedRunId: string | undefined; let taskList: TaskListPresenter | undefined; + let nativeAgent: NativeAgentPresenter | undefined; const ack = inc.unprompted ? undefined : createAckPresenter({ @@ -292,6 +296,7 @@ export function createTurnHandler(deps: { } const settleAck = async (): Promise => { await ack?.settle().catch(swallowAs("slack: ack settle", undefined)); + await nativeAgent?.settle().catch(swallowAs("slack: native agent settle", undefined)); }; if (inc.kind === "channel") { @@ -403,6 +408,35 @@ export function createTurnHandler(deps: { if (inc.unprompted && !text.trim() && attachments.length === 0) return; + if (!inc.unprompted) { + const nativeThreadTs = replyThreadTs ?? inc.ts; + nativeAgent = createNativeAgentPresenter({ + setStatus: (status) => + client.assistant.threads + .setStatus({ channel_id: inc.channel, thread_ts: nativeThreadTs, status }) + .then(() => {}), + start: async (chunks) => { + const response = await client.chat.startStream({ + channel: inc.channel, + thread_ts: nativeThreadTs, + chunks, + task_display_mode: "timeline", + ...(inc.kind === "channel" ? { recipient_team_id: ids.ownTeamId, recipient_user_id: inc.userId } : {}), + }); + return typeof response.ts === "string" ? response.ts : undefined; + }, + append: (ts, chunks) => client.chat.appendStream({ channel: inc.channel, ts, chunks }).then(() => {}), + stop: (ts) => client.chat.stopStream({ channel: inc.channel, ts }).then(() => {}), + checkpoint: async (ts) => { + if (queuedRunId) await checkpointRunEditRef(queuedRunId, ts); + }, + onSurfacePosted: () => ack?.onSurfacePosted(), + onError: (error) => console.error("[slack-plugin] native agent update failed:", errMessage(error)), + }); + await nativeAgent.begin(); + } + const streamingReply = createStreamingReplyFilter(); + const turn: Omit = { actor, conversation: { @@ -455,8 +489,12 @@ export function createTurnHandler(deps: { onSteered: () => inc.ackGate?.persisted(), ...(ack ? { + onDelta: async (delta: string) => { + const visible = streamingReply.push(delta); + if (visible) await nativeAgent?.onDelta(visible); + }, onFirstBlock: (blockText: string) => { - ack.onFirstBlock(cleanAgentReplyForSlack(blockText).text); + if (!nativeAgent?.ownsSurface()) ack.onFirstBlock(cleanAgentReplyForSlack(blockText).text); }, onSurfacePosted: () => ack.onSurfacePosted(), } @@ -465,6 +503,7 @@ export function createTurnHandler(deps: { ? { onTasks: async (tasks: RunTaskView[]) => { await ack?.drain(); + if (await nativeAgent?.onTasks(tasks)) return; await taskList?.onTasks(tasks); }, } @@ -523,6 +562,9 @@ export function createTurnHandler(deps: { const postText = reply; const tDeliverStart = performance.now(); let finalizedTaskList = false; + const finalStreamDelta = streamingReply.flush(); + if (finalStreamDelta) await nativeAgent?.onDelta(finalStreamDelta); + const finalizedNative = (await nativeAgent?.finalize()) ?? false; if (result.attachments?.length) { let uploadError: unknown; try { @@ -539,13 +581,13 @@ export function createTurnHandler(deps: { console.error("[slack-plugin] file upload failed:", (err as Error).message); } await settleAck(); - if (postText) finalizedTaskList = (await taskList?.finalize(postText)) ?? false; - if (postText && !finalizedTaskList) await postReply(postText); + if (postText && !finalizedNative) finalizedTaskList = (await taskList?.finalize(postText)) ?? false; + if (postText && !finalizedNative && !finalizedTaskList) await postReply(postText); if (uploadError) await postReply(uploadFailureNote(uploadError)); } else { await settleAck(); - if (postText) finalizedTaskList = (await taskList?.finalize(postText)) ?? false; - if (postText && !finalizedTaskList) await postReply(postText); + if (postText && !finalizedNative) finalizedTaskList = (await taskList?.finalize(postText)) ?? false; + if (postText && !finalizedNative && !finalizedTaskList) await postReply(postText); } if (queuedRunId) { reportTurnMetrics(queuedRunId, { diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts index a5bc2a65..50d35d4a 100644 --- a/test/slack-index.integration.test.ts +++ b/test/slack-index.integration.test.ts @@ -18,6 +18,10 @@ class FakeSlackClient { readonly deletes: any[] = []; readonly reactionsAdded: any[] = []; readonly reactionsRemoved: any[] = []; + readonly statuses: any[] = []; + readonly streamsStarted: any[] = []; + readonly streamsAppended: any[] = []; + readonly streamsStopped: any[] = []; readonly usersById = new Map(); readonly channelsById = new Map(); readonly membersByChannel = new Map(); @@ -87,6 +91,26 @@ class FakeSlackClient { this.deletes.push(body); return { ok: true }; }, + startStream: async (body: any) => { + this.streamsStarted.push(body); + return { ok: true, ts: "stream-1" }; + }, + appendStream: async (body: any) => { + this.streamsAppended.push(body); + return { ok: true, ts: body.ts }; + }, + stopStream: async (body: any) => { + this.streamsStopped.push(body); + return { ok: true, ts: body.ts }; + }, + }; + readonly assistant = { + threads: { + setStatus: async (body: any) => { + this.statuses.push(body); + return { ok: true }; + }, + }, }; readonly reactions = { add: async (body: any) => { @@ -199,6 +223,8 @@ class FakeCore implements SlackCoreClient { private runGate: Promise | undefined; private releaseRun: (() => void) | undefined; readonly modelChangeListeners: Array<(scope: any) => void> = []; + streamDeltas: string[] = []; + streamTasks: Array<{ id: string; title: string; status: "pending" | "in_progress" | "completed" }> = []; async externalSlackParticipants(): Promise { return this.externalParticipants; @@ -238,9 +264,11 @@ class FakeCore implements SlackCoreClient { } return this.result; } - async waitRun(runId: string): Promise { + async waitRun(runId: string, hooks: any = {}): Promise { this.polled.push(runId); if (this.runGate) await this.runGate; + for (const delta of this.streamDeltas) hooks.onDelta?.(delta); + if (this.streamTasks.length) await hooks.onTasks?.(this.streamTasks); return this.result; } /** Enqueue `runId` on the first submit and hold waitRun open; every later submit is a @@ -401,6 +429,42 @@ test("a DM becomes one scoped live turn and one Slack reply", async () => { } }); +test("a queued DM uses Slack native status, reply streaming, and task updates without a duplicate post", async () => { + const f = await fixture(); + try { + f.core.queuedRunId = "R-stream"; + f.core.streamDeltas = ["Checking ", "now."]; + f.core.streamTasks = [{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]; + f.core.result = { status: "ok", reply: "Checking now." }; + + await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "check it", ts: "100.9" }); + + assert.deepEqual(f.client.statuses, [ + { channel_id: "D1", thread_ts: "100.9", status: "Thinking…" }, + { channel_id: "D1", thread_ts: "100.9", status: "" }, + ]); + assert.deepEqual(f.client.streamsStarted, [ + { + channel: "D1", + thread_ts: "100.9", + chunks: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }], + task_display_mode: "timeline", + }, + ]); + assert.deepEqual(f.client.streamsAppended, [ + { + channel: "D1", + ts: "stream-1", + chunks: [{ type: "markdown_text", text: "Checking now." }], + }, + ]); + assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]); + assert.equal(f.client.posts.length, 0); + } finally { + await f.stop(); + } +}); + test("a human's DM sets the conversation header to the serving model + web surface", async () => { const f = await fixture({ webUiPublicUrl: "https://claw.example.dev" }); try { diff --git a/test/slack-messaging.test.ts b/test/slack-messaging.test.ts new file mode 100644 index 00000000..8fb5e136 --- /dev/null +++ b/test/slack-messaging.test.ts @@ -0,0 +1,30 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { createStreamingReplyFilter } from "../src/slack/messaging.ts"; + +test("streaming reply filter emits prose incrementally and never exposes internal directives", () => { + const filter = createStreamingReplyFilter(); + const visible = [ + filter.push("Here is the answer. "), + filter.push("[[rea"), + filter.push("ct: eyes]]"), + filter.push(" Done."), + filter.flush(), + ].join(""); + + assert.equal(visible, "Here is the answer. Done."); + assert.equal(visible.includes("[[react"), false); +}); + +test("streaming reply filter holds a split agent request directive out of Slack", () => { + const filter = createStreamingReplyFilter(); + const visible = [ + filter.push("I need help. [[ask-"), + filter.push("agent: <@U2> | inspect token=secret"), + filter.push("]] Thanks."), + filter.flush(), + ].join(""); + + assert.equal(visible, "I need help. Thanks."); + assert.equal(visible.includes("token=secret"), false); +}); diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts index f6aa2b5c..7378e039 100644 --- a/test/slack-presenters.test.ts +++ b/test/slack-presenters.test.ts @@ -1,6 +1,85 @@ import { test } from "node:test"; import assert from "node:assert/strict"; -import { renderTaskList, createTaskListPresenter, createAckPresenter, stripAckPrefix } from "../src/slack/lib.ts"; +import { + renderTaskList, + createTaskListPresenter, + createAckPresenter, + createNativeAgentPresenter, + stripAckPrefix, +} from "../src/slack/lib.ts"; + +test("native agent presenter streams text and public task progress into one message", async () => { + const calls: Array<{ method: string; body: unknown }> = []; + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push({ method: "status", body: status }); + }, + start: async (chunks) => { + calls.push({ method: "start", body: chunks }); + return "171.1"; + }, + append: async (ts, chunks) => { + calls.push({ method: `append:${ts}`, body: chunks }); + }, + stop: async (ts) => { + calls.push({ method: `stop:${ts}`, body: null }); + }, + checkpoint: async (ts) => { + calls.push({ method: "checkpoint", body: ts }); + }, + onSurfacePosted: () => calls.push({ method: "surface", body: null }), + }); + + await presenter.begin(); + assert.equal(await presenter.onDelta("Checking "), true); + assert.equal(await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]), true); + assert.equal(await presenter.onDelta("now."), true); + assert.equal(await presenter.finalize(), true); + + assert.deepEqual(calls, [ + { method: "status", body: "Thinking…" }, + { method: "start", body: [{ type: "markdown_text", text: "Checking " }] }, + { method: "checkpoint", body: "171.1" }, + { method: "surface", body: null }, + { + method: "append:171.1", + body: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }], + }, + { method: "append:171.1", body: [{ type: "markdown_text", text: "now." }] }, + { method: "stop:171.1", body: null }, + { method: "status", body: "" }, + ]); +}); + +test("native agent presenter falls back cleanly when Slack streaming is unavailable", async () => { + const calls: string[] = []; + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push(`status:${status}`); + }, + start: async () => { + calls.push("start"); + throw new Error("unknown_method"); + }, + append: async () => { + calls.push("append"); + }, + stop: async () => { + calls.push("stop"); + }, + checkpoint: async () => { + calls.push("checkpoint"); + }, + onSurfacePosted: () => calls.push("surface"), + onError: (error) => calls.push(`error:${(error as Error).message}`), + }); + + await presenter.begin(); + assert.equal(await presenter.onDelta("Answer"), false); + assert.equal(await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]), false); + assert.equal(await presenter.finalize(), false); + assert.deepEqual(calls, ["status:Thinking…", "start", "error:unknown_method", "status:"]); +}); test("renderTaskList renders every terminal state", () => { assert.equal( diff --git a/test/turn-stream.test.ts b/test/turn-stream.test.ts index b4687307..86a8516b 100644 --- a/test/turn-stream.test.ts +++ b/test/turn-stream.test.ts @@ -21,6 +21,22 @@ test("accumulates deltas per run and isolates runs", () => { assert.equal(s.snapshot("r2"), "world"); }); +test("subscribers receive public reply deltas in order", () => { + const s = createTurnStream(); + const deltas: string[] = []; + const unsubscribe = s.subscribe("r1", { + onDelta: (delta) => { + deltas.push(delta); + }, + }); + s.publish("r1", "Hel"); + s.publish("r1", "lo"); + s.publish("r2", "private to another run"); + unsubscribe(); + s.publish("r1", "!"); + assert.deepEqual(deltas, ["Hel", "lo"]); +}); + test("ignores empty deltas", () => { const s = createTurnStream(); s.publish("r1", ""); From 11b21dd051358b0e114c2b91020f58fc81e68aae Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 11:37:08 -0400 Subject: [PATCH 02/13] fix(slack): clean failed stream checkpoints --- src/slack/presenters.ts | 11 ++++++++ src/slack/turn-handler.ts | 3 ++- test/slack-presenters.test.ts | 48 ++++++++++++++++++++++++++++++++++- 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index ac206861..14243665 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -198,6 +198,7 @@ export function createNativeAgentPresenter(deps: { start(chunks: NativeAgentChunk[]): Promise; append(ts: string, chunks: NativeAgentChunk[]): Promise; stop(ts: string): Promise; + remove(ts: string): Promise; checkpoint(ts: string): Promise; onSurfacePosted(): void; onError?(error: unknown): void; @@ -226,11 +227,14 @@ export function createNativeAgentPresenter(deps: { if (!messageTs && state === "idle") state = "starting"; await enqueue(async () => { if (state === "disabled" || state === "stopped") return; + let uncheckpointedTs: string | undefined; try { if (!messageTs) { const ts = await deps.start(chunks); if (!ts) throw new Error("chat.startStream returned no message timestamp"); + uncheckpointedTs = ts; await deps.checkpoint(ts); + uncheckpointedTs = undefined; messageTs = ts; state = "active"; deps.onSurfacePosted(); @@ -238,6 +242,13 @@ export function createNativeAgentPresenter(deps: { } await deps.append(messageTs, chunks); } catch (error) { + if (uncheckpointedTs) { + try { + await deps.remove(uncheckpointedTs); + } catch (removeError) { + deps.onError?.(removeError); + } + } state = "disabled"; deps.onError?.(error); } diff --git a/src/slack/turn-handler.ts b/src/slack/turn-handler.ts index c7eb087b..e45fa6fe 100644 --- a/src/slack/turn-handler.ts +++ b/src/slack/turn-handler.ts @@ -427,13 +427,14 @@ export function createTurnHandler(deps: { }, append: (ts, chunks) => client.chat.appendStream({ channel: inc.channel, ts, chunks }).then(() => {}), stop: (ts) => client.chat.stopStream({ channel: inc.channel, ts }).then(() => {}), + remove: (ts) => client.chat.delete({ channel: inc.channel, ts }).then(() => {}), checkpoint: async (ts) => { if (queuedRunId) await checkpointRunEditRef(queuedRunId, ts); }, onSurfacePosted: () => ack?.onSurfacePosted(), onError: (error) => console.error("[slack-plugin] native agent update failed:", errMessage(error)), }); - await nativeAgent.begin(); + void nativeAgent.begin(); } const streamingReply = createStreamingReplyFilter(); diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts index 7378e039..0a55972a 100644 --- a/test/slack-presenters.test.ts +++ b/test/slack-presenters.test.ts @@ -24,6 +24,9 @@ test("native agent presenter streams text and public task progress into one mess stop: async (ts) => { calls.push({ method: `stop:${ts}`, body: null }); }, + remove: async (ts) => { + calls.push({ method: `remove:${ts}`, body: null }); + }, checkpoint: async (ts) => { calls.push({ method: "checkpoint", body: ts }); }, @@ -67,11 +70,14 @@ test("native agent presenter falls back cleanly when Slack streaming is unavaila stop: async () => { calls.push("stop"); }, + remove: async () => { + calls.push("remove"); + }, checkpoint: async () => { calls.push("checkpoint"); }, onSurfacePosted: () => calls.push("surface"), - onError: (error) => calls.push(`error:${(error as Error).message}`), + onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`), }); await presenter.begin(); @@ -81,6 +87,46 @@ test("native agent presenter falls back cleanly when Slack streaming is unavaila assert.deepEqual(calls, ["status:Thinking…", "start", "error:unknown_method", "status:"]); }); +test("native agent presenter removes an uncheckpointed stream before falling back", async () => { + const calls: string[] = []; + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push(`status:${status}`); + }, + start: async () => { + calls.push("start"); + return "171.7"; + }, + append: async () => { + calls.push("append"); + }, + stop: async () => { + calls.push("stop"); + }, + remove: async (ts) => { + calls.push(`remove:${ts}`); + }, + checkpoint: async () => { + calls.push("checkpoint"); + throw new Error("core unavailable"); + }, + onSurfacePosted: () => calls.push("surface"), + onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`), + }); + + await presenter.begin(); + assert.equal(await presenter.onDelta("Answer"), false); + assert.equal(await presenter.finalize(), false); + assert.deepEqual(calls, [ + "status:Thinking…", + "start", + "checkpoint", + "remove:171.7", + "error:core unavailable", + "status:", + ]); +}); + test("renderTaskList renders every terminal state", () => { assert.equal( renderTaskList([ From 17367787dd174b3c897562978d092a6f92c2428e Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 11:43:10 -0400 Subject: [PATCH 03/13] fix(slack): harden native stream fallback --- src/slack/messaging.ts | 49 ++++++++++++--- src/slack/presenters.ts | 32 ++++++---- test/slack-index.integration.test.ts | 5 +- test/slack-messaging.test.ts | 9 +++ test/slack-presenters.test.ts | 91 ++++++++++++++++++++++++++++ 5 files changed, 164 insertions(+), 22 deletions(-) diff --git a/src/slack/messaging.ts b/src/slack/messaging.ts index e056b6b9..1795129e 100644 --- a/src/slack/messaging.ts +++ b/src/slack/messaging.ts @@ -80,19 +80,50 @@ export interface StreamingReplyFilter { } export function createStreamingReplyFilter(): StreamingReplyFilter { - let raw = ""; - let emitted = ""; + const directiveStarts = ["[[react:", "[[ask-agent:"]; + let pending = ""; + let insideDirective = false; const take = (flush: boolean): string => { - const cleaned = stripSlackDirectives(raw); - const limit = flush ? cleaned.length : Math.max(0, cleaned.length - 16); - if (limit <= emitted.length) return ""; - const next = cleaned.slice(emitted.length, limit); - emitted += next; - return next; + let visible = ""; + for (;;) { + if (insideDirective) { + const end = pending.indexOf("]]"); + if (end === -1) { + if (flush) pending = ""; + return visible; + } + pending = pending.slice(end + 2); + insideDirective = false; + continue; + } + const possibleStart = pending.indexOf("[["); + if (possibleStart === -1) { + const held = !flush && pending.endsWith("[") ? 1 : 0; + visible += pending.slice(0, pending.length - held); + pending = pending.slice(pending.length - held); + return visible; + } + visible += pending.slice(0, possibleStart); + pending = pending.slice(possibleStart); + const lower = pending.toLowerCase(); + if (directiveStarts.some((start) => lower.startsWith(start))) { + insideDirective = true; + continue; + } + if (directiveStarts.some((start) => start.startsWith(lower))) { + if (flush) { + visible += pending; + pending = ""; + } + return visible; + } + visible += pending[0]; + pending = pending.slice(1); + } }; return { push(delta) { - raw += delta; + pending += delta; return take(false); }, flush() { diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index 14243665..6a84edea 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -203,7 +203,7 @@ export function createNativeAgentPresenter(deps: { onSurfacePosted(): void; onError?(error: unknown): void; }): NativeAgentPresenter { - let state: "idle" | "starting" | "active" | "disabled" | "stopped" = "idle"; + let state: "idle" | "starting" | "active" | "disabled" | "orphaned" | "stopped" = "idle"; let messageTs: string | undefined; let chain = Promise.resolve(); let statusStarted = false; @@ -242,14 +242,18 @@ export function createNativeAgentPresenter(deps: { } await deps.append(messageTs, chunks); } catch (error) { - if (uncheckpointedTs) { + const failedStreamTs = uncheckpointedTs ?? messageTs; + let cleanupFailed = false; + if (failedStreamTs) { try { - await deps.remove(uncheckpointedTs); + await deps.remove(failedStreamTs); + messageTs = undefined; } catch (removeError) { + cleanupFailed = true; deps.onError?.(removeError); } } - state = "disabled"; + state = cleanupFailed ? "orphaned" : "disabled"; deps.onError?.(error); } }); @@ -257,12 +261,14 @@ export function createNativeAgentPresenter(deps: { }; return { async begin() { - try { - await deps.setStatus("Thinking…"); - statusStarted = true; - } catch (error) { - deps.onError?.(error); - } + statusStarted = true; + await enqueue(async () => { + try { + await deps.setStatus("Thinking…"); + } catch (error) { + deps.onError?.(error); + } + }); }, onDelta(delta) { if (!delta) return Promise.resolve(state === "active"); @@ -284,6 +290,10 @@ export function createNativeAgentPresenter(deps: { }, async finalize() { await chain; + if (state === "orphaned") { + await clearStatus(); + return true; + } if (state !== "active" || !messageTs) { await clearStatus(); return false; @@ -312,7 +322,7 @@ export function createNativeAgentPresenter(deps: { await clearStatus(); }, ownsSurface() { - return state === "starting" || state === "active" || state === "stopped"; + return state === "starting" || state === "active" || state === "orphaned" || state === "stopped"; }, }; } diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts index 50d35d4a..4566f1f7 100644 --- a/test/slack-index.integration.test.ts +++ b/test/slack-index.integration.test.ts @@ -447,15 +447,16 @@ test("a queued DM uses Slack native status, reply streaming, and task updates wi { channel: "D1", thread_ts: "100.9", - chunks: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }], + chunks: [{ type: "markdown_text", text: "Checking " }], task_display_mode: "timeline", }, ]); assert.deepEqual(f.client.streamsAppended, [ + { channel: "D1", ts: "stream-1", chunks: [{ type: "markdown_text", text: "now." }] }, { channel: "D1", ts: "stream-1", - chunks: [{ type: "markdown_text", text: "Checking now." }], + chunks: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }], }, ]); assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]); diff --git a/test/slack-messaging.test.ts b/test/slack-messaging.test.ts index 8fb5e136..aed0e523 100644 --- a/test/slack-messaging.test.ts +++ b/test/slack-messaging.test.ts @@ -28,3 +28,12 @@ test("streaming reply filter holds a split agent request directive out of Slack" assert.equal(visible, "I need help. Thanks."); assert.equal(visible.includes("token=secret"), false); }); + +test("streaming reply filter never leaks a long directive delivered one character at a time", () => { + const filter = createStreamingReplyFilter(); + const raw = "Public. [[react: internal-tool-argument-secret-1234567890]] Done."; + const visible = [...raw].map((character) => filter.push(character)).join("") + filter.flush(); + + assert.equal(visible, "Public. Done."); + assert.equal(visible.includes("internal-tool-argument"), false); +}); diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts index 0a55972a..b17a2b5c 100644 --- a/test/slack-presenters.test.ts +++ b/test/slack-presenters.test.ts @@ -127,6 +127,97 @@ test("native agent presenter removes an uncheckpointed stream before falling bac ]); }); +test("native agent presenter deletes a partial stream before fallback after append fails", async () => { + const calls: string[] = []; + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push(`status:${status}`); + }, + start: async () => { + calls.push("start"); + return "171.8"; + }, + append: async () => { + calls.push("append"); + throw new Error("stream disconnected"); + }, + stop: async () => { + calls.push("stop"); + }, + remove: async (ts) => { + calls.push(`remove:${ts}`); + }, + checkpoint: async () => { + calls.push("checkpoint"); + }, + onSurfacePosted: () => calls.push("surface"), + onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`), + }); + + await presenter.begin(); + assert.equal(await presenter.onDelta("Partial"), true); + assert.equal(await presenter.onDelta(" answer"), false); + assert.equal(await presenter.finalize(), false); + assert.deepEqual(calls, [ + "status:Thinking…", + "start", + "checkpoint", + "surface", + "append", + "remove:171.8", + "error:stream disconnected", + "status:", + ]); +}); + +test("native agent presenter clears status after a delayed begin", async () => { + const calls: string[] = []; + let releaseStatus: (() => void) | undefined; + const statusGate = new Promise((resolve) => { + releaseStatus = resolve; + }); + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push(`status:${status}`); + if (status) await statusGate; + }, + start: async () => undefined, + append: async () => {}, + stop: async () => {}, + remove: async () => {}, + checkpoint: async () => {}, + onSurfacePosted: () => {}, + }); + + const beginning = presenter.begin(); + const finalizing = presenter.finalize(); + releaseStatus?.(); + await beginning; + assert.equal(await finalizing, false); + assert.deepEqual(calls, ["status:Thinking…", "status:"]); +}); + +test("native agent presenter suppresses fallback when a failed partial stream cannot be removed", async () => { + const presenter = createNativeAgentPresenter({ + setStatus: async () => {}, + start: async () => "171.9", + append: async () => { + throw new Error("stream disconnected"); + }, + stop: async () => {}, + remove: async () => { + throw new Error("delete failed"); + }, + checkpoint: async () => {}, + onSurfacePosted: () => {}, + }); + + await presenter.begin(); + assert.equal(await presenter.onDelta("Partial"), true); + assert.equal(await presenter.onDelta(" answer"), false); + assert.equal(await presenter.finalize(), true); +}); + test("renderTaskList renders every terminal state", () => { assert.equal( renderTaskList([ From e6fdf687a9b088c8ccd1c334504c21f7fcd32d83 Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 14:09:14 -0400 Subject: [PATCH 04/13] fix(connectors): enable token-backed native skills --- src/core/orchestrator.ts | 32 ++++++++++++++++++----------- src/credentials/connector-status.ts | 11 ++++++++++ test/connector-status.test.ts | 13 ++++++++++++ 3 files changed, 44 insertions(+), 12 deletions(-) diff --git a/src/core/orchestrator.ts b/src/core/orchestrator.ts index 373b496a..c1f1f5c7 100644 --- a/src/core/orchestrator.ts +++ b/src/core/orchestrator.ts @@ -36,6 +36,7 @@ import { configuredConnectorProviders, connectorStatusIsStale, refreshConnectorStatus, + usableConnectorProviders, } from "../credentials/connector-status.ts"; import { renderComputerBlock, renderResidentLoginsBlock, renderConnectedAppsBlock } from "./environment-facts.ts"; import { PROVIDERS } from "../connectors/oauth.ts"; @@ -826,11 +827,28 @@ export function createOrchestrator(deps: OrchestratorDeps): Orchestrator { : ""; await deps.skillsReady; - const configuredProviders = deps.resolveConnectorClient + let connectorStatus = null; + if (!strictReadOnly && conversation.kind === "dm") { + try { + connectorStatus = deps.connectorStatusCache ? await deps.connectorStatusCache.get(actor.id) : null; + if ( + deps.connectorTokens && + deps.connectorStatusCache && + connectorStatusIsStale(connectorStatus, Date.now()) + ) { + connectorStatus = await refreshConnectorStatus(deps.connectorTokens, actor.id, Date.now()); + await deps.connectorStatusCache.put(connectorStatus); + } + } catch (e) { + swallow("orchestrator: connected-app status", e); + } + } + const oauthConfiguredProviders = deps.resolveConnectorClient ? await configuredConnectorProviders(deps.resolveConnectorClient).catch( swallowAs("orchestrator: configured connector providers", []), ) : []; + const configuredProviders = usableConnectorProviders(oauthConfiguredProviders, connectorStatus); const visibleSkillsForTurn = async (): Promise => filterConnectorSkills((await deps.skills?.visibleFor(skillScopes)) ?? [], configuredProviders); const visibleSkills = await visibleSkillsForTurn(); @@ -1487,18 +1505,8 @@ export function createOrchestrator(deps: OrchestratorDeps): Orchestrator { } } if (!strictReadOnly && deps.resolveConnectorClient && conversation.kind === "dm") { - let status = null; - try { - status = deps.connectorStatusCache ? await deps.connectorStatusCache.get(actor.id) : null; - if (deps.connectorTokens && deps.connectorStatusCache && connectorStatusIsStale(status, Date.now())) { - status = await refreshConnectorStatus(deps.connectorTokens, actor.id, Date.now()); - await deps.connectorStatusCache.put(status); - } - } catch (e) { - swallow("orchestrator: connected-app status", e); - } const connectionsUrl = deps.publicWebUrl ? `${deps.publicWebUrl.replace(/\/$/, "")}/keychain` : undefined; - systemPrompt += `\n\n${renderConnectedAppsBlock(status, configuredProviders, connectionsUrl)}`; + systemPrompt += `\n\n${renderConnectedAppsBlock(connectorStatus, configuredProviders, connectionsUrl)}`; } systemPrompt += memoryBlock; if (onboardingBlock) systemPrompt += `\n\n${onboardingBlock}`; diff --git a/src/credentials/connector-status.ts b/src/credentials/connector-status.ts index 5effbeaa..dd1d3403 100644 --- a/src/credentials/connector-status.ts +++ b/src/credentials/connector-status.ts @@ -106,6 +106,17 @@ export function connectorLabel(name: string): string { return PROVIDER_LABELS[name] ?? name.charAt(0).toUpperCase() + name.slice(1); } +export function usableConnectorProviders( + configuredProviders: readonly string[], + status: ConnectorStatusRecord | null, +): string[] { + const usable = new Set(configuredProviders); + for (const [provider, entry] of Object.entries(status?.providers ?? {})) { + if (entry.connected && !entry.needsReconnect) usable.add(provider); + } + return [...usable]; +} + export async function configuredConnectorProviders(resolveClient: OAuthClientResolver): Promise { const configured = await Promise.all( Object.keys(PROVIDERS).map(async (provider) => { diff --git a/test/connector-status.test.ts b/test/connector-status.test.ts index 306d1577..aa9ff2f2 100644 --- a/test/connector-status.test.ts +++ b/test/connector-status.test.ts @@ -6,6 +6,7 @@ import { connectorStatusIsStale, createConnectorStatusCache, refreshConnectorStatus, + usableConnectorProviders, type ConnectorStatusRecord, } from "../src/credentials/connector-status.ts"; import type { ConnectorTokenStore, OAuthTokenStatus } from "../src/credentials/keychain.ts"; @@ -135,3 +136,15 @@ test("labels: known providers get friendly names; unknown is capitalized", () => assert.equal(connectorLabel("github"), "GitHub"); assert.equal(connectorLabel("acme"), "Acme"); }); + +test("usable providers include valid manual tokens without requiring an OAuth client", () => { + const status: ConnectorStatusRecord = { + principalId: "U1", + checkedAt: 1, + providers: { + linear: { connected: true }, + github: { connected: true, needsReconnect: true }, + }, + }; + assert.deepEqual(usableConnectorProviders(["google"], status), ["google", "linear"]); +}); From fbf07e31d2a3e450610b3ece465bd5145414e0f3 Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 14:56:00 -0400 Subject: [PATCH 05/13] fix(slack): buffer native stream deltas --- src/slack/presenters.ts | 26 ++++++++++++---- test/slack-index.integration.test.ts | 3 +- test/slack-presenters.test.ts | 45 ++++++++++++++++++++++++---- 3 files changed, 61 insertions(+), 13 deletions(-) diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index 6a84edea..b0ad82b1 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -184,6 +184,8 @@ type NativeAgentChunk = status: "pending" | "in_progress" | "complete" | "error"; }; +const NATIVE_STREAM_BUFFER_SIZE = 256; + export interface NativeAgentPresenter { begin(): Promise; onDelta(delta: string): Promise; @@ -208,6 +210,7 @@ export function createNativeAgentPresenter(deps: { let chain = Promise.resolve(); let statusStarted = false; let statusCleared = false; + let pendingText = ""; const nativeTaskStatus = (status: RunTaskStatus): "pending" | "in_progress" | "complete" | "error" => { if (status === "completed" || status === "skipped") return "complete"; if (status === "failed") return "error"; @@ -259,6 +262,12 @@ export function createNativeAgentPresenter(deps: { }); return state === "active"; }; + const flushPendingText = (): Promise => { + if (!pendingText) return Promise.resolve(state === "starting" || state === "active"); + const text = pendingText; + pendingText = ""; + return send([{ type: "markdown_text", text }]); + }; return { async begin() { statusStarted = true; @@ -270,12 +279,17 @@ export function createNativeAgentPresenter(deps: { } }); }, - onDelta(delta) { - if (!delta) return Promise.resolve(state === "active"); - return send([{ type: "markdown_text", text: delta }]); + async onDelta(delta) { + if (!delta) return state === "starting" || state === "active"; + if (state === "disabled" || state === "stopped") return false; + pendingText += delta; + if (state === "idle") state = "starting"; + if (pendingText.length < NATIVE_STREAM_BUFFER_SIZE) return true; + return flushPendingText(); }, - onTasks(tasks) { - if (!tasks.length) return Promise.resolve(state === "active"); + async onTasks(tasks) { + if (!tasks.length) return state === "starting" || state === "active"; + if (pendingText && !(await flushPendingText())) return false; const chunks: NativeAgentChunk[] = tasks.slice(0, 20).map((task) => ({ type: "task_update", id: task.id, @@ -289,6 +303,7 @@ export function createNativeAgentPresenter(deps: { return send(chunks); }, async finalize() { + if (pendingText) await flushPendingText(); await chain; if (state === "orphaned") { await clearStatus(); @@ -310,6 +325,7 @@ export function createNativeAgentPresenter(deps: { } }, async settle() { + pendingText = ""; await chain; if (state === "active" && messageTs) { try { diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts index 4566f1f7..5c7b4318 100644 --- a/test/slack-index.integration.test.ts +++ b/test/slack-index.integration.test.ts @@ -447,12 +447,11 @@ test("a queued DM uses Slack native status, reply streaming, and task updates wi { channel: "D1", thread_ts: "100.9", - chunks: [{ type: "markdown_text", text: "Checking " }], + chunks: [{ type: "markdown_text", text: "Checking now." }], task_display_mode: "timeline", }, ]); assert.deepEqual(f.client.streamsAppended, [ - { channel: "D1", ts: "stream-1", chunks: [{ type: "markdown_text", text: "now." }] }, { channel: "D1", ts: "stream-1", diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts index b17a2b5c..aa85a25f 100644 --- a/test/slack-presenters.test.ts +++ b/test/slack-presenters.test.ts @@ -54,6 +54,39 @@ test("native agent presenter streams text and public task progress into one mess ]); }); +test("native agent presenter coalesces token deltas into Slack-sized stream updates", async () => { + const calls: Array<{ method: string; text?: string }> = []; + const presenter = createNativeAgentPresenter({ + setStatus: async () => {}, + start: async (chunks) => { + calls.push({ method: "start", text: chunks[0]?.type === "markdown_text" ? chunks[0].text : undefined }); + return "171.6"; + }, + append: async (_ts, chunks) => { + calls.push({ method: "append", text: chunks[0]?.type === "markdown_text" ? chunks[0].text : undefined }); + }, + stop: async () => { + calls.push({ method: "stop" }); + }, + remove: async () => {}, + checkpoint: async () => {}, + onSurfacePosted: () => {}, + }); + + for (const character of "x".repeat(600)) assert.equal(await presenter.onDelta(character), true); + assert.equal(await presenter.finalize(), true); + + assert.deepEqual( + calls.map(({ method, text }) => ({ method, length: text?.length })), + [ + { method: "start", length: 256 }, + { method: "append", length: 256 }, + { method: "append", length: 88 }, + { method: "stop", length: undefined }, + ], + ); +}); + test("native agent presenter falls back cleanly when Slack streaming is unavailable", async () => { const calls: string[] = []; const presenter = createNativeAgentPresenter({ @@ -81,7 +114,7 @@ test("native agent presenter falls back cleanly when Slack streaming is unavaila }); await presenter.begin(); - assert.equal(await presenter.onDelta("Answer"), false); + assert.equal(await presenter.onDelta("x".repeat(256)), false); assert.equal(await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]), false); assert.equal(await presenter.finalize(), false); assert.deepEqual(calls, ["status:Thinking…", "start", "error:unknown_method", "status:"]); @@ -115,7 +148,7 @@ test("native agent presenter removes an uncheckpointed stream before falling bac }); await presenter.begin(); - assert.equal(await presenter.onDelta("Answer"), false); + assert.equal(await presenter.onDelta("x".repeat(256)), false); assert.equal(await presenter.finalize(), false); assert.deepEqual(calls, [ "status:Thinking…", @@ -155,8 +188,8 @@ test("native agent presenter deletes a partial stream before fallback after appe }); await presenter.begin(); - assert.equal(await presenter.onDelta("Partial"), true); - assert.equal(await presenter.onDelta(" answer"), false); + assert.equal(await presenter.onDelta("x".repeat(256)), true); + assert.equal(await presenter.onDelta("y".repeat(256)), false); assert.equal(await presenter.finalize(), false); assert.deepEqual(calls, [ "status:Thinking…", @@ -213,8 +246,8 @@ test("native agent presenter suppresses fallback when a failed partial stream ca }); await presenter.begin(); - assert.equal(await presenter.onDelta("Partial"), true); - assert.equal(await presenter.onDelta(" answer"), false); + assert.equal(await presenter.onDelta("x".repeat(256)), true); + assert.equal(await presenter.onDelta("y".repeat(256)), false); assert.equal(await presenter.finalize(), true); }); From a8692c9000625b9d5746987b8920f0bbf0d09bfc Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 15:11:03 -0400 Subject: [PATCH 06/13] fix(slack): replay buffered stream prefix --- src/runs/turn-stream.ts | 2 ++ test/turn-stream.test.ts | 29 +++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/runs/turn-stream.ts b/src/runs/turn-stream.ts index 722bcef8..3319256f 100644 --- a/src/runs/turn-stream.ts +++ b/src/runs/turn-stream.ts @@ -172,6 +172,8 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream { listeners.set(runId, set); } set.add(listener); + const buffered = runs.get(runId)?.text; + if (buffered) listener.onDelta?.(buffered); return () => { set.delete(listener); if (set.size === 0 && listeners.get(runId) === set) listeners.delete(runId); diff --git a/test/turn-stream.test.ts b/test/turn-stream.test.ts index 86a8516b..6e941b45 100644 --- a/test/turn-stream.test.ts +++ b/test/turn-stream.test.ts @@ -6,6 +6,7 @@ import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { createTurnStream } from "../src/runs/turn-stream.ts"; +import { createStreamingReplyFilter } from "../src/slack/messaging.ts"; import { buildApp } from "../src/wiring.ts"; import { testConfig } from "./support/test-config.ts"; @@ -37,6 +38,34 @@ test("subscribers receive public reply deltas in order", () => { assert.deepEqual(deltas, ["Hel", "lo"]); }); +test("a late subscriber receives the buffered prefix before live deltas", () => { + const s = createTurnStream(); + s.publish("r1", "Hel"); + const deltas: string[] = []; + s.subscribe("r1", { + onDelta: (delta) => { + deltas.push(delta); + }, + }); + s.publish("r1", "lo"); + assert.deepEqual(deltas, ["Hel", "lo"]); +}); + +test("a directive split across buffered and live deltas stays private", () => { + const s = createTurnStream(); + const filter = createStreamingReplyFilter(); + const publicDeltas: string[] = []; + s.publish("r1", "Public. [[ask-agent:"); + s.subscribe("r1", { + onDelta: (delta) => { + publicDeltas.push(filter.push(delta)); + }, + }); + s.publish("r1", " <@U2> | private context]] Done."); + publicDeltas.push(filter.flush()); + assert.equal(publicDeltas.join(""), "Public. Done."); +}); + test("ignores empty deltas", () => { const s = createTurnStream(); s.publish("r1", ""); From d08295d374fe778d9372066f75db8813b8bc7290 Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 18:01:34 -0400 Subject: [PATCH 07/13] fix(github): support per-user OAuth in native skill --- skills-seed/github-gitlab/SKILL.md | 20 ++++++++++++++++---- test/skills-seed.test.ts | 9 +++++++++ 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/skills-seed/github-gitlab/SKILL.md b/skills-seed/github-gitlab/SKILL.md index 20fcd8d3..676a284d 100644 --- a/skills-seed/github-gitlab/SKILL.md +++ b/skills-seed/github-gitlab/SKILL.md @@ -12,9 +12,20 @@ requiredCapabilities: Use this skill when the user asks to inspect repos, issues, pull requests, merge requests, code history, branches, or to make a small code change in a hosted repo. -This is a resident-machine-auth connector. Prefer the native CLIs (`gh`, `glab`) and -`git`, using the agent computer's logged-in state. Do not ask the user to paste tokens, -and do not rely on proxy bearer-token injection. +Prefer the native CLIs (`gh`, `glab`) and `git`. GitHub supports either the requesting +user's product OAuth token or resident machine auth. When +`$VAULT_TOKEN_API_GITHUB_COM` is present, pass it to `gh` only through `GH_TOKEN` for +the command being run: + +```bash +GH_TOKEN="$VAULT_TOKEN_API_GITHUB_COM" gh auth status +GH_TOKEN="$VAULT_TOKEN_API_GITHUB_COM" gh repo view OWNER/REPO --json name,description,url,defaultBranchRef +``` + +Never print either variable, persist it in a file, or use another principal's token. If +the vault variable is absent in a direct DM, the user has not connected GitHub through +the product; use resident login only when the computer profile says durable process +sessions are supported. Do not ask the user to paste a token. One exception: if the system prompt lists a shared org credential for a Git remote, the token is broker-only and never appears on the computer. For clone/fetch/push, use the @@ -24,7 +35,8 @@ server-side. ## Logging in -If `gh auth status` (or `glab auth status`) fails, log in with the native command: +When no product OAuth token is available and `gh auth status` (or `glab auth status`) +fails, log in with the native command only on a computer with durable process sessions: ```bash gh auth login diff --git a/test/skills-seed.test.ts b/test/skills-seed.test.ts index 02d56e09..cb934d3f 100644 --- a/test/skills-seed.test.ts +++ b/test/skills-seed.test.ts @@ -206,6 +206,15 @@ test("a fresh app advertises and materializes only admin-enabled connector skill } as TurnRequest); assert.match(drive.reply ?? "", /Google Drive \/ Docs \/ Sheets \/ Slides/); assert.match(drive.reply ?? "", /sheets\.googleapis\.com/); + + const github = await app.turn({ + surface: "test", + actor, + conversation: { kind: "dm", threadRef: "dm:U1:seeded-github-read" }, + text: "!read skills/github-gitlab/SKILL.md", + } as TurnRequest); + assert.match(github.reply ?? "", /VAULT_TOKEN_API_GITHUB_COM/); + assert.match(github.reply ?? "", /GH_TOKEN/); }); function fakeSandbox() { From bd0eaec14b1b39844ce9d7223cef64664eaca35e Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 20:46:01 -0400 Subject: [PATCH 08/13] fix(slack): keep draft tokens out of process args --- .../slack-drafts/scripts/slack_drafts.py | 24 +++++++++++-------- test/skill-conformance.test.ts | 7 ++++++ 2 files changed, 21 insertions(+), 10 deletions(-) diff --git a/skills-seed/slack-drafts/scripts/slack_drafts.py b/skills-seed/slack-drafts/scripts/slack_drafts.py index 982bc8b4..83294ca1 100755 --- a/skills-seed/slack-drafts/scripts/slack_drafts.py +++ b/skills-seed/slack-drafts/scripts/slack_drafts.py @@ -15,9 +15,10 @@ import json import os import re -import subprocess import sys +import urllib.error import urllib.parse +import urllib.request import uuid API = "https://slack.com/api" @@ -28,18 +29,21 @@ def call(method: str, body: dict | None = None, query: dict | None = None): if not tok: sys.exit("no Slack token: ask the user to connect Slack") url = f"{API}/{method}" + (f"?{urllib.parse.urlencode(query, doseq=True)}" if query else "") - cmd = ["curl", "-sS", "--fail-with-body", "--max-time", "60", - "-H", f"Authorization: Bearer {tok}", url] + headers = {"Authorization": f"Bearer {tok}"} + data = None if body is not None: - cmd += ["-H", "Content-Type: application/json; charset=utf-8", "--data-binary", "@-"] - proc = subprocess.run(cmd, input=json.dumps(body) if body is not None else None, - capture_output=True, text=True) - if proc.returncode != 0: - sys.exit(f"slack api unreachable on {method}: {proc.stderr.strip()[:300]}") + headers["Content-Type"] = "application/json; charset=utf-8" + data = json.dumps(body).encode("utf-8") + request = urllib.request.Request(url, data=data, headers=headers) try: - payload = json.loads(proc.stdout) + with urllib.request.urlopen(request, timeout=60) as response: + response_text = response.read().decode("utf-8") + except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as error: + sys.exit(f"slack api unreachable on {method}: {str(error)[:300]}") + try: + payload = json.loads(response_text) except ValueError: - sys.exit(f"slack api returned non-JSON on {method}: {proc.stdout[:300]}") + sys.exit(f"slack api returned non-JSON on {method}: {response_text[:300]}") if not payload.get("ok"): err = payload.get("error") hints = { diff --git a/test/skill-conformance.test.ts b/test/skill-conformance.test.ts index c67b8f57..b2144f11 100644 --- a/test/skill-conformance.test.ts +++ b/test/skill-conformance.test.ts @@ -21,3 +21,10 @@ test("every seed SKILL.md parses with a name, description, and body", () => { assert.ok(m.name && m.description && m.body.trim(), `${path} is missing a required field`); } }); + +test("Slack draft OAuth tokens never enter subprocess arguments", () => { + const script = readFileSync(join(SEED_DIR, "slack-drafts", "scripts", "slack_drafts.py"), "utf8"); + assert.doesNotMatch(script, /subprocess|\["curl"/); + assert.match(script, /urllib\.request\.Request/); + assert.match(script, /"Authorization": f"Bearer \{tok\}"/); +}); From a87dadda0620246249f0a8975e07686ea6e5d3d3 Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 20:59:44 -0400 Subject: [PATCH 09/13] fix(slack): preserve task-only terminal replies --- src/slack/turn-handler.ts | 8 ++++++-- test/slack-index.integration.test.ts | 26 ++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/src/slack/turn-handler.ts b/src/slack/turn-handler.ts index e45fa6fe..fc5097bd 100644 --- a/src/slack/turn-handler.ts +++ b/src/slack/turn-handler.ts @@ -437,6 +437,7 @@ export function createTurnHandler(deps: { void nativeAgent.begin(); } const streamingReply = createStreamingReplyFilter(); + let streamedNativeText = false; const turn: Omit = { actor, @@ -492,7 +493,7 @@ export function createTurnHandler(deps: { ? { onDelta: async (delta: string) => { const visible = streamingReply.push(delta); - if (visible) await nativeAgent?.onDelta(visible); + if (visible && (await nativeAgent?.onDelta(visible))) streamedNativeText = true; }, onFirstBlock: (blockText: string) => { if (!nativeAgent?.ownsSurface()) ack.onFirstBlock(cleanAgentReplyForSlack(blockText).text); @@ -564,7 +565,10 @@ export function createTurnHandler(deps: { const tDeliverStart = performance.now(); let finalizedTaskList = false; const finalStreamDelta = streamingReply.flush(); - if (finalStreamDelta) await nativeAgent?.onDelta(finalStreamDelta); + if (finalStreamDelta && (await nativeAgent?.onDelta(finalStreamDelta))) streamedNativeText = true; + if (postText && !streamedNativeText && nativeAgent?.ownsSurface()) { + await nativeAgent.onDelta(postText); + } const finalizedNative = (await nativeAgent?.finalize()) ?? false; if (result.attachments?.length) { let uploadError: unknown; diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts index 5c7b4318..bbbe0109 100644 --- a/test/slack-index.integration.test.ts +++ b/test/slack-index.integration.test.ts @@ -465,6 +465,32 @@ test("a queued DM uses Slack native status, reply streaming, and task updates wi } }); +test("a task-only native stream appends the terminal reply before stopping", async () => { + const f = await fixture(); + try { + f.core.queuedRunId = "R-task-only"; + f.core.streamTasks = [{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]; + f.core.result = { status: "ok", reply: "The deployment is healthy." }; + + await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "check it", ts: "100.10" }); + + assert.deepEqual(f.client.streamsStarted[0]?.chunks, [ + { type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }, + ]); + assert.deepEqual(f.client.streamsAppended, [ + { + channel: "D1", + ts: "stream-1", + chunks: [{ type: "markdown_text", text: "The deployment is healthy." }], + }, + ]); + assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]); + assert.equal(f.client.posts.length, 0); + } finally { + await f.stop(); + } +}); + test("a human's DM sets the conversation header to the serving model + web surface", async () => { const f = await fixture({ webUiPublicUrl: "https://claw.example.dev" }); try { From a7738354b5babd2d03eb95103112b68733375c41 Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 21:10:36 -0400 Subject: [PATCH 10/13] fix(stream): drain async delta listeners --- src/api/slack-core-client.ts | 1 + src/runs/turn-stream.ts | 41 ++++++++++++++++++++++++++++++++++-- test/turn-stream.test.ts | 31 ++++++++++++++++++++++++--- 3 files changed, 68 insertions(+), 5 deletions(-) diff --git a/src/api/slack-core-client.ts b/src/api/slack-core-client.ts index 988b98b1..6e979b55 100644 --- a/src/api/slack-core-client.ts +++ b/src/api/slack-core-client.ts @@ -223,6 +223,7 @@ export function createSlackCoreClient(deps: SlackCoreClientDeps): SlackCoreClien if (isTerminal(run.status)) { const view = await deps.app.getRun(runId); await emitTasks().catch(swallowAs("slack-core-client: terminal task refresh", undefined)); + await deps.turnStream.drain(runId); if (view?.surfacePosted) signalSurface(); return (view?.result as TurnResult | null | undefined) ?? null; } diff --git a/src/runs/turn-stream.ts b/src/runs/turn-stream.ts index 3319256f..97d3587d 100644 --- a/src/runs/turn-stream.ts +++ b/src/runs/turn-stream.ts @@ -9,6 +9,7 @@ export interface TurnStream { markSurfacePosted(runId: string): void; surfacePosted(runId: string): boolean; snapshot(runId: string): string | null; + drain(runId: string): Promise; markReplyDone(runId: string): void; isReplyDone(runId: string): boolean; end(runId: string): void; @@ -33,6 +34,11 @@ interface Entry { timer: ReturnType | null; } +interface ListenerDelivery { + tail: Promise; + error?: unknown; +} + export interface TurnStreamOptions { maxChars?: number; graceMs?: number; @@ -63,6 +69,19 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream { const graceMs = opts.graceMs ?? DEFAULT_GRACE_MS; const runs = new Map(); const listeners = new Map>(); + const deliveries = new Map>(); + + const enqueueDelta = (runId: string, listener: TurnStreamListener, delta: string): void => { + const delivery = deliveries.get(runId)?.get(listener); + if (!delivery || !listener.onDelta) return; + delivery.tail = delivery.tail.then(async () => { + try { + await listener.onDelta?.(delta); + } catch (error) { + delivery.error ??= error; + } + }); + }; const ensure = (runId: string): Entry => { let entry = runs.get(runId); @@ -102,7 +121,7 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream { if (entry.firstBlockOpen && entry.firstBlock.length < FIRST_BLOCK_MAX_CHARS) entry.firstBlock = (entry.firstBlock + delta).slice(0, FIRST_BLOCK_MAX_CHARS); if (entry.text.length < maxChars) entry.text = (entry.text + delta).slice(0, maxChars); - for (const l of listeners.get(runId) ?? []) l.onDelta?.(delta); + for (const l of listeners.get(runId) ?? []) enqueueDelta(runId, l, delta); }, publishBlockStart(runId) { @@ -145,6 +164,13 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream { return text ? text : null; }, + async drain(runId) { + const pending = [...(deliveries.get(runId)?.values() ?? [])]; + await Promise.all(pending.map((delivery) => delivery.tail)); + const failed = pending.find((delivery) => delivery.error !== undefined); + if (failed) throw failed.error; + }, + markReplyDone(runId) { const entry = runs.get(runId); if (entry) entry.replyDone = true; @@ -172,11 +198,22 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream { listeners.set(runId, set); } set.add(listener); + let deliveryMap = deliveries.get(runId); + if (!deliveryMap) { + deliveryMap = new Map(); + deliveries.set(runId, deliveryMap); + } + deliveryMap.set(listener, { tail: Promise.resolve() }); const buffered = runs.get(runId)?.text; - if (buffered) listener.onDelta?.(buffered); + if (buffered) enqueueDelta(runId, listener, buffered); return () => { set.delete(listener); if (set.size === 0 && listeners.get(runId) === set) listeners.delete(runId); + const delivery = deliveryMap.get(listener); + void delivery?.tail.finally(() => { + if (deliveryMap.get(listener) === delivery) deliveryMap.delete(listener); + if (deliveryMap.size === 0 && deliveries.get(runId) === deliveryMap) deliveries.delete(runId); + }); }; }, }; diff --git a/test/turn-stream.test.ts b/test/turn-stream.test.ts index 6e941b45..95cf9c93 100644 --- a/test/turn-stream.test.ts +++ b/test/turn-stream.test.ts @@ -22,7 +22,7 @@ test("accumulates deltas per run and isolates runs", () => { assert.equal(s.snapshot("r2"), "world"); }); -test("subscribers receive public reply deltas in order", () => { +test("subscribers receive public reply deltas in order", async () => { const s = createTurnStream(); const deltas: string[] = []; const unsubscribe = s.subscribe("r1", { @@ -35,10 +35,11 @@ test("subscribers receive public reply deltas in order", () => { s.publish("r2", "private to another run"); unsubscribe(); s.publish("r1", "!"); + await s.drain("r1"); assert.deepEqual(deltas, ["Hel", "lo"]); }); -test("a late subscriber receives the buffered prefix before live deltas", () => { +test("a late subscriber receives the buffered prefix before live deltas", async () => { const s = createTurnStream(); s.publish("r1", "Hel"); const deltas: string[] = []; @@ -48,10 +49,33 @@ test("a late subscriber receives the buffered prefix before live deltas", () => }, }); s.publish("r1", "lo"); + await s.drain("r1"); assert.deepEqual(deltas, ["Hel", "lo"]); }); -test("a directive split across buffered and live deltas stays private", () => { +test("drain waits for asynchronous delta listeners and preserves delivery order", async () => { + const s = createTurnStream(); + const deltas: string[] = []; + let releaseFirst: (() => void) | undefined; + const firstGate = new Promise((resolve) => (releaseFirst = resolve)); + s.subscribe("r1", { + onDelta: async (delta) => { + if (delta === "Hel") await firstGate; + deltas.push(delta); + }, + }); + + s.publish("r1", "Hel"); + s.publish("r1", "lo"); + const draining = s.drain("r1"); + await Promise.resolve(); + assert.deepEqual(deltas, []); + releaseFirst?.(); + await draining; + assert.deepEqual(deltas, ["Hel", "lo"]); +}); + +test("a directive split across buffered and live deltas stays private", async () => { const s = createTurnStream(); const filter = createStreamingReplyFilter(); const publicDeltas: string[] = []; @@ -62,6 +86,7 @@ test("a directive split across buffered and live deltas stays private", () => { }, }); s.publish("r1", " <@U2> | private context]] Done."); + await s.drain("r1"); publicDeltas.push(filter.flush()); assert.equal(publicDeltas.join(""), "Public. Done."); }); From b7896c76d6c48938b47f40cfb6b50633948bad2d Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 21:45:37 -0400 Subject: [PATCH 11/13] fix(slack): bound native stream chunks --- src/slack/presenters.ts | 14 ++++++++----- test/slack-index.integration.test.ts | 30 +++++++++++++++++++++++++++- 2 files changed, 38 insertions(+), 6 deletions(-) diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index b0ad82b1..b1acb12e 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -185,6 +185,7 @@ type NativeAgentChunk = }; const NATIVE_STREAM_BUFFER_SIZE = 256; +const NATIVE_STREAM_CHUNK_SIZE = 12_000; export interface NativeAgentPresenter { begin(): Promise; @@ -262,11 +263,14 @@ export function createNativeAgentPresenter(deps: { }); return state === "active"; }; - const flushPendingText = (): Promise => { - if (!pendingText) return Promise.resolve(state === "starting" || state === "active"); - const text = pendingText; - pendingText = ""; - return send([{ type: "markdown_text", text }]); + const flushPendingText = async (): Promise => { + if (!pendingText) return state === "starting" || state === "active"; + while (pendingText) { + const text = pendingText.slice(0, NATIVE_STREAM_CHUNK_SIZE); + pendingText = pendingText.slice(NATIVE_STREAM_CHUNK_SIZE); + if (!(await send([{ type: "markdown_text", text }]))) return false; + } + return true; }; return { async begin() { diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts index bbbe0109..b5bfd423 100644 --- a/test/slack-index.integration.test.ts +++ b/test/slack-index.integration.test.ts @@ -267,7 +267,7 @@ class FakeCore implements SlackCoreClient { async waitRun(runId: string, hooks: any = {}): Promise { this.polled.push(runId); if (this.runGate) await this.runGate; - for (const delta of this.streamDeltas) hooks.onDelta?.(delta); + for (const delta of this.streamDeltas) await hooks.onDelta?.(delta); if (this.streamTasks.length) await hooks.onTasks?.(this.streamTasks); return this.result; } @@ -491,6 +491,34 @@ test("a task-only native stream appends the terminal reply before stopping", asy } }); +test("native Slack streaming splits large model deltas at the API chunk limit", async () => { + const f = await fixture(); + try { + const reply = "x".repeat(24_001); + f.core.queuedRunId = "R-large-delta"; + f.core.streamDeltas = [reply]; + f.core.result = { status: "ok", reply }; + + await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "write it", ts: "100.11" }); + + const texts = [ + ...f.client.streamsStarted.flatMap((request) => request.chunks), + ...f.client.streamsAppended.flatMap((request) => request.chunks), + ] + .filter((chunk) => chunk.type === "markdown_text") + .map((chunk) => chunk.text); + assert.deepEqual( + texts.map((text) => text.length), + [12_000, 12_000, 1], + ); + assert.equal(texts.join(""), reply); + assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]); + assert.equal(f.client.posts.length, 0); + } finally { + await f.stop(); + } +}); + test("a human's DM sets the conversation header to the serving model + web surface", async () => { const f = await fixture({ webUiPublicUrl: "https://claw.example.dev" }); try { From c40f0a3a81875d8dcac70b3a545fed2fb6987e9d Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 22:27:00 -0400 Subject: [PATCH 12/13] fix(slack): hold reply text until terminal approval --- src/slack/presenters.ts | 4 ++- src/slack/turn-handler.ts | 23 ++------------- test/slack-index.integration.test.ts | 41 +++++++++++++++++++++------ test/slack-presenters.test.ts | 42 ++++++++++++++++++++++++++++ 4 files changed, 81 insertions(+), 29 deletions(-) diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index b1acb12e..1ccb86d3 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -333,9 +333,11 @@ export function createNativeAgentPresenter(deps: { await chain; if (state === "active" && messageTs) { try { - await deps.stop(messageTs); + await deps.remove(messageTs); + messageTs = undefined; state = "stopped"; } catch (error) { + state = "orphaned"; deps.onError?.(error); } } diff --git a/src/slack/turn-handler.ts b/src/slack/turn-handler.ts index fc5097bd..7a10dc66 100644 --- a/src/slack/turn-handler.ts +++ b/src/slack/turn-handler.ts @@ -61,7 +61,6 @@ import { type SlackConversationKind, applyAndLogReactions, cleanAgentReplyForSlack, - createStreamingReplyFilter, conversationPlaceLabel, slackSurfaceInstructions, } from "./messaging.ts"; @@ -436,9 +435,6 @@ export function createTurnHandler(deps: { }); void nativeAgent.begin(); } - const streamingReply = createStreamingReplyFilter(); - let streamedNativeText = false; - const turn: Omit = { actor, conversation: { @@ -489,18 +485,7 @@ export function createTurnHandler(deps: { // Folded into a live run: the envelope is durably accepted just the same, but the run // stays pinned to its own handler — claiming it here would unpin it on the way out. onSteered: () => inc.ackGate?.persisted(), - ...(ack - ? { - onDelta: async (delta: string) => { - const visible = streamingReply.push(delta); - if (visible && (await nativeAgent?.onDelta(visible))) streamedNativeText = true; - }, - onFirstBlock: (blockText: string) => { - if (!nativeAgent?.ownsSurface()) ack.onFirstBlock(cleanAgentReplyForSlack(blockText).text); - }, - onSurfacePosted: () => ack.onSurfacePosted(), - } - : {}), + ...(ack ? { onSurfacePosted: () => ack.onSurfacePosted() } : {}), ...(taskList ? { onTasks: async (tasks: RunTaskView[]) => { @@ -564,10 +549,8 @@ export function createTurnHandler(deps: { const postText = reply; const tDeliverStart = performance.now(); let finalizedTaskList = false; - const finalStreamDelta = streamingReply.flush(); - if (finalStreamDelta && (await nativeAgent?.onDelta(finalStreamDelta))) streamedNativeText = true; - if (postText && !streamedNativeText && nativeAgent?.ownsSurface()) { - await nativeAgent.onDelta(postText); + if (postText && (queuedRunId || nativeAgent?.ownsSurface())) { + await nativeAgent?.onDelta(postText); } const finalizedNative = (await nativeAgent?.finalize()) ?? false; if (result.attachments?.length) { diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts index b5bfd423..753e6c76 100644 --- a/test/slack-index.integration.test.ts +++ b/test/slack-index.integration.test.ts @@ -395,11 +395,12 @@ test("a mid-turn message that STEERS the live run does not post the reply twice" f.core.finishRun({ status: "ok", reply: "agent reply" }); await Promise.all([first, steer]); - assert.equal( - f.client.posts.filter((p) => p.text === "agent reply").length, - 1, - "the shared run's reply is posted once, by the handler that owns it", - ); + const deliveredReplies = [ + ...f.client.posts.map((post) => post.text), + ...f.client.streamsStarted.flatMap((request) => request.chunks.map((chunk: { text?: string }) => chunk.text)), + ...f.client.streamsAppended.flatMap((request) => request.chunks.map((chunk: { text?: string }) => chunk.text)), + ].filter((text) => text === "agent reply"); + assert.equal(deliveredReplies.length, 1, "the shared run's reply is delivered once, by the handler that owns it"); } finally { await f.stop(); } @@ -429,7 +430,7 @@ test("a DM becomes one scoped live turn and one Slack reply", async () => { } }); -test("a queued DM uses Slack native status, reply streaming, and task updates without a duplicate post", async () => { +test("a queued DM keeps model text private until terminal approval and uses native task updates", async () => { const f = await fixture(); try { f.core.queuedRunId = "R-stream"; @@ -447,7 +448,7 @@ test("a queued DM uses Slack native status, reply streaming, and task updates wi { channel: "D1", thread_ts: "100.9", - chunks: [{ type: "markdown_text", text: "Checking now." }], + chunks: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }], task_display_mode: "timeline", }, ]); @@ -455,7 +456,7 @@ test("a queued DM uses Slack native status, reply streaming, and task updates wi { channel: "D1", ts: "stream-1", - chunks: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }], + chunks: [{ type: "markdown_text", text: "Checking now." }], }, ]); assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]); @@ -465,6 +466,30 @@ test("a queued DM uses Slack native status, reply streaming, and task updates wi } }); +test("a rejected queued turn deletes its provisional native task stream", async () => { + const f = await fixture(); + try { + f.core.queuedRunId = "R-silent"; + f.core.streamDeltas = ["must never be shown"]; + f.core.streamTasks = [{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]; + f.core.result = { status: "silent" }; + + await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "check it", ts: "100.12" }); + + assert.equal( + f.client.streamsStarted.some((request) => + request.chunks.some((chunk: { type: string; text?: string }) => chunk.text === "must never be shown"), + ), + false, + ); + assert.deepEqual(f.client.deletes, [{ channel: "D1", ts: "stream-1" }]); + assert.deepEqual(f.client.streamsStopped, []); + assert.equal(f.client.posts.length, 0); + } finally { + await f.stop(); + } +}); + test("a task-only native stream appends the terminal reply before stopping", async () => { const f = await fixture(); try { diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts index aa85a25f..9913d769 100644 --- a/test/slack-presenters.test.ts +++ b/test/slack-presenters.test.ts @@ -251,6 +251,48 @@ test("native agent presenter suppresses fallback when a failed partial stream ca assert.equal(await presenter.finalize(), true); }); +test("native agent presenter deletes a provisional stream when the turn is abandoned", async () => { + const calls: string[] = []; + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push(`status:${status}`); + }, + start: async () => { + calls.push("start"); + return "171.10"; + }, + append: async () => { + calls.push("append"); + }, + stop: async () => { + calls.push("stop"); + }, + remove: async (ts) => { + calls.push(`remove:${ts}`); + }, + checkpoint: async () => { + calls.push("checkpoint"); + }, + onSurfacePosted: () => calls.push("surface"), + }); + + await presenter.begin(); + assert.equal( + await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]), + true, + ); + await presenter.settle(); + + assert.deepEqual(calls, [ + "status:Thinking…", + "start", + "checkpoint", + "surface", + "remove:171.10", + "status:", + ]); +}); + test("renderTaskList renders every terminal state", () => { assert.equal( renderTaskList([ From 13af32e371d6c5adfcbb57ad925eb546590dd83e Mon Sep 17 00:00:00 2001 From: puffcooks Date: Tue, 11 Aug 2026 22:50:07 -0400 Subject: [PATCH 13/13] fix(slack): preserve approved stream on stop failure --- src/slack/presenters.ts | 1 + test/slack-presenters.test.ts | 38 +++++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts index 1ccb86d3..b94a74f9 100644 --- a/src/slack/presenters.ts +++ b/src/slack/presenters.ts @@ -323,6 +323,7 @@ export function createNativeAgentPresenter(deps: { await clearStatus(); return true; } catch (error) { + state = "orphaned"; deps.onError?.(error); await clearStatus(); return true; diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts index 9913d769..97caa69a 100644 --- a/test/slack-presenters.test.ts +++ b/test/slack-presenters.test.ts @@ -251,6 +251,44 @@ test("native agent presenter suppresses fallback when a failed partial stream ca assert.equal(await presenter.finalize(), true); }); +test("native agent presenter preserves approved text when stopping its stream fails", async () => { + const calls: string[] = []; + const presenter = createNativeAgentPresenter({ + setStatus: async (status) => { + calls.push(`status:${status}`); + }, + start: async () => { + calls.push("start"); + return "171.11"; + }, + append: async () => {}, + stop: async (ts) => { + calls.push(`stop:${ts}`); + throw new Error("stop failed"); + }, + remove: async (ts) => { + calls.push(`remove:${ts}`); + }, + checkpoint: async () => { + calls.push("checkpoint"); + }, + onSurfacePosted: () => calls.push("surface"), + onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`), + }); + + assert.equal(await presenter.onDelta("approved".repeat(40)), true); + assert.equal(await presenter.finalize(), true); + await presenter.settle(); + + assert.deepEqual(calls, [ + "start", + "checkpoint", + "surface", + "stop:171.11", + "error:stop failed", + ]); +}); + test("native agent presenter deletes a provisional stream when the turn is abandoned", async () => { const calls: string[] = []; const presenter = createNativeAgentPresenter({