Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,10 @@ describe("Slack Turn Steering", () => {
"succeeded",
"succeeded",
]);
// Each person whose mention got its own turn joins the conversation.
expect(conversation.participantSlackUserIds).toEqual(
expect.arrayContaining(["U0SAM", "U0RIO", "U0KAI"]),
);
});

test("when cross-actor steering is configured, another person's mention joins the running turn", async ({
Expand Down
5 changes: 5 additions & 0 deletions packages/junior-evals/src/fixture/agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ import {
readCallEvents,
readFailedCallRun,
readConversationDetail,
participantSlackUserIds,
slackCallReplies,
toHarnessRun,
VIEWER_HEADER,
Expand Down Expand Up @@ -138,6 +139,8 @@ export interface Conversation extends HarnessRun {
threadTitles: string[];
/** The title that the dashboard shows after the call. */
title: string;
/** Slack user ids of the participants that the dashboard lists after the call. */
participantSlackUserIds: string[];
/**
* The label of the connect prompt that the dashboard shows to the person
* after the call, when a turn waits for their authorization.
Expand Down Expand Up @@ -745,6 +748,7 @@ export async function createFixtureAgent(
reactions,
replies,
statuses,
participantSlackUserIds: participantSlackUserIds(detail),
threadTitles,
title: detail.displayTitle,
toolCalls: events.toolCalls,
Expand Down Expand Up @@ -855,6 +859,7 @@ export async function createFixtureAgent(
reactions: [],
replies: [],
statuses: [],
participantSlackUserIds: participantSlackUserIds(detail),
threadTitles: [],
title: detail.displayTitle,
toolCalls: [],
Expand Down
7 changes: 7 additions & 0 deletions packages/junior-evals/src/fixture/results.ts
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,13 @@ export async function readConversationDetail(
return conversationDetailReportSchema.parse(await response.json());
}

/** Slack user ids of the participants that the dashboard lists. */
export function participantSlackUserIds(detail: ConversationDetail): string[] {
return (detail.participants ?? []).flatMap((participant) =>
participant.slackUserId ? [participant.slackUserId] : [],
);
}

type CallTranscriptSource = {
conversationId: string;
viewerEmail: string;
Expand Down
27 changes: 20 additions & 7 deletions packages/junior/src/api/conversations/participants.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import { and, asc, eq, inArray, ne } from "drizzle-orm";
import { and, asc, eq, inArray, sql } from "drizzle-orm";
import {
identityAvatarUrl,
readUserAvatarUrls,
} from "@/chat/identities/avatar";
import type { JuniorDatabase } from "@/db/db";
import {
juniorConversationEvents,
juniorConversationParticipants,
juniorConversations,
juniorIdentities,
juniorUsers,
Expand Down Expand Up @@ -96,15 +97,27 @@ export async function readConversationParticipants(
eq(juniorIdentities.id, juniorConversationEvents.actorIdentityId),
)
.leftJoin(juniorUsers, eq(juniorUsers.id, juniorIdentities.userId))
.where(
.innerJoin(
juniorConversations,
eq(
juniorConversations.conversationId,
juniorConversationEvents.conversationId,
),
)
// Message authors show only when they joined by starting a Turn.
.innerJoin(
juniorConversationParticipants,
and(
inArray(juniorConversationEvents.conversationId, [
...conversationIds,
]),
// System Messages, such as events, carry the system Identity.
ne(juniorIdentities.kind, "system"),
eq(juniorConversationParticipants.userId, juniorIdentities.userId),
eq(
juniorConversationParticipants.rootConversationId,
sql`coalesce(${juniorConversations.rootConversationId}, ${juniorConversations.conversationId})`,
),
),
)
.where(
inArray(juniorConversationEvents.conversationId, [...conversationIds]),
)
.orderBy(
asc(juniorConversationEvents.conversationId),
asc(juniorConversationEvents.actorIdentityId),
Expand Down
49 changes: 42 additions & 7 deletions packages/junior/src/chat/conversations/sql/history.ts
Original file line number Diff line number Diff line change
Expand Up @@ -239,16 +239,16 @@ class SqlConversationEventStore implements ConversationEventStore {
actorIdentityId,
),
);
if (actorIdentityId) {
await recordConversationParticipant(this.executor, {
actorIdentityId,
}
await this.executor.db().insert(juniorConversationEvents).values(rows);
for (const event of pending) {
if (event.data.type === "turn_started") {
await this.recordTurnInputParticipants(
conversationId,
atMs: event.createdAtMs,
restoreArchive: eventUnarchivesConversation(event.data),
});
event.data.inputMessageIds,
);
}
}
await this.executor.db().insert(juniorConversationEvents).values(rows);
return rows.map((row) => ({
seq: row.seq,
historyVersion: row.historyVersion,
Expand All @@ -257,6 +257,41 @@ class SqlConversationEventStore implements ConversationEventStore {
);
}

/**
* Add the authors of Turn input messages to the conversation.
* Only people who start a Turn join it. Thread context, such as backfilled
* or skipped messages, keeps its author without membership.
*/
private async recordTurnInputParticipants(
conversationId: string,
inputMessageIds: readonly string[],
): Promise<void> {
const rows = await this.executor
.db()
.select()
.from(juniorConversationEvents)
.where(
and(
eq(juniorConversationEvents.conversationId, conversationId),
eq(juniorConversationEvents.type, "message"),
inArray(
sql<string>`${juniorConversationEvents.payload}->>'messageId'`,
[...inputMessageIds],
),
),
);
for (const row of rows) {
const { data } = eventFromRow(row);
if (!row.actorIdentityId || data.type !== "message") continue;
await recordConversationParticipant(this.executor, {
actorIdentityId: row.actorIdentityId,
conversationId,
atMs: row.createdAt.getTime(),
restoreArchive: eventUnarchivesConversation(data),
});
}
}

async replaceHistory(
conversationId: string,
replacement: HistoryReplacement,
Expand Down
4 changes: 2 additions & 2 deletions packages/junior/src/db/schema/conversation-participants.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ import { timestamptz } from "./timestamps";

/**
* Materialized personal-feed membership for one linked user on one root
* conversation. Rows come from the root actor and from durable human user
* message authors.
* conversation. Rows come from the root actor and from the authors of Turn
* input messages. Thread context authors do not get a row.
*/
export const juniorConversationParticipants = pgTable(
"junior_conversation_participants",
Expand Down
46 changes: 31 additions & 15 deletions packages/junior/tests/integration/api/conversations/list.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -938,21 +938,37 @@ describe("conversation list API", () => {
.limit(1);
expect(event?.actorIdentityId).toBe("identity-dashboard-participant");

const participantRows = await fixture.sql
.db()
.select({
userId: juniorConversationParticipants.userId,
})
.from(juniorConversationParticipants)
.where(
eq(
juniorConversationParticipants.rootConversationId,
"slack:C1:dashboard-author",
),
);
expect(participantRows.map((row) => row.userId)).toContain(
participantUserId,
);
const readParticipantUserIds = async () =>
(
await fixture.sql
.db()
.select({
userId: juniorConversationParticipants.userId,
})
.from(juniorConversationParticipants)
.where(
eq(
juniorConversationParticipants.rootConversationId,
"slack:C1:dashboard-author",
),
)
).map((row) => row.userId);
// A message that does not start a Turn is context, not membership.
expect(await readParticipantUserIds()).not.toContain(participantUserId);

await eventStore.append("slack:C1:dashboard-author", [
{
createdAtMs: 5_600,
data: {
inputMessageIds: ["dashboard-msg-1"],
surface: "api",
turnId: "turn-dashboard-msg-1",
type: "turn_started",
},
idempotencyKey: "turn:turn-dashboard-msg-1:started",
},
]);
expect(await readParticipantUserIds()).toContain(participantUserId);

const feed = await readConversationFeedFromSql({
viewer: {
Expand Down
Loading