feat(messaging): route messages to live runs via pi steer

Incoming messages (human → agent, agent → agent, channel broadcast)
persisted to the inbox and were only projected into the recipient's
transcript on the next `prepareRun`. When the recipient was already
running, every message waited for the next wake — a Slack-like
system operating at heartbeat cadence, not in real time.

`RuntimeMessageInjector` closes the gap: it looks up the recipient's
active run through `RuntimeBridge.findActiveRunForAgent`, projects
the persisted `AgentMessageRecord` via the existing inbox-projection
helper, and posts a `steer-run` ParentMessage into the worker. The
worker's runner enqueues the projection on the Agent's steering
queue; pi drains it at the next turn boundary and emits the usual
`message_end` → `persistMessage` sequence, so the session store
shadow is preserved without double-writing.

`dispatchAgentMessage` gains an optional `injectRuntimeMessage` hook
used by every API-side dispatch. Injection succeeds → we mark the
inbox row delivered and skip `wakeAgentIfPossible`. Injection fails
(no active run) → the legacy inbox + wake path still fires, so
sleeping agents continue to receive their messages on the next turn.

The injector is owned by the service graph alongside `workerService`
and `piRuntimeRef`, which gives it both the worker postMessage
channel and the RuntimeBridge active-run map without additional
coupling. Internal dispatch call sites (sweepStaleReviews,
automation actions) leave the hook unset; those system messages
target idle agents by design and do not need the runtime path.

src/main/agent-runtime/index.ts

@@ -1,3 +1,4 @@
 export { SqliteSessionStore } from "./sqlite-session-store";
 export { CostWriter } from "./cost-writer";
 export { RuntimeBridge } from "./runtime-bridge";
+export { RuntimeMessageInjector } from "./runtime-message-injector";

src/main/agent-runtime/runtime-message-injector.ts

@@ -0,0 +1,91 @@
+/**
+ * Runtime message injector — the inward half of the Slack-like
+ * messaging system. When an agent has an active pi run, incoming
+ * direct / channel / mention messages must reach the agent *during*
+ * the loop, not on the next heartbeat. Pre-Task-3 the only delivery
+ * path was a batched `prepareRun` projection, so any message that
+ * arrived while the recipient was working sat dark until the next
+ * wake.
+ *
+ * This module closes that gap by routing a projected pi AgentMessage
+ * into the worker's active Agent via the `steer-run` ParentMessage.
+ * Pi's steering queue drains at the next turn boundary; the agent
+ * observes the new user turn and can react in-loop.
+ *
+ * Delivery is best-effort and marked as delivered upon post. The
+ * session-store shadow still writes via the normal pi message_end
+ * event so the transcript remains durable. Callers fall back to the
+ * existing inbox+wake path when `tryInject` returns false.
+ */
+
+import type { AppDatabase } from "@main/database";
+import type { AgentMessageRecord } from "@shared/types";
+import { projectInboxMessageToPi } from "./inbox-projection";
+import type { RuntimeBridge } from "./runtime-bridge";
+import type { ParentMessage } from "@main/worker/worker-helpers";
+
+export interface RuntimeMessageInjectorDeps {
+  db: AppDatabase;
+  runtimeBridge: RuntimeBridge;
+  postToWorker: (message: ParentMessage) => void;
+  logger: {
+    info: (...args: unknown[]) => void;
+    warn: (...args: unknown[]) => void;
+  };
+}
+
+export class RuntimeMessageInjector {
+  constructor(private readonly deps: RuntimeMessageInjectorDeps) {}
+
+  /**
+   * Attempts to deliver a persisted inbox message into the recipient's
+   * live pi run. Returns true when the message was dispatched to the
+   * worker, false when no active run exists for the recipient.
+   *
+   * The caller owns the "post-delivery" decision: on true, it should
+   * skip the regular `wakeAgentIfPossible` path for this recipient; on
+   * false, it falls back to the legacy inbox+wake flow.
+   */
+  tryInject(recipientAgentId: string, companyId: string, message: AgentMessageRecord): boolean {
+    const ctx = this.deps.runtimeBridge.findActiveRunForAgent(recipientAgentId);
+    if (!ctx) return false;
+    if (ctx.companyId !== companyId) {
+      this.deps.logger.warn(
+        `[injector] refusing cross-company steer: agent ${recipientAgentId} active in ${ctx.companyId}, message from ${companyId}`,
+      );
+      return false;
+    }
+
+    const projected = projectInboxMessageToPi(message, {
+      getAgentName: (id) => this.resolveAgentName(id),
+    });
+
+    try {
+      this.deps.postToWorker({ type: "steer-run", runId: ctx.runId, message: projected });
+    } catch (error) {
+      this.deps.logger.warn(
+        `[injector] postToWorker steer-run failed for run ${ctx.runId}: ${error instanceof Error ? error.message : String(error)}`,
+      );
+      return false;
+    }
+
+    try {
+      this.deps.db.markAgentInboxDelivered(recipientAgentId, [message.id]);
+    } catch (error) {
+      this.deps.logger.warn(
+        `[injector] markAgentInboxDelivered failed for ${message.id}: ${error instanceof Error ? error.message : String(error)}`,
+      );
+    }
+    return true;
+  }
+
+  private resolveAgentName(agentId: string): string {
+    try {
+      const name = this.deps.db.findAgentName(agentId);
+      if (name && name.length > 0) return name;
+    } catch {
+      // fall through
+    }
+    return agentId;
+  }
+}

