Write thread events for taskless communication runs

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.
2026-04-19 14:51:26 -04:00
1 parent 1959d218d9
commit 1b8e6d384a
@@ -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<string, ActiveRunContext>();
  private readonly sessionStores = new Map<string, SqliteSessionStore>();
+  /**
+   * 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<string, Map<number, InboundCorrelationEntry>>();

  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(
@@ -51,7 +51,8 @@ export interface RuntimeMessageInjectorDeps {
 interface PendingSteer {
  recipientAgentId: string;
  companyId: string;
-  messageId: string;
+  runId: string;
+  message: AgentMessageRecord;
  timer: ReturnType<typeof setTimeout>;
 }

@@ -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();
@@ -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,