From de0cf8d293443fd65ecad9be7028a1704b8b17d4 Mon Sep 17 00:00:00 2001 From: xingmiao Date: Sat, 4 Jul 2026 22:22:28 +0800 Subject: [PATCH 1/2] Fix channel realtime refresh for agent replies --- apps/web/features/channels/channels-page-client.test.tsx | 1 + apps/web/features/channels/channels-page-client.tsx | 6 +++--- apps/web/next.config.mjs | 2 ++ 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/apps/web/features/channels/channels-page-client.test.tsx b/apps/web/features/channels/channels-page-client.test.tsx index d140daa..6daa9fe 100644 --- a/apps/web/features/channels/channels-page-client.test.tsx +++ b/apps/web/features/channels/channels-page-client.test.tsx @@ -402,6 +402,7 @@ describe("ChannelsPageClient", () => { const composer = screen.getByPlaceholderText("发送到 tour visit"); fireEvent.change(composer, { target: { value: "草稿", selectionStart: 2 } }); + composer.focus(); expect(eventSources[0]?.url).toBe("/api/workspaces/workspace-1/channels/tour%20visit/events"); const activeSource = eventSources.at(-1); diff --git a/apps/web/features/channels/channels-page-client.tsx b/apps/web/features/channels/channels-page-client.tsx index 14b7003..def2d9f 100644 --- a/apps/web/features/channels/channels-page-client.tsx +++ b/apps/web/features/channels/channels-page-client.tsx @@ -822,11 +822,11 @@ export function ChannelsPageClient({ const selectedChannel = selectedChannelId ? visibleChannelById.get(selectedChannelId) ?? null : null; const selectedConversationChannelName = resolveSelectedChannelName(selectedChannel); - const refreshChannelData = useCallback(() => { + const refreshChannelData = useCallback((options?: { allowWhileInputActive?: boolean }) => { if (typeof document !== "undefined" && document.visibilityState === "hidden") { return; } - if (isDocumentInputActive()) { + if (!options?.allowWhileInputActive && isDocumentInputActive()) { return; } if (refreshInFlightRef.current || transitionPendingRef.current) { @@ -943,7 +943,7 @@ export function ChannelsPageClient({ Boolean(selectedConversationChannelName) && !selectedChannelRequiresAccess, onInvalidation, - refresh: refreshChannelData, + refresh: () => refreshChannelData({ allowWhileInputActive: true }), }); useEffect(() => { diff --git a/apps/web/next.config.mjs b/apps/web/next.config.mjs index ceab4c3..1268b0b 100644 --- a/apps/web/next.config.mjs +++ b/apps/web/next.config.mjs @@ -16,6 +16,7 @@ const nextConfig = { transpilePackages: [ "@agent-space/db", "@agent-space/domain", + "@agent-space/sandbox", "@agent-space/services", "agent-space-daemon", ], @@ -63,6 +64,7 @@ const nextConfig = { "@agent-space/db": "../../packages/db/src/index.ts", "@agent-space/domain": "../../packages/domain/src/index.ts", "@agent-space/domain/workspace": "../../packages/domain/src/workspace.ts", + "@agent-space/sandbox": "../../packages/sandbox/src/index.ts", "@agent-space/services": "../../packages/services/src/index.ts", "agent-space-daemon": "../../packages/daemon/src/index.ts", "agent-space-daemon/agent-router": "../../packages/daemon/src/agent-router/index.ts", From a4d99523f51fd0b24d33082e1c38c8d4f4d43e65 Mon Sep 17 00:00:00 2001 From: xingmiao Date: Sat, 4 Jul 2026 23:10:03 +0800 Subject: [PATCH 2/2] Stream agent reply deltas in channels --- .../app/api/daemon/client.integration.test.ts | 31 +++++++ .../daemon/tasks/[taskId]/messages/route.ts | 40 ++++++++- .../[channelName]/events/route.test.ts | 20 ++++- .../channels/[channelName]/events/route.ts | 8 ++ .../channels/channels-page-client.test.tsx | 7 +- .../channels/channels-page-client.tsx | 2 + .../features/chat/chat-primitives.test.tsx | 82 ++++++++++++++++- apps/web/features/chat/chat-primitives.tsx | 76 ++++++++++++++-- .../src/agent-router/adapters/opencode.ts | 33 ++++++- packages/daemon/src/provider-runtime.test.ts | 48 +++++++++- packages/daemon/src/remote-daemon.ts | 41 +++++++-- packages/services/src/index.ts | 2 + .../services/src/messages/messages.test.ts | 71 +++++++++++++++ packages/services/src/messages/messages.ts | 87 +++++++++++++++++++ packages/services/src/realtime/events.ts | 26 ++++++ 15 files changed, 547 insertions(+), 27 deletions(-) diff --git a/apps/web/app/api/daemon/client.integration.test.ts b/apps/web/app/api/daemon/client.integration.test.ts index 012b172..8126e3a 100644 --- a/apps/web/app/api/daemon/client.integration.test.ts +++ b/apps/web/app/api/daemon/client.integration.test.ts @@ -171,6 +171,37 @@ describe("remote daemon client integration", () => { ], }); + await client.reportMessages(claimed.task.id, { + messages: [ + { + type: "text", + content: "我先", + }, + ], + }); + await client.reportMessages(claimed.task.id, { + messages: [ + { + type: "text", + content: "给你整理。", + }, + ], + }); + + const streamingState = readWorkspaceStateSync(); + const streamingDirectChannel = streamingState.channels.find( + (channel) => channel.kind === "direct" && channel.employeeNames.some((name) => name === "Atlas"), + ); + const streamingPending = streamingState.messages.find( + (message) => + message.channel === streamingDirectChannel?.name && + message.role === "agent" && + message.status === "pending" && + message.speaker === "Atlas", + ); + expect(streamingPending?.summary).toBe("我先给你整理。"); + expect(streamingPending?.data?.stream_started).toBe("true"); + await client.uploadOutputBundle(claimed.task.id, { version: 1, format: "json-inline-v1", diff --git a/apps/web/app/api/daemon/tasks/[taskId]/messages/route.ts b/apps/web/app/api/daemon/tasks/[taskId]/messages/route.ts index 7b70844..7300de4 100644 --- a/apps/web/app/api/daemon/tasks/[taskId]/messages/route.ts +++ b/apps/web/app/api/daemon/tasks/[taskId]/messages/route.ts @@ -5,6 +5,8 @@ import { type QueuedTaskRecord, } from "@agent-space/db"; import type { DaemonProvider, DaemonTaskMessageInput, ReportTaskMessagesRequest } from "@agent-space/domain"; +import { parseTaskPayload } from "agent-space-daemon"; +import { streamAgentChannelReplyDeltaSync } from "@agent-space/services"; import { readTaskForWorkspace, requireDaemonAuth } from "../../../_lib/auth"; export const runtime = "nodejs"; @@ -30,14 +32,20 @@ export async function POST( return Response.json({ error: "messages[] is required." }, { status: 400 }); } - const appended = body.messages.map((message) => appendSingleMessage(task, message)); + const payload = parseTaskPayload(task); + const appended = body.messages.map((message) => appendSingleMessage(task, payload, message)); return Response.json({ messages: appended }); } -function appendSingleMessage(task: QueuedTaskRecord, message: DaemonTaskMessageInput) { +function appendSingleMessage( + task: QueuedTaskRecord, + payload: ReturnType, + message: DaemonTaskMessageInput, +) { if (message.type === "provider_session_invalid") { handleProviderSessionInvalid(task, message); } + streamChannelReplyDeltaBestEffort(task, payload, message); return appendTaskMessageSync({ taskId: task.id, type: message.type === "provider_session_invalid" ? "status" : message.type, @@ -48,6 +56,34 @@ function appendSingleMessage(task: QueuedTaskRecord, message: DaemonTaskMessageI }); } +function streamChannelReplyDeltaBestEffort( + task: QueuedTaskRecord, + payload: ReturnType, + message: DaemonTaskMessageInput, +): void { + if (message.type !== "text" || !message.content) { + return; + } + const channelName = payload.channelName ?? payload.channel; + const pendingSpeaker = payload.assignee ?? payload.contactId ?? task.agentId; + if (!channelName || !pendingSpeaker) { + return; + } + + try { + streamAgentChannelReplyDeltaSync({ + channel: channelName, + pendingSpeaker, + delta: message.content, + sourceTaskQueueId: task.id, + }, task.workspaceId); + } catch (error) { + console.error( + `Failed to stream channel reply delta for ${task.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } +} + function handleProviderSessionInvalid(task: QueuedTaskRecord, message: DaemonTaskMessageInput): void { if (!task.routerSessionId) { return; diff --git a/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.test.ts b/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.test.ts index b284be5..545f421 100644 --- a/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.test.ts +++ b/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.test.ts @@ -57,12 +57,13 @@ describe("channel realtime events route", () => { it("streams matching channel events without leaking other channels", async () => { let listener: ((event: { - type: "channel.message.created"; + type: "channel.message.created" | "channel.message.updated"; workspaceId: string; channelName: string; messageId: string; sequence: number; - createdAt: string; + createdAt?: string; + updatedAt?: string; }) => void) | null = null; mockSubscribeWorkspaceRealtimeEvents.mockImplementation((_workspaceId, nextListener) => { listener = nextListener; @@ -104,6 +105,21 @@ describe("channel realtime events route", () => { expect(eventText).toContain("event: channel.message.created"); expect(eventText).toContain("message-visible"); expect(eventText).not.toContain("message-hidden"); + + emit({ + type: "channel.message.updated", + workspaceId: "workspace-1", + channelName: "general", + messageId: "message-visible", + sequence: 3, + updatedAt: "2026-05-01T00:00:02.000Z", + }); + + const updateChunk = await reader.read(); + const updateText = decoder.decode(updateChunk.value); + expect(updateText).toContain("event: channel.message.updated"); + expect(updateText).toContain("message-visible"); + expect(updateText).toContain("2026-05-01T00:00:02.000Z"); await reader.cancel(); }); }); diff --git a/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.ts b/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.ts index 0b50fc2..7b6165d 100644 --- a/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.ts +++ b/apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.ts @@ -74,6 +74,14 @@ export async function GET( sequence: event.sequence, createdAt: event.createdAt, } + : event.type === "channel.message.updated" + ? { + type: event.type, + channelName: event.channelName, + messageId: event.messageId, + sequence: event.sequence, + updatedAt: event.updatedAt, + } : event.type === "channel.thread.changed" ? { type: event.type, diff --git a/apps/web/features/channels/channels-page-client.test.tsx b/apps/web/features/channels/channels-page-client.test.tsx index 6daa9fe..c3aaf83 100644 --- a/apps/web/features/channels/channels-page-client.test.tsx +++ b/apps/web/features/channels/channels-page-client.test.tsx @@ -411,10 +411,15 @@ describe("ChannelsPageClient", () => { messageId: "message-2", sequence: 1, }); + activeSource?.emit("channel.message.updated", { + channelName: "tour visit", + messageId: "message-2", + sequence: 2, + }); activeSource?.emit("channel.message.created", { channelName: "tour visit", messageId: "message-3", - sequence: 2, + sequence: 3, }); expect(routerRefreshMock).not.toHaveBeenCalled(); diff --git a/apps/web/features/channels/channels-page-client.tsx b/apps/web/features/channels/channels-page-client.tsx index def2d9f..f3ac16b 100644 --- a/apps/web/features/channels/channels-page-client.tsx +++ b/apps/web/features/channels/channels-page-client.tsx @@ -420,10 +420,12 @@ function useChannelRealtimeRefresh({ }; source.addEventListener("channel.message.created", scheduleRefresh as EventListener); + source.addEventListener("channel.message.updated", scheduleRefresh as EventListener); source.addEventListener("channel.thread.changed", scheduleRefresh as EventListener); return () => { source.removeEventListener("channel.message.created", scheduleRefresh as EventListener); + source.removeEventListener("channel.message.updated", scheduleRefresh as EventListener); source.removeEventListener("channel.thread.changed", scheduleRefresh as EventListener); source.close(); if (refreshTimerRef.current !== null) { diff --git a/apps/web/features/chat/chat-primitives.test.tsx b/apps/web/features/chat/chat-primitives.test.tsx index ca1aa5f..77fdacb 100644 --- a/apps/web/features/chat/chat-primitives.test.tsx +++ b/apps/web/features/chat/chat-primitives.test.tsx @@ -1,6 +1,6 @@ -import { fireEvent, render, screen } from "@testing-library/react"; +import { act, fireEvent, render, screen } from "@testing-library/react"; import userEvent from "@testing-library/user-event"; -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ChatAttachmentRow, ConversationMessageBubble } from "@/features/chat/chat-primitives"; import { LanguageProvider } from "@/features/i18n/language-provider"; import type { MessageAttachment } from "@/shared/types/workspace"; @@ -53,6 +53,10 @@ describe("ConversationMessageBubble", () => { window.localStorage.clear(); }); + afterEach(() => { + vi.useRealTimers(); + }); + it("translates the system speaker label in English", () => { render( @@ -116,7 +120,7 @@ describe("ConversationMessageBubble", () => { const onReviewApproval = vi.fn(async () => {}); render( - + { expect(onReviewApproval).toHaveBeenCalledWith("approval-1", "approved"); }); + + it("reveals streamed pending agent reply content instead of hiding it behind dots", () => { + vi.useFakeTimers(); + + const { container } = render( + + + , + ); + + expect(screen.queryByText("我正在整理第一版。")).not.toBeInTheDocument(); + + act(() => { + vi.advanceTimersByTime(16); + }); + expect(screen.getByText("我")).toBeInTheDocument(); + expect(container.querySelector(".contacts-pending-dots")).toBeInTheDocument(); + + act(() => { + vi.advanceTimersByTime(16 * 8); + }); + expect(screen.getByText("我正在整理第一版。")).toBeInTheDocument(); + expect(container.querySelector(".contacts-pending-dots")).not.toBeInTheDocument(); + }); + + it("reveals completed streamed agent replies progressively", () => { + vi.useFakeTimers(); + + render( + + + , + ); + + expect(screen.queryByText("abcdef")).not.toBeInTheDocument(); + + act(() => { + vi.advanceTimersByTime(16); + }); + expect(screen.getByText("a")).toBeInTheDocument(); + + act(() => { + vi.advanceTimersByTime(16 * 2); + }); + expect(screen.getByText("abc")).toBeInTheDocument(); + + act(() => { + vi.advanceTimersByTime(16 * 3); + }); + expect(screen.getByText("abcdef")).toBeInTheDocument(); + }); }); diff --git a/apps/web/features/chat/chat-primitives.tsx b/apps/web/features/chat/chat-primitives.tsx index 90f75c1..858c1a4 100644 --- a/apps/web/features/chat/chat-primitives.tsx +++ b/apps/web/features/chat/chat-primitives.tsx @@ -1,6 +1,6 @@ "use client"; -import { memo, useState } from "react"; +import { memo, useEffect, useState } from "react"; import Link from "next/link"; import type { MessageAttachment, MessageMention } from "@/shared/types/workspace"; import { useLanguage } from "@/features/i18n/language-provider"; @@ -127,6 +127,22 @@ export const ConversationMessageBubble = memo(function ConversationMessageBubble const isPendingMessage = message.status === "pending"; const isError = message.status === "error"; const isProcessMessage = message.kind === "process"; + const isStreamedAgentReply = message.role === "agent" && message.data?.stream_started === "true"; + const shouldRevealStreamedContent = isStreamedAgentReply && !isError; + const renderedContent = isStreamedAgentReply + ? message.content + : translateWorkspaceMessageSummary(message, tx); + const visibleContent = useTypewriterContent({ + content: renderedContent, + enabled: shouldRevealStreamedContent, + speedMs: 16, + step: 1, + }); + const isStreamedContentRevealed = + !shouldRevealStreamedContent || visibleContent.length >= renderedContent.length; + const shouldShowPendingDots = + isPendingMessage && + (!isStreamedAgentReply || !renderedContent.trim() || !isStreamedContentRevealed); const speakerLabel = translateSystemSpeaker(message.speaker, tx); const replyToSpeakerLabel = replyToMessage ? translateSystemSpeaker(replyToMessage.speaker, tx) : ""; const approvalAction = buildRuntimeApprovalAction(message, tx); @@ -196,13 +212,26 @@ export const ConversationMessageBubble = memo(function ConversationMessageBubble {isPendingMessage ? tx("思考中", "Thinking") : renderMessageTimestamp(message.timestamp)} {isPendingMessage ? ( -
- - - -
+ isStreamedAgentReply && renderedContent.trim() ? ( + <> +

{renderMessageContent(visibleContent, message.mentions)}

+ {shouldShowPendingDots ? ( +
+ + + +
+ ) : null} + + ) : ( +
+ + + +
+ ) ) : ( -

{renderMessageContent(translateWorkspaceMessageSummary(message, tx), message.mentions)}

+

{renderMessageContent(visibleContent, message.mentions)}

)} {approvalAction ? (
@@ -295,6 +324,39 @@ export const ConversationMessageBubble = memo(function ConversationMessageBubble } }); +function useTypewriterContent(input: { + content: string; + enabled: boolean; + speedMs: number; + step: number; +}): string { + const { content, enabled, speedMs, step } = input; + const [visibleLength, setVisibleLength] = useState(() => (enabled ? 0 : content.length)); + + useEffect(() => { + if (!enabled || content.length === 0) { + setVisibleLength(content.length); + return; + } + + setVisibleLength((current) => Math.min(current, content.length)); + const timer = window.setInterval(() => { + setVisibleLength((current) => { + const next = Math.min(content.length, current + step); + if (next >= content.length) { + window.clearInterval(timer); + } + return next; + }); + }, speedMs); + return () => { + window.clearInterval(timer); + }; + }, [content, enabled, speedMs, step]); + + return enabled ? content.slice(0, visibleLength) : content; +} + function renderMessageTimestamp(value: string): string { return formatCompactTimestamp(value, { emptyFallback: value }); } diff --git a/packages/daemon/src/agent-router/adapters/opencode.ts b/packages/daemon/src/agent-router/adapters/opencode.ts index eefd5a2..26b5cf5 100644 --- a/packages/daemon/src/agent-router/adapters/opencode.ts +++ b/packages/daemon/src/agent-router/adapters/opencode.ts @@ -83,10 +83,38 @@ async function runOpenCode( observer: AgentRouterObserver, request: AgentRouterRunRequest, ): Promise { + let stdoutBuffer = ""; + let streamedEventCount = 0; + const streamNativeEvent = (event: Record, runObserver: AgentRouterObserver): void => { + streamedEventCount += 1; + for (const mapped of mapOpenCodeNativeEvent(event)) { + runObserver.emit(mapped); + } + }; + return runNativeHarness("opencode", plan, observer, request, { emptyMessage: "OpenCode returned an empty response.", nonZeroMessage: (exitCode) => `OpenCode exited with code ${exitCode}.`, timeoutMessage: (timeoutMs) => `OpenCode timed out after ${timeoutMs}ms.`, + onStdout: (chunk, runObserver) => { + stdoutBuffer += chunk; + const lines = stdoutBuffer.split(/\r?\n/); + stdoutBuffer = lines.pop() ?? ""; + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed.startsWith("{")) { + continue; + } + try { + const parsed = JSON.parse(trimmed); + if (parsed && typeof parsed === "object") { + streamNativeEvent(parsed as Record, runObserver); + } + } catch { + // Final stdout parsing emits protocol diagnostics for invalid JSON. + } + } + }, parseEvents: (stdout, stderr, runObserver) => { const parsed = parseJsonEventOutput(stdout); const diagnostics = [...parsed.diagnostics]; @@ -99,10 +127,13 @@ async function runOpenCode( })); } - for (const event of parsed.events) { + for (const event of parsed.events.slice(streamedEventCount)) { for (const mapped of mapOpenCodeNativeEvent(event)) { runObserver.emit(mapped); } + } + + for (const event of parsed.events) { const finalText = extractOpenCodeFinalText(event); if (finalText) { outputText = appendLine(outputText, finalText); diff --git a/packages/daemon/src/provider-runtime.test.ts b/packages/daemon/src/provider-runtime.test.ts index 2780bb4..d3442b5 100644 --- a/packages/daemon/src/provider-runtime.test.ts +++ b/packages/daemon/src/provider-runtime.test.ts @@ -945,6 +945,7 @@ test("runProviderTask routes OpenCode through AgentRouter with model, session, a const fakeCliPath = join(toolBinDir, "fake-cli"); const argsPath = join(workDir, "opencode-args.txt"); const seenPathFile = join(workDir, "seen-path.txt"); + const continueFile = join(workDir, "continue"); const originalOpenCodeModel = process.env.OPENCODE_MODEL; const originalDaemonBin = process.env.AGENT_SPACE_DAEMON_BIN; mkdirSync(providerBinDir, { recursive: true }); @@ -959,6 +960,7 @@ test("runProviderTask routes OpenCode through AgentRouter with model, session, a "if command -v fake-cli >/dev/null 2>&1 && command -v agent-space-daemon >/dev/null 2>&1; then", " printf '%s\\n' '{\"type\":\"step_start\",\"sessionID\":\"opencode-session\",\"part\":{\"text\":\"working\"}}'", " printf '%s\\n' '{\"type\":\"text\",\"sessionID\":\"opencode-session\",\"part\":{\"text\":\"opencode provider output\"}}'", + " while [ ! -f \"$OPENCODE_CONTINUE_FILE\" ]; do sleep 0.05; done", " printf '%s\\n' '{\"type\":\"step_finish\",\"sessionID\":\"opencode-session\",\"part\":{\"tokens\":{\"input\":3,\"output\":4}}}'", "else", " printf '%s\\n' 'missing runtime path'", @@ -989,11 +991,16 @@ test("runProviderTask routes OpenCode through AgentRouter with model, session, a process.env.AGENT_SPACE_DAEMON_BIN = daemonBinPath; process.env.OPENCODE_MODEL = "openrouter/openai/gpt-4.1"; const events: Array<{ type: string; content?: string }> = []; - const result = await runProviderTask(runtime, "write a short reply", workDir, { + let resolveStreamedText: (() => void) | undefined; + const streamedText = new Promise((resolve) => { + resolveStreamedText = resolve; + }); + const resultPromise = runProviderTask(runtime, "write a short reply", workDir, { sessionId: "previous-opencode-session", contextEnv: { OPENCODE_ARGS_PATH: argsPath, SEEN_PATH_FILE: seenPathFile, + OPENCODE_CONTINUE_FILE: continueFile, }, runtimeToolCapabilities: [{ id: "fake-cli", @@ -1003,9 +1010,26 @@ test("runProviderTask routes OpenCode through AgentRouter with model, session, a allowedShellPatterns: ["fake-cli *"], source: "runtime", }], - onEvent: (event) => events.push(event), + onEvent: (event) => { + events.push(event); + if (event.type === "text" && event.content === "opencode provider output") { + resolveStreamedText?.(); + } + }, taskTimeoutMs: 1_000, }); + let streamError: unknown; + try { + await waitForPromise(streamedText, 500, "OpenCode text event was not streamed before the process exited."); + } catch (error) { + streamError = error; + } finally { + writeFileSync(continueFile, "ok\n", "utf8"); + } + const result = await resultPromise; + if (streamError) { + throw streamError; + } const args = readFileSync(argsPath, "utf8").trim().split(/\r?\n/); const seenPath = readFileSync(seenPathFile, "utf8").split(delimiter); @@ -1024,6 +1048,10 @@ test("runProviderTask routes OpenCode through AgentRouter with model, session, a assert.equal(seenPath.includes(daemonBinDir), true); assert.equal(seenPath.includes(toolBinDir), true); assert.equal(events.some((event) => event.type === "usage"), true); + assert.equal( + events.filter((event) => event.type === "text" && event.content === "opencode provider output").length, + 1, + ); } finally { if (originalOpenCodeModel === undefined) { delete process.env.OPENCODE_MODEL; @@ -1039,6 +1067,22 @@ test("runProviderTask routes OpenCode through AgentRouter with model, session, a } }); +function waitForPromise(promise: Promise, timeoutMs: number, message: string): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error(message)), timeoutMs); + promise.then( + (value) => { + clearTimeout(timeout); + resolve(value); + }, + (error: unknown) => { + clearTimeout(timeout); + reject(error); + }, + ); + }); +} + test("runProviderTask maps Hermes capability and empty-response diagnostics to provider errors", async () => { const workDir = mkdtempSync(join(tmpdir(), "agent-space-hermes-diagnostics-")); const binPath = join(workDir, "hermes"); diff --git a/packages/daemon/src/remote-daemon.ts b/packages/daemon/src/remote-daemon.ts index 4524987..2045167 100644 --- a/packages/daemon/src/remote-daemon.ts +++ b/packages/daemon/src/remote-daemon.ts @@ -3,7 +3,14 @@ import { spawn } from "node:child_process"; import { join, resolve } from "node:path"; import { getDaemonChannelWorkDirPath, getDaemonTaskWorkDirPath } from "@agent-space/db"; import { getStringFlag, parseArgs } from "./args.ts"; -import type { ClaimedDaemonTask, ClaimedRuntimeAppOperation, DaemonTaskInputBundle, HeartbeatDaemonResponse, RegisterDaemonResponse } from "./daemon-api.ts"; +import type { + ClaimedDaemonTask, + ClaimedRuntimeAppOperation, + DaemonTaskInputBundle, + DaemonTaskMessageInput, + HeartbeatDaemonResponse, + RegisterDaemonResponse, +} from "./daemon-api.ts"; import { collectRuntimeOutputBundle, clearTaskOutputArtifacts, materializeInputBundle } from "./bundle.ts"; import { HttpDaemonClient } from "./daemon-client.ts"; import { prepareSkillImportOperationArtifacts } from "./skill-imports.ts"; @@ -61,6 +68,10 @@ interface DaemonStatusSummary { stateDir: string; } +type OrderedTaskMessageReporter = ((messages: DaemonTaskMessageInput[]) => Promise) & { + flush: () => Promise; +}; + export async function runRemoteDaemonCommand(subcommand: string | undefined, args: string[]): Promise { if (subcommand === "start") { return runRemoteDaemonStart(args); @@ -555,6 +566,7 @@ async function executeRemoteTask( await client.startTask(task.id); const bundle = await client.getInputBundle(task.id); materializeInputBundle(workDir, bundle); + const reportTaskMessages = createOrderedTaskMessageReporter(client, task.id); const result = await runProviderTask( runtime, @@ -571,21 +583,16 @@ async function executeRemoteTask( runtimeApps: bundle.metadata.runtimeApps?.apps ?? [], runtimeToolCapabilities: bundle.metadata.runtimeToolCapabilities?.capabilities ?? [], onEvent: (event) => { - void client.reportMessages(task.id, { messages: [event] }).catch((error) => { - const message = error instanceof Error ? error.message : String(error); - console.error(`Failed to report remote task message for ${task.id}: ${message}`); - }); + reportTaskMessages([event]); }, onApprovalRequest: (request) => waitForRuntimeApproval(client, task.id, request), }, ); + await reportTaskMessages.flush(); const preparedSkillImports = prepareSkillImportOperationArtifacts(workDir); for (const warning of preparedSkillImports.warnings) { - await client.reportMessages(task.id, { messages: [{ type: "status", content: warning }] }).catch((error) => { - const message = error instanceof Error ? error.message : String(error); - console.error(`Failed to report skill import warning for ${task.id}: ${message}`); - }); + await reportTaskMessages([{ type: "status", content: warning }]); } const outputBundle = collectRuntimeOutputBundle(workDir); @@ -619,6 +626,22 @@ async function executeRemoteTask( } } +function createOrderedTaskMessageReporter(client: HttpDaemonClient, taskId: string): OrderedTaskMessageReporter { + let chain = Promise.resolve(); + + const report = ((messages: DaemonTaskMessageInput[]) => { + chain = chain + .then(() => client.reportMessages(taskId, { messages })) + .catch((error) => { + const message = error instanceof Error ? error.message : String(error); + console.error(`Failed to report remote task message for ${taskId}: ${message}`); + }); + return chain; + }) as OrderedTaskMessageReporter; + report.flush = () => chain; + return report; +} + function buildRuntimeContextEnv( base: Record, googleWorkspace: DaemonTaskInputBundle["metadata"]["googleWorkspace"], diff --git a/packages/services/src/index.ts b/packages/services/src/index.ts index 5ac904f..c3e7808 100644 --- a/packages/services/src/index.ts +++ b/packages/services/src/index.ts @@ -253,6 +253,7 @@ export { postMessageSync, sendChannelHumanMessageSync, replacePendingChannelMessageSync, + streamAgentChannelReplyDeltaSync, pinMessageSync, unpinMessageSync, acknowledgeMessageSync, @@ -261,6 +262,7 @@ export { // Realtime export { publishChannelMessageCreatedEvent, + publishChannelMessageUpdatedEvent, publishChannelThreadChangedEvent, subscribeWorkspaceRealtimeEvents, type WorkspaceRealtimeEvent, diff --git a/packages/services/src/messages/messages.test.ts b/packages/services/src/messages/messages.test.ts index b8db155..17e43ea 100644 --- a/packages/services/src/messages/messages.test.ts +++ b/packages/services/src/messages/messages.test.ts @@ -34,6 +34,7 @@ import { sendContactMessageSync, sendHumanDirectMessageSync, setEmployeeChannelMemberAccessSync, + streamAgentChannelReplyDeltaSync, stopAutoContinuationSync, subscribeWorkspaceRealtimeEvents, unpinMessageSync, @@ -1196,6 +1197,76 @@ test("replacePendingChannelMessageSync publishes realtime thread change events", assert.equal(events[0]?.channelName, "tour visit"); }); +test("streamAgentChannelReplyDeltaSync updates the pending message and publishes realtime update events", () => { + seedWorkspace(); + postMessageSync({ + channel: "tour visit", + speaker: "Atlas", + role: "agent", + summary: "Thinking", + code: "agent.pending", + status: "pending", + }); + const events: Array<{ type: string; channelName: string; messageId?: string }> = []; + const unsubscribe = subscribeWorkspaceRealtimeEvents(DEFAULT_WORKSPACE_ID, (event) => events.push(event)); + + const first = streamAgentChannelReplyDeltaSync({ + channel: "tour visit", + pendingSpeaker: "Atlas", + delta: "你好", + sourceTaskQueueId: "queue-stream", + }); + const second = streamAgentChannelReplyDeltaSync({ + channel: "tour visit", + pendingSpeaker: "Atlas", + delta: "，我正在处理。", + sourceTaskQueueId: "queue-stream", + }); + unsubscribe(); + + const streamedMessage = readWorkspaceStateSync().messages.find((message) => message.id === first?.id); + assert.ok(streamedMessage); + assert.equal(streamedMessage.summary, "你好，我正在处理。"); + assert.equal(streamedMessage.status, "pending"); + assert.equal(streamedMessage.data?.stream_started, "true"); + assert.equal(streamedMessage.data?.source_task_queue_id, "queue-stream"); + assert.equal(second?.id, first?.id); + assert.equal(events.length, 2); + assert.equal(events[0]?.type, "channel.message.updated"); + assert.equal(events[0]?.messageId, first?.id); + assert.equal(events[1]?.type, "channel.message.updated"); +}); + +test("completeAgentChannelReplySync preserves stream metadata from the pending reply", () => { + seedWorkspace(); + postMessageSync({ + channel: "tour visit", + speaker: "Atlas", + role: "agent", + summary: "Thinking", + code: "agent.pending", + status: "pending", + }); + streamAgentChannelReplyDeltaSync({ + channel: "tour visit", + pendingSpeaker: "Atlas", + delta: "你好", + sourceTaskQueueId: "queue-stream", + }); + + const result = completeAgentChannelReplySync({ + channel: "tour visit", + pendingSpeaker: "Atlas", + speaker: "Atlas", + summary: "你好，任务完成。", + sourceTaskQueueId: "queue-stream", + }); + + assert.equal(result.message.status, "completed"); + assert.equal(result.message.data?.stream_started, "true"); + assert.equal(result.message.data?.source_task_queue_id, "queue-stream"); +}); + test("runtime tool approval requests are visible and reviewable in channel messages", () => { seedWorkspace(); const approval = createRuntimeToolApprovalRequestSync({ diff --git a/packages/services/src/messages/messages.ts b/packages/services/src/messages/messages.ts index eb789d1..0771d4b 100644 --- a/packages/services/src/messages/messages.ts +++ b/packages/services/src/messages/messages.ts @@ -45,6 +45,7 @@ import { import { canReadChannelForActorSync } from "../channel-access/channel-access.ts"; import { publishChannelMessageCreatedEvent, + publishChannelMessageUpdatedEvent, publishChannelThreadChangedEvent, } from "../realtime/events.ts"; @@ -673,6 +674,16 @@ export function completeAgentChannelReplySync(input: { throw new Error(`Channel "${input.channel}" does not exist.`); } + const pendingMessage = input.pendingSpeaker?.trim() + ? state.messages.find( + (message) => + sameValue(message.channel ?? "", channel.name) && + message.role === "agent" && + message.status === "pending" && + sameValue(message.speaker, input.pendingSpeaker ?? ""), + ) + : undefined; + if (input.pendingSpeaker?.trim()) { state.messages = state.messages.filter( (message) => @@ -697,6 +708,14 @@ export function completeAgentChannelReplySync(input: { status: "completed", attachments: input.attachments, mentions: mentionParse.allMentions, + data: pendingMessage?.data?.stream_started === "true" + ? { + stream_started: "true", + ...(pendingMessage.data.source_task_queue_id + ? { source_task_queue_id: pendingMessage.data.source_task_queue_id } + : {}), + } + : undefined, }, effectiveWorkspaceId); const dispatchResult = shouldProcessMentions @@ -771,6 +790,13 @@ export function replacePendingChannelMessageSync(input: { attachments?: MessageAttachment[]; }, workspaceId?: string): AgentSpaceState { const state = ensureWorkspaceStateSync(workspaceId); + const pendingMessage = state.messages.find( + (message) => + sameValue(message.channel ?? "", input.channel) && + message.role === "agent" && + message.status === "pending" && + sameValue(message.speaker, input.pendingSpeaker), + ); state.messages = state.messages.filter( (message) => @@ -788,6 +814,14 @@ export function replacePendingChannelMessageSync(input: { summary: input.summary, status: input.status ?? "completed", attachments: input.attachments, + data: pendingMessage?.data?.stream_started === "true" + ? { + stream_started: "true", + ...(pendingMessage.data.source_task_queue_id + ? { source_task_queue_id: pendingMessage.data.source_task_queue_id } + : {}), + } + : undefined, }, workspaceId); const nextState = writeWorkspaceStateSync(state, workspaceId); @@ -799,6 +833,59 @@ export function replacePendingChannelMessageSync(input: { return nextState; } +export function streamAgentChannelReplyDeltaSync(input: { + channel: string; + pendingSpeaker: string; + delta: string; + sourceTaskQueueId?: string; +}, workspaceId?: string): WorkspaceMessage | null { + const delta = input.delta; + if (!delta) { + return null; + } + + const state = ensureWorkspaceStateSync(workspaceId); + const effectiveWorkspaceId = workspaceId ?? DEFAULT_WORKSPACE_ID; + const channelName = input.channel.trim(); + const pendingSpeaker = input.pendingSpeaker.trim(); + if (!channelName || !pendingSpeaker) { + return null; + } + + const message = state.messages.find((item) => + sameValue(item.channel ?? "", channelName) && + item.role === "agent" && + item.status === "pending" && + sameValue(item.speaker, pendingSpeaker) + ); + if (!message) { + return null; + } + + const streamStarted = message.data?.stream_started === "true"; + const baseSummary = streamStarted + ? message.summary + : message.code === "agent.pending" + ? "" + : message.summary; + const updatedAt = new Date().toISOString(); + message.summary = `${baseSummary}${delta}`; + message.data = { + ...(message.data ?? {}), + stream_started: "true", + ...(input.sourceTaskQueueId ? { source_task_queue_id: input.sourceTaskQueueId } : {}), + }; + + writeWorkspaceStateSync(state, effectiveWorkspaceId); + publishChannelMessageUpdatedEvent({ + workspaceId: effectiveWorkspaceId, + channelName, + messageId: message.id, + updatedAt, + }); + return message; +} + function buildAgentOutputMentionParseWarnings(parse: ChannelMentionParseResult): string[] { const warnings: string[] = []; for (const token of parse.unknownMentions) { diff --git a/packages/services/src/realtime/events.ts b/packages/services/src/realtime/events.ts index 92b46db..cc24ade 100644 --- a/packages/services/src/realtime/events.ts +++ b/packages/services/src/realtime/events.ts @@ -9,6 +9,14 @@ export type WorkspaceRealtimeEvent = sequence: number; createdAt: string; } + | { + type: "channel.message.updated"; + workspaceId: string; + channelName: string; + messageId: string; + sequence: number; + updatedAt: string; + } | { type: "channel.thread.changed"; workspaceId: string; @@ -50,6 +58,24 @@ export function publishChannelMessageCreatedEvent(input: { return event; } +export function publishChannelMessageUpdatedEvent(input: { + workspaceId: string; + channelName: string; + messageId: string; + updatedAt: string; +}): WorkspaceRealtimeEvent { + const event: WorkspaceRealtimeEvent = { + type: "channel.message.updated", + workspaceId: input.workspaceId, + channelName: input.channelName, + messageId: input.messageId, + sequence: nextSequence(), + updatedAt: input.updatedAt, + }; + emitter.emit(event.workspaceId, event); + return event; +} + export function publishChannelThreadChangedEvent(input: { workspaceId: string; channelName: string;