diff --git a/src/main/agent-runtime/inbox-projection.ts b/src/main/agent-runtime/inbox-projection.ts new file mode 100644 index 00000000..1f426fc1 --- /dev/null +++ b/src/main/agent-runtime/inbox-projection.ts @@ -0,0 +1,96 @@ +/** + * Inbox projection — converts a stored AgentMessageRecord into a pi + * AgentMessage (user role) so inter-agent communication enters the + * recipient's transcript as natural conversation rather than a prompt + * block re-rendered every turn. + * + * The projection is the single source of truth for how an inbound message + * looks inside a recipient's context: `[Message from {sender}{· channel}]` + * header, optional subject line, then the body. Content is normalized + * (control characters stripped, oversized bodies truncated) via the same + * helpers used by the legacy prompt-builder boundary. + */ + +import type { AgentMessage } from "@mariozechner/pi-agent-core"; +import type { AgentMessageRecord } from "@shared/types"; + +const MAX_MESSAGE_BODY_LENGTH = 4000; +const MAX_SUBJECT_LENGTH = 200; + +export interface AgentDirectory { + getAgentName(agentId: string): string; +} + +/** + * Strip control characters (except newlines/tabs) and collapse excessive + * whitespace runs. Mirrors `message-normalize.ts#cleanText` to keep the + * transcript resilient against garbage from upstream senders. + */ +function cleanText(text: string): string { + // eslint-disable-next-line no-control-regex + return text.replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, "") + .replace(/[ \t]{4,}/g, " ") + .trim(); +} + +function truncate(text: string, maxLen: number): string { + if (text.length <= maxLen) return text; + return text.slice(0, maxLen - 1) + "…"; +} + +function formatChannelLabel(record: AgentMessageRecord): string | null { + if (record.channel === "direct") return null; + if (record.channelTargetId) return `${record.channel}:${record.channelTargetId}`; + return record.channel; +} + +function resolveSenderLabel( + record: AgentMessageRecord, + directory: AgentDirectory, +): string { + switch (record.senderKind) { + case "human": + // The board-reader virtual account is the only non-agent human sender + // surfaced in practice; render it as "Board" so agents know where + // strategic directives originate. + if (record.fromAgentId === "__board__") return "Board"; + return "User"; + case "system": + return "System"; + case "agent": + default: + return directory.getAgentName(record.fromAgentId); + } +} + +/** + * Projects a stored `AgentMessageRecord` into a pi `AgentMessage` + * (role: "user") suitable for appending to the recipient's session + * transcript. `timestamp` carries the message's `createdAt` in + * ms-since-epoch so transcript ordering aligns with wall-clock time. + */ +export function projectInboxMessageToPi( + record: AgentMessageRecord, + directory: AgentDirectory, +): AgentMessage { + const senderLabel = resolveSenderLabel(record, directory); + const channelLabel = formatChannelLabel(record); + const header = channelLabel + ? `[Message from ${senderLabel} · ${channelLabel}]` + : `[Message from ${senderLabel}]`; + + const subject = record.subject ? truncate(cleanText(record.subject), MAX_SUBJECT_LENGTH) : ""; + const body = truncate(cleanText(record.body), MAX_MESSAGE_BODY_LENGTH); + + const parts: string[] = [header]; + if (subject.length > 0) parts.push(subject); + if (body.length > 0) parts.push(body); + const text = parts.join("\n\n"); + + const timestamp = Date.parse(record.createdAt); + return { + role: "user", + content: [{ type: "text", text }], + timestamp: Number.isFinite(timestamp) ? timestamp : Date.now(), + } as AgentMessage; +} diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index d218c9c0..921230fe 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -6,6 +6,7 @@ import { SqliteSessionStore } from "./sqlite-session-store"; import { CostWriter } from "./cost-writer"; import type { CredentialResolver } from "@main/credentials"; import { getConnectorDefinition } from "@main/connectors"; +import { projectInboxMessageToPi } from "./inbox-projection"; /** * Projects a pi AgentEvent to a human-readable log line for the Console. @@ -168,6 +169,23 @@ export class RuntimeBridge { return store; } + /** + * Resolve a display name for an agent id used in inbox header lines. + * Falls back to the id when the lookup fails so the transcript remains + * unambiguous instead of silently rendering "undefined". + */ + private resolveAgentName(agentId: string): string { + try { + const name = this.deps.db.findAgentName(agentId); + if (name && name.length > 0) return name; + } catch (error) { + this.deps.logger.warn( + `[runtime-bridge] agent name lookup failed for ${agentId}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + return agentId; + } + /** Receives an `agent-event` envelope from the worker; emits to renderer. */ handleAgentEvent(envelope: { runId: string; @@ -303,7 +321,29 @@ export class RuntimeBridge { }); const definition = getConnectorDefinition(payload.connectorId); const resolved = await this.deps.credentialResolver.resolve(payload.connectorId); - const priorMessages = this.getOrCreateStore(payload.taskSessionId).loadMessages(); + + // Project any pending inbox messages into the session transcript + // *before* loading priorMessages — the recipient must observe them + // as plain user turns in its context, and the transcript row is + // the single source of memory (recentMessages prompt block is + // gone). Append fails closed: if persistence throws, the inbox + // row stays unread and we surface the error so the worker aborts. + const store = this.getOrCreateStore(payload.taskSessionId); + const inboxRecords = this.deps.db.listAgentInboxUnread(payload.companyId, payload.agentId); + for (const record of inboxRecords) { + const projected = projectInboxMessageToPi(record, { + getAgentName: (id) => this.resolveAgentName(id), + }); + store.appendMessage(projected); + } + if (inboxRecords.length > 0) { + this.deps.db.markAgentInboxDelivered( + payload.agentId, + inboxRecords.map((r) => r.id), + ); + } + + const priorMessages = store.loadMessages(); return { ok: true, result: { diff --git a/src/main/database-base.ts b/src/main/database-base.ts index 141aefca..1ca72841 100644 --- a/src/main/database-base.ts +++ b/src/main/database-base.ts @@ -120,6 +120,7 @@ export class AppDatabaseBase { this.messages = buildMessageMethods(getDb, ra, { getAgent: (id) => this.getAgent(id), addActivity: (input) => this.addActivity(input), + transaction: tx, }); this.runs = buildRunMethods(getDb, ra, tx, { logsDir: this.logsDir }); this.workspacesMod = buildWorkspaceMethods(getDb, ra, tx); diff --git a/src/main/database.ts b/src/main/database.ts index a28e1b2f..51a7f25a 100644 --- a/src/main/database.ts +++ b/src/main/database.ts @@ -19,6 +19,8 @@ export class AppDatabase extends AppDatabaseBase { listAgentMessages(companyId: string, opts?: { agentId?: string; messageId?: string; channel?: string; channelTargetId?: string | null; unreadOnly?: boolean; limit?: number }) { return this.messages.listAgentMessages(companyId, opts); } searchMessages(companyId: string, query: string, opts?: { channel?: string; channelTargetId?: string | null; limit?: number }) { return this.messages.searchMessages(companyId, query, opts); } markMessageReadForReader(messageId: string, companyId: string, readerAgentId?: string | null) { return this.messages.markMessageReadForReader(messageId, companyId, readerAgentId); } + listAgentInboxUnread(companyId: string, agentId: string) { return this.messages.listAgentInboxUnread(companyId, agentId); } + markAgentInboxDelivered(agentId: string, messageIds: string[]) { return this.messages.markAgentInboxDelivered(agentId, messageIds); } // ─── Social & Browser Actions ────────────────────────────────────────────── diff --git a/src/main/database/messages.ts b/src/main/database/messages.ts index affe4a4a..8a332d66 100644 --- a/src/main/database/messages.ts +++ b/src/main/database/messages.ts @@ -4,6 +4,7 @@ import type { AgentMessageRecord, AgentRecord } from "@shared/types"; import { BOARD_READER_ID, type ReadAll, + type Transaction, nowIso, } from "./helpers"; @@ -13,6 +14,7 @@ export function buildMessageMethods( deps: { getAgent: (id: string) => AgentRecord; addActivity: (input: { companyId: string; actor: string; action: string; entityType: string; entityId: string; detail: string }) => void; + transaction: Transaction; }, ) { return { @@ -253,5 +255,109 @@ export function buildMessageMethods( ) .run(readerAgentId ?? BOARD_READER_ID, ts, messageId, companyId); }, + + /** + * List unread inbox messages for an agent, chronologically oldest-first, + * for pi-native transcript injection. Uses the same channel-subscription + * visibility rules as listAgentMessages (direct/company/incident always, + * department when agent belongs, project when assigned/lead). + * + * "Unread" means: + * - Direct messages to this agent: `agent_messages.read_at IS NULL`. + * - Channel messages: no matching row in `agent_message_reads` for + * (message_id, agent_id). + * + * Messages where `from_agent_id = agentId` are excluded (agents don't + * inject their own sent messages back into their transcript). + */ + listAgentInboxUnread(companyId: string, agentId: string): AgentMessageRecord[] { + let department: string | null = null; + try { + const agent = deps.getAgent(agentId); + if (agent.companyId === companyId) { + department = agent.department; + } + } catch { + department = null; + } + + const relevantProjectIds = new Set(); + for (const row of readAll<{ projectId: string | null }>( + "select distinct project_id as projectId from tasks where company_id = ? and assignee_agent_id = ? and project_id is not null", + companyId, + agentId, + )) { + if (row.projectId) relevantProjectIds.add(row.projectId); + } + for (const row of readAll<{ id: string }>( + "select id from projects where company_id = ? and lead_agent_id = ?", + companyId, + agentId, + )) { + relevantProjectIds.add(row.id); + } + + const visibilityClauses: string[] = [ + "(channel = 'direct' and to_agent_id = ?)", + "channel in ('company','incident')", + ]; + const visibilityParams: SQLInputValue[] = [agentId]; + if (department) { + visibilityClauses.push("(channel = 'department' and channel_target_id = ?)"); + visibilityParams.push(department); + } + if (relevantProjectIds.size > 0) { + const placeholders = Array.from(relevantProjectIds).map(() => "?").join(","); + visibilityClauses.push(`(channel = 'project' and channel_target_id in (${placeholders}))`); + visibilityParams.push(...Array.from(relevantProjectIds)); + } + + const selectCols = + "select id, company_id as companyId, from_agent_id as fromAgentId, to_agent_id as toAgentId, sender_kind as senderKind, channel, channel_target_id as channelTargetId, subject, body, priority, read_at as readAt, parent_message_id as parentMessageId, attachments_json as attachmentsJson, created_at as createdAt from agent_messages"; + const sql = `${selectCols} where company_id = ? and from_agent_id != ? and (${visibilityClauses.join(" or ")}) order by created_at asc`; + const rows = readAll(sql, companyId, agentId, ...visibilityParams); + + const readIds = new Set( + readAll<{ messageId: string }>( + `select reads.message_id as messageId + from agent_message_reads reads + join agent_messages messages on messages.id = reads.message_id + where reads.agent_id = ? and messages.company_id = ?`, + agentId, + companyId, + ).map((row) => row.messageId), + ); + + return rows.filter((message) => { + if (message.channel === "direct" && message.toAgentId === agentId) { + return !message.readAt; + } + return !readIds.has(message.id); + }); + }, + + /** + * Transactionally mark a batch of messages as delivered to `agentId`. + * - Direct messages: sets agent_messages.read_at = now() if currently null. + * - All messages (direct + channel): insert into agent_message_reads with + * ON CONFLICT DO NOTHING so reprocessing is idempotent. + */ + markAgentInboxDelivered(agentId: string, messageIds: string[]): void { + if (messageIds.length === 0) return; + const ts = nowIso(); + deps.transaction(() => { + const db = getDb(); + const directUpdate = db.prepare( + "update agent_messages set read_at = ? where id = ? and to_agent_id = ? and read_at is null", + ); + const receiptInsert = db.prepare( + "insert into agent_message_reads (message_id, agent_id, read_at) values (?, ?, ?) on conflict(message_id, agent_id) do nothing", + ); + for (const id of messageIds) { + directUpdate.run(ts, id, agentId); + receiptInsert.run(id, agentId, ts); + } + }); + }, }; } diff --git a/src/main/message-normalize.ts b/src/main/message-normalize.ts index d79c6eea..5c6fd123 100644 --- a/src/main/message-normalize.ts +++ b/src/main/message-normalize.ts @@ -1,19 +1,18 @@ /** * Message Normalization Layer * - * Normalizes messages and comments before they are injected into agent prompts. + * Normalizes comments before they are injected into agent prompts. * Prevents malformed, empty, or excessively long content from degrading * model output quality. Inspired by Reference/src/utils/messages.ts. * * Applied at the prompt-builder boundary — the last point before content * enters the model's context window. + * + * Inter-agent messages no longer flow through a prompt block; they are + * projected directly into the recipient's pi session transcript. See + * `agent-runtime/inbox-projection.ts`. */ -import type { AgentMessageRecord } from "@shared/types"; - -/** Maximum body length for a single message in prompt context. */ -const MAX_MESSAGE_BODY_LENGTH = 800; - /** Maximum comment body length in prompt context. */ const MAX_COMMENT_BODY_LENGTH = 500; @@ -36,48 +35,6 @@ function truncate(text: string, maxLen: number): string { return text.slice(0, maxLen - 1) + "…"; } -/** - * Normalize agent messages before prompt injection. - * - Removes empty-body messages - * - Cleans control characters - * - Truncates overly long bodies - * - Deduplicates consecutive identical messages from the same sender - */ -export function normalizeMessagesForPrompt( - messages: AgentMessageRecord[], - limit: number = 15, -): AgentMessageRecord[] { - const result: AgentMessageRecord[] = []; - let prevKey = ""; - - for (const msg of messages) { - const cleaned = cleanText(msg.body); - if (cleaned.length === 0) continue; - - // Deduplicate consecutive identical messages from the same sender - const key = `${msg.fromAgentId}:${cleaned}`; - if (key === prevKey) continue; - prevKey = key; - - result.push({ - ...msg, - body: truncate(cleaned, MAX_MESSAGE_BODY_LENGTH), - subject: msg.subject ? truncate(cleanText(msg.subject), 120) : msg.subject, - }); - - if (result.length >= limit) break; - } - - return result; -} - -// --------------------------------------------------------------------------- -// Transcript Repair — for session handoff / resume scenarios -// Inspired by Reference/src/utils/messages.ts tool pairing logic. -// Applied when session state is restored, not at every message send. -// --------------------------------------------------------------------------- - - export interface CommentRecord { taskId: string; taskTitle: string; diff --git a/src/main/orchestration/turn-context.ts b/src/main/orchestration/turn-context.ts index 51a54917..336f6e91 100644 --- a/src/main/orchestration/turn-context.ts +++ b/src/main/orchestration/turn-context.ts @@ -11,7 +11,7 @@ import type { AppDatabase } from "../database"; import type { ProjectedSkill } from "../skill-types"; import type { ConnectorId, ExecutionMode } from "@shared/types"; import type { WorkspaceLockMode, EnqueueRunPayload } from "./types"; -import { normalizeMessagesForPrompt, normalizeCommentsForPrompt } from "../message-normalize"; +import { normalizeCommentsForPrompt } from "../message-normalize"; import { getConnectorModelPolicy, validateConnectorModelSelection } from "../connector-model-policy"; export interface RunDispatchParams { @@ -291,7 +291,6 @@ export function buildRunPayload(params: RunDispatchParams, deps: TurnContextDeps status: account.status, requireApproval: account.requireApproval, })), - recentMessages: normalizeMessagesForPrompt(db.listAgentMessages(companyId, { agentId: agent.id, limit: 20 })), upcomingMeetings: filterUpcomingMeetings(snapshot.meetings, companyId, agent.id), relevantKnowledge: filterRelevantKnowledge(snapshot.knowledgeBase, companyId), activeSprint: snapshot.sprints.find((sprint) => sprint.companyId === companyId && sprint.status === "active") ?? null, diff --git a/src/main/prompt-builder/attachment-resolver.ts b/src/main/prompt-builder/attachment-resolver.ts deleted file mode 100644 index 673fdb47..00000000 --- a/src/main/prompt-builder/attachment-resolver.ts +++ /dev/null @@ -1,65 +0,0 @@ -/** - * Attachment Content Resolver - * - * Resolves document content for message attachments so agents can actually - * read attached documents instead of only seeing titles. - * - * Previously, attachments were rendered as [Attached: title] with no content. - * Now the full document text is included, truncated within a per-message budget. - */ - -import type { DocumentRecord } from "@shared/types"; - -interface AttachmentEntry { - type?: string; - id?: string; - title?: string; -} - -/** - * Resolve attachment content from document records. - * Returns a formatted block of document content for prompt injection, - * with each document truncated proportionally within the total budget. - */ -export function resolveAttachmentContent( - attachmentsJson: string, - documents: DocumentRecord[], - maxTotalChars: number = 8000, -): string { - let entries: AttachmentEntry[]; - try { - const parsed = JSON.parse(attachmentsJson); - if (!Array.isArray(parsed) || parsed.length === 0) return ""; - entries = parsed; - } catch { - return ""; - } - - const docEntries = entries.filter( - (e): e is typeof e & { id: string } => e.type === "document" && typeof e.id === "string", - ); - if (docEntries.length === 0) return ""; - - const docMap = new Map(documents.map((d) => [d.id, d])); - const resolved: Array<{ title: string; content: string }> = []; - - for (const entry of docEntries) { - const doc = docMap.get(entry.id); - if (!doc || !doc.content) continue; - resolved.push({ title: doc.title || entry.title || "Untitled", content: doc.content }); - } - - if (resolved.length === 0) return ""; - - const budgetPerDoc = Math.floor(maxTotalChars / resolved.length); - const parts: string[] = []; - - for (const { title, content } of resolved) { - const truncated = content.length > budgetPerDoc - ? content.slice(0, budgetPerDoc) + "..." - : content; - parts.push(`--- ${title} ---\n${truncated}\n---`); - } - - return parts.join("\n\n"); -} diff --git a/src/main/prompt-builder/context-injection.ts b/src/main/prompt-builder/context-injection.ts index 8ce5bd68..03127ca5 100644 --- a/src/main/prompt-builder/context-injection.ts +++ b/src/main/prompt-builder/context-injection.ts @@ -1,7 +1,6 @@ import log from "electron-log/main.js"; -import type { AgentRecord, AgentMessageRecord, ApprovalRecord, DocumentRecord, KnowledgeEntryRecord, MeetingRecord, TaskRecord } from "@shared/types"; +import type { AgentRecord, ApprovalRecord, DocumentRecord, KnowledgeEntryRecord, MeetingRecord, TaskRecord } from "@shared/types"; import { inferTaskType } from "../connector-matching"; -import { resolveAttachmentContent } from "./attachment-resolver"; interface TaskSectionCtx { task: TaskRecord; @@ -89,53 +88,6 @@ REVIEW WORKFLOW — This is your HIGHEST PRIORITY after resolving blockers: `; } -interface MessagesSectionCtx { - recentMessages: AgentMessageRecord[]; - allAgents: AgentRecord[]; - companyDocuments?: DocumentRecord[]; -} - -export function buildMessagesSection(ctx: MessagesSectionCtx): string | null { - if (ctx.recentMessages.length === 0) return null; - - const agentMap = new Map(ctx.allAgents.map((a) => [a.id, a.name])); - const msgLines = ctx.recentMessages.slice(0, 15).map((m) => { - const from = m.fromAgentId === "__board__" ? "Board (__board__)" : (agentMap.get(m.fromAgentId) ?? m.fromAgentId); - const channelLabel = m.channel === "direct" - ? `DM with ${m.fromAgentId === "__board__" ? "Board (__board__)" : from}` - : `#${m.channel}${m.channelTargetId ? `-${m.channelTargetId}` : ""}`; - const isUnread = !m.readAt; - let attachmentNote = ""; - let attachmentContent = ""; - if (m.attachmentsJson && m.attachmentsJson !== "[]") { - try { - const atts = JSON.parse(m.attachmentsJson) as Array<{ title?: string }>; - if (Array.isArray(atts) && atts.length > 0) { - attachmentNote = ` [Attached: ${atts.map((a) => a.title ?? "document").join(", ")}]`; - } - } catch (error) { - log.warn( - `[prompt-builder] malformed attachmentsJson on message ${m.id}: ${error instanceof Error ? error.message : String(error)}`, - ); - } - if (ctx.companyDocuments && ctx.companyDocuments.length > 0) { - attachmentContent = resolveAttachmentContent(m.attachmentsJson, ctx.companyDocuments, 4000); - } - } - const responseHint = m.channel === "direct" - ? ` Reply with channel=direct, toAgentId="${m.fromAgentId}", parentMessageId="${m.id}".` - : ` Reply in channel "${m.channel}"${m.channelTargetId ? ` with channelTargetId="${m.channelTargetId}"` : ""}, parentMessageId="${m.id}".` - const line = `- ${isUnread ? "[UNREAD] " : ""}[ID:${m.id}] [${channelLabel}] ${from}: "${m.subject}" — ${m.body.slice(0, 120)}${m.body.length > 120 ? "..." : ""}${attachmentNote}${responseHint}`; - return attachmentContent ? `${line}\n${attachmentContent}` : line; - }); - const unreadCount = ctx.recentMessages.filter((m) => !m.readAt).length; - - return `# Your Messages${unreadCount > 0 ? ` (${unreadCount} unread)` : ""} -${msgLines.join("\n")} -${unreadCount > 0 ? "\n**ACTION REQUIRED**: Read and respond to unread messages before starting new work. Mark messages as read after processing: POST /api/companies/{companyId}/messages/{messageId}/read" : ""} -`; -} - interface MeetingsSectionCtx { upcomingMeetings: MeetingRecord[]; allAgents: AgentRecord[]; diff --git a/src/main/prompt-builder/index.ts b/src/main/prompt-builder/index.ts index 3d1f81aa..20d9adc6 100644 --- a/src/main/prompt-builder/index.ts +++ b/src/main/prompt-builder/index.ts @@ -1,4 +1,4 @@ -import type { AgentRecord, AgentMessageRecord, ApprovalRecord, DocumentRecord, ExecutionMode, GoalRecord, KnowledgeEntryRecord, MeetingRecord, ProjectRecord, SprintRecord, TaskRecord } from "@shared/types"; +import type { AgentRecord, ApprovalRecord, DocumentRecord, ExecutionMode, GoalRecord, KnowledgeEntryRecord, MeetingRecord, ProjectRecord, SprintRecord, TaskRecord } from "@shared/types"; import type { AppLocale } from "@shared/locale"; import type { ProjectedSkill } from "../skill-types"; import { buildSystemPrompt } from "./system-prompt"; @@ -28,7 +28,6 @@ export interface PromptContext { recentComments: Array<{ taskId: string; taskTitle: string; authorName: string; body: string; createdAt: string }>; companyDescription: string; socialAccounts: Array<{ id: string; platform: string; accountName: string; status: string; requireApproval: boolean }>; - recentMessages?: AgentMessageRecord[]; upcomingMeetings?: MeetingRecord[]; relevantKnowledge?: KnowledgeEntryRecord[]; activeSprint?: SprintRecord | null; diff --git a/src/main/prompt-builder/system-prompt.ts b/src/main/prompt-builder/system-prompt.ts index 94b63e11..dcdc2e1d 100644 --- a/src/main/prompt-builder/system-prompt.ts +++ b/src/main/prompt-builder/system-prompt.ts @@ -4,7 +4,6 @@ import { formatSkillsForPrompt } from "@mariozechner/pi-coding-agent"; import { buildTaskSection, buildReviewSection, - buildMessagesSection, buildMeetingsSection, buildKnowledgeSection, buildDocumentsSection, @@ -209,13 +208,6 @@ export function buildSystemPrompt(ctx: PromptContext): string { sections.push(m.recentCommentsSection(lines)); } - const msgsSection = buildMessagesSection({ - recentMessages: ctx.recentMessages ?? [], - allAgents: ctx.allAgents, - companyDocuments: ctx.companyDocuments ?? [], - }); - if (msgsSection) sections.push(msgsSection); - sections.push(buildDepartmentSection(ctx)); sections.push(buildOperatingInstructions(ctx, isCEO, isManager)); sections.push(buildDecisionAuthoritySection(ctx, isCEO, isManager));