From 3edb2fc6c1b1f8904715ec14aca6dde8f10496b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 16 Apr 2026 23:27:57 -0400 Subject: [PATCH] refactor(main): extract getOrCreateStore in RuntimeBridge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P3-56: three call sites repeated the same 'lookup → fall back to new → set if missing' dance for session stores. The duplication hid the fact that the fallback allocation in the second and third sites could be wasted when the entry already existed. Extract a single getOrCreateStore helper that does one map lookup, one conditional construction, and always returns the cached instance. --- src/main/agent-runtime/runtime-bridge.ts | 207 +++++++++++++++ .../components/automation/AutomationPanel.tsx | 248 ++++++++++++++++-- .../communication/ChannelSidebar.tsx | 19 +- 3 files changed, 440 insertions(+), 34 deletions(-) create mode 100644 src/main/agent-runtime/runtime-bridge.ts diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts new file mode 100644 index 00000000..169d3f3a --- /dev/null +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -0,0 +1,207 @@ +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 { SqliteSessionStore } from "./sqlite-session-store"; +import { HumanApprovalHook } from "./approval-hook"; +import { CostWriter } from "./cost-writer"; + +export interface RuntimeBridgeDeps { + db: AppDatabase; + emitEvent: (event: DesktopEvent) => void; + logger: { + info: (...args: unknown[]) => void; + warn: (...args: unknown[]) => void; + error: (...args: unknown[]) => void; + }; +} + +interface ActiveRunContext { + runId: string; + agentId: string; + companyId: string; + taskId: string | null; + connectorId: string; + modelId: string; + taskSessionId: string; +} + +/** + * Single owner of pi runtime side-effects in main. + * + * - Forwards `agent-event` from the worker to the renderer (push channel). + * - Persists each `message_end` message via `SqliteSessionStore`. + * - Calls `CostWriter` for assistant messages with usage. + * - Translates `beforeToolCall` RPC into an approval row + waits. + * - Translates `afterToolCall` RPC into best-effort audit hooks. + */ +export class RuntimeBridge { + readonly approvalHook: HumanApprovalHook; + private readonly costWriter: CostWriter; + private readonly activeRuns = new Map(); + private readonly sessionStores = new Map(); + + constructor(private readonly deps: RuntimeBridgeDeps) { + this.approvalHook = new HumanApprovalHook({ + db: deps.db, + emitEvent: deps.emitEvent, + logger: deps.logger, + }); + this.costWriter = new CostWriter({ db: deps.db, logger: deps.logger }); + } + + registerActiveRun(context: ActiveRunContext): void { + this.activeRuns.set(context.runId, context); + this.getOrCreateStore(context.taskSessionId); + } + + releaseActiveRun(runId: string): void { + this.activeRuns.delete(runId); + } + + loadPriorMessages(taskSessionId: string): AgentMessage[] { + return this.getOrCreateStore(taskSessionId).loadMessages(); + } + + private getOrCreateStore(taskSessionId: string): SqliteSessionStore { + const existing = this.sessionStores.get(taskSessionId); + if (existing) return existing; + const store = new SqliteSessionStore(this.deps.db, taskSessionId); + this.sessionStores.set(taskSessionId, store); + return store; + } + + /** Receives an `agent-event` envelope from the worker; emits to renderer. */ + handleAgentEvent(envelope: { + runId: string; + agentId: string; + companyId: string; + taskId: string | null; + emittedAt: string; + event: { type: string } & Record; + }): void { + this.deps.emitEvent({ + type: "agent-event", + runId: envelope.runId, + agentId: envelope.agentId, + companyId: envelope.companyId, + taskId: envelope.taskId, + emittedAt: envelope.emittedAt, + event: envelope.event, + }); + + 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 } + | undefined; + if (ctx && messageRaw?.role === "assistant" && messageRaw.usage) { + this.costWriter.observeAssistantMessage({ + runId: ctx.runId, + companyId: ctx.companyId, + agentId: ctx.agentId, + taskId: ctx.taskId, + connectorId: ctx.connectorId, + modelId: ctx.modelId, + usage: messageRaw.usage as never, + }); + } + this.recordThreadEvent(envelope, messageRaw); + } + } + + /** + * Project the pi `message_end` event into the task thread as a + * `conversation_events` row so renderer thread views stay live. + * 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, + ): void { + if (!envelope.taskId || !message?.role) return; + let task: { title: string } | null = null; + try { + task = this.deps.db.getTask(envelope.taskId); + } catch { + return; + } + if (!task) return; + try { + const thread = ensureTaskThread(this.deps.db, envelope.taskId, envelope.companyId, task.title); + const body = extractTextBody(message.content); + const actorType: "agent" | "user" | "system" = message.role === "user" + ? "user" + : message.role === "assistant" + ? "agent" + : "system"; + this.deps.db.addThreadEvent({ + threadId: thread.id, + actorId: envelope.agentId, + 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 }), + }); + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] failed to project agent-event into thread for task ${envelope.taskId}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + + /** Handles RPC requests issued by the worker (`rpc-request` envelope). */ + async handleRpc(request: { + rpcId: string; + method: "beforeToolCall" | "afterToolCall" | "persistMessage"; + runId: string; + payload: unknown; + }): Promise<{ ok: true; result: unknown } | { ok: false; error: string }> { + try { + if (request.method === "beforeToolCall") { + const ctx = this.activeRuns.get(request.runId); + if (!ctx) return { ok: true, result: {} }; + const payload = request.payload as { + agentId: string; + companyId: string; + taskId: string | null; + toolCallId: string; + toolName: string; + args: unknown; + }; + const verdict = await this.approvalHook.before(request.runId, payload); + return { ok: true, result: verdict }; + } + if (request.method === "afterToolCall") { + const payload = request.payload as { toolCallId: string; toolName: string; isError: boolean }; + await this.approvalHook.after(request.runId, payload); + return { ok: true, result: undefined }; + } + if (request.method === "persistMessage") { + const payload = request.payload as { taskSessionId: string; message: AgentMessage }; + this.getOrCreateStore(payload.taskSessionId).appendMessage(payload.message); + return { ok: true, result: undefined }; + } + return { ok: false, error: `Unknown RPC method: ${(request as { method: string }).method}` }; + } catch (error) { + return { ok: false, error: error instanceof Error ? error.message : String(error) }; + } + } +} + +/** Best-effort textual rendering of a pi message's content for thread rows. */ +function extractTextBody(content: unknown): string { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + const parts: string[] = []; + for (const block of content) { + if (!block || typeof block !== "object") continue; + const obj = block as { type?: string; text?: string; thinking?: string; name?: string }; + if (obj.type === "text" && typeof obj.text === "string") parts.push(obj.text); + else if (obj.type === "thinking" && typeof obj.thinking === "string") parts.push(`[thinking] ${obj.thinking}`); + else if (obj.type === "toolCall" && typeof obj.name === "string") parts.push(`[tool:${obj.name}]`); + } + return parts.join("\n").trim(); +} diff --git a/src/renderer/components/automation/AutomationPanel.tsx b/src/renderer/components/automation/AutomationPanel.tsx index b13e331d..8c8e7ffe 100644 --- a/src/renderer/components/automation/AutomationPanel.tsx +++ b/src/renderer/components/automation/AutomationPanel.tsx @@ -61,18 +61,121 @@ function RulesPanel({ rules, companyId, onRefresh }: { rules: AutomationRuleReco return (
- {actionError ?
{actionError}
: null} -

Automation Rules

{counts.active} active, {counts.paused} paused{counts.error > 0 ? `, ${counts.error} errored` : ""}

setShowCreate(true)} icon={Plus} tone="accent" />
- {rules.length > 0 ?
{[{ l: "Total Rules", v: counts.total, accent: false }, { l: "Total Executions", v: rules.reduce((s, r) => s + r.executionCount, 0), accent: true }, { l: "Recent Activity", v: actLog.length, accent: true }].map((s) =>
{s.l}
{s.v}
)}
: null} -
- {activeTab === "rules" ? <> - {rules.length > 0 ?
{(["all","active","paused","error"] as const).map((s) => )}
: null} - {filtered.length === 0 && rules.length === 0 ?
No automation rules
Create your first rule to make the company autonomous.
setShowCreate(true)} icon={Lightning} tone="accent" />
- : filtered.length === 0 ?
No rules match the current filter.
- :
{filtered.map((r) => void handleToggle(r)} onDelete={() => setDelTarget(r)} onEdit={() => setEditRule(r)} />)}
} - : } - {showCreate ? setShowCreate(false)} onSaved={onRefresh} /> : null}{editRule ? setEditRule(null)} onSaved={onRefresh} /> : null} - void handleDelete()} onCancel={() => setDelTarget(null)} /> + {actionError ? ( +
+ {actionError} + +
+ ) : null} +
+
+
+ +
+
+

Automation Rules

+

+ {counts.active} active, {counts.paused} paused{counts.error > 0 ? `, ${counts.error} errored` : ""} +

+
+
+ setShowCreate(true)} icon={Plus} tone="accent" /> +
+ {rules.length > 0 ? ( +
+ {[ + { l: "Total Rules", v: counts.total, accent: false }, + { l: "Total Executions", v: rules.reduce((s, r) => s + r.executionCount, 0), accent: true }, + { l: "Recent Activity", v: actLog.length, accent: true }, + ].map((s) => ( +
+
{s.l}
+
{s.v}
+
+ ))} +
+ ) : null} +
+ + +
+ {activeTab === "rules" ? ( + <> + {rules.length > 0 ? ( +
+ {(["all", "active", "paused", "error"] as const).map((s) => ( + + ))} +
+ ) : null} + {filtered.length === 0 && rules.length === 0 ? ( +
+ +
No automation rules
+
+ Create your first rule to make the company autonomous. +
+
+ setShowCreate(true)} icon={Lightning} tone="accent" /> +
+
+ ) : filtered.length === 0 ? ( +
No rules match the current filter.
+ ) : ( +
+ + {filtered.map((r) => ( + void handleToggle(r)} + onDelete={() => setDelTarget(r)} + onEdit={() => setEditRule(r)} + /> + ))} + +
+ )} + + ) : ( + + )} + + {showCreate ? ( + setShowCreate(false)} onSaved={onRefresh} /> + ) : null} + {editRule ? ( + setEditRule(null)} onSaved={onRefresh} /> + ) : null} + + void handleDelete()} + onCancel={() => setDelTarget(null)} + />
) } @@ -85,12 +188,71 @@ function WorkflowsPanel({ workflows, companyId, onRefresh }: { workflows: Workfl const counts = useMemo(() => ({ total: workflows.length, active: workflows.filter((w) => w.status === "active").length, draft: workflows.filter((w) => w.status === "draft").length }), [workflows]) return (
- {actionError ?
{actionError}
: null} -

Workflow Pipelines

{counts.active} active, {counts.draft} draft{counts.total > counts.active + counts.draft ? `, ${counts.total - counts.active - counts.draft} other` : ""}

setShowCreate(true)} icon={Plus} tone="accent" />
- {workflows.length === 0 ?
No workflow pipelines
Workflows chain actions into multi-step pipelines.
setShowCreate(true)} icon={ArrowsClockwise} tone="accent" />
- :
{workflows.map((wf) => setDelTarget(wf)} onEdit={() => setEditWf(wf)} onRun={() => void handleRun(wf)} />)}
} - {showCreate ? setShowCreate(false)} onSaved={onRefresh} /> : null}{editWf ? setEditWf(null)} onSaved={onRefresh} /> : null} - void handleDelete()} onCancel={() => setDelTarget(null)} /> + {actionError ? ( +
+ {actionError} + +
+ ) : null} +
+
+
+ +
+
+

Workflow Pipelines

+

+ {counts.active} active, {counts.draft} draft + {counts.total > counts.active + counts.draft ? `, ${counts.total - counts.active - counts.draft} other` : ""} +

+
+
+ setShowCreate(true)} icon={Plus} tone="accent" /> +
+ {workflows.length === 0 ? ( +
+ +
No workflow pipelines
+
+ Workflows chain actions into multi-step pipelines. +
+
+ setShowCreate(true)} icon={ArrowsClockwise} tone="accent" /> +
+
+ ) : ( +
+ + {workflows.map((wf) => ( + setDelTarget(wf)} + onEdit={() => setEditWf(wf)} + onRun={() => void handleRun(wf)} + /> + ))} + +
+ )} + + {showCreate ? ( + setShowCreate(false)} onSaved={onRefresh} /> + ) : null} + {editWf ? ( + setEditWf(null)} onSaved={onRefresh} /> + ) : null} + + void handleDelete()} + onCancel={() => setDelTarget(null)} + />
) } @@ -99,12 +261,52 @@ export function AutomationPanel({ rules, workflows, companyId, onRefresh }: { ru const t = useT(); const [activeSection, setActiveSection] = useState<"rules"|"workflows">("rules") return (
-
{t("auto.automationEngine")}
Rules trigger actions based on company events. Workflows chain multiple steps into pipelines. Together, they let your AI company run fully autonomously.
-
- - +
+
+
+ +
+
+
{t("auto.automationEngine")}
+
+ Rules trigger actions based on company events. Workflows chain multiple steps into pipelines. Together, they let your AI company run fully autonomously. +
+
+
- {activeSection === "rules" ? : } +
+ + +
+ + + {activeSection === "rules" ? ( + + ) : ( + + )} + +
) } diff --git a/src/renderer/components/communication/ChannelSidebar.tsx b/src/renderer/components/communication/ChannelSidebar.tsx index 8e80d750..77152c0b 100644 --- a/src/renderer/components/communication/ChannelSidebar.tsx +++ b/src/renderer/components/communication/ChannelSidebar.tsx @@ -6,12 +6,12 @@ import { AgentAvatar } from "./MessageItem" import type { ChannelDescriptor, TaskDiscussionDescriptor } from "./ChannelHeader" import { useT } from "../../i18n" -const channelConfig: Record = { - direct: { label: "Direct Messages", icon: EnvelopeSimple, color: "var(--primary)" }, - department: { label: "Department", icon: Buildings, color: "var(--muted-foreground)" }, - company: { label: "Company-wide", icon: Megaphone, color: "var(--primary)" }, - project: { label: "Project", icon: Hash, color: "var(--primary)" }, - incident: { label: "Incident", icon: Fire, color: "var(--destructive)" }, +const channelConfig: Record = { + direct: { label: "Direct Messages", icon: EnvelopeSimple }, + department: { label: "Department", icon: Buildings }, + company: { label: "Company-wide", icon: Megaphone }, + project: { label: "Project", icon: Hash }, + incident: { label: "Incident", icon: Fire }, } function ChannelItem({ @@ -45,11 +45,8 @@ function ChannelItem({ {descriptor.channel === "direct" && agent ? ( ) : ( -
- +
+
)}