src/main/api-server/routes-messages.ts

@@ -109,6 +109,7 @@ export function registerMessageRoutes(server: AgentApiServer) {
           if (!agentId) return null;
           return onWakeAgent(agentId, cid, trigger);
         },
+        injectRuntimeMessage: server.callbacks.injectRuntimeMessage,
         onHumanMessage: server.callbacks.onHumanMessage,
       },
       parsed.data,

src/main/api-server/server-core.ts

@@ -2,7 +2,7 @@ import { createServer, type IncomingMessage, type ServerResponse } from "node:ht
 import log from "electron-log/main.js";
 import { validateAgentToken, type TokenPayload } from "../agent-auth";
 import type { AppDatabase } from "../database";
-import type { DesktopEvent, SectionId } from "@shared/types";
+import type { AgentMessageRecord, DesktopEvent, SectionId } from "@shared/types";
 
 /**
  * Decode an HTTP request body to a UTF-8 string, handling the common Windows
@@ -64,6 +64,13 @@ export interface ApiServerEventCallbacks {
   onDocumentCreated: (companyId: string, authorAgentId: string | null, projectId: string | null, documentId: string, documentTitle: string, documentType: string, taskId: string | null) => void;
   onAutoAssignTask: (taskId: string, companyId: string) => string | null;
   onHumanMessage: (companyId: string, toAgentId?: string | null, threadId?: string | null, waitStateId?: string | null) => void;
+  /**
+   * Tries to hand a freshly persisted message to the recipient's live
+   * pi run via the runtime steer channel. Returns true when the worker
+   * accepted the steer; dispatch should then skip the regular wake
+   * pipeline for this recipient.
+   */
+  injectRuntimeMessage: (recipientAgentId: string, companyId: string, message: AgentMessageRecord) => boolean;
   emitEvent: (event: DesktopEvent) => void;
   notify: (options: { title: string; body: string; urgency: "critical" | "informational"; navigation?: { section: SectionId; entityId?: string }; batchKey?: string }) => void;
 }
@@ -145,6 +152,7 @@ export class AgentApiServer {
       onDocumentCreated: callbacks?.onDocumentCreated ?? (() => {}),
       onAutoAssignTask: callbacks?.onAutoAssignTask ?? (() => null),
       onHumanMessage: callbacks?.onHumanMessage ?? (() => {}),
+      injectRuntimeMessage: callbacks?.injectRuntimeMessage ?? (() => false),
       emitEvent: callbacks?.emitEvent ?? (() => {}),
       notify: callbacks?.notify ?? (() => {}),
     };

src/main/bootstrap/wire-api-server.ts

@@ -1,7 +1,7 @@
 import log from "electron-log/main.js";
 import type { AgentApiServer } from "../api-server";
 import type { AppDatabase } from "../database";
-import type { DesktopEvent, SectionId } from "@shared/types";
+import type { AgentMessageRecord, DesktopEvent, SectionId } from "@shared/types";
 import type { NotificationOptions } from "../notification-service";
 import type { createServiceGraph } from "../service-graph";
 import type { createThreadIntegration } from "../thread-integration";
@@ -24,6 +24,7 @@ export interface CreateApiServerFn {
     onDocumentCreated: (companyId: string, authorAgentId: string | null, projectId: string | null, documentId: string, documentTitle: string, documentType: string, taskId: string | null) => void;
     onAutoAssignTask: (taskId: string, companyId: string) => string | null;
     onHumanMessage: (companyId: string, toAgentId?: string | null, threadId?: string | null, waitStateId?: string | null) => void;
+    injectRuntimeMessage: (recipientAgentId: string, companyId: string, message: AgentMessageRecord) => boolean;
     emitEvent: (event: DesktopEvent) => void;
     notify: (options: { title: string; body: string; urgency: "critical" | "informational"; navigation?: { section: SectionId; entityId?: string }; batchKey?: string }) => void;
   }, allowedOrigins: string[]): AgentApiServer;
@@ -108,6 +109,18 @@ export function wireApiServer(deps: WireApiServerDeps): AgentApiServer {
       deps.threadIntegration.onHumanMessage(companyId, toAgentId, threadId, waitStateId);
       deps.publishDomainChanged();
     },
