feat: add relay message ID support for improved message tracking

- Introduced `relayOfMessageId` to various message handling components, allowing for better tracking of relayed messages.
- Updated `buildMessage`, `TeamInboxReader`, `TeamInboxWriter`, and related services to accommodate the new relay ID.
- Enhanced filtering logic in `filterTeamMessages` to hide relay copies when the original message is visible, improving message clarity.
- Added tests to validate the functionality of relay message handling and ensure proper integration across services.
This commit is contained in:
iliya 2026-03-15 18:10:37 +02:00
parent 194bd1bf1e
commit 0c0088c871
15 changed files with 339 additions and 10 deletions

View file

@ -91,6 +91,9 @@ function buildMessage(flags, defaults) {
...(typeof flags.summary === 'string' && flags.summary.trim()
? { summary: flags.summary.trim() }
: {}),
...(typeof flags.relayOfMessageId === 'string' && flags.relayOfMessageId.trim()
? { relayOfMessageId: flags.relayOfMessageId.trim() }
: {}),
...(typeof flags.source === 'string' && flags.source.trim() ? { source: flags.source.trim() } : {}),
...(typeof flags.leadSessionId === 'string' && flags.leadSessionId.trim()
? { leadSessionId: flags.leadSessionId.trim() }

View file

@ -539,6 +539,7 @@ describe('agent-teams-controller API', () => {
from: 'team-lead',
text: 'Need your review',
summary: 'Review request',
relayOfMessageId: 'm-original-1',
source: 'system_notification',
leadSessionId: 'session-42',
attachments: [{ id: 'a1', filename: 'note.txt', mimeType: 'text/plain', size: 7 }],
@ -551,6 +552,7 @@ describe('agent-teams-controller API', () => {
const rows = JSON.parse(fs.readFileSync(inboxPath, 'utf8'));
expect(rows).toHaveLength(1);
expect(rows[0].source).toBe('system_notification');
expect(rows[0].relayOfMessageId).toBe('m-original-1');
expect(rows[0].leadSessionId).toBe('session-42');
expect(rows[0].attachments[0].filename).toBe('note.txt');
});

View file

@ -102,6 +102,8 @@ export class TeamInboxReader {
summary: typeof row.summary === 'string' ? row.summary : undefined,
color: typeof row.color === 'string' ? row.color : undefined,
messageId: row.messageId,
relayOfMessageId:
typeof row.relayOfMessageId === 'string' ? row.relayOfMessageId : undefined,
source: typeof row.source === 'string' ? (row.source as InboxMessage['source']) : undefined,
leadSessionId: typeof row.leadSessionId === 'string' ? row.leadSessionId : undefined,
conversationId: typeof row.conversationId === 'string' ? row.conversationId : undefined,

View file

@ -30,6 +30,7 @@ export class TeamInboxWriter {
taskRefs: request.taskRefs?.length ? request.taskRefs : undefined,
summary: request.summary,
messageId,
...(request.relayOfMessageId && { relayOfMessageId: request.relayOfMessageId }),
attachments: attachmentMeta?.length ? attachmentMeta : undefined,
...(request.source && { source: request.source }),
...(request.leadSessionId && { leadSessionId: request.leadSessionId }),

View file

@ -250,9 +250,15 @@ interface ProvisioningRun {
* When set, the current stdin-injected turn is an internal "forward user DM to teammate"
* request triggered by the UI. We suppress any leaduser echo for that turn.
*/
silentUserDmForward: { target: string; startedAt: string } | null;
silentUserDmForward: {
target: string;
startedAt: string;
mode: 'user_dm' | 'member_inbox_relay';
} | null;
/** Safety valve: clears silentUserDmForward if turn never completes. */
silentUserDmForwardClearHandle: NodeJS.Timeout | null;
/** Exact inbox rows currently being bridged into the live teammate process. */
pendingInboxRelayCandidates: PendingInboxRelayCandidate[];
/** Accumulates assistant text during provisioning phase for live UI preview. */
provisioningOutputParts: string[];
/** Session ID detected from stream-json output (result.session_id or message.session_id). */
@ -1343,9 +1349,18 @@ function isTransientProbeWarning(warning: string): boolean {
);
}
interface PendingInboxRelayCandidate {
recipient: string;
sourceMessageId: string;
normalizedText: string;
normalizedSummary: string;
queuedAtMs: number;
}
export class TeamProvisioningService {
private static readonly CLAUDE_LOG_LINES_LIMIT = 50_000;
private static readonly RECENT_CROSS_TEAM_DELIVERY_TTL_MS = 10 * 60 * 1000;
private static readonly PENDING_INBOX_RELAY_TTL_MS = 2 * 60 * 1000;
private readonly runs = new Map<string, ProvisioningRun>();
private readonly provisioningRunByTeam = new Map<string, string>();
@ -1904,6 +1919,7 @@ export class TeamProvisioningService {
timestamp: message.timestamp,
summary: message.summary,
messageId: message.messageId,
relayOfMessageId: message.relayOfMessageId,
source: message.source,
leadSessionId: message.leadSessionId,
conversationId: message.conversationId,
@ -1931,6 +1947,7 @@ export class TeamProvisioningService {
timestamp: message.timestamp,
summary: message.summary,
messageId: message.messageId,
relayOfMessageId: message.relayOfMessageId,
source: message.source,
leadSessionId: message.leadSessionId,
conversationId: message.conversationId,
@ -1950,8 +1967,100 @@ export class TeamProvisioningService {
return `${teamName}:${memberName.trim()}`;
}
private armSilentTeammateForward(run: ProvisioningRun, teammateName: string): void {
run.silentUserDmForward = { target: teammateName, startedAt: nowIso() };
private normalizeRelayCandidateText(text: string): string {
return stripAgentBlocks(String(text)).trim().replace(/\r\n/g, '\n');
}
private normalizeRelayCandidateSummary(summary?: string): string {
return typeof summary === 'string' ? summary.trim() : '';
}
private prunePendingInboxRelayCandidates(run: ProvisioningRun): PendingInboxRelayCandidate[] {
const cutoff = Date.now() - TeamProvisioningService.PENDING_INBOX_RELAY_TTL_MS;
run.pendingInboxRelayCandidates = (run.pendingInboxRelayCandidates ?? []).filter(
(candidate) => candidate.queuedAtMs >= cutoff
);
return run.pendingInboxRelayCandidates;
}
private rememberPendingInboxRelayCandidates(
run: ProvisioningRun,
recipient: string,
messages: Array<Pick<InboxMessage, 'messageId' | 'text' | 'summary'>>
): string[] {
const candidates = this.prunePendingInboxRelayCandidates(run);
const queuedAtMs = Date.now();
const rememberedIds: string[] = [];
for (const message of messages) {
const sourceMessageId = typeof message.messageId === 'string' ? message.messageId.trim() : '';
const normalizedText = this.normalizeRelayCandidateText(message.text);
if (!sourceMessageId || !normalizedText) {
continue;
}
candidates.push({
recipient,
sourceMessageId,
normalizedText,
normalizedSummary: this.normalizeRelayCandidateSummary(message.summary),
queuedAtMs,
});
rememberedIds.push(sourceMessageId);
}
return rememberedIds;
}
private forgetPendingInboxRelayCandidates(
run: ProvisioningRun,
recipient: string,
sourceMessageIds: readonly string[]
): void {
if (sourceMessageIds.length === 0) {
return;
}
const idSet = new Set(sourceMessageIds);
run.pendingInboxRelayCandidates = this.prunePendingInboxRelayCandidates(run).filter(
(candidate) => !(candidate.recipient === recipient && idSet.has(candidate.sourceMessageId))
);
}
private consumePendingInboxRelayCandidate(
run: ProvisioningRun,
recipient: string,
text: string,
summary?: string
): string | undefined {
const normalizedText = this.normalizeRelayCandidateText(text);
if (!normalizedText) {
return undefined;
}
const normalizedSummary = this.normalizeRelayCandidateSummary(summary);
const candidates = this.prunePendingInboxRelayCandidates(run);
const exactSummaryIdx = candidates.findIndex(
(candidate) =>
candidate.recipient === recipient &&
candidate.normalizedText === normalizedText &&
candidate.normalizedSummary === normalizedSummary
);
const fallbackIdx =
exactSummaryIdx >= 0
? exactSummaryIdx
: candidates.findIndex(
(candidate) =>
candidate.recipient === recipient && candidate.normalizedText === normalizedText
);
if (fallbackIdx < 0) {
return undefined;
}
const [matched] = candidates.splice(fallbackIdx, 1);
return matched?.sourceMessageId;
}
private armSilentTeammateForward(
run: ProvisioningRun,
teammateName: string,
mode: 'user_dm' | 'member_inbox_relay'
): void {
run.silentUserDmForward = { target: teammateName, startedAt: nowIso(), mode };
if (run.silentUserDmForwardClearHandle) {
clearTimeout(run.silentUserDmForwardClearHandle);
run.silentUserDmForwardClearHandle = null;
@ -2732,6 +2841,7 @@ export class TeamProvisioningService {
lastLeadTextEmitMs: 0,
silentUserDmForward: null,
silentUserDmForwardClearHandle: null,
pendingInboxRelayCandidates: [],
provisioningOutputParts: [],
detectedSessionId: null,
leadActivityState: 'active',
@ -3095,6 +3205,7 @@ export class TeamProvisioningService {
lastLeadTextEmitMs: 0,
silentUserDmForward: null,
silentUserDmForwardClearHandle: null,
pendingInboxRelayCandidates: [],
provisioningOutputParts: [],
detectedSessionId: null,
leadActivityState: 'active',
@ -3388,7 +3499,7 @@ export class TeamProvisioningService {
return;
}
this.armSilentTeammateForward(run, teammateName);
this.armSilentTeammateForward(run, teammateName, 'user_dm');
const summaryLine = userSummary?.trim() ? `Summary: ${userSummary.trim()}` : null;
const internal = wrapInAgentBlock(
@ -3466,7 +3577,8 @@ export class TeamProvisioningService {
const MAX_RELAY = 10;
const batch = actionableUnread.slice(0, MAX_RELAY);
this.armSilentTeammateForward(run, memberName);
this.armSilentTeammateForward(run, memberName, 'member_inbox_relay');
const rememberedRelayIds = this.rememberPendingInboxRelayCandidates(run, memberName, batch);
const message = [
`Relay inbox messages to teammate "${memberName}".`,
@ -3517,6 +3629,7 @@ export class TeamProvisioningService {
try {
await this.sendMessageToTeam(teamName, message);
} catch {
this.forgetPendingInboxRelayCandidates(run, memberName, rememberedRelayIds);
return 0;
}
@ -4213,6 +4326,16 @@ export class TeamProvisioningService {
continue;
}
const relayOfMessageId =
recipient !== 'user'
? this.consumePendingInboxRelayCandidate(
run,
recipient,
strippedCrossTeamContent,
summary
)
: undefined;
const msg: InboxMessage = {
from: leadName,
to: recipient,
@ -4224,6 +4347,7 @@ export class TeamProvisioningService {
? (summary || strippedCrossTeamContent).slice(0, 57) + '...'
: summary || strippedCrossTeamContent,
messageId: `lead-sendmsg-${run.runId}-${Date.now()}`,
...(relayOfMessageId ? { relayOfMessageId } : {}),
source: 'lead_process',
};
@ -4521,7 +4645,7 @@ export class TeamProvisioningService {
// Capture SendMessage tool_use blocks from assistant output.
// Works in both pre-ready and post-ready phases so outbound runtime messages
// are visible in our team message artifacts even if Claude's own routing drifts.
if (!run.silentUserDmForward) {
if (!run.silentUserDmForward || run.silentUserDmForward.mode === 'member_inbox_relay') {
this.captureSendMessages(run, content ?? []);
}
@ -4683,6 +4807,7 @@ export class TeamProvisioningService {
}
// Clear silent relay flag after any successful turn.
run.activeCrossTeamReplyHints = [];
run.pendingInboxRelayCandidates = [];
run.silentUserDmForward = null;
if (run.silentUserDmForwardClearHandle) {
clearTimeout(run.silentUserDmForwardClearHandle);
@ -4719,6 +4844,7 @@ export class TeamProvisioningService {
// Clear silent relay flag after any errored turn.
run.pendingDirectCrossTeamSendRefresh = false;
run.activeCrossTeamReplyHints = [];
run.pendingInboxRelayCandidates = [];
run.silentUserDmForward = null;
if (run.silentUserDmForwardClearHandle) {
clearTimeout(run.silentUserDmForwardClearHandle);
@ -5463,6 +5589,7 @@ export class TeamProvisioningService {
this.pendingCrossTeamFirstReplies.delete(run.teamName);
this.recentCrossTeamLeadDeliveryMessageIds.delete(run.teamName);
run.activeCrossTeamReplyHints = [];
run.pendingInboxRelayCandidates = [];
for (const key of Array.from(this.memberInboxRelayInFlight.keys())) {
if (key.startsWith(`${run.teamName}:`)) {
this.memberInboxRelayInFlight.delete(key);

View file

@ -75,6 +75,8 @@ export class TeamSentMessagesStore {
taskRefs: Array.isArray(row.taskRefs) ? row.taskRefs : undefined,
summary: typeof row.summary === 'string' ? row.summary : undefined,
messageId: row.messageId,
relayOfMessageId:
typeof row.relayOfMessageId === 'string' ? row.relayOfMessageId : undefined,
color: typeof row.color === 'string' ? row.color : undefined,
attachments: Array.isArray(row.attachments) ? row.attachments : undefined,
source: typeof row.source === 'string' ? (row.source as InboxMessage['source']) : undefined,

View file

@ -1,3 +1,7 @@
import { useMemo } from 'react';
import { filterTeamMessages } from '@renderer/utils/teamMessageFiltering';
import { ActivityItem } from '../activity/ActivityItem';
import type { InboxMessage } from '@shared/types';
@ -15,7 +19,15 @@ export const MemberMessagesTab = ({
teamName,
onCreateTask,
}: MemberMessagesTabProps): React.JSX.Element => {
const displayMessages = messages.slice(0, MAX_MESSAGES);
const displayMessages = useMemo(
() =>
filterTeamMessages(messages, {
timeWindow: null,
filter: { from: new Set(), to: new Set(), showNoise: true },
searchQuery: '',
}).slice(0, MAX_MESSAGES),
[messages]
);
if (displayMessages.length === 0) {
return (

View file

@ -103,6 +103,7 @@ export function areInboxMessagesEquivalentForRender(
if (prev.summary !== next.summary) return false;
if (prev.color !== next.color) return false;
if (prev.read !== next.read) return false;
if (prev.relayOfMessageId !== next.relayOfMessageId) return false;
if (prev.source !== next.source) return false;
if (prev.leadSessionId !== next.leadSessionId) return false;
if (prev.toolSummary !== next.toolSummary) return false;

View file

@ -56,5 +56,22 @@ export function filterTeamMessages(
});
}
return list;
const visibleMessageIds = new Set(
list
.map((m) => (typeof m.messageId === 'string' ? m.messageId.trim() : ''))
.filter((id) => id.length > 0)
);
return list.filter((m) => {
const relayOfMessageId =
typeof m.relayOfMessageId === 'string' ? m.relayOfMessageId.trim() : '';
if (!relayOfMessageId) {
return true;
}
const ownMessageId = typeof m.messageId === 'string' ? m.messageId.trim() : '';
if (relayOfMessageId === ownMessageId) {
return true;
}
return !visibleMessageIds.has(relayOfMessageId);
});
}

View file

@ -258,6 +258,8 @@ export interface InboxMessage {
summary?: string;
color?: string;
messageId?: string;
/** Original inbox messageId when this row is only a relay/delivery bridge copy. */
relayOfMessageId?: string;
source?:
| 'inbox'
| 'lead_session'
@ -290,6 +292,7 @@ export interface SendMessageRequest {
from?: string;
timestamp?: string;
messageId?: string;
relayOfMessageId?: string;
/** Override the `to` field in the stored message (defaults to `member`). */
to?: string;
color?: string;

View file

@ -165,6 +165,7 @@ describe('TeamInboxWriter', () => {
text: 'Hello cross-team',
summary: 'Cross-team response',
messageId: 'lead-sendmsg-run-1-123',
relayOfMessageId: 'msg-original-1',
timestamp: '2026-03-10T00:33:55.000Z',
source: 'lead_process',
color: 'purple',
@ -180,6 +181,7 @@ describe('TeamInboxWriter', () => {
text: 'Hello cross-team',
summary: 'Cross-team response',
messageId: 'lead-sendmsg-run-1-123',
relayOfMessageId: 'msg-original-1',
timestamp: '2026-03-10T00:33:55.000Z',
source: 'lead_process',
color: 'purple',

View file

@ -131,7 +131,9 @@ interface RunLike {
pendingDirectCrossTeamSendRefresh: boolean;
lastLeadTextEmitMs: number;
leadRelayCapture: null;
silentUserDmForward: null;
silentUserDmForward:
| null
| { target: string; startedAt: string; mode: 'user_dm' | 'member_inbox_relay' };
suppressPostCompactReminderOutput?: boolean;
child: Record<string, unknown> | null;
processKilled: boolean;
@ -139,6 +141,7 @@ interface RunLike {
provisioningOutputParts: string[];
request: { members: { name: string; role?: string }[] };
activeCrossTeamReplyHints?: Array<{ toTeam: string; conversationId: string }>;
pendingInboxRelayCandidates?: unknown[];
}
/**
@ -161,6 +164,7 @@ function attachRun(
lastLeadTextEmitMs: 0,
leadRelayCapture: null,
silentUserDmForward: null,
pendingInboxRelayCandidates: [],
child: { stdin: { writable: true, write: vi.fn(), end: vi.fn() } },
processKilled: false,
cancelRequested: false,

View file

@ -221,7 +221,11 @@ describe('TeamProvisioningService post-compact lifecycle', () => {
it('injectPostCompactReminder defers when silentUserDmForward is active', async () => {
const { svc, run, runId } = await setupRunningTeam('compact-test-6');
run.pendingPostCompactReminder = true;
run.silentUserDmForward = { target: 'alice', startedAt: new Date().toISOString() };
run.silentUserDmForward = {
target: 'alice',
startedAt: new Date().toISOString(),
mode: 'user_dm',
};
await (svc as any).injectPostCompactReminder(run);

View file

@ -162,6 +162,18 @@ function attachAliveRun(
(service as unknown as { runs: Map<string, unknown> }).runs.set(runId, {
runId,
teamName,
request: {
teamName,
members: [{ name: 'team-lead', role: 'team-lead' }],
},
leadMsgSeq: 0,
pendingToolCalls: [],
pendingDirectCrossTeamSendRefresh: false,
lastLeadTextEmitMs: 0,
activeCrossTeamReplyHints: [],
pendingInboxRelayCandidates: [],
silentUserDmForward: null,
silentUserDmForwardClearHandle: null,
child: {
stdin: {
writable,
@ -198,6 +210,7 @@ describe('TeamProvisioningService relayLeadInboxMessages', () => {
hoisted.readFile.mockClear();
hoisted.atomicWrite.mockClear();
hoisted.appendSentMessage.mockClear();
hoisted.sendInboxMessage.mockClear();
hoisted.setAtomicWriteShouldFail(false);
});
@ -610,6 +623,94 @@ describe('TeamProvisioningService relayLeadInboxMessages', () => {
expect(payload).toContain('Please retry with logging enabled.');
});
it('marks exact teammate relay copies with relayOfMessageId', async () => {
const service = new TeamProvisioningService();
const teamName = 'my-team';
seedConfig(teamName);
seedMemberInbox(teamName, 'alice', [
{
from: 'team-lead',
text:
`**Comment on task #abcd1234**\n> Investigate\n\n> Please retry with logging enabled.\n\n` +
'<agent-block>\nReply using task_add_comment\n</agent-block>',
timestamp: '2026-02-23T10:00:00.000Z',
read: false,
summary: 'Comment on #abcd1234',
messageId: 'm-alice-1',
source: 'system_notification',
},
]);
attachAliveRun(service, teamName);
const relayed = await service.relayMemberInboxMessages(teamName, 'alice');
expect(relayed).toBe(1);
const run = (service as unknown as { runs: Map<string, unknown> }).runs.get('run-1') as unknown;
expect(run).toBeTruthy();
(service as any).handleStreamJsonMessage(run, {
type: 'assistant',
content: [
{
type: 'tool_use',
name: 'SendMessage',
input: {
recipient: 'alice',
summary: 'Comment on #abcd1234',
content:
`**Comment on task #abcd1234**\n> Investigate\n\n> Please retry with logging enabled.\n\n` +
'<agent-block>\nHidden internal instructions\n</agent-block>',
},
},
],
});
const inbox = JSON.parse(
hoisted.files.get(`/mock/teams/${teamName}/inboxes/alice.json`) ?? '[]'
) as Array<{ messageId?: string; relayOfMessageId?: string; source?: string }>;
const relayedCopy = inbox.find((row) => row.messageId?.startsWith('lead-sendmsg-run-1-'));
expect(relayedCopy).toMatchObject({
source: 'lead_process',
relayOfMessageId: 'm-alice-1',
});
});
it('does not capture user-dm silent forwards as extra lead_process messages', () => {
const service = new TeamProvisioningService();
const teamName = 'my-team';
seedConfig(teamName);
attachAliveRun(service, teamName);
const run = (service as unknown as { runs: Map<string, unknown> }).runs.get('run-1') as {
silentUserDmForward: { target: string; startedAt: string; mode: 'user_dm' | 'member_inbox_relay' } | null;
};
run.silentUserDmForward = {
target: 'alice',
startedAt: new Date().toISOString(),
mode: 'user_dm',
};
(service as any).handleStreamJsonMessage(run, {
type: 'assistant',
content: [
{
type: 'tool_use',
name: 'SendMessage',
input: {
recipient: 'alice',
summary: 'Forwarded DM',
content: 'User DM payload',
},
},
],
});
const inbox = JSON.parse(
hoisted.files.get(`/mock/teams/${teamName}/inboxes/alice.json`) ?? '[]'
) as Array<{ messageId?: string; source?: string }>;
expect(inbox).toHaveLength(0);
});
it('does not relay pseudo cross-team member inboxes as teammates', async () => {
const service = new TeamProvisioningService();
const teamName = 'my-team';

View file

@ -37,6 +37,54 @@ describe('filterTeamMessages', () => {
expect(result[0].source).toBe('lead_process');
});
it('hides relay bridge copies when the original message is visible', () => {
const messages = [
makeMessage({
messageId: 'orig-1',
to: 'alice',
source: 'system_notification',
text: 'Original inbox notification',
}),
makeMessage({
messageId: 'relay-1',
to: 'alice',
source: 'lead_process',
text: 'Original inbox notification',
relayOfMessageId: 'orig-1',
}),
];
const result = filterTeamMessages(messages, {
timeWindow: null,
filter: { from: new Set(), to: new Set(), showNoise: true },
searchQuery: '',
});
expect(result).toHaveLength(1);
expect(result[0].messageId).toBe('orig-1');
});
it('keeps relay bridge copies when the original message is not visible', () => {
const messages = [
makeMessage({
messageId: 'relay-1',
to: 'alice',
source: 'lead_process',
text: 'Original inbox notification',
relayOfMessageId: 'orig-1',
}),
];
const result = filterTeamMessages(messages, {
timeWindow: null,
filter: { from: new Set(), to: new Set(), showNoise: true },
searchQuery: '',
});
expect(result).toHaveLength(1);
expect(result[0].messageId).toBe('relay-1');
});
it('still filters noise messages when showNoise is false', () => {
const messages = [
makeMessage({