diff --git a/tauri/src/main.ts b/tauri/src/main.ts index 26dc6df..1bbcd5f 100644 --- a/tauri/src/main.ts +++ b/tauri/src/main.ts @@ -4,6 +4,7 @@ import { type ConversationItem, type Envelope, type JsonObject, + parseEnvelopeJSON, } from "./protocol"; import { ConversationStore } from "./runtime/conversation-store"; import { InteractionKernel } from "./runtime/interaction-kernel"; @@ -25,6 +26,7 @@ const DEFAULT_API = "http://127.0.0.1:8090"; let session: Session = { api: localStorage.getItem("lineup.api") || DEFAULT_API, uid: "", agentUID: "agent_hermes_main", channelID: "agent_default_channel", channelType: 2, lastSeq: 0, active: false }; let transport: TransportAdapter | undefined; let polling = false; +let flushingOutbox = false; const interactionKernel = new InteractionKernel(); const conversationStore = new ConversationStore(localStorage); @@ -98,13 +100,19 @@ const rendererContext = new TrustedDOMRendererContext(messages, presence, { submit(callID, submission) { const transition = interactionKernel.submitToolCall(callID, new Date().toISOString(), submission); conversationStore.replaceToolCalls(interactionKernel.snapshot().tool_calls); - if (transition.disposition === "accepted") conversationStore.clearToolDraft(callID); + if (transition.disposition === "accepted") { + conversationStore.clearToolDraft(callID); + queueToolAction("tool-result", callID, { call_id: callID, status: "completed", result: submission }); + } return transition.disposition === "accepted"; }, cancel(callID, reason) { const transition = interactionKernel.cancelToolCall(callID, reason, new Date().toISOString()); conversationStore.replaceToolCalls(interactionKernel.snapshot().tool_calls); - if (transition.disposition === "accepted") conversationStore.clearToolDraft(callID); + if (transition.disposition === "accepted") { + conversationStore.clearToolDraft(callID); + queueToolAction("tool-cancel", callID, { call_id: callID, reason }); + } return transition.disposition === "accepted"; }, expire(callID) { @@ -145,6 +153,64 @@ function currentConversationKey(): string { return `${session.uid}:${session.channelType}:${session.channelID}`; } +function queueToolAction(kind: "tool-result" | "tool-cancel", callID: string, payload: JsonObject): void { + if (!session.uid) { + rendererContext.renderSystemMessage("交互结果已保留,将在下次登录后发送。"); + return; + } + const createdAt = new Date().toISOString(); + const localID = newLocalID("tool"); + const type = kind === "tool-result" ? "lineup.v1.tool.result" : "lineup.v1.tool.cancel"; + const envelope = makeProtocolEvent( + type, payload, currentConversationKey(), + { kind: "human", id: session.uid }, { kind: "agent", id: session.agentUID }, + ); + try { + // Durable enqueue precedes the HTTP request: a reload or transient + // network failure cannot turn one accepted UI action into an untracked + // second submission. + conversationStore.enqueueToolAction(localID, kind, JSON.stringify(envelope), createdAt); + } catch (reason) { + rendererContext.renderSystemMessage(storageFailureMessage(reason)); + return; + } + void flushOutbox(); +} + +async function flushOutbox(): Promise { + if (flushingOutbox || !session.active || !transport || !session.uid) return; + flushingOutbox = true; + try { + for (const entry of conversationStore.snapshot().outbox) { + try { + const sequence = await transport.sendText({ + from_uid: session.uid, + channel_id: session.channelID, + channel_type: session.channelType, + payload: entry.payload, + }); + if (entry.kind === "text") { + if (sequence) conversationStore.markSubmitted(entry.local_id, sequence, new Date().toISOString()); + rendererContext.updateUserDelivery(entry.local_id, { status: "submitted", updated_at: new Date().toISOString(), ...(sequence ? { transport_id: String(sequence) } : {}) }); + } else { + conversationStore.acknowledgeOutbox(entry.local_id); + } + } catch (reason) { + if (entry.kind === "text") { + const message = reason instanceof Error ? reason.message : "发送失败"; + conversationStore.markFailed(entry.local_id, message, new Date().toISOString()); + rendererContext.updateUserDelivery(entry.local_id, { status: "failed", updated_at: new Date().toISOString(), error: message, retryable: true }); + } + // Keep this and later entries durable; the next sync/login cycle + // retries in original order. + return; + } + } + } finally { + flushingOutbox = false; + } +} + function decodeBase64(value: string): string { try { return decodeURIComponent([...atob(value)].map(c => `%${c.charCodeAt(0).toString(16).padStart(2, "0")}`).join("")); } catch { return value; } } @@ -182,6 +248,12 @@ async function syncLoop(): Promise { conversationStore.advanceCursor(message.message_seq); const payload = decodeBase64(message.payload); if (message.from_uid === session.uid) { + const ownEnvelope = parseEnvelopeJSON(payload); + if (ownEnvelope.ok && (ownEnvelope.envelope.type === "lineup.v1.tool.result" || ownEnvelope.envelope.type === "lineup.v1.tool.cancel")) { + // Tool action echoes advance the inclusive cursor but are never + // rendered as raw JSON user messages or replayed as a new action. + continue; + } const restored = conversationStore.recordSyncedUserText(message.message_seq, payload, new Date().toISOString()); if (restored) renderIncoming(restored.item); } else { @@ -250,7 +322,7 @@ $("#login-form").addEventListener("submit", async event => { messages.replaceChildren(); rendererContext.reset(); loginView.hidden = true; chatView.hidden = false; for (const stored of snapshot.items) renderIncoming(stored.item); - rendererContext.renderSystemMessage("已进入与 AI Agent 的对话"); rendererContext.setPresence("正在同步消息…"); void syncLoop(); + rendererContext.renderSystemMessage("已进入与 AI Agent 的对话"); rendererContext.setPresence("正在同步消息…"); void flushOutbox(); void syncLoop(); } catch (reason) { error.textContent = loginFailureMessage(reason); } finally { submit.disabled = false; submit.textContent = "进入 LineUp"; } }); diff --git a/tauri/src/runtime/conversation-store.test.ts b/tauri/src/runtime/conversation-store.test.ts index 88b1334..b688ac6 100644 --- a/tauri/src/runtime/conversation-store.test.ts +++ b/tauri/src/runtime/conversation-store.test.ts @@ -55,7 +55,7 @@ describe("ConversationStore", () => { const snapshot = reopened.open(conversationID); expect(snapshot.cursor).toBe(0); - expect(snapshot.outbox).toEqual([expect.objectContaining({ local_id: "local_002", payload: "persist me" })]); + expect(snapshot.outbox).toEqual([expect.objectContaining({ local_id: "local_002", kind: "text", payload: "persist me" })]); expect(snapshot.items).toEqual([expect.objectContaining({ local_id: "local_002", item: expect.objectContaining({ kind: "user-message", text: "persist me" }), @@ -110,4 +110,32 @@ describe("ConversationStore", () => { expect(new ConversationStore(storage).open(conversationID).tasks).toEqual([expect.objectContaining({ operation_id: "task_store_001", percent: 70 })]); }); + + it("durably queues a tool result without creating a user-message history item", () => { + const storage = new MemoryStorage(); + const first = new ConversationStore(storage); + first.open(conversationID); + first.enqueueToolAction("tool_001", "tool-result", '{"type":"lineup.v1.tool.result"}', createdAt); + + const reopened = new ConversationStore(storage).open(conversationID); + expect(reopened.items).toEqual([]); + expect(reopened.outbox).toEqual([{ + local_id: "tool_001", kind: "tool-result", payload: '{"type":"lineup.v1.tool.result"}', created_at: createdAt, + }]); + + const current = new ConversationStore(storage); + current.open(conversationID); + current.acknowledgeOutbox("tool_001"); + expect(current.snapshot().outbox).toEqual([]); + }); + + it("upgrades v5 plaintext outbox entries to typed text entries", () => { + const storage = new MemoryStorage(); + storage.setItem("lineup.conversation.v1:user_test:2:agent_channel", JSON.stringify({ + version: 5, cursor: 8, items: [], outbox: [{ local_id: "legacy_001", payload: "hello", created_at: createdAt }], tool_calls: [], tool_drafts: {}, tasks: [], + })); + expect(new ConversationStore(storage).open(conversationID).outbox).toEqual([{ + local_id: "legacy_001", kind: "text", payload: "hello", created_at: createdAt, + }]); + }); }); diff --git a/tauri/src/runtime/conversation-store.ts b/tauri/src/runtime/conversation-store.ts index dfb3476..aeab396 100644 --- a/tauri/src/runtime/conversation-store.ts +++ b/tauri/src/runtime/conversation-store.ts @@ -11,6 +11,7 @@ export type StoredConversationItem = { export type OutboxEntry = { local_id: string; + kind: "text" | "tool-result" | "tool-cancel"; payload: string; created_at: string; }; @@ -26,7 +27,7 @@ export type ConversationSnapshot = { }; type PersistedConversation = { - version: 5; + version: 6; cursor: number; items: StoredConversationItem[]; outbox: OutboxEntry[]; @@ -109,21 +110,29 @@ export class ConversationStore { item: { kind: "user-message", id: localID, text, delivery }, }; this.state.items.push(stored); - this.state.outbox.push({ local_id: localID, payload: text, created_at: createdAt }); + this.state.outbox.push({ local_id: localID, kind: "text", payload: text, created_at: createdAt }); this.persist(); return { ...stored }; } + /** Queues a standard interaction response before any network attempt. */ + public enqueueToolAction(localID: string, kind: "tool-result" | "tool-cancel", payload: string, createdAt: string): void { + if (this.state.outbox.some(entry => entry.local_id === localID)) return; + this.state.outbox.push({ local_id: localID, kind, payload, created_at: createdAt }); + this.persist(); + } + /** Transport acknowledgement gives the local message its durable IM sequence. */ public markSubmitted(localID: string, messageSeq: number, updatedAt: string): void { const stored = this.state.items.find(item => item.local_id === localID); - if (!stored || stored.item.kind !== "user-message") return; - stored.message_seq = messageSeq; - stored.item = { - ...stored.item, - delivery: { status: "submitted", updated_at: updatedAt, transport_id: String(messageSeq) }, - }; - this.state.outbox = this.state.outbox.filter(entry => entry.local_id !== localID); + if (stored?.item.kind === "user-message") { + stored.message_seq = messageSeq; + stored.item = { + ...stored.item, + delivery: { status: "submitted", updated_at: updatedAt, transport_id: String(messageSeq) }, + }; + } + this.acknowledgeOutbox(localID); // A send acknowledgement is not a sync acknowledgement. The legacy // `/messages/sync` cursor is inclusive, so advancing it here would make a // refresh start at `messageSeq + 1` and permanently skip this user's echo. @@ -132,6 +141,13 @@ export class ConversationStore { this.persist(); } + /** Removes a delivered non-text action without creating a chat bubble. */ + public acknowledgeOutbox(localID: string): void { + const before = this.state.outbox.length; + this.state.outbox = this.state.outbox.filter(entry => entry.local_id !== localID); + if (this.state.outbox.length !== before) this.persist(); + } + public markFailed(localID: string, error: string, updatedAt: string): void { const stored = this.state.items.find(item => item.local_id === localID); if (!stored || stored.item.kind !== "user-message") return; @@ -198,31 +214,31 @@ export class ConversationStore { const parsed: unknown = JSON.parse(this.storage.getItem(this.key(conversationID)) ?? "null"); if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return emptyConversation(); const candidate = parsed as { version?: unknown; cursor?: unknown; items?: unknown; outbox?: unknown; tool_calls?: unknown; tool_drafts?: unknown; tasks?: unknown }; - if ((candidate.version !== 1 && candidate.version !== 2 && candidate.version !== 3 && candidate.version !== 4 && candidate.version !== 5) || !Array.isArray(candidate.items) || !Array.isArray(candidate.outbox) || typeof candidate.cursor !== "number") { + if ((candidate.version !== 1 && candidate.version !== 2 && candidate.version !== 3 && candidate.version !== 4 && candidate.version !== 5 && candidate.version !== 6) || !Array.isArray(candidate.items) || !Array.isArray(candidate.outbox) || typeof candidate.cursor !== "number") { return emptyConversation(); } - if (candidate.version !== 5) { + if (candidate.version !== 6) { // v1 advanced its cursor from a send acknowledgement. Some browsers // had already been upgraded to v2 before that correction, retaining a // high cursor with holes for their own echoes. Preserve every valid // local item but replay the server history once to repair those holes; // the next persist upgrades this record to v5. return { - version: 5, - cursor: candidate.version === 4 ? Math.max(0, candidate.cursor) : 0, + version: 6, + cursor: candidate.version === 4 || candidate.version === 5 ? Math.max(0, candidate.cursor) : 0, items: candidate.items as StoredConversationItem[], - outbox: candidate.outbox as OutboxEntry[], - tool_calls: candidate.version === 4 && Array.isArray(candidate.tool_calls) ? candidate.tool_calls as ToolCallRecord[] : [], - tool_drafts: candidate.version === 4 && isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {}, - tasks: [], + outbox: normalizeOutbox(candidate.outbox), + tool_calls: (candidate.version === 4 || candidate.version === 5) && Array.isArray(candidate.tool_calls) ? candidate.tool_calls as ToolCallRecord[] : [], + tool_drafts: (candidate.version === 4 || candidate.version === 5) && isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {}, + tasks: candidate.version === 5 && Array.isArray(candidate.tasks) ? candidate.tasks as TaskRecord[] : [], }; } return { - version: 5, + version: 6, cursor: Math.max(0, candidate.cursor), items: candidate.items as StoredConversationItem[], - outbox: candidate.outbox as OutboxEntry[], + outbox: normalizeOutbox(candidate.outbox), tool_calls: Array.isArray(candidate.tool_calls) ? candidate.tool_calls as ToolCallRecord[] : [], tool_drafts: isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {}, tasks: Array.isArray(candidate.tasks) ? candidate.tasks as TaskRecord[] : [], @@ -247,7 +263,18 @@ export class ConversationStore { } function emptyConversation(): PersistedConversation { - return { version: 5, cursor: 0, items: [], outbox: [], tool_calls: [], tool_drafts: {}, tasks: [] }; + return { version: 6, cursor: 0, items: [], outbox: [], tool_calls: [], tool_drafts: {}, tasks: [] }; +} + +function normalizeOutbox(value: unknown): OutboxEntry[] { + if (!Array.isArray(value)) return []; + return value.flatMap(entry => { + if (!entry || typeof entry !== "object" || Array.isArray(entry)) return []; + const candidate = entry as Partial; + if (typeof candidate.local_id !== "string" || typeof candidate.payload !== "string" || typeof candidate.created_at !== "string") return []; + const kind = candidate.kind === "tool-result" || candidate.kind === "tool-cancel" ? candidate.kind : "text"; + return [{ local_id: candidate.local_id, kind, payload: candidate.payload, created_at: candidate.created_at }]; + }); } function isDraftMap(value: unknown): value is Record {