feat: reliably deliver tool interaction responses
This commit is contained in:
+75
-3
@@ -4,6 +4,7 @@ import {
|
|||||||
type ConversationItem,
|
type ConversationItem,
|
||||||
type Envelope,
|
type Envelope,
|
||||||
type JsonObject,
|
type JsonObject,
|
||||||
|
parseEnvelopeJSON,
|
||||||
} from "./protocol";
|
} from "./protocol";
|
||||||
import { ConversationStore } from "./runtime/conversation-store";
|
import { ConversationStore } from "./runtime/conversation-store";
|
||||||
import { InteractionKernel } from "./runtime/interaction-kernel";
|
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 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 transport: TransportAdapter | undefined;
|
||||||
let polling = false;
|
let polling = false;
|
||||||
|
let flushingOutbox = false;
|
||||||
const interactionKernel = new InteractionKernel();
|
const interactionKernel = new InteractionKernel();
|
||||||
const conversationStore = new ConversationStore(localStorage);
|
const conversationStore = new ConversationStore(localStorage);
|
||||||
|
|
||||||
@@ -98,13 +100,19 @@ const rendererContext = new TrustedDOMRendererContext(messages, presence, {
|
|||||||
submit(callID, submission) {
|
submit(callID, submission) {
|
||||||
const transition = interactionKernel.submitToolCall(callID, new Date().toISOString(), submission);
|
const transition = interactionKernel.submitToolCall(callID, new Date().toISOString(), submission);
|
||||||
conversationStore.replaceToolCalls(interactionKernel.snapshot().tool_calls);
|
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";
|
return transition.disposition === "accepted";
|
||||||
},
|
},
|
||||||
cancel(callID, reason) {
|
cancel(callID, reason) {
|
||||||
const transition = interactionKernel.cancelToolCall(callID, reason, new Date().toISOString());
|
const transition = interactionKernel.cancelToolCall(callID, reason, new Date().toISOString());
|
||||||
conversationStore.replaceToolCalls(interactionKernel.snapshot().tool_calls);
|
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";
|
return transition.disposition === "accepted";
|
||||||
},
|
},
|
||||||
expire(callID) {
|
expire(callID) {
|
||||||
@@ -145,6 +153,64 @@ function currentConversationKey(): string {
|
|||||||
return `${session.uid}:${session.channelType}:${session.channelID}`;
|
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<void> {
|
||||||
|
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 {
|
function decodeBase64(value: string): string {
|
||||||
try { return decodeURIComponent([...atob(value)].map(c => `%${c.charCodeAt(0).toString(16).padStart(2, "0")}`).join("")); } catch { return value; }
|
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<void> {
|
|||||||
conversationStore.advanceCursor(message.message_seq);
|
conversationStore.advanceCursor(message.message_seq);
|
||||||
const payload = decodeBase64(message.payload);
|
const payload = decodeBase64(message.payload);
|
||||||
if (message.from_uid === session.uid) {
|
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());
|
const restored = conversationStore.recordSyncedUserText(message.message_seq, payload, new Date().toISOString());
|
||||||
if (restored) renderIncoming(restored.item);
|
if (restored) renderIncoming(restored.item);
|
||||||
} else {
|
} else {
|
||||||
@@ -250,7 +322,7 @@ $<HTMLFormElement>("#login-form").addEventListener("submit", async event => {
|
|||||||
messages.replaceChildren(); rendererContext.reset();
|
messages.replaceChildren(); rendererContext.reset();
|
||||||
loginView.hidden = true; chatView.hidden = false;
|
loginView.hidden = true; chatView.hidden = false;
|
||||||
for (const stored of snapshot.items) renderIncoming(stored.item);
|
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); }
|
} catch (reason) { error.textContent = loginFailureMessage(reason); }
|
||||||
finally { submit.disabled = false; submit.textContent = "进入 LineUp"; }
|
finally { submit.disabled = false; submit.textContent = "进入 LineUp"; }
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -55,7 +55,7 @@ describe("ConversationStore", () => {
|
|||||||
const snapshot = reopened.open(conversationID);
|
const snapshot = reopened.open(conversationID);
|
||||||
|
|
||||||
expect(snapshot.cursor).toBe(0);
|
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({
|
expect(snapshot.items).toEqual([expect.objectContaining({
|
||||||
local_id: "local_002",
|
local_id: "local_002",
|
||||||
item: expect.objectContaining({ kind: "user-message", text: "persist me" }),
|
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 })]);
|
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,
|
||||||
|
}]);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ export type StoredConversationItem = {
|
|||||||
|
|
||||||
export type OutboxEntry = {
|
export type OutboxEntry = {
|
||||||
local_id: string;
|
local_id: string;
|
||||||
|
kind: "text" | "tool-result" | "tool-cancel";
|
||||||
payload: string;
|
payload: string;
|
||||||
created_at: string;
|
created_at: string;
|
||||||
};
|
};
|
||||||
@@ -26,7 +27,7 @@ export type ConversationSnapshot = {
|
|||||||
};
|
};
|
||||||
|
|
||||||
type PersistedConversation = {
|
type PersistedConversation = {
|
||||||
version: 5;
|
version: 6;
|
||||||
cursor: number;
|
cursor: number;
|
||||||
items: StoredConversationItem[];
|
items: StoredConversationItem[];
|
||||||
outbox: OutboxEntry[];
|
outbox: OutboxEntry[];
|
||||||
@@ -109,21 +110,29 @@ export class ConversationStore {
|
|||||||
item: { kind: "user-message", id: localID, text, delivery },
|
item: { kind: "user-message", id: localID, text, delivery },
|
||||||
};
|
};
|
||||||
this.state.items.push(stored);
|
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();
|
this.persist();
|
||||||
return { ...stored };
|
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. */
|
/** Transport acknowledgement gives the local message its durable IM sequence. */
|
||||||
public markSubmitted(localID: string, messageSeq: number, updatedAt: string): void {
|
public markSubmitted(localID: string, messageSeq: number, updatedAt: string): void {
|
||||||
const stored = this.state.items.find(item => item.local_id === localID);
|
const stored = this.state.items.find(item => item.local_id === localID);
|
||||||
if (!stored || stored.item.kind !== "user-message") return;
|
if (stored?.item.kind === "user-message") {
|
||||||
stored.message_seq = messageSeq;
|
stored.message_seq = messageSeq;
|
||||||
stored.item = {
|
stored.item = {
|
||||||
...stored.item,
|
...stored.item,
|
||||||
delivery: { status: "submitted", updated_at: updatedAt, transport_id: String(messageSeq) },
|
delivery: { status: "submitted", updated_at: updatedAt, transport_id: String(messageSeq) },
|
||||||
};
|
};
|
||||||
this.state.outbox = this.state.outbox.filter(entry => entry.local_id !== localID);
|
}
|
||||||
|
this.acknowledgeOutbox(localID);
|
||||||
// A send acknowledgement is not a sync acknowledgement. The legacy
|
// A send acknowledgement is not a sync acknowledgement. The legacy
|
||||||
// `/messages/sync` cursor is inclusive, so advancing it here would make a
|
// `/messages/sync` cursor is inclusive, so advancing it here would make a
|
||||||
// refresh start at `messageSeq + 1` and permanently skip this user's echo.
|
// refresh start at `messageSeq + 1` and permanently skip this user's echo.
|
||||||
@@ -132,6 +141,13 @@ export class ConversationStore {
|
|||||||
this.persist();
|
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 {
|
public markFailed(localID: string, error: string, updatedAt: string): void {
|
||||||
const stored = this.state.items.find(item => item.local_id === localID);
|
const stored = this.state.items.find(item => item.local_id === localID);
|
||||||
if (!stored || stored.item.kind !== "user-message") return;
|
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");
|
const parsed: unknown = JSON.parse(this.storage.getItem(this.key(conversationID)) ?? "null");
|
||||||
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return emptyConversation();
|
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 };
|
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();
|
return emptyConversation();
|
||||||
}
|
}
|
||||||
|
|
||||||
if (candidate.version !== 5) {
|
if (candidate.version !== 6) {
|
||||||
// v1 advanced its cursor from a send acknowledgement. Some browsers
|
// v1 advanced its cursor from a send acknowledgement. Some browsers
|
||||||
// had already been upgraded to v2 before that correction, retaining a
|
// had already been upgraded to v2 before that correction, retaining a
|
||||||
// high cursor with holes for their own echoes. Preserve every valid
|
// high cursor with holes for their own echoes. Preserve every valid
|
||||||
// local item but replay the server history once to repair those holes;
|
// local item but replay the server history once to repair those holes;
|
||||||
// the next persist upgrades this record to v5.
|
// the next persist upgrades this record to v5.
|
||||||
return {
|
return {
|
||||||
version: 5,
|
version: 6,
|
||||||
cursor: candidate.version === 4 ? Math.max(0, candidate.cursor) : 0,
|
cursor: candidate.version === 4 || candidate.version === 5 ? Math.max(0, candidate.cursor) : 0,
|
||||||
items: candidate.items as StoredConversationItem[],
|
items: candidate.items as StoredConversationItem[],
|
||||||
outbox: candidate.outbox as OutboxEntry[],
|
outbox: normalizeOutbox(candidate.outbox),
|
||||||
tool_calls: candidate.version === 4 && Array.isArray(candidate.tool_calls) ? candidate.tool_calls as ToolCallRecord[] : [],
|
tool_calls: (candidate.version === 4 || candidate.version === 5) && Array.isArray(candidate.tool_calls) ? candidate.tool_calls as ToolCallRecord[] : [],
|
||||||
tool_drafts: candidate.version === 4 && isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {},
|
tool_drafts: (candidate.version === 4 || candidate.version === 5) && isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {},
|
||||||
tasks: [],
|
tasks: candidate.version === 5 && Array.isArray(candidate.tasks) ? candidate.tasks as TaskRecord[] : [],
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
return {
|
return {
|
||||||
version: 5,
|
version: 6,
|
||||||
cursor: Math.max(0, candidate.cursor),
|
cursor: Math.max(0, candidate.cursor),
|
||||||
items: candidate.items as StoredConversationItem[],
|
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_calls: Array.isArray(candidate.tool_calls) ? candidate.tool_calls as ToolCallRecord[] : [],
|
||||||
tool_drafts: isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {},
|
tool_drafts: isDraftMap(candidate.tool_drafts) ? candidate.tool_drafts : {},
|
||||||
tasks: Array.isArray(candidate.tasks) ? candidate.tasks as TaskRecord[] : [],
|
tasks: Array.isArray(candidate.tasks) ? candidate.tasks as TaskRecord[] : [],
|
||||||
@@ -247,7 +263,18 @@ export class ConversationStore {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function emptyConversation(): PersistedConversation {
|
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<OutboxEntry>;
|
||||||
|
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<string, JsonObject> {
|
function isDraftMap(value: unknown): value is Record<string, JsonObject> {
|
||||||
|
|||||||
Reference in New Issue
Block a user