import { OBSERVER_STATUSES, type CommitmentEvent, type ObservedThread, type ObserverOutput, type Registry, type Thread, type ThreadStatus } from "./types.ts";
export const emptyRegistry = (): Registry => ({ threads: [], unreconciled: [], created: 0 });
export const isFinal = (thread: Pick<Thread, "status">): boolean =>
thread.status === "done" ||
thread.status === "cancelled" ||
thread.status === "superseded" ||
thread.status === "dismissed";
type LiveThread = Thread & { status: Exclude<ThreadStatus, "done" | "cancelled" | "superseded" | "dismissed"> };
export const liveThreads = (registry: Registry): LiveThread[] =>
registry.threads.filter((thread): thread is LiveThread => !isFinal(thread));
export function observationChangesState(registry: Registry, output: ObserverOutput): boolean {
if (output.additions.length > 0) return true;
const existing = new Map(output.existing.map((thread) => [thread.id, thread]));
const live = liveThreads(registry);
return (
existing.size !== live.length ||
live.some((thread) => {
const observed = existing.get(thread.id);
return (
observed === undefined ||
observed.canonicalRequest !== thread.canonicalRequest ||
observed.status !== thread.status
);
})
);
}
const patchThread = (thread: Thread, observed: ObservedThread, at: number): Thread => ({
...thread,
canonicalRequest: observed.canonicalRequest,
status: observed.status,
updatedAt: at,
});
const applyObservation = (
registry: Registry,
event: Extract<CommitmentEvent, { kind: "observation" }>,
): Registry => {
const byId = new Map(event.output.existing.map((thread) => [thread.id, thread]));
const threads = registry.threads.map((thread) => {
const observed = byId.get(thread.id);
return observed && !isFinal(thread) ? patchThread(thread, observed, event.at) : thread;
});
const additions = event.output.additions.map((observed, index): Thread => ({
id: `t${registry.created + index + 1}`,
canonicalRequest: observed.canonicalRequest,
status: observed.status,
createdAt: event.at,
updatedAt: event.at,
}));
const reconciled = new Set(event.reconciled);
return {
threads: [...threads, ...additions],
unreconciled: registry.unreconciled.filter((message) => !reconciled.has(message.messageId)),
created: registry.created + additions.length,
};
};
const applyUserAction = (
registry: Registry,
event: Extract<CommitmentEvent, { kind: "user_action" }>,
): Registry => {
if (event.action === "drop") {
return { ...registry, threads: registry.threads.filter((thread) => thread.id !== event.threadId) };
}
const status: ThreadStatus =
event.action === "accept" ? "done" : event.action === "dismiss" ? "dismissed" : "open";
return {
...registry,
threads: registry.threads.map((thread) =>
thread.id === event.threadId ? { ...thread, status, updatedAt: event.at } : thread,
),
};
};
export function replay(events: CommitmentEvent[]): Registry {
return events.reduce<Registry>((registry, event) => {
switch (event.kind) {
case "user_message":
return {
...registry,
unreconciled: [...registry.unreconciled, event],
};
case "observation":
return applyObservation(registry, event);
case "user_action":
return applyUserAction(registry, event);
case "correction":
return {
...registry,
threads: registry.threads.map((thread) =>
thread.id === event.threadId
? {
...thread,
...(event.canonicalRequest === undefined
? {}
: { canonicalRequest: event.canonicalRequest }),
...(event.status === undefined ? {} : { status: event.status }),
updatedAt: event.at,
}
: thread,
),
};
case "observer_failure":
return registry;
}
}, emptyRegistry());
}
const isRecord = (value: unknown): value is Record<string, unknown> =>
typeof value === "object" && value !== null && !Array.isArray(value);
const statuses = new Set<string>([...OBSERVER_STATUSES, "dismissed"]);
const observerStatuses = new Set<string>(OBSERVER_STATUSES);
const userActions = new Set(["accept", "dismiss", "reopen", "drop"]);
const strings = (value: unknown): value is string[] =>
Array.isArray(value) && value.every((item) => typeof item === "string");
const observed = (value: unknown, requireId: boolean): boolean =>
isRecord(value) &&
typeof value.canonicalRequest === "string" &&
typeof value.status === "string" &&
observerStatuses.has(value.status) &&
(requireId ? typeof value.id === "string" : value.id === undefined);
export function isCommitmentEvent(value: unknown): value is CommitmentEvent {
if (!isRecord(value) || typeof value.kind !== "string" || typeof value.at !== "number") return false;
switch (value.kind) {
case "user_message":
return typeof value.messageId === "string" && typeof value.text === "string";
case "observation":
return (
strings(value.reconciled) &&
isRecord(value.output) &&
Array.isArray(value.output.existing) &&
value.output.existing.every((thread) => observed(thread, true)) &&
Array.isArray(value.output.additions) &&
value.output.additions.every((thread) => observed(thread, false))
);
case "user_action":
return (
typeof value.threadId === "string" &&
typeof value.action === "string" &&
userActions.has(value.action)
);
case "correction":
return (
typeof value.threadId === "string" &&
(value.canonicalRequest === undefined || typeof value.canonicalRequest === "string") &&
(value.status === undefined || (typeof value.status === "string" && statuses.has(value.status)))
);
case "observer_failure":
return strings(value.messageIds) && typeof value.error === "string";
default:
return false;
}
}