Skip to content

Commit 55dfc5f

Browse files
Bil0000Bil0000
authored
fix(composer): read dropped threads across environments (#15210)
1 parent ec1313c commit 55dfc5f

21 files changed

Lines changed: 1164 additions & 108 deletions

‎apps/server/integration/transferBudgetV2.integration.test.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import {
1010
AuthSessionId,
1111
AuthOrchestrationReadScope,
1212
EnvironmentHttpApi,
13+
EnvironmentId,
1314
EnvironmentAuthenticatedAuth,
1415
EnvironmentAuthenticatedPrincipal,
1516
ORCHESTRATION_V2_WS_METHODS,
@@ -35,6 +36,7 @@ import * as HttpApi from "effect/http-api/HttpApi";
3536
import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
3637
import { HttpRouter, HttpServer } from "effect/http";
3738
import { Rpc, RpcGroup, RpcServer, RpcSerialization } from "effect/rpc";
39+
import * as ServerEnvironment from "../src/environment/ServerEnvironment.ts";
3840
import * as SqlitePersistence from "../src/persistence/Sqlite.ts";
3941
import * as OrchestrationEventStore from "../src/persistence/OrchestrationEventStore.ts";
4042
import * as EventStore from "../src/orchestration-v2/EventStore.ts";
@@ -111,6 +113,11 @@ const layerEnrichment = Layer.unwrap(
111113
);
112114
// The transfer history has no project events, so shell streams never read a project shell.
113115
const layerServices = layerManagement.pipe(
116+
Layer.provideMerge(
117+
Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
118+
getEnvironmentId: Effect.succeed(EnvironmentId.make("transfer-test")),
119+
}),
120+
),
114121
Layer.provideMerge(ProjectStore.layer),
115122
Layer.provideMerge(Layer.mock(ProjectService.ProjectService)({})),
116123
Layer.provideMerge(layerEnrichment),
@@ -237,6 +244,7 @@ it.live(
237244
Layer.provide(Layer.succeedContext(context)),
238245
Layer.provideMerge(
239246
NodeHttpServer.layer(NodeHttp.createServer, {
247+
host: "127.0.0.1",
240248
port: 0,
241249
websocket: { perMessageDeflate: true },
242250
}),

‎apps/server/src/orchestration-v2/ThreadManagementService.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,16 +23,19 @@ import {
2323
type OrchestrationV2Run,
2424
type OrchestrationV2ThreadShellSnapshot,
2525
type OrchestrationV2ThreadProjection,
26+
type OrchestrationV2ThreadTranscript,
2627
type OrchestrationV2ThreadShell,
2728
type OrchestrationV2TurnItem,
2829
ProjectId,
30+
PROVIDER_SEND_TURN_MAX_FILE_BYTES,
2931
ProviderInstanceId,
3032
resolveProjectProviderInstanceEnabled,
3133
RunId,
3234
type ScheduledTaskId,
3335
ThreadId,
3436
type TurnItemId,
3537
} from "@t3tools/contracts";
38+
import { threadTranscriptHeader } from "@t3tools/shared/threadTranscript";
3639
import * as Context from "effect/Context";
3740
import * as DateTime from "effect/DateTime";
3841
import * as Duration from "effect/Duration";
@@ -45,6 +48,8 @@ import * as Stream from "effect/Stream";
4548

4649
import * as CommandReceiptStore from "./CommandReceiptStore.ts";
4750
import * as Orchestrator from "./Orchestrator.ts";
51+
import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
52+
import { projectedRowEncodedBytes } from "./threadHistoryPaging.ts";
4853
import * as ProviderSessionManager from "./ProviderSessionManager.ts";
4954
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
5055
import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimit.ts";
@@ -186,6 +191,15 @@ export class ThreadManagementThreadNotFoundError extends Schema.TaggedError<Thre
186191
}
187192
}
188193

194+
export class ThreadTranscriptTooLargeError extends Schema.TaggedError<ThreadTranscriptTooLargeError>()(
195+
"ThreadTranscriptTooLargeError",
196+
{ threadId: ThreadId },
197+
) {
198+
override get message(): string {
199+
return "The thread transcript exceeds the file attachment limit.";
200+
}
201+
}
202+
189203
export class ThreadManagementRunNotFoundError extends Schema.TaggedError<ThreadManagementRunNotFoundError>()(
190204
"ThreadManagementRunNotFoundError",
191205
{
@@ -336,6 +350,13 @@ export interface ThreadManagementServiceShape {
336350
) => Effect.Effect<OrchestrationV2ThreadProjection, Orchestrator.OrchestratorV2Error>;
337351
readonly getCheckpointContext: Orchestrator.OrchestratorV2["Service"]["getCheckpointContext"];
338352
readonly getThreadSnapshot: Orchestrator.OrchestratorV2["Service"]["getThreadSnapshot"];
353+
readonly getThreadTranscript: (
354+
threadId: ThreadId,
355+
) => Effect.Effect<
356+
OrchestrationV2ThreadTranscript,
357+
Orchestrator.OrchestratorV2Error | ThreadTranscriptTooLargeError,
358+
ServerEnvironment.ServerEnvironmentIdentity
359+
>;
339360
readonly getThreadSnapshotWindow: Orchestrator.OrchestratorV2["Service"]["getThreadSnapshotWindow"];
340361
readonly getProjectThreadRecords: <K extends ProjectionRecordField>(
341362
input: { readonly projectId: ProjectId; readonly threadId: ThreadId },
@@ -627,6 +648,32 @@ const make = Effect.gen(function* () {
627648
ensureProjectionTranscript(threadId).pipe(
628649
Effect.andThen(orchestrator.getThreadSnapshot(threadId)),
629650
);
651+
const getThreadTranscript = Effect.fn("orchestrationV2.threadManagement.getThreadTranscript")(
652+
function* (threadId: ThreadId) {
653+
const { projection } = yield* getThreadSnapshot(threadId);
654+
const environment = yield* ServerEnvironment.ServerEnvironmentIdentity;
655+
const transcript = {
656+
threadId: projection.thread.id,
657+
title: projection.thread.title,
658+
updatedAt: projection.updatedAt,
659+
items: projection.visibleTurnItems,
660+
};
661+
let sizeBytes = Buffer.byteLength(
662+
threadTranscriptHeader(yield* environment.getEnvironmentId, transcript),
663+
"utf8",
664+
);
665+
if (sizeBytes > PROVIDER_SEND_TURN_MAX_FILE_BYTES) {
666+
return yield* new ThreadTranscriptTooLargeError({ threadId });
667+
}
668+
for (const row of transcript.items) {
669+
sizeBytes += projectedRowEncodedBytes(row) + 1;
670+
if (sizeBytes > PROVIDER_SEND_TURN_MAX_FILE_BYTES) {
671+
return yield* new ThreadTranscriptTooLargeError({ threadId });
672+
}
673+
}
674+
return transcript;
675+
},
676+
);
630677
const getThreadSnapshotWindow: ThreadManagementServiceShape["getThreadSnapshotWindow"] = (
631678
threadId,
632679
options,
@@ -1161,6 +1208,7 @@ const make = Effect.gen(function* () {
11611208
getThreadProjection,
11621209
getCheckpointContext,
11631210
getThreadSnapshot,
1211+
getThreadTranscript,
11641212
getThreadSnapshotWindow,
11651213
getProjectThreadRecords,
11661214
getProjectThread,

‎apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import {
55
EnvironmentAuthenticatedAuth,
66
EnvironmentAuthenticatedPrincipal,
77
EnvironmentHttpApi,
8+
EnvironmentId,
89
EventId,
910
MessageId,
1011
NodeId,
@@ -34,6 +35,7 @@ import * as HttpApi from "effect/http-api/HttpApi";
3435
import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
3536
import { Etag, HttpRouter } from "effect/http";
3637

38+
import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
3739
import * as SqlitePersistence from "../persistence/Sqlite.ts";
3840
import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
3941
import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
@@ -226,6 +228,9 @@ const seed = Effect.gen(function* () {
226228
// thread routes never touch projects.
227229
const TestLayer = Layer.mergeAll(
228230
management,
231+
Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
232+
getEnvironmentId: Effect.succeed(EnvironmentId.make("compact-transport")),
233+
}),
229234
Layer.effectDiscard(seed),
230235
Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}),
231236
Layer.mock(ProjectStore.ProjectStoreV2)({}),

0 commit comments

Comments
 (0)