+    injectRuntimeMessage: (recipientAgentId, companyId, message) => {
+      const injector = deps.serviceGraph.getRuntimeMessageInjector();
+      if (!injector) return false;
+      try {
+        return injector.tryInject(recipientAgentId, companyId, message);
+      } catch (error) {
+        log.warn(
+          `[wire-api-server] runtime message injection failed for agent ${recipientAgentId}: ${error instanceof Error ? error.message : String(error)}`,
+        );
+        return false;
+      }
+    },
     emitEvent: deps.emitEvent,
     notify: deps.notify,
   }, allowedOrigins);

src/main/message-dispatch.ts

@@ -64,6 +64,14 @@ export interface DispatchAgentMessageDependencies {
   db: AppDatabase;
   publishDomainChanged?: () => void;
   wakeAgentIfPossible?: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null;
+  /**
+   * Attempts to route a persisted message into the recipient's live pi
+   * run via `agent.steer()`. Returns true when delivery happened, false
+   * when the recipient has no active run. When true, dispatch skips
+   * `wakeAgentIfPossible` for that recipient because the message is
+   * already in-flight inside the agent loop.
+   */
+  injectRuntimeMessage?: (recipientAgentId: string, companyId: string, message: AgentMessageRecord) => boolean;
   emitEvent?: (event: DesktopEvent) => void;
   notify?: (options: { title: string; body: string; urgency: "critical" | "informational"; navigation?: { section: SectionId; entityId?: string }; batchKey?: string }) => void;
   /** Called when a human user sends a message. Resolves thread wait states scoped by waitStateId, threadId, or recipient. */
@@ -303,6 +311,18 @@ export function dispatchAgentMessage(
     const agentNameMap = new Map(agentNameRows.map((a) => [a.id, a.name]));
     for (const wakeTarget of wakeTargets) {
       const agentName = agentNameMap.get(wakeTarget) ?? wakeTarget;
+      // Prefer live-loop injection: if the recipient is currently running
+      // a pi turn, the message lands in its steering queue and is
+      // observed within one turn. When no active run exists, fall back to
+      // the inbox + wake pipeline so a sleeping agent is woken with the
+      // message queued for its next `prepareRun` cycle.
+      const injected = message && deps.injectRuntimeMessage
+        ? deps.injectRuntimeMessage(wakeTarget, normalized.companyId, message)
+        : false;
+      if (injected) {
+        mentionResults.push({ agentId: wakeTarget, agentName, outcome: "woken" });
+        continue;
+      }
       const runId = deps.wakeAgentIfPossible(wakeTarget, normalized.companyId, wakeTrigger);
       const hasQueuedWake = !runId && deps.db.listWakeupRequests(normalized.companyId, {
         agentId: wakeTarget,

src/main/service-graph.ts

@@ -6,6 +6,7 @@ import { BrowserManager } from "./browser-manager";
 import type { DesktopResult } from "@shared/contracts";
 import { createConnectorHealthService } from "./connector-health-service";
 import { createPiRuntimeServices, type PiRuntimeServices } from "./service-graph-pi-runtime";
+import { RuntimeMessageInjector } from "./agent-runtime";
 import type { DesktopEvent } from "@shared/types";
 import type { NotificationOptions } from "./notification-service";
 import type { createAutomationService } from "./automation-service";
@@ -98,6 +99,7 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
   let updaterService: ReturnType<typeof createUpdaterService> | null = null;
   let appLifecycleService: ReturnType<typeof createLifecycle> | null = null;
   let leaf: ReturnType<typeof createLeafServices> | null = null;
+  let runtimeMessageInjector: RuntimeMessageInjector | null = null;
   const piRuntimeRef: { current: PiRuntimeServices | null } = { current: null };
 
   // Late-bound slots resolve circular construction: worker needs orchestration
@@ -171,6 +173,17 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
       piRuntimeRef,
     });
 
+    runtimeMessageInjector = new RuntimeMessageInjector({
+      db,
+      runtimeBridge: piRuntimeRef.current.runtimeBridge,
+      postToWorker: (message) => {
+        const worker = workerService?.getWorker();
+        if (!worker) throw new Error("Worker is not initialised — cannot deliver steer message.");
+        worker.postMessage(message);
+      },
+      logger,
+    });
+
     orchestrationService = buildOrchestrationService({
       db,
       logger,
@@ -233,6 +246,7 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
         orchestrationService = null;
         updaterService = null;
         appLifecycleService = null;
+        runtimeMessageInjector = null;
         leaf = null;
       },
       lateBindings: [
@@ -259,5 +273,6 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
     getOrCreateBudgetEnforcementService: () => requireLeaf().budgetEnforcementService,
     getOrCreateSessionManagementService: () => requireLeaf().sessionManagementService,
     getPiRuntime: () => piRuntimeRef.current,
+    getRuntimeMessageInjector: () => runtimeMessageInjector,
   };
 }