Skip to content
34 changes: 15 additions & 19 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import {
type OrchestrationSessionStatus,
ThreadId,
} from "@t3tools/contracts";
import { compareDateTimeStrings } from "@t3tools/shared/dateTime";
import { selectFirstByDateTime } from "@t3tools/shared/dateTime";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
Expand Down Expand Up @@ -247,19 +247,17 @@ function retainProjectionMessagesAfterRevert(
).length;
const missingUserCount = Math.max(0, turnCount - retainedUserCount);
if (missingUserCount > 0) {
const fallbackUserMessages = messages
.filter(
const fallbackUserMessages = selectFirstByDateTime(
messages.filter(
(message) =>
message.role === "user" &&
!retainedMessageIds.has(message.messageId) &&
(message.turnId === null || retainedTurnIds.has(message.turnId)),
)
.toSorted(
(left, right) =>
compareDateTimeStrings(left.createdAt, right.createdAt) ||
left.messageId.localeCompare(right.messageId),
)
.slice(0, missingUserCount);
),
(message) => message.createdAt,
missingUserCount,
(left, right) => left.messageId.localeCompare(right.messageId),
);
for (const message of fallbackUserMessages) {
retainedMessageIds.add(message.messageId);
}
Expand All @@ -273,19 +271,17 @@ function retainProjectionMessagesAfterRevert(
).length;
const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount);
if (missingAssistantCount > 0) {
const fallbackAssistantMessages = messages
.filter(
const fallbackAssistantMessages = selectFirstByDateTime(
messages.filter(
(message) =>
message.role === "assistant" &&
!retainedMessageIds.has(message.messageId) &&
(message.turnId === null || retainedTurnIds.has(message.turnId)),
)
.toSorted(
(left, right) =>
compareDateTimeStrings(left.createdAt, right.createdAt) ||
left.messageId.localeCompare(right.messageId),
)
.slice(0, missingAssistantCount);
),
(message) => message.createdAt,
missingAssistantCount,
(left, right) => left.messageId.localeCompare(right.messageId),
);
for (const message of fallbackAssistantMessages) {
retainedMessageIds.add(message.messageId);
}
Expand Down
34 changes: 15 additions & 19 deletions apps/server/src/orchestration/projector.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import {
legacyThreadPullRequestKey,
threadPullRequestKeysEqual,
} from "@t3tools/shared/threadPullRequests";
import { compareDateTimeStrings } from "@t3tools/shared/dateTime";
import { selectFirstByDateTime } from "@t3tools/shared/dateTime";
import * as Effect from "effect/Effect";
import * as Schema from "effect/Schema";
import * as Predicate from "effect/Predicate";
Expand Down Expand Up @@ -230,19 +230,17 @@ function retainThreadMessagesAfterRevert(
).length;
const missingUserCount = Math.max(0, turnCount - retainedUserCount);
if (missingUserCount > 0) {
const fallbackUserMessages = messages
.filter(
const fallbackUserMessages = selectFirstByDateTime(
messages.filter(
(message) =>
message.role === "user" &&
!retainedMessageIds.has(message.id) &&
(message.turnId === null || retainedTurnIds.has(message.turnId)),
)
.toSorted(
(left, right) =>
compareDateTimeStrings(left.createdAt, right.createdAt) ||
left.id.localeCompare(right.id),
)
.slice(0, missingUserCount);
),
(message) => message.createdAt,
missingUserCount,
(left, right) => left.id.localeCompare(right.id),
);
for (const message of fallbackUserMessages) {
retainedMessageIds.add(message.id);
}
Expand All @@ -256,19 +254,17 @@ function retainThreadMessagesAfterRevert(
).length;
const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount);
if (missingAssistantCount > 0) {
const fallbackAssistantMessages = messages
.filter(
const fallbackAssistantMessages = selectFirstByDateTime(
messages.filter(
(message) =>
message.role === "assistant" &&
!retainedMessageIds.has(message.id) &&
(message.turnId === null || retainedTurnIds.has(message.turnId)),
)
.toSorted(
(left, right) =>
compareDateTimeStrings(left.createdAt, right.createdAt) ||
left.id.localeCompare(right.id),
)
.slice(0, missingAssistantCount);
),
(message) => message.createdAt,
missingAssistantCount,
(left, right) => left.id.localeCompare(right.id),
);
for (const message of fallbackAssistantMessages) {
retainedMessageIds.add(message.id);
}
Expand Down
21 changes: 9 additions & 12 deletions packages/client-runtime/src/state/threadReducer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ import type {
} from "@t3tools/contracts";
import { threadPullRequestKeysEqual } from "@t3tools/shared/threadPullRequests";
import { isImportedAgentSessionMessageId } from "@t3tools/contracts";
import { compareDateTimeStrings } from "@t3tools/shared/dateTime";
import { selectFirstByDateTime } from "@t3tools/shared/dateTime";

export type ThreadDetailReducerResult =
| { readonly kind: "updated"; readonly thread: OrchestrationThread }
Expand Down Expand Up @@ -840,21 +840,18 @@ function retainMessagesAfterRevert(
retainedMessageIds.has(message.id),
).length;
const missingCount = Math.max(0, turnCount - retainedCount);
const fallbackMessages = messages
.filter(
if (missingCount === 0) continue;
const fallbackMessages = selectFirstByDateTime(
messages.filter(
(message) =>
message.role === role &&
!retainedMessageIds.has(message.id) &&
(message.turnId === null || retainedTurnIds.has(message.turnId)),
)
// `.sort()`, not `.toSorted()`: `.filter()` above already returned a fresh array, and
// this is shared with mobile, which runs on Hermes and has no ES2023 array methods.
.sort(
(left, right) =>
compareDateTimeStrings(left.createdAt, right.createdAt) ||
left.id.localeCompare(right.id),
)
.slice(0, missingCount);
),
(message) => message.createdAt,
missingCount,
(left, right) => left.id.localeCompare(right.id),
);
for (const message of fallbackMessages) {
retainedMessageIds.add(message.id);
}
Expand Down
87 changes: 86 additions & 1 deletion packages/shared/src/dateTime.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { describe, expect, it } from "@effect/vitest";

import { compareDateTimeStrings } from "./dateTime.ts";
import { compareDateTimeStrings, selectFirstByDateTime } from "./dateTime.ts";

describe("compareDateTimeStrings", () => {
it("compares valid date-time strings by absolute time", () => {
Expand Down Expand Up @@ -90,3 +90,88 @@ describe("compareDateTimeStrings", () => {
}
});
});

describe("selectFirstByDateTime", () => {
const items = [
{ id: "e", createdAt: "2026-09-01T12:00:01.000Z" },
{ id: "c", createdAt: "invalid" },
{ id: "a", createdAt: "2026-09-01T12:00:00.000Z" },
{ id: "d", createdAt: "2024-02-30T00:00:00.000Z" },
{ id: "b", createdAt: "2026-09-01T05:00:00.000-07:00" },
];
const byId = (left: { id: string }, right: { id: string }) => left.id.localeCompare(right.id);
const expectedOrder = [...items]
.sort(
(left, right) => compareDateTimeStrings(left.createdAt, right.createdAt) || byId(left, right),
)
.map((item) => item.id);

it("selects the same items as a sort and slice for every limit", () => {
for (const limit of [0, 1, 2, 3, 5, 9]) {
const actual = selectFirstByDateTime(items, (item) => item.createdAt, limit, byId)
.map((item) => item.id)
.sort();
expect(actual).toEqual(expectedOrder.slice(0, limit).sort());
}
});

it("returns the selected items in date-time order", () => {
expect(
selectFirstByDateTime(items, (item) => item.createdAt, 3, byId).map((item) => item.id),
).toEqual(expectedOrder.slice(0, 3));
});

it("breaks timestamp ties with the tie-break comparator", () => {
expect(
selectFirstByDateTime(items, (item) => item.createdAt, 2, byId).map((item) => item.id),
).toEqual(["d", "c"]);
expect(
selectFirstByDateTime(items, (item) => item.createdAt, 4, byId).map((item) => item.id),
).toEqual(["d", "c", "a", "b"]);
});

it("returns an empty array for a non-positive limit", () => {
expect(selectFirstByDateTime(items, (item) => item.createdAt, 0, byId)).toEqual([]);
expect(selectFirstByDateTime(items, (item) => item.createdAt, -1, byId)).toEqual([]);
});

it("does not mutate its input", () => {
const before = items.map((item) => item.id);
selectFirstByDateTime(items, (item) => item.createdAt, 2, byId);
expect(items.map((item) => item.id)).toEqual(before);
});

it("sorts a generated corpus like compareDateTimeStrings", () => {
const offsets = ["Z", "+00:00", "-07:00", "+05:30", "+14:00", "-12:00"];
const instants = [
"2000-02-29T00:00:00.100",
"2024-02-29T12:00:00.000",
"2024-03-01T00:00:00.000",
"2025-01-01T00:00:00.000",
"2025-06-15T08:30:00.500",
"2025-12-31T23:59:59.999",
"2026-09-01T12:00:00.000",
"2026-09-01T24:00:00.000",
];
const generated = instants.flatMap((instant, instantIndex) =>
offsets.map((offset, offsetIndex) => ({
id: `i-${instantIndex}-${offsetIndex}`,
createdAt: `${instant}${offset}`,
})),
);
const expected = [...generated]
.sort(
(left, right) =>
compareDateTimeStrings(left.createdAt, right.createdAt) || byId(left, right),
)
.map((item) => item.id);
expect(
selectFirstByDateTime(generated, (item) => item.createdAt, generated.length, byId).map(
(item) => item.id,
),
).toEqual(expected);
expect(
selectFirstByDateTime(generated, (item) => item.createdAt, 7, byId).map((item) => item.id),
).toEqual(expected.slice(0, 7));
});
});
95 changes: 95 additions & 0 deletions packages/shared/src/dateTime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,3 +36,98 @@ export function compareDateTimeStrings(left: string, right: string): number {
if (leftIsValid) return leftTimestamp - rightTimestamp;
return left < right ? -1 : left > right ? 1 : 0;
}

interface DateEntry<T> {
readonly item: T;
readonly raw: string;
readonly timestamp: number;
}

function entryOf<T>(item: T, getDate: (item: T) => string): DateEntry<T> {
const raw = getDate(item);
return { item, raw, timestamp: parseTimestamp(raw) };
}

function compareDateEntries<T>(
left: DateEntry<T>,
right: DateEntry<T>,
compareTie: (left: T, right: T) => number,
): number {
const leftValid = !Number.isNaN(left.timestamp);
const rightValid = !Number.isNaN(right.timestamp);
if (leftValid !== rightValid) return leftValid ? 1 : -1;
if (leftValid) {
const byTime = left.timestamp - right.timestamp;
return byTime !== 0 ? byTime : compareTie(left.item, right.item);
}
const byRaw = left.raw < right.raw ? -1 : left.raw > right.raw ? 1 : 0;
return byRaw !== 0 ? byRaw : compareTie(left.item, right.item);
}

function siftUp<T>(
heap: Array<DateEntry<T>>,
index: number,
compareTie: (left: T, right: T) => number,
): void {
while (index > 0) {
const parent = (index - 1) >> 1;
if (compareDateEntries(heap[index]!, heap[parent]!, compareTie) <= 0) return;
const swap = heap[index]!;
heap[index] = heap[parent]!;
heap[parent] = swap;
index = parent;
}
}

function siftDown<T>(
heap: Array<DateEntry<T>>,
index: number,
compareTie: (left: T, right: T) => number,
): void {
for (;;) {
const left = index * 2 + 1;
const right = left + 1;
let largest = index;
if (left < heap.length && compareDateEntries(heap[left]!, heap[largest]!, compareTie) > 0) {
largest = left;
}
if (right < heap.length && compareDateEntries(heap[right]!, heap[largest]!, compareTie) > 0) {
largest = right;
}
if (largest === index) return;
const swap = heap[index]!;
heap[index] = heap[largest]!;
heap[largest] = swap;
index = largest;
}
}

export function selectFirstByDateTime<T>(
items: readonly T[],
getDate: (item: T) => string,
limit: number,
compareTie: (left: T, right: T) => number,
): T[] {
if (limit <= 0) return [];
const sortedItems = (entries: Array<DateEntry<T>>): T[] =>
// `.sort()`, not `.toSorted()`: the entries array is created here, and this module is
// shared with mobile, which runs on Hermes and has no ES2023 array methods.
entries
.sort((left, right) => compareDateEntries(left, right, compareTie))
.map((entry) => entry.item);
if (limit >= items.length) {
return sortedItems(items.map((item) => entryOf(item, getDate)));
}
const heap: Array<DateEntry<T>> = [];
for (const item of items) {
const entry = entryOf(item, getDate);
if (heap.length < limit) {
heap.push(entry);
siftUp(heap, heap.length - 1, compareTie);
} else if (compareDateEntries(entry, heap[0]!, compareTie) < 0) {
heap[0] = entry;
siftDown(heap, 0, compareTie);
}
}
return sortedItems(heap);
}
Loading