refactor(messaging): project inter-agent messages into pi session transcript

Inter-agent messages previously entered the recipient's context as a
separate prompt block (recentMessages), rebuilt every turn. This double-
charged tokens and kept messages outside pi's native AgentMessage stream.

Now at prepareRun: pull the recipient's unread inbox, project each
AgentMessageRecord into a pi user-role AgentMessage, persist to
SqliteSessionStore, and mark delivered. The recipient reads messages
as natural conversation turns in the transcript.

- db: listAgentInboxUnread + markAgentInboxDelivered (transactional)
- agent-runtime/inbox-projection.ts: AgentMessageRecord -> pi AgentMessage
- runtime-bridge.prepareRun: inject + deliver before loadMessages
- prompt-builder: delete recentMessages context block + buildMessagesSection
- message-normalize: drop unused normalizeMessagesForPrompt
- delete prompt-builder/attachment-resolver.ts (orphan after section removal)

Send path (api-server/routes-messages, ipc, UI) unchanged — inbox table
remains the UI source. Transactional delivery marking prevents duplicate
injection across runs. Failed append -> no mark -> retry on next run.
This commit is contained in:
ZacharyZhang-NY committed 2026-04-17 14:48:22 -04:00
1 parent 5547c29491
commit b50716d4e8
11 files changed
+254 -175

No files matched your search

@@ -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;
}
+41 -1
View File
@@ -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: {
+1
View File
@@ -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);
+2
View File
@@ -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 ──────────────────────────────────────────────
+106
View File
@@ -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<string>();
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<AgentMessageRecord>(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);
}
});
},
};
}
+5 -48
View File
@@ -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;
+1 -2
View File
@@ -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,
@@ -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");
}
+1 -49
View File
@@ -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[];
+1 -2
View File
@@ -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;
-8
View File
@@ -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));