From 1b8e6d384ae33761bb44ef8aa21464fb821fe513 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Sun, 19 Apr 2026 14:51:26 -0400 Subject: [PATCH] Write thread events for taskless communication runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit recordThreadEvent used to drop the event whenever `envelope.taskId` was null, so communication wakes produced nothing in the thread view even while the agent actively replied to peers. Taskless runs now write into a per-agent general thread (`ensureAgentCommunicationThread`), so the UI surfaces peer conversations alongside task activity. Peer-authored user turns inside the transcript used to be labelled with the recipient's id — pi erases the sender once the message is projected. A new `inboundCorrelations` map on `RuntimeBridge` pins each pi-message timestamp to its source `agent_messages` row at injection time, so the thread-event writer can recover the real author id and `senderKind` on `message_end`. prepareRun projections (no `message_end` fires for priorMessages) write their thread event directly as the projection lands so they are not silently lost. --- src/main/agent-runtime/runtime-bridge.ts | 236 +++++++++++++++--- .../agent-runtime/runtime-message-injector.ts | 19 +- src/main/thread-integration/thread-helpers.ts | 36 +++ 3 files changed, 250 insertions(+), 41 deletions(-) diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index 84ec4fc2..c4f8d609 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -1,7 +1,10 @@ import type { AgentMessage } from "@mariozechner/pi-agent-core"; import type { AppDatabase } from "@main/database"; -import type { DesktopEvent } from "@shared/types"; -import { ensureTaskThread } from "@main/thread-integration/thread-helpers"; +import type { AgentMessageRecord, DesktopEvent } from "@shared/types"; +import { + ensureAgentCommunicationThread, + ensureTaskThread, +} from "@main/thread-integration/thread-helpers"; import { SqliteSessionStore } from "./sqlite-session-store"; import { CostWriter } from "./cost-writer"; import type { CredentialResolver } from "@main/credentials"; @@ -52,10 +55,26 @@ type RpcMethod = * Policy itself lives in `hook-handlers.ts`; this class is a pure IPC * router plus an `ActiveRunContext` cache. */ +interface InboundCorrelationEntry { + fromAgentId: string; + senderKind: AgentMessageRecord["senderKind"]; + messageId: string; +} + export class RuntimeBridge { private readonly costWriter: CostWriter; private readonly activeRuns = new Map(); private readonly sessionStores = new Map(); + /** + * Per-run mapping from pi-message timestamp to the inbound message that + * sourced it. The transcript only carries a rendered `[Message from X]` + * header, which is insufficient to recover the original author when the + * thread-event writer fires on `message_end`. Writers (inbox projection + * during prepareRun, steer injection during the run) register the entry + * before pi surfaces the message so the author-id survives the + * round-trip. + */ + private readonly inboundCorrelations = new Map>(); constructor(private readonly deps: RuntimeBridgeDeps) { this.costWriter = new CostWriter({ db: deps.db }); @@ -67,16 +86,40 @@ export class RuntimeBridge { } /** - * Evicts the per-run `activeRuns` context and the per-session store cache. - * Callers must invoke after a `run-finished` worker message so the maps - * stay bounded by concurrent-run count rather than process lifetime. + * Evicts the per-run `activeRuns` context, the per-session store cache, + * and the inbound-correlation log. Callers must invoke after a + * `run-finished` worker message so the maps stay bounded by concurrent- + * run count rather than process lifetime. */ releaseRunResources(runId: string): void { const ctx = this.activeRuns.get(runId); this.activeRuns.delete(runId); + this.inboundCorrelations.delete(runId); if (ctx) this.sessionStores.delete(ctx.taskSessionId); } + /** + * Register an inbound message that is about to enter (or has already + * entered) the recipient's transcript as a user-role turn. Called from + * the prepareRun inbox projection and from `RuntimeMessageInjector` on + * steer-ack success so the `recordThreadEvent` hook can recover the + * original author instead of labelling peer messages as the recipient. + */ + registerInboundInjection(runId: string, record: AgentMessageRecord): void { + const timestamp = Date.parse(record.createdAt); + if (!Number.isFinite(timestamp)) return; + let runMap = this.inboundCorrelations.get(runId); + if (!runMap) { + runMap = new Map(); + this.inboundCorrelations.set(runId, runMap); + } + runMap.set(timestamp, { + fromAgentId: record.fromAgentId, + senderKind: record.senderKind, + messageId: record.id, + }); + } + /** Returns the active run context for a given agent, if any. Used by * main-side services (inbox dispatcher) to decide whether a new inbox * row should route to `agent.steer()` instead of waiting for the next @@ -165,7 +208,7 @@ export class RuntimeBridge { if (envelope.event.type === "message_end") { const ctx = this.activeRuns.get(envelope.runId); const messageRaw = envelope.event.message as - | { role?: string; usage?: unknown; content?: unknown; toolCallId?: string } + | { role?: string; usage?: unknown; content?: unknown; toolCallId?: string; timestamp?: number } | undefined; if (ctx && messageRaw?.role === "assistant" && messageRaw.usage) { try { @@ -189,50 +232,158 @@ export class RuntimeBridge { } /** - * Project the pi `message_end` event into the task thread as a - * `conversation_events` row so renderer thread views stay live. + * Project the pi `message_end` event into a `conversation_events` row + * so renderer thread views stay live. Resolves the thread from the run + * context: task runs write into the task thread; taskless runs — used + * by communication wakes — write into a per-agent general thread so + * peer conversations remain visible in the UI. + * + * User-role messages whose pi timestamp matches a registered inbound + * injection are attributed back to the original author via the + * correlation map; without that, the envelope's `agentId` (the + * recipient) would be recorded as the sender and every peer message + * would collapse into a self-reply row. + * * Best-effort: thread persistence failures must not interrupt the run. */ private recordThreadEvent( envelope: { runId: string; agentId: string; companyId: string; taskId: string | null }, - message: { role?: string; content?: unknown; toolCallId?: string } | undefined, + message: { role?: string; content?: unknown; toolCallId?: string; timestamp?: number } | undefined, ): void { - if (!envelope.taskId || !message?.role) return; - let task: { title: string } | null = null; + if (!message?.role) return; + + const thread = this.resolveThreadForRun(envelope); + if (!thread) return; + try { - task = this.deps.db.getTask(envelope.taskId); - } catch (err) { - this.deps.logger.warn(`[runtime-bridge] getTask failed for ${envelope.taskId} in recordThreadEvent:`, err instanceof Error ? err.message : String(err)); - return; - } - if (!task) return; - try { - const thread = ensureTaskThread(this.deps.db, envelope.taskId, envelope.companyId, task.title); const body = extractTextBody(message.content); - const isUserMessage = message.role === "user"; - const actorType: "agent" | "user" | "system" = isUserMessage - ? "user" - : message.role === "assistant" - ? "agent" - : "system"; - // User-role messages inside the transcript come from either a human - // (inbox projection → Board/User) or a peer agent (via steer). The - // envelope's agentId is the *recipient*, never the author; recording - // it there creates fake "CEO replied to CEO" rows. For user-authored - // rows we record the virtual user actor id; peer-authored user rows - // are labeled "system" since pi has already erased the sender. - const actorId = isUserMessage ? "user" : envelope.agentId; + const actor = this.resolveThreadEventActor(envelope, message); this.deps.db.addThreadEvent({ threadId: thread.id, - actorId, - actorType, + actorId: actor.actorId, + actorType: actor.actorType, eventType: message.role === "toolResult" ? "system_note" : "message", body: body.length > 4000 ? `${body.slice(0, 3997)}...` : body, - metadataJson: JSON.stringify({ runId: envelope.runId, role: message.role }), + metadataJson: JSON.stringify({ + runId: envelope.runId, + role: message.role, + sourceMessageId: actor.sourceMessageId ?? undefined, + }), }); } catch (error) { this.deps.logger.warn( - `[runtime-bridge] failed to project agent-event into thread for task ${envelope.taskId}: ${ + `[runtime-bridge] failed to project agent-event into thread ${thread.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + + private resolveThreadForRun( + envelope: { runId: string; agentId: string; companyId: string; taskId: string | null }, + ) { + if (envelope.taskId) { + try { + const task = this.deps.db.getTask(envelope.taskId); + return ensureTaskThread(this.deps.db, envelope.taskId, envelope.companyId, task.title); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] getTask failed for ${envelope.taskId}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + try { + const agentLabel = this.resolveAgentName(envelope.agentId); + return ensureAgentCommunicationThread( + this.deps.db, + envelope.agentId, + envelope.companyId, + agentLabel, + ); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] ensureAgentCommunicationThread failed for ${envelope.agentId}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + + private resolveThreadEventActor( + envelope: { runId: string; agentId: string }, + message: { role?: string; timestamp?: number }, + ): { actorId: string; actorType: "agent" | "user" | "system"; sourceMessageId: string | null } { + if (message.role === "user") { + const correlation = this.lookupInboundCorrelation(envelope.runId, message.timestamp); + if (correlation) { + const actorType: "agent" | "user" | "system" = + correlation.senderKind === "human" + ? "user" + : correlation.senderKind === "system" + ? "system" + : "agent"; + return { + actorId: correlation.fromAgentId, + actorType, + sourceMessageId: correlation.messageId, + }; + } + return { actorId: "user", actorType: "user", sourceMessageId: null }; + } + if (message.role === "assistant") { + return { actorId: envelope.agentId, actorType: "agent", sourceMessageId: null }; + } + return { actorId: envelope.agentId, actorType: "system", sourceMessageId: null }; + } + + private lookupInboundCorrelation( + runId: string, + timestamp: number | undefined, + ): InboundCorrelationEntry | null { + if (typeof timestamp !== "number" || !Number.isFinite(timestamp)) return null; + const runMap = this.inboundCorrelations.get(runId); + if (!runMap) return null; + return runMap.get(timestamp) ?? null; + } + + /** + * Write the conversation-event row for an inbound peer message that + * was projected onto the recipient's transcript as a prior-turn (via + * `prepareRun`). priorMessages never fire `message_end`, so this is + * the only path that surfaces them in the thread view. Mid-run steer + * injections still go through `recordThreadEvent` off `message_end` + * and resolve their author via `inboundCorrelations`. + */ + private recordInboundMessageAsThreadEvent( + envelope: { runId: string; agentId: string; companyId: string; taskId: string | null }, + record: AgentMessageRecord, + ): void { + const thread = this.resolveThreadForRun(envelope); + if (!thread) return; + try { + const body = record.body.length > 4000 ? `${record.body.slice(0, 3997)}...` : record.body; + const actorType: "agent" | "user" | "system" = + record.senderKind === "human" + ? "user" + : record.senderKind === "system" + ? "system" + : "agent"; + this.deps.db.addThreadEvent({ + threadId: thread.id, + actorId: record.fromAgentId, + actorType, + eventType: "message", + body, + metadataJson: JSON.stringify({ + runId: envelope.runId, + role: "user", + sourceMessageId: record.id, + projection: "prepare_run", + }), + }); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] inbound thread-event write failed for message ${record.id}: ${ error instanceof Error ? error.message : String(error) }`, ); @@ -306,6 +457,19 @@ export class RuntimeBridge { getAgentName: (id) => this.resolveAgentName(id), }); store.appendMessage(projected); + this.registerInboundInjection(runId, record); + // priorMessages do not fire `message_end`, so write the thread + // event directly. Steer-injected messages (mid-run) go through + // `recordThreadEvent` via the pi event loop instead. + this.recordInboundMessageAsThreadEvent( + { + runId, + agentId: payload.agentId, + companyId: payload.companyId, + taskId: payload.taskId, + }, + record, + ); } if (inboxRecords.length > 0) { this.deps.db.markAgentInboxDelivered( diff --git a/src/main/agent-runtime/runtime-message-injector.ts b/src/main/agent-runtime/runtime-message-injector.ts index 252d8a9c..dd72544f 100644 --- a/src/main/agent-runtime/runtime-message-injector.ts +++ b/src/main/agent-runtime/runtime-message-injector.ts @@ -51,7 +51,8 @@ export interface RuntimeMessageInjectorDeps { interface PendingSteer { recipientAgentId: string; companyId: string; - messageId: string; + runId: string; + message: AgentMessageRecord; timer: ReturnType; } @@ -116,7 +117,8 @@ export class RuntimeMessageInjector { this.pending.set(correlationId, { recipientAgentId, companyId, - messageId: message.id, + runId: ctx.runId, + message, timer, }); return true; @@ -145,17 +147,24 @@ export class RuntimeMessageInjector { if (accepted) { try { - this.deps.db.markAgentInboxDelivered(entry.recipientAgentId, [entry.messageId]); + this.deps.runtimeBridge.registerInboundInjection(entry.runId, entry.message); } catch (error) { this.deps.logger.warn( - `[injector] markAgentInboxDelivered failed for ${entry.messageId}: ${error instanceof Error ? error.message : String(error)}`, + `[injector] registerInboundInjection failed for ${entry.message.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + try { + this.deps.db.markAgentInboxDelivered(entry.recipientAgentId, [entry.message.id]); + } catch (error) { + this.deps.logger.warn( + `[injector] markAgentInboxDelivered failed for ${entry.message.id}: ${error instanceof Error ? error.message : String(error)}`, ); } return; } this.deps.logger.warn( - `[injector] steer rejected (${source}) for message ${entry.messageId} to agent ${entry.recipientAgentId}; falling back to wake + inbox projection`, + `[injector] steer rejected (${source}) for message ${entry.message.id} to agent ${entry.recipientAgentId}; falling back to wake + inbox projection`, ); const wake = this.deps.getWakeAgent(); diff --git a/src/main/thread-integration/thread-helpers.ts b/src/main/thread-integration/thread-helpers.ts index d107d5f7..efdb2763 100644 --- a/src/main/thread-integration/thread-helpers.ts +++ b/src/main/thread-integration/thread-helpers.ts @@ -35,6 +35,42 @@ export function ensureTaskThread(db: AppDatabase, taskId: string, companyId: str }); } +export function findAgentCommunicationThread(db: AppDatabase, agentId: string, companyId: string) { + const existing = db.listThreads({ + companyId, + type: "general" as ThreadType, + relatedEntityType: "agent", + relatedEntityId: agentId, + limit: 1, + }); + return existing[0] ?? null; +} + +/** + * Returns (or creates) the single "general" thread that accumulates an + * agent's taskless communication activity. Task runs already write into + * a task-scoped thread; taskless communication runs would otherwise drop + * their conversation events on the floor, leaving the thread view blank + * even while the agent actively replies to peers. + */ +export function ensureAgentCommunicationThread( + db: AppDatabase, + agentId: string, + companyId: string, + agentLabel: string, +) { + const existing = findAgentCommunicationThread(db, agentId, companyId); + if (existing) return existing; + + return db.createThread({ + companyId, + title: `${agentLabel} · communication`, + type: "general" as ThreadType, + relatedEntityType: "agent", + relatedEntityId: agentId, + }); +} + export function findApprovalThread(db: AppDatabase, approvalId: string, companyId: string) { const existing = db.listThreads({ companyId,