Skip to content
Closed
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
11 changes: 7 additions & 4 deletions apps/mobile/src/connection/environment-cache-store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -204,11 +204,14 @@ describe("mobile SQLite environment cache store", () => {
const store = yield* make().pipe(
Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
);
const resume = { key: "resume-key", syncedAt: 1_753_790_400_000 };

yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT);
yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT, resume);
yield* store.saveThread(ENVIRONMENT_ID, THREAD_SNAPSHOT);

const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID));
const cached = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID));
const shell = cached.snapshot;
expect(cached.resume).toEqual(resume);
const thread = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID));

expect(DateTime.formatIso(shell.threads[0]!.updatedAt)).toBe("2026-07-29T12:00:00.000Z");
Expand Down Expand Up @@ -257,7 +260,7 @@ describe("mobile SQLite environment cache store", () => {
threads: [linked, unlinked],
});

const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID));
const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)).snapshot;
expect(shell.threads.map((thread) => thread.pullRequests)).toEqual([undefined, undefined]);
const links = yield* shell.loadPullRequests!;
expect([...links]).toEqual([[THREAD_ID, [link]]]);
Expand Down Expand Up @@ -298,7 +301,7 @@ describe("mobile SQLite environment cache store", () => {
payload.snapshot.threads[0].pullRequests = [{ number: -1 }];
memory.values.set(id, JSON.stringify(payload));

const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID));
const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)).snapshot;
expect(shell.threads.map((thread) => thread.id)).toEqual([THREAD_ID, readable.id]);
expect([...(yield* shell.loadPullRequests!)]).toEqual([[readable.id, [link]]]);
expect(messages).toHaveLength(1);
Expand Down
36 changes: 25 additions & 11 deletions apps/mobile/src/connection/environment-cache-store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,12 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {
operation: "load-shell",
decode: decodeStoredShellSnapshot,
select: (stored) =>
stored.environmentId === environmentId ? Option.some(stored.snapshot) : Option.none(),
stored.environmentId === environmentId
? Option.some({
snapshot: stored.snapshot,
...(stored.resume === undefined ? {} : { resume: stored.resume }),
})
: Option.none(),
}).pipe(
Effect.zipWith(
Effect.promise(() => projectFaviconDatabaseCache.hydrate()),
Expand All @@ -120,16 +125,25 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {
),
),
),
saveShell: Effect.fn("MobileEnvironmentCache.saveShell")(function* (environmentId, snapshot) {
const payload = yield* encodeStoredShellSnapshot({
schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
environmentId,
snapshot,
}).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause)));
yield* database
.saveCache(environmentId, "shell", "snapshot", ORCHESTRATION_CACHE_SCHEMA_VERSION, payload)
.pipe(Effect.mapError(mapDatabaseError("save-shell")));
}),
saveShell: Effect.fn("MobileEnvironmentCache.saveShell")(
function* (environmentId, snapshot, resume) {
const payload = yield* encodeStoredShellSnapshot({
schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION,
environmentId,
snapshot,
...(resume === undefined ? {} : { resume }),
}).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause)));
yield* database
.saveCache(
environmentId,
"shell",
"snapshot",
ORCHESTRATION_CACHE_SCHEMA_VERSION,
payload,
)
.pipe(Effect.mapError(mapDatabaseError("save-shell")));
},
),
loadThread: Effect.fn("MobileEnvironmentCache.loadThread")((environmentId, threadId) =>
loadDecodedCache({
database,
Expand Down
8 changes: 8 additions & 0 deletions apps/mobile/src/connection/platform.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,10 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
import * as Stream from "effect/Stream";
import * as Application from "expo-application";
import Constants from "expo-constants";
import * as Network from "expo-network";
import * as Updates from "expo-updates";
import { AppState } from "react-native";

import { authClientMetadata } from "../lib/authClientMetadata";
Expand Down Expand Up @@ -213,6 +215,12 @@ const layerCapabilities = Layer.effectContext(
ClientCapabilities.ClientPresentation,
ClientCapabilities.ClientPresentation.of({
metadata: authClientMetadata(Constants.expoConfig?.version),
// Store builds can share a version, and over-the-air updates keep the native build.
build: [
Constants.expoConfig?.version,
Application.nativeBuildVersion,
Updates.updateId ?? "embedded",
].join(" "),
}),
),
Context.add(
Expand Down
14 changes: 14 additions & 0 deletions apps/server/src/orchestration-v2/ShellStream.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import {
omitOlderSettledShells,
shellStreamItemFromEnrichmentRefresh,
shellStreamItemsFromInitialSnapshot,
ShellResumeKey,
shouldPublishShellApplicationEvent,
skipUnchangedThreadShells,
} from "./ShellStream.ts";
Expand Down Expand Up @@ -662,3 +663,16 @@ describe("skipUnchangedThreadShells", () => {
}),
);
});

describe("shell resume key", () => {
it.effect("stays one value for the whole server process", () =>
Effect.gen(function* () {
// Each socket reads it on its own fiber, which falls back to the cached default.
const read = Effect.gen(function* () {
return yield* ShellResumeKey;
});
const [first, second] = yield* Effect.all([read, read], { concurrency: "unbounded" });
expect(second).toBe(first);
}),
);
});
47 changes: 38 additions & 9 deletions apps/server/src/orchestration-v2/ShellStream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,19 +9,20 @@ import type {
OrchestrationV2ThreadShellSnapshot,
OrchestrationV2ShellStreamItem,
OrchestrationV2StoredEvent,
ProjectId,
ThreadId,
} from "@t3tools/contracts";
import {
isLiveSubagentShell,
isOlderSettledThreadShell,
OrchestrationProjectShell as ProjectShellSchema,
OrchestrationV2ThreadShell as ThreadShellSchema,
ProjectId,
ThreadId,
} from "@t3tools/contracts";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import type * as Arr from "effect/Array";
import type * as Cause from "effect/Cause";
import * as Clock from "effect/Clock";
import * as Context from "effect/Context";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
Expand All @@ -31,6 +32,7 @@ import * as Schema from "effect/Schema";
import type * as SqlClient from "effect/sql/SqlClient";
import * as Stream from "effect/Stream";

import type { ShellReplayHead } from "../persistence/OrchestrationEventStore.ts";
import { bufferLatestLiveStream } from "./LiveStreamBudget.ts";

type ShellDelta = Extract<OrchestrationV2ShellStreamItem, { readonly sequence: number }>;
Expand Down Expand Up @@ -114,7 +116,7 @@ export function activeShellRowsForSubscriber(
};
}

const NEWEST_SETTLED_KEPT = 25;
export const NEWEST_SETTLED_KEPT = 25;

/**
* Opted-in snapshots keep the 25 newest settled top-level threads (and ties) and count the rest per
Expand Down Expand Up @@ -282,7 +284,14 @@ export type ShellApplicationEvent =
readonly event: Pick<OrchestrationV2StoredEvent["event"], "threadId">;
};

/** Focused subagents receive these updates through their thread subscription. */
/** Subagent events the shell skips; focused subagents receive them through their thread subscription. */
export const SUBAGENT_TRANSCRIPT_EVENT_TYPES: ReadonlySet<string> = new Set([
"node.updated",
"turn-item.updated",
"provider-turn.updated",
"message.updated",
]);

export function shouldPublishShellApplicationEvent(
stored: ApplicationStoredEvent,
subagentThreadIds: Set<string>,
Expand All @@ -300,14 +309,22 @@ export function shouldPublishShellApplicationEvent(
return true;
}
return !(
subagentThreadIds.has(event.threadId) &&
(event.type === "node.updated" ||
event.type === "turn-item.updated" ||
event.type === "provider-turn.updated" ||
event.type === "message.updated")
subagentThreadIds.has(event.threadId) && SUBAGENT_TRANSCRIPT_EVENT_TYPES.has(event.type)
);
}

/**
* `ServerConfig.shellResumeKey`. The default is computed once and cached on the reference, so it
* is one random value per server process; tests provide their own.
*/
export class ShellResumeKey extends Context.Reference<string>(
"@t3tools/server/orchestration-v2/ShellStream/ShellResumeKey",
{
// @effect-diagnostics-next-line cryptoRandomUUID:off - A reference default runs outside Effect.
defaultValue: () => crypto.randomUUID(),
},
) {}

/** Shell updates refetch an aggregate; drop transcript bodies before retaining an event. */
export function toShellApplicationEvent(stored: ApplicationStoredEvent): ShellApplicationEvent {
return "aggregateKind" in stored
Expand All @@ -320,6 +337,18 @@ export function toShellApplicationEvent(stored: ApplicationStoredEvent): ShellAp
: { sequence: stored.sequence, event: { threadId: stored.event.threadId } };
}

export function shellApplicationEventFromReplayHead(head: ShellReplayHead): ShellApplicationEvent {
return head.aggregateKind === "project"
? {
aggregateKind: "project",
aggregateId: ProjectId.make(head.aggregateId),
// Only a deletion skips the row read; any other project event, even an unknown one, reads it.
type: head.eventType === "project.deleted" ? "project.deleted" : "project.meta-updated",
sequence: head.sequence,
}
: { sequence: head.sequence, event: { threadId: ThreadId.make(head.aggregateId) } };
}

export function coalesceShellApplicationEvents<A extends ShellApplicationEvent>(
events: ReadonlyArray<A>,
): ReadonlyArray<A> {
Expand Down
78 changes: 68 additions & 10 deletions apps/server/src/persistence/OrchestrationEventStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -209,11 +209,6 @@ layer("OrchestrationEventStore", (it) => {
maxEvents: events.length,
});
assert.isAbove(stats.rawPayloadBytes, prompt.length * 6);
const applicationStats = yield* store.getReplayStats({
afterSequence,
throughSequence: yield* store.latestApplicationSequence,
});
assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6);
}),
);

Expand Down Expand Up @@ -456,14 +451,77 @@ layer("OrchestrationEventStore", (it) => {
rawPayloadBytes: 0,
hasCreateEvent: false,
});
assert.deepEqual(
yield* store.getReplayStats({ afterSequence: baseline, throughSequence: head }),
}),
);

it.effect("finds the newest shell event of each project and thread in a range", () =>
Effect.gen(function* () {
const store = yield* OrchestrationEventStore.OrchestrationEventStore;
const sql = yield* SqlClient.SqlClient;
const baseline = yield* store.latestApplicationSequence;
const versions = new Map<string, number>();
const insert = Effect.fn(function* (
aggregateKind: "project" | "thread",
streamId: string,
eventType: string,
applicationEventVersion = 2,
) {
const streamVersion = versions.get(streamId) ?? 0;
versions.set(streamId, streamVersion + 1);
const rows = yield* sql<{ readonly sequence: number }>`
INSERT INTO orchestration_events (
event_id, aggregate_kind, stream_id, stream_version, event_type,
occurred_at, actor_kind, payload_json, metadata_json, application_event_version
) VALUES (
${`heads:${streamId}:${streamVersion}`}, ${aggregateKind}, ${streamId}, ${streamVersion},
${eventType}, '2026-09-04T00:00:00.000Z', 'server', 'not JSON', '{}',
${applicationEventVersion}
) RETURNING sequence
`;
return rows[0]!.sequence;
});
yield* insert("thread", "heads:parent", "thread.created");
const projectHead = yield* insert("project", "heads:project", "project.deleted");
const childHead = yield* insert("thread", "heads:child", "thread.metadata-updated");
yield* insert("thread", "heads:child", "turn-item.updated");
// Legacy rows are not application events.
yield* insert("thread", "heads:legacy", "thread.metadata-updated", 1);
const parentHead = yield* insert("thread", "heads:parent", "future.event");
const head = yield* store.latestApplicationSequence;
const read = (input: { readonly afterSequence?: number; readonly limit?: number } = {}) =>
store.readShellReplayHeads({
afterSequence: input.afterSequence ?? baseline,
throughSequence: head,
transcriptEventTypes: ["turn-item.updated"],
skipTranscriptOf: ["heads:child"],
limit: input.limit ?? 10,
});

assert.deepEqual(yield* read(), [
{
eventCount: 3,
rawPayloadBytes:
Buffer.byteLength(unicodePayload) + 2 * Buffer.byteLength(oversizedPayload),
aggregateKind: "project",
aggregateId: "heads:project",
sequence: projectHead,
eventType: "project.deleted",
},
{
aggregateKind: "thread",
aggregateId: "heads:child",
sequence: childHead,
eventType: "thread.metadata-updated",
},
{
aggregateKind: "thread",
aggregateId: "heads:parent",
sequence: parentHead,
eventType: "future.event",
},
]);
assert.deepEqual(
(yield* read({ limit: 2 })).map((row) => row.aggregateId),
["heads:project", "heads:child"],
);
assert.deepEqual(yield* read({ afterSequence: parentHead }), []);
}),
);

Expand Down
Loading
Loading