diff --git a/src/main/agent-runtime/index.ts b/src/main/agent-runtime/index.ts index 667a9095..bfa688fb 100644 --- a/src/main/agent-runtime/index.ts +++ b/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"; diff --git a/src/main/agent-runtime/runtime-message-injector.ts b/src/main/agent-runtime/runtime-message-injector.ts new file mode 100644 index 00000000..b629bf54 --- /dev/null +++ b/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; + } +} diff --git a/src/main/api-server/routes-messages.ts b/src/main/api-server/routes-messages.ts index b530b42b..a3221eb4 100644 --- a/src/main/api-server/routes-messages.ts +++ b/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, diff --git a/src/main/api-server/server-core.ts b/src/main/api-server/server-core.ts index 695cc4a8..21842711 100644 --- a/src/main/api-server/server-core.ts +++ b/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 ?? (() => {}), }; diff --git a/src/main/bootstrap/wire-api-server.ts b/src/main/bootstrap/wire-api-server.ts index 2a3735e1..5f0e298d 100644 --- a/src/main/bootstrap/wire-api-server.ts +++ b/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); diff --git a/src/main/message-dispatch.ts b/src/main/message-dispatch.ts index d05d8fcc..2cb1302f 100644 --- a/src/main/message-dispatch.ts +++ b/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, diff --git a/src/main/service-graph.ts b/src/main/service-graph.ts index f40aaea2..b61f37eb 100644 --- a/src/main/service-graph.ts +++ b/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 | null = null; let appLifecycleService: ReturnType | null = null; let leaf: ReturnType | 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, }; }