diff --git a/tests/runner-e2e/chat-catalog-descriptors.json b/tests/runner-e2e/chat-catalog-descriptors.json index 9e2ae1708ce..687cc12b500 100644 --- a/tests/runner-e2e/chat-catalog-descriptors.json +++ b/tests/runner-e2e/chat-catalog-descriptors.json @@ -3328,6 +3328,7 @@ "chat-catalog-domains/internal-inbox-search.ts", "chat-scenarios/inbox-search.yaml", "chat-catalog-domains/internal-notification-audience.ts", + "chat-catalog-domains/internal-notification-events.ts", "chat-catalog-domains/internal-privacy.ts", "chat-catalog-domains/identity-auth.ts", "../../packages/shared/src/project-mentions.ts" diff --git a/tests/runner-e2e/chat-catalog-domains/internal-notification-audience.ts b/tests/runner-e2e/chat-catalog-domains/internal-notification-audience.ts index 15315b44ad8..374e946d67d 100644 --- a/tests/runner-e2e/chat-catalog-domains/internal-notification-audience.ts +++ b/tests/runner-e2e/chat-catalog-domains/internal-notification-audience.ts @@ -1,3 +1,4 @@ +import { bindNotificationPublicEvents, publicNotificationFrame, type NotificationPublicFrame, type NotificationBoundFrame, type NotificationObserverInput } from "./internal-notification-events.js"; import { expect, type Page } from "@playwright/test"; import { extractUserMentionIds } from "../../../packages/shared/src/project-mentions.js"; import type { ChatCatalogCase } from "../chat-catalog-contract.js"; @@ -5,21 +6,21 @@ import { pollUntil } from "../api.js"; import { makeDefinition } from "./internal-definition.js"; import { mainPerson, observerPerson, personRead } from "./internal-privacy.js"; import { checks, entity, hasLink, observation, observeInternal, project, putDocument, sameTask, task, type Entity, type InternalObservation } from "./internal-support.js"; -interface Frame { principalId: string; commentId: string; taskId: string; eligible: boolean; at: number } +type Frame = NotificationBoundFrame; interface ToastReceipt { text: string; hrefs: string[]; at: number } interface ObserverEvidence { principalId: string; visibility: string; frames: Frame[]; toasts: ToastReceipt[]; sockets: string[] } interface AudienceObservation extends InternalObservation { alex: ObserverEvidence; riley: ObserverEvidence; observedAt: number } -interface ObserverProbe { page: Page; principalId: string; frames: Frame[]; sockets: string[] } +interface ObserverProbe { page: Page; principalId: string; companyId: string; taskId: string; origin: string; frames: NotificationPublicFrame[]; sockets: string[]; overflow: boolean; baseline?: { sessionPrincipalId: string; activityIds: string[]; commentIds: string[] } } const negativeWindowMs = 2000; declare global { interface Window { __paperclipAudienceToastReceipts?: ToastReceipt[] } } -async function startObserver(page: Page, principalId: string, target: Entity, href: string): Promise { - const probe: ObserverProbe = { page, principalId, frames: [], sockets: [] }; +async function startObserver(page: Page, principalId: string, target: Entity, href: string, companyId: string): Promise { + const probe: ObserverProbe = { page, principalId, companyId, taskId: target.id, origin: new URL(href).origin, frames: [], sockets: [], overflow: false }; page.on("websocket", socket => { const path = new URL(socket.url()).pathname; if (!/\/events\/ws$/.test(path)) return; probe.sockets.push(path); socket.on("framereceived", frame => { - try { const message = JSON.parse(String(frame.payload)); const payload = message.payload; - if (message.type === "activity.logged" && payload?.action === "issue.comment_added" && payload.entityId === target.id && typeof payload.details?.commentId === "string" && typeof message.toastEligible === "boolean") - probe.frames.push({ principalId, commentId: payload.details.commentId, taskId: target.id, eligible: message.toastEligible, at: Date.now() }); + try { const message = JSON.parse(String(frame.payload)); + const actual = publicNotificationFrame(message, companyId, target.id, Date.now()); + if (actual) { if (probe.frames.length >= 128) probe.overflow = true; else probe.frames.push(actual); } } catch { /* Non-JSON transport frames are not task/comment evidence. */ } }); }); @@ -37,9 +38,35 @@ async function startObserver(page: Page, principalId: string, target: Entity, hr }, { id: target.id, identifier: target.identifier }); return probe; } -async function observeProbe(probe: ObserverProbe): Promise { +async function publicProbeRows(probe: ObserverProbe) { + const read = async (route: string) => { + const response = await probe.page.request.get(new URL(route, probe.origin).href, { timeout: 15000, maxRedirects: 0 }); + try { if (response.status() !== 200) throw new Error("Notification observer lacks authenticated saved evidence"); return await response.json(); } + finally { await response.dispose(); } + }; + const [activity, comments] = await Promise.all([read(`/api/issues/${probe.taskId}/activity`), read(`/api/issues/${probe.taskId}/comments?order=asc`)]); + if (!Array.isArray(activity) || !Array.isArray(comments)) throw new Error("Notification observer saved evidence is not a list"); + return { activity: activity.map((row: Record) => ({ id: row.id, companyId: row.companyId, entityType: row.entityType, entityId: row.entityId, + action: row.action, actorType: row.actorType, actorId: row.actorId, createdAt: row.createdAt, details: { commentId: (row.details as Record | null)?.commentId } })), + comments: comments.map((row: Record) => ({ id: row.id, companyId: row.companyId, issueId: row.issueId, createdAt: row.createdAt, + authorAgentId: row.authorAgentId, authorUserId: row.authorUserId, deletedAt: row.deletedAt })) }; +} +async function armProbe(probe: ObserverProbe) { + const session = await probe.page.request.get(new URL("/api/auth/get-session", probe.origin).href, { timeout: 15000, maxRedirects: 0 }); + let personId: unknown; + try { if (session.status() !== 200) throw new Error("Notification observer has no actual session"); personId = (await session.json())?.user?.id; } + finally { await session.dispose(); } + if (personId !== probe.principalId) throw new Error("Notification observer session principal mismatch"); + const rows = await publicProbeRows(probe); + probe.baseline = { sessionPrincipalId: probe.principalId, activityIds: rows.activity.map(row => String(row.id)), commentIds: rows.comments.map(row => String(row.id)) }; + probe.frames.length = 0; +} +async function observeProbe(probe: ObserverProbe) { const browser = await probe.page.evaluate(() => ({ visibility: document.visibilityState, toasts: window.__paperclipAudienceToastReceipts ?? [] })); - return { principalId: probe.principalId, frames: structuredClone(probe.frames), sockets: [...probe.sockets], ...browser }; + const input: NotificationObserverInput | undefined = probe.baseline ? { principalId: probe.principalId, sessionPrincipalId: probe.baseline.sessionPrincipalId, + frames: structuredClone(probe.frames), overflow: probe.overflow, baselineActivityIds: probe.baseline.activityIds, baselineCommentIds: probe.baseline.commentIds, + ...await publicProbeRows(probe) } : undefined; + return { principalId: probe.principalId, sockets: [...probe.sockets], ...browser, input }; } function commentEvent(person: ObserverEvidence, commentId: string, taskId: string) { return person.frames.find(frame => frame.principalId === person.principalId && frame.commentId === commentId && frame.taskId === taskId); } function actualToast(person: ObserverEvidence, since: number, target: Entity, origin: string) { @@ -80,6 +107,14 @@ export function gradeNotificationAudience(value: unknown) { } function notificationAudience(): ChatCatalogCase { let riley: Awaited> | undefined; let alexProbe: ObserverProbe | undefined; let rileyProbe: ObserverProbe | undefined; + let latestBinding: ReturnType | undefined; + const observeAudienceProbes = async () => { + if (!alexProbe || !rileyProbe) throw new Error("Missing actual observer pages"); + const [alex, riley] = await Promise.all([observeProbe(alexProbe), observeProbe(rileyProbe)]); + latestBinding = alex.input && riley.input ? bindNotificationPublicEvents(alexProbe.companyId, alexProbe.taskId, [alex.input, riley.input]) : undefined; + return { alex: { principalId: alex.principalId, visibility: alex.visibility, sockets: alex.sockets, toasts: alex.toasts, frames: latestBinding?.frames[0] ?? [] }, + riley: { principalId: riley.principalId, visibility: riley.visibility, sockets: riley.sockets, toasts: riley.toasts, frames: latestBinding?.frames[1] ?? [] } }; + }; let observe: (ctx: Parameters[0], state: Record) => Promise; const base = makeDefinition("inbox-search.notification-audience", async ctx => { const alexId = await mainPerson(ctx); riley = await observerPerson(ctx); @@ -88,25 +123,26 @@ function notificationAudience(): ChatCatalogCase { await putDocument(ctx, t.id, "# Launch draft\nThe Acorn Notes draft is ready for human review.\n"); const read = await personRead(riley, `/api/issues/${t.id}`); if (read.status !== 200) throw new Error("The nonparticipant is not an actual authorized task reader"); const origin = ctx.input.api.baseURL; const href = `${origin}/${ctx.input.fixtures.company.issuePrefix}/activity`; - alexProbe = await startObserver(await ctx.input.page.context().newPage(), alexId, t, href); rileyProbe = await startObserver(await riley.context.newPage(), riley.id, t, href); + alexProbe = await startObserver(await ctx.input.page.context().newPage(), alexId, t, href, ctx.input.fixtures.company.id); rileyProbe = await startObserver(await riley.context.newPage(), riley.id, t, href, ctx.input.fixtures.company.id); await pollUntil({ label: "Observer page visibility and production socket registration", deadlineAt: Date.now() + 15000, - load: async () => ({ alex: await observeProbe(alexProbe!), riley: await observeProbe(rileyProbe!) }), + load: observeAudienceProbes, accept: proof => proof.alex.visibility === "visible" && proof.riley.visibility === "visible" && proof.alex.sockets.length > 0 && proof.riley.sockets.length > 0 }); + await Promise.all([armProbe(alexProbe), armProbe(rileyProbe)]); await rileyProbe.page.waitForTimeout(250); const participation = await ctx.input.api.post(`/api/issues/${t.id}/comments`, { body: "I am following the launch draft. Please post an update here when it is ready." }); await pollUntil({ label: "Both real signed-in task cache subscriptions", deadlineAt: Date.now() + 30000, - load: async () => ({ alex: await observeProbe(alexProbe!), riley: await observeProbe(rileyProbe!) }), + load: observeAudienceProbes, accept: proof => proof.alex.visibility === "visible" && proof.riley.visibility === "visible" && !!commentEvent(proof.alex, participation.id, t.id) && !!commentEvent(proof.riley, participation.id, t.id) }); return { project: p, task: t, alexId, rileyId: riley.id, participation, negativeWindowMs }; - } catch (error) { await alexProbe?.page.close(); await rileyProbe?.page.close(); await riley.dispose(); riley = undefined; throw error; } + } catch (error) { await ctx.input.evidence("chat-notification-public-event-binding.json", { binding: latestBinding ?? null }).catch(() => undefined); await alexProbe?.page.close(); await rileyProbe?.page.close(); await riley.dispose(); riley = undefined; throw error; } }, gradeNotificationAudience); observe = async (ctx, state) => { if (!alexProbe || !rileyProbe) throw new Error("Missing actual observer pages"); - const o = await observeInternal(ctx, state); const [alex, riley] = await Promise.all([observeProbe(alexProbe), observeProbe(rileyProbe)]); + const o = await observeInternal(ctx, state); const { alex, riley } = await observeAudienceProbes(); return { ...o, alex, riley, observedAt: Date.now() }; }; return { ...base, authentication: "human-session", expectedRunCount: 2, minimumRunCount: 2, expectedHumanMessageCount: 2, attemptTimeoutMs: 480000, - sourceFiles: [...base.sourceFiles, "chat-catalog-domains/internal-notification-audience.ts", "chat-catalog-domains/internal-privacy.ts", "chat-catalog-domains/identity-auth.ts", "../../packages/shared/src/project-mentions.ts"], observe, + sourceFiles: [...base.sourceFiles, "chat-catalog-domains/internal-notification-audience.ts", "chat-catalog-domains/internal-notification-events.ts", "chat-catalog-domains/internal-privacy.ts", "chat-catalog-domains/identity-auth.ts", "../../packages/shared/src/project-mentions.ts"], observe, async act(ctx, state) { const first = await pollUntil({ label: "Normal participant-only actual browser notification", deadlineAt: Date.now() + 180000, load: () => observe(ctx, state), accept: o => o.runs.length === 1 && o.runs[0]?.status === "succeeded" && ctx.issue()?.conversationState === "waiting" && firstAudiencePasses(o) }); diff --git a/tests/runner-e2e/chat-catalog-domains/internal-notification-events.test.ts b/tests/runner-e2e/chat-catalog-domains/internal-notification-events.test.ts new file mode 100644 index 00000000000..3944186d6b0 --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/internal-notification-events.test.ts @@ -0,0 +1,49 @@ +import { expect, it } from "vitest"; +import { bindNotificationPublicEvents, publicNotificationFrame, type NotificationObserverInput } from "./internal-notification-events.js"; +const base=Date.parse("2026-10-28T00:00:00.000Z"),iso=(n:number)=>new Date(base+n).toISOString(); +function example():[NotificationObserverInput,NotificationObserverInput]{ + const activity=[1,2].map(n=>({id:`audit-${n}`,companyId:"company",entityType:"issue",entityId:"task",action:"issue.comment_added",actorType:"agent",actorId:"ceo",createdAt:iso(n*3000+1),details:{commentId:`comment-${n}`}})); + const comments=[1,2].map(n=>({id:`comment-${n}`,companyId:"company",issueId:"task",authorAgentId:"ceo",authorUserId:null,createdAt:iso(n*3000),deletedAt:null})); + const person=(principalId:string):NotificationObserverInput=>({ + principalId,sessionPrincipalId:principalId,overflow:false,baselineActivityIds:[],baselineCommentIds:[], + activity:structuredClone(activity),comments:structuredClone(comments), + frames:[1,2].map(n=>({eventId:n+100,companyId:"company",taskId:"task",actorType:"agent" as const,actorId:"ceo", + eventCreatedAt:iso(n*3000+2),eligible:principalId==="alex"||n===2,at:base+n*3000+3})), + }); + return[person("alex"),person("riley")]; +} +it("binds actual public shapes without details to exact saved comments and common per-person event identities",()=>{ + const observers=example(),result=bindNotificationPublicEvents("company","task",observers);expect(result.complete).toBe(true);expect(result.frames[0]!.map(f=>[f.eventId,f.commentId,f.activityId,f.eligible])).toEqual([[101,"comment-1","audit-1",true],[102,"comment-2","audit-2",true]]);expect(result.frames[1]!.map(f=>[f.eventId,f.commentId,f.eligible])).toEqual([[101,"comment-1",false],[102,"comment-2",true]]); + const message={id:101,companyId:"company",type:"activity.logged",createdAt:iso(3002),toastEligible:false,payload:{action:"issue.comment_added",entityType:"issue",entityId:"task",actorType:"agent",actorId:"ceo"}}; + expect(publicNotificationFrame(message,"company","task",base+3003)).toEqual(observers[1].frames[0]);expect(publicNotificationFrame({...message,payload:{...message.payload,details:{commentId:"untrusted-private-field"}}},"company","task",base+3003)).toEqual(observers[1].frames[0]); +}); +it.each([ + ["wrong principal",(o:ReturnType)=>{o[1].principalId="alex";}], + ["wrong signed-in session",(o:ReturnType)=>{o[1].sessionPrincipalId="other";}], + ["capture overflow",(o:ReturnType)=>{o[0].overflow=true;}], + ["missing recipient event",(o:ReturnType)=>{o[1].frames.pop();}], + ["different received event ID",(o:ReturnType)=>{o[1].frames[0]!.eventId=99;}], + ["duplicate received event",(o:ReturnType)=>{o[0].frames.push(structuredClone(o[0].frames[0]!));}], + ["wrong task event",(o:ReturnType)=>{o.forEach(p=>p.frames[0]!.taskId="other");}], + ["wrong company event",(o:ReturnType)=>{o.forEach(p=>p.frames[0]!.companyId="other");}], + ["wrong event actor",(o:ReturnType)=>{o.forEach(p=>p.frames[0]!.actorId="other");}], + ["missing audit comment",(o:ReturnType)=>{o.forEach(p=>p.activity[0]!.details={});}], + ["duplicate saved activity",(o:ReturnType)=>{o.forEach(p=>p.activity.push(structuredClone(p.activity[0]!)));}], + ["duplicate saved comment",(o:ReturnType)=>{o.forEach(p=>p.comments.push(structuredClone(p.comments[0]!)));}], + ["wrong saved comment author",(o:ReturnType)=>{o.forEach(p=>p.comments[0]!.authorAgentId="other");}], + ["wrong saved activity actor",(o:ReturnType)=>{o.forEach(p=>p.activity[0]!.actorId="other");}], + ["deleted saved comment",(o:ReturnType)=>{o.forEach(p=>p.comments[0]!.deletedAt=iso(9999));}], + ["references old baseline comment",(o:ReturnType)=>{o.forEach(p=>p.baselineCommentIds.push("comment-1"));}], + ["saved comment without activity",(o:ReturnType)=>{o.forEach(p=>p.activity.pop());}], + ["saved activity without event",(o:ReturnType)=>{o.forEach(p=>p.frames.pop());}], + ["ambiguous audit order",(o:ReturnType)=>{o.forEach(p=>p.activity[1]!.createdAt=p.activity[0]!.createdAt);}], + ["ambiguous comment order",(o:ReturnType)=>{o.forEach(p=>p.comments[1]!.createdAt=p.comments[0]!.createdAt);}], + ["event before saved audit",(o:ReturnType)=>{o.forEach(p=>p.frames[0]!.eventCreatedAt=iso(1));}], + ["different authorized evidence",(o:ReturnType)=>{o[1].activity[0]!.id="different-audit";}], +] as const)("rejects %s instead of inventing a comment/event binding",(_label,mutate)=>{const observers=example();mutate(observers);expect(bindNotificationPublicEvents("company","task",observers)).toMatchObject({complete:false,frames:[[],[]]});}); +it("rejects unrelated task events, missing eligibility, malformed timestamps and wrong scope at capture",()=>{ + const message={id:1,companyId:"company",type:"activity.logged",createdAt:iso(1),toastEligible:true,payload:{action:"issue.comment_added",entityType:"issue",entityId:"task",actorType:"user",actorId:"alex"}}; + for(const changed of [{...message,toastEligible:undefined},{...message,id:"1"},{...message,createdAt:"unavailable"},{...message,type:"heartbeat.run.status"},{...message,payload:{...message.payload,action:"issue.updated"}},{...message,payload:{...message.payload,entityId:"other"}}])expect(publicNotificationFrame(changed,"company","task",base+2)).toBeNull(); +}); + +it("keeps empty observations unavailable rather than synthesizing a comment/event proof",()=>{const observers=example();for(const o of observers){o.frames=[];o.comments=[];o.activity=[];}expect(bindNotificationPublicEvents("company","task",observers)).toMatchObject({complete:false,reason:"no-new-comment-events",frames:[[],[]]});}); diff --git a/tests/runner-e2e/chat-catalog-domains/internal-notification-events.ts b/tests/runner-e2e/chat-catalog-domains/internal-notification-events.ts new file mode 100644 index 00000000000..402e60b2c72 --- /dev/null +++ b/tests/runner-e2e/chat-catalog-domains/internal-notification-events.ts @@ -0,0 +1,80 @@ +export interface NotificationPublicFrame { + eventId: number; companyId: string; taskId: string; actorType: "user" | "agent"; actorId: string; + eventCreatedAt: string; eligible: boolean; at: number; +} +export interface NotificationBoundFrame { + principalId: string; commentId: string; taskId: string; eligible: boolean; at: number; + eventId: number; activityId: string; actorType: "user" | "agent"; actorId: string; + eventCreatedAt: string; activityCreatedAt: string; commentCreatedAt: string; +} +export interface NotificationObserverInput { + principalId: string; sessionPrincipalId: string; frames: NotificationPublicFrame[]; overflow: boolean; + baselineActivityIds: string[]; baselineCommentIds: string[]; + activity: Array>; comments: Array>; +} +const nonempty = (v: unknown): v is string => typeof v === "string" && v.length > 0; +const time = (v: unknown) => typeof v === "string" ? Date.parse(v) : NaN; +export function publicNotificationFrame(value: unknown, companyId: string, taskId: string, at: number): NotificationPublicFrame | null { + const message = value as Record | null; + const payload = message?.payload as Record | undefined; + if (!message || !payload || message.companyId !== companyId || message.type !== "activity.logged" || payload.action !== "issue.comment_added" + || payload.entityType !== "issue" || payload.entityId !== taskId || !Number.isSafeInteger(message.id) || Number(message.id) <= 0 + || !["user", "agent"].includes(String(payload.actorType)) || !nonempty(payload.actorId) || typeof message.toastEligible !== "boolean" + || !Number.isFinite(time(message.createdAt)) || !Number.isFinite(at) || at < time(message.createdAt)) return null; + return { eventId: Number(message.id), companyId, taskId, actorType: payload.actorType as "user" | "agent", actorId: payload.actorId, + eventCreatedAt: String(message.createdAt), eligible: message.toastEligible, at }; +} +export function bindNotificationPublicEvents(companyId: string, taskId: string, observers: [NotificationObserverInput, NotificationObserverInput]) { + const reject = (reason: string) => ({ complete: false, reason, frames: [[], []] as NotificationBoundFrame[][] }); + if (!companyId || !taskId || observers[0].principalId === observers[1].principalId + || observers.some(o => !o.principalId || o.principalId !== o.sessionPrincipalId || o.overflow)) return reject("principal-or-capture-invalid"); + const normalized = []; + for (const observer of observers) { + if (new Set(observer.frames.map(f => f.eventId)).size !== observer.frames.length) return reject("duplicate-public-event"); + if (observer.frames.some(f => f.companyId !== companyId || f.taskId !== taskId || !Number.isSafeInteger(f.eventId) || f.eventId <= 0 + || !["user", "agent"].includes(f.actorType) || !f.actorId || typeof f.eligible !== "boolean" || !Number.isFinite(time(f.eventCreatedAt)) + || !Number.isFinite(f.at) || f.at < time(f.eventCreatedAt))) return reject("public-event-scope-invalid"); + const comments = observer.comments.filter(c => !observer.baselineCommentIds.includes(String(c.id)) && c.companyId === companyId && c.issueId === taskId); + const activity = observer.activity.filter(a => !observer.baselineActivityIds.includes(String(a.id)) && a.companyId === companyId + && a.entityType === "issue" && a.entityId === taskId && a.action === "issue.comment_added"); + if (new Set(comments.map(c => c.id)).size !== comments.length || new Set(activity.map(a => a.id)).size !== activity.length) return reject("duplicate-saved-evidence"); + const rows = []; + for (const a of activity) { + const commentId = (a.details as Record | undefined)?.commentId; + const c = comments.find(c => c.id === commentId); + if (!nonempty(a.id) || !nonempty(commentId) || !["user", "agent"].includes(String(a.actorType)) || !nonempty(a.actorId) || !c || c.deletedAt + || (a.actorType === "agent" ? c.authorAgentId !== a.actorId || !!c.authorUserId : c.authorUserId !== a.actorId || !!c.authorAgentId) + || !Number.isFinite(time(a.createdAt)) || !Number.isFinite(time(c.createdAt)) || time(c.createdAt) > time(a.createdAt)) return reject("saved-comment-audit-binding-invalid"); + rows.push({ activityId: a.id, commentId, actorType: a.actorType as "user" | "agent", actorId: a.actorId, + activityCreatedAt: String(a.createdAt), commentCreatedAt: String(c.createdAt) }); + } + if (rows.length !== comments.length || new Set(rows.map(r => r.commentId)).size !== rows.length) return reject("saved-comment-audit-not-bijective"); + normalized.push({ rows, frames: [...observer.frames].sort((a,b) => a.eventId-b.eventId) }); + } + const [left, right] = normalized as [typeof normalized[number], typeof normalized[number]]; + if (!left.rows.length || !left.frames.length) return reject("no-new-comment-events"); + const sortRows = (rows: typeof left.rows) => [...rows].sort((a,b) => time(a.activityCreatedAt)-time(b.activityCreatedAt)); + if (JSON.stringify(sortRows(left.rows)) !== JSON.stringify(sortRows(right.rows)) || left.frames.length !== right.frames.length + || left.frames.some((f,i) => { const g = right.frames[i]; return !g || f.eventId !== g.eventId || f.actorType !== g.actorType + || f.actorId !== g.actorId || f.eventCreatedAt !== g.eventCreatedAt; })) return reject("principals-did-not-observe-same-complete-events"); + const bound: NotificationBoundFrame[][] = [[], []]; + const actors = new Set(left.rows.map(r => `${r.actorType}:${r.actorId}`)); + for (const frame of left.frames) actors.add(`${frame.actorType}:${frame.actorId}`); + for (const actor of actors) { + const rows = sortRows(left.rows.filter(r => `${r.actorType}:${r.actorId}` === actor)); + const events = left.frames.filter(f => `${f.actorType}:${f.actorId}` === actor); + if (rows.length !== events.length || rows.some((r,i) => i > 0 && (time(r.activityCreatedAt) <= time(rows[i-1]!.activityCreatedAt) + || time(r.commentCreatedAt) <= time(rows[i-1]!.commentCreatedAt))) || events.some((f,i) => i > 0 && time(f.eventCreatedAt) <= time(events[i-1]!.eventCreatedAt))) return reject("actor-event-order-ambiguous"); + for (const [i, row] of rows.entries()) { + const event = events[i]!; + if (time(event.eventCreatedAt) < time(row.activityCreatedAt)) return reject("event-precedes-saved-activity"); + for (const [index, observer] of observers.entries()) { + const actual = observer.frames.find(f => f.eventId === event.eventId)!; + bound[index]!.push({ principalId: observer.principalId, commentId: row.commentId, taskId, eligible: actual.eligible, at: actual.at, + eventId: actual.eventId, activityId: row.activityId, actorType: row.actorType, actorId: row.actorId, + eventCreatedAt: actual.eventCreatedAt, activityCreatedAt: row.activityCreatedAt, commentCreatedAt: row.commentCreatedAt }); + } + } + } + return { complete: true, reason: null, frames: bound }; +}