Q144: each message says where the reply it asked for stands
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
075dea725a
commit
2a89d1e883
18 files changed
+211
-5
No files matched your search
@@ -122,7 +122,10 @@ export class ChatAgentWakeService {
|
||||
// newest messages when it executes, so waking again before it starts would
|
||||
// spend a whole run to answer the same conversation twice.
|
||||
const pending = this.pendingWakeRun(input.orgId, agentId, threadId);
|
||||
if (pending) return { targetId: agentId, taskId: pending.taskId, runId: pending.runId };
|
||||
if (pending) {
|
||||
this.recordReplyRequest(messageId, agentId, pending.taskId, now);
|
||||
return { targetId: agentId, taskId: pending.taskId, runId: pending.runId };
|
||||
}
|
||||
const workspace = this.workspaceForAgent(input.orgId, agent.workspace_id);
|
||||
const from = this.principalName(input.orgId, input.senderType, input.senderId);
|
||||
const space = this.spaceRow(input.orgId, input.spaceId);
|
||||
@@ -142,6 +145,7 @@ export class ChatAgentWakeService {
|
||||
origin: "chat_wake",
|
||||
});
|
||||
const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, senderType: input.senderType, now });
|
||||
this.recordReplyRequest(messageId, agent.id, taskId, now);
|
||||
this.recordPresence(agent.id, { command: "mention", args: "" }, now);
|
||||
return { targetId: agent.id, taskId, runId };
|
||||
}
|
||||
@@ -168,10 +172,16 @@ export class ChatAgentWakeService {
|
||||
origin: "chat_wake",
|
||||
});
|
||||
const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, senderType: input.senderType, now });
|
||||
this.recordReplyRequest(messageId, agent.id, taskId, now);
|
||||
this.recordPresence(agent.id, slash, now);
|
||||
return { targetId: agent.id, taskId, runId };
|
||||
}
|
||||
|
||||
/** Links the message to the wake task that answers it, so the message can show where its reply stands. */
|
||||
private recordReplyRequest(messageId: string, agentId: string, taskId: string, now: string): void {
|
||||
this.db.prepare("insert or ignore into chat_reply_requests(message_id, agent_id, task_id, created_at) values (?, ?, ?, ?)").run(messageId, agentId, taskId, now);
|
||||
}
|
||||
|
||||
private pendingWakeRun(orgId: string, agentId: string, threadId: string): { taskId: string; runId: string } | null {
|
||||
const row = this.db.prepare(`
|
||||
select id, task_id from task_runs
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { SpaceHistoryPage, ThreadHistory } from "../../shared/chat-domain";
|
||||
import { camel } from "./snapshot-service";
|
||||
import { chatReplyStatuses } from "./chat-reply-status";
|
||||
import { taskHolds } from "./task-holds";
|
||||
|
||||
/** Every message of one thread with its attachments, mentions and commands, however far back it started. */
|
||||
export function threadHistory(db: DatabaseSync, orgId: string, threadId: string): ThreadHistory {
|
||||
@@ -55,6 +57,7 @@ export function spaceHistory(db: DatabaseSync, orgId: string, spaceId: string, b
|
||||
mentions: carried("select * from chat_mentions where message_id in (select value from json_each(?)) order by message_id, offset_start"),
|
||||
slashInvocations: carried("select * from chat_slash_invocations where message_id in (select value from json_each(?)) order by message_id"),
|
||||
threads: carried("select * from chat_threads where id in (select thread_id from chat_messages where id in (select value from json_each(?)))"),
|
||||
replyStatuses: chatReplyStatuses(db, page.map((row) => row.id as string), taskHolds(db, new Date())),
|
||||
hasMore,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { ChatReplyStatus } from "../../shared/chat-domain";
|
||||
import type { TaskHold } from "../../shared/domain";
|
||||
|
||||
const EXPIRED_PREFIX = "Expired:";
|
||||
|
||||
/** Where each agent reply a loaded message asked for stands: its wake task, through every run (checkpoints and continuations too). */
|
||||
export function chatReplyStatuses(db: DatabaseSync, messageIds: string[], holds: Map<string, TaskHold>): ChatReplyStatus[] {
|
||||
const requests = db.prepare(`
|
||||
select q.message_id, q.agent_id, q.task_id, t.status as task_status, r.state as run_state, r.error_message
|
||||
from chat_reply_requests q
|
||||
join tasks t on t.id = q.task_id
|
||||
left join task_runs r on r.id = (select id from task_runs where task_id = q.task_id order by created_at desc, rowid desc limit 1)
|
||||
where q.message_id in (select value from json_each(?))
|
||||
order by q.created_at, q.rowid
|
||||
`).all(JSON.stringify(messageIds)) as Array<{
|
||||
message_id: string; agent_id: string; task_id: string; task_status: string; run_state: string | null; error_message: string | null;
|
||||
}>;
|
||||
return requests.map((request): ChatReplyStatus => {
|
||||
const base = { messageId: request.message_id, agentId: request.agent_id };
|
||||
if (request.run_state === "running") return { ...base, state: "running", detail: null, resumesAt: null };
|
||||
if (request.run_state === "failed") {
|
||||
if (request.error_message === null) throw new Error(`The reply to message ${request.message_id} failed without a recorded reason.`);
|
||||
return { ...base, state: request.error_message.startsWith(EXPIRED_PREFIX) ? "expired" : "failed", detail: request.error_message, resumesAt: null };
|
||||
}
|
||||
if (request.run_state === "stopped" || request.run_state === "superseded") return { ...base, state: "stopped", detail: request.error_message, resumesAt: null };
|
||||
// A completed run replied only if it closed the wake task; a step-limit checkpoint leaves it open for the next session.
|
||||
if (request.task_status === "done") return { ...base, state: "replied", detail: null, resumesAt: null };
|
||||
const hold = holds.get(request.task_id);
|
||||
return hold ? { ...base, state: "waiting", detail: hold.reason, resumesAt: hold.resumesAt } : { ...base, state: "queued", detail: null, resumesAt: null };
|
||||
});
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import { IntegrationKeyStore } from "./integration-key-store";
|
||||
import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain";
|
||||
import { buildUsageStats } from "./usage-stats";
|
||||
import { taskHolds } from "./task-holds";
|
||||
import { chatReplyStatuses } from "./chat-reply-status";
|
||||
|
||||
type Row = Record<string, unknown>;
|
||||
type ProviderUsageWindowRow = Pick<
|
||||
@@ -51,6 +52,7 @@ export class SnapshotService {
|
||||
const agents = list<AppSnapshot["agents"][number]>(this.db, "agents", "created_at asc")
|
||||
.map((agent) => ({ ...agent, heartbeatEnabled: toBool(agent.heartbeatEnabled) }));
|
||||
const holds = taskHolds(this.db, new Date());
|
||||
const chatMessages = recentList<AppSnapshot["chatMessages"][number]>(this.db, "chat_messages", "created_at desc, rowid desc", 500).reverse();
|
||||
const tasks = list<AppSnapshot["tasks"][number]>(this.db, "tasks", "created_at asc")
|
||||
.map((task) => ({ ...task, requiresFilesystem: toBool(task.requiresFilesystem), hold: holds.get(task.id) ?? null }));
|
||||
const workspaces = list<AppSnapshot["workspaces"][number]>(this.db, "workspaces", "created_at asc")
|
||||
@@ -96,7 +98,8 @@ export class SnapshotService {
|
||||
approvals: list(this.db, "approvals", "created_at desc"),
|
||||
chatSpaces: list(this.db, "chat_spaces", "created_at asc"),
|
||||
chatMemberships: list(this.db, "chat_memberships", "space_id asc, principal_type asc, principal_id asc"),
|
||||
chatMessages: recentList<AppSnapshot["chatMessages"][number]>(this.db, "chat_messages", "created_at desc, rowid desc", 500).reverse(),
|
||||
chatMessages,
|
||||
chatReplyStatuses: chatReplyStatuses(this.db, chatMessages.map((message) => message.id), holds),
|
||||
// Every thread a loaded message belongs to, plus the most recently active ones, so no shown message loses its thread.
|
||||
chatThreads: all<Row>(this.db.prepare(`
|
||||
select * from chat_threads
|
||||
|
||||
Reference in new issue
Block a user