authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-01 00:38:27-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-01 20:06:21-08:00
log8f42d34db921fa623715b9524aa3894919dec8bb
tree1748f1f8e228514d0fc1a8e6697e211a921886ea
parent70857ce299941ac18a4509b10c725aba7e15a6b5

Inject test ServerConfig across orchestration and git layers

- add `ServerConfig.layerTest(cwd, stateDir)` to standardize test wiring - require `ServerConfig` in `CodexTextGeneration` attachment path resolution - update orchestration/projection tests and harness to use managed runtimes with explicit config

11 files changed, 586 insertions(+), 624 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+9-13
...@@ -56,6 +56,7 @@ import {...@@ -56,6 +56,7 @@ import {
56 makeTestProviderAdapterHarness,56 makeTestProviderAdapterHarness,
57 type TestProviderAdapterHarness,57 type TestProviderAdapterHarness,
58} from "./TestProviderAdapter.integration.ts";58} from "./TestProviderAdapter.integration.ts";
59import { ServerConfig } from "../src/config.ts";
5960
60function runGit(cwd: string, args: ReadonlyArray<string>) {61function runGit(cwd: string, args: ReadonlyArray<string>) {
61 return execFileSync("git", args, {62 return execFileSync("git", args, {
...@@ -229,19 +230,13 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {...@@ -229,19 +230,13 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {
229 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(230 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
230 Layer.provideMerge(runtimeServicesLayer),231 Layer.provideMerge(runtimeServicesLayer),
231 );232 );
232 const gitCoreLayer = Layer.succeed(233 const gitCoreLayer = Layer.succeed(GitCore, {
233 GitCore,234 renameBranch: (input: Parameters<GitCoreShape["renameBranch"]>[0]) =>
234 {235 Effect.succeed({ branch: input.newBranch }),
235 renameBranch: (input: Parameters<GitCoreShape["renameBranch"]>[0]) =>236 } as unknown as GitCoreShape);
236 Effect.succeed({ branch: input.newBranch }),237 const textGenerationLayer = Layer.succeed(TextGeneration, {
237 } as unknown as GitCoreShape,238 generateBranchName: () => Effect.succeed({ branch: null }),
238 );239 } as unknown as TextGenerationShape);
239 const textGenerationLayer = Layer.succeed(
240 TextGeneration,
241 {
242 generateBranchName: () => Effect.succeed({ branch: null }),
243 } as unknown as TextGenerationShape,
244 );
245 const providerCommandReactorLayer = ProviderCommandReactorLive.pipe(240 const providerCommandReactorLayer = ProviderCommandReactorLive.pipe(
246 Layer.provideMerge(runtimeServicesLayer),241 Layer.provideMerge(runtimeServicesLayer),
247 Layer.provideMerge(gitCoreLayer),242 Layer.provideMerge(gitCoreLayer),
...@@ -257,6 +252,7 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {...@@ -257,6 +252,7 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {
257 );252 );
258 const layer = orchestrationReactorLayer.pipe(253 const layer = orchestrationReactorLayer.pipe(
259 Layer.provide(persistenceLayer),254 Layer.provide(persistenceLayer),
255 Layer.provideMerge(ServerConfig.layerTest(workspaceDir, stateDir)),
260 Layer.provideMerge(NodeServices.layer),256 Layer.provideMerge(NodeServices.layer),
261 );257 );
262258
apps/server/src/config.ts+23-1
...@@ -36,7 +36,29 @@ export interface ServerConfigShape {...@@ -36,7 +36,29 @@ export interface ServerConfigShape {
36 */36 */
37export class ServerConfig extends ServiceMap.Service<ServerConfig, ServerConfigShape>()(37export class ServerConfig extends ServiceMap.Service<ServerConfig, ServerConfigShape>()(
38 "t3/config/ServerConfig",38 "t3/config/ServerConfig",
39) {}39) {
40 static readonly layerTest = (cwd: string, statedir: string) =>
41 Layer.effect(
42 ServerConfig,
43 Effect.gen(function* () {
44 const path = yield* Path.Path;
45 return {
46 cwd,
47 stateDir: statedir,
48 mode: "web",
49 autoBootstrapProjectFromCwd: false,
50 logWebSocketEvents: false,
51 port: 0,
52 host: undefined,
53 authToken: undefined,
54 keybindingsConfigPath: path.join(statedir, "keybindings.json"),
55 staticDir: undefined,
56 devUrl: undefined,
57 noBrowser: false,
58 };
59 }),
60 );
61}
4062
41// Helpers63// Helpers
4264
apps/server/src/git/Layers/CodexTextGeneration.test.ts+47-59
...@@ -3,27 +3,16 @@ import { it } from "@effect/vitest";...@@ -3,27 +3,16 @@ import { it } from "@effect/vitest";
3import { Effect, FileSystem, Layer, Path } from "effect";3import { Effect, FileSystem, Layer, Path } from "effect";
4import { expect } from "vitest";4import { expect } from "vitest";
55
6import { type ServerConfigShape, ServerConfig } from "../../config.ts";6import { ServerConfig } from "../../config.ts";
7import { CodexTextGenerationLive } from "./CodexTextGeneration.ts";7import { CodexTextGenerationLive } from "./CodexTextGeneration.ts";
8import { TextGenerationError } from "../Errors.ts";8import { TextGenerationError } from "../Errors.ts";
9import { TextGeneration } from "../Services/TextGeneration.ts";9import { TextGeneration } from "../Services/TextGeneration.ts";
1010
11function makeTestServerConfig(stateDir: string): ServerConfigShape {11const makeCodexTextGenerationTestLayer = (stateDir: string) =>
12 return {12 CodexTextGenerationLive.pipe(
13 mode: "web",13 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)),
14 port: 0,14 Layer.provideMerge(NodeServices.layer),
15 host: undefined,15 );
16 cwd: process.cwd(),
17 keybindingsConfigPath: "",
18 stateDir,
19 staticDir: undefined,
20 devUrl: undefined,
21 noBrowser: true,
22 authToken: undefined,
23 autoBootstrapProjectFromCwd: false,
24 logWebSocketEvents: false,
25 };
26}
2716
28function makeFakeCodexBinary(dir: string) {17function makeFakeCodexBinary(dir: string) {
29 return Effect.gen(function* () {18 return Effect.gen(function* () {
...@@ -197,10 +186,7 @@ function withFakeCodexEnv<A, E, R>(...@@ -197,10 +186,7 @@ function withFakeCodexEnv<A, E, R>(
197 );186 );
198}187}
199188
200const CodexTextGenerationTestLayer = Layer.provideMerge(189const CodexTextGenerationTestLayer = makeCodexTextGenerationTestLayer(process.cwd());
201 CodexTextGenerationLive,
202 NodeServices.layer,
203);
204190
205it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {191it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
206 it.effect("generates and sanitizes commit messages", () =>192 it.effect("generates and sanitizes commit messages", () =>
...@@ -340,14 +326,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -340,14 +326,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
340 Effect.gen(function* () {326 Effect.gen(function* () {
341 const fs = yield* FileSystem.FileSystem;327 const fs = yield* FileSystem.FileSystem;
342 const path = yield* Path.Path;328 const path = yield* Path.Path;
343 const stateDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-state-textgen-" });329 const uniqueSuffix = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
344 const imagePath = path.join(330 const imageFileName = `message-1-${uniqueSuffix}.png`;
345 stateDir,331 const imagePath = path.join(process.cwd(), "attachments", "thread-1", imageFileName);
346 "attachments",332 yield* fs.makeDirectory(path.join(process.cwd(), "attachments", "thread-1"), {
347 "thread-1",
348 "message-1-0.png",
349 );
350 yield* fs.makeDirectory(path.join(stateDir, "attachments", "thread-1"), {
351 recursive: true,333 recursive: true,
352 });334 });
353 yield* fs.writeFile(imagePath, Buffer.from("hello"));335 yield* fs.writeFile(imagePath, Buffer.from("hello"));
...@@ -363,46 +345,52 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -363,46 +345,52 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
363 name: "bug.png",345 name: "bug.png",
364 mimeType: "image/png",346 mimeType: "image/png",
365 sizeBytes: 5,347 sizeBytes: 5,
366 dataUrl: "/attachments/thread-1/message-1-0.png",348 dataUrl: `/attachments/thread-1/${imageFileName}`,
367 },349 },
368 ],350 ],
369 })351 })
370 .pipe(Effect.provideService(ServerConfig, makeTestServerConfig(stateDir)));352 .pipe(
353 Effect.ensuring(
354 fs.remove(imagePath).pipe(Effect.catch(() => Effect.void)),
355 ),
356 );
371357
372 expect(generated.branch).toBe("fix/ui-regression");358 expect(generated.branch).toBe("fix/ui-regression");
373 }),359 }),
374 ),360 ),
375 );361 );
376362
377 it.effect("fails with typed TextGenerationError when codex returns wrong branch payload shape", () =>363 it.effect(
378 withFakeCodexEnv(364 "fails with typed TextGenerationError when codex returns wrong branch payload shape",
379 {365 () =>
380 output: JSON.stringify({366 withFakeCodexEnv(
381 title: "This is not a branch payload",367 {
368 output: JSON.stringify({
369 title: "This is not a branch payload",
370 }),
371 },
372 Effect.gen(function* () {
373 const textGeneration = yield* TextGeneration;
374
375 const result = yield* textGeneration
376 .generateBranchName({
377 cwd: process.cwd(),
378 message: "Fix websocket reconnect flake",
379 })
380 .pipe(
381 Effect.match({
382 onFailure: (error) => ({ _tag: "Left" as const, left: error }),
383 onSuccess: (value) => ({ _tag: "Right" as const, right: value }),
384 }),
385 );
386
387 expect(result._tag).toBe("Left");
388 if (result._tag === "Left") {
389 expect(result.left).toBeInstanceOf(TextGenerationError);
390 expect(result.left.message).toContain("Codex returned invalid structured output");
391 }
382 }),392 }),
383 },393 ),
384 Effect.gen(function* () {
385 const textGeneration = yield* TextGeneration;
386
387 const result = yield* textGeneration
388 .generateBranchName({
389 cwd: process.cwd(),
390 message: "Fix websocket reconnect flake",
391 })
392 .pipe(
393 Effect.match({
394 onFailure: (error) => ({ _tag: "Left" as const, left: error }),
395 onSuccess: (value) => ({ _tag: "Right" as const, right: value }),
396 }),
397 );
398
399 expect(result._tag).toBe("Left");
400 if (result._tag === "Left") {
401 expect(result.left).toBeInstanceOf(TextGenerationError);
402 expect(result.left.message).toContain("Codex returned invalid structured output");
403 }
404 }),
405 ),
406 );394 );
407395
408 it.effect("returns typed TextGenerationError when codex exits non-zero", () =>396 it.effect("returns typed TextGenerationError when codex exits non-zero", () =>
apps/server/src/git/Layers/CodexTextGeneration.ts+5-7
...@@ -105,6 +105,7 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -105,6 +105,7 @@ const makeCodexTextGeneration = Effect.gen(function* () {
105 const fileSystem = yield* FileSystem.FileSystem;105 const fileSystem = yield* FileSystem.FileSystem;
106 const path = yield* Path.Path;106 const path = yield* Path.Path;
107 const commandSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;107 const commandSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
108 const serverConfig = yield* Effect.service(ServerConfig);
108109
109 const readStreamAsString = <E>(110 const readStreamAsString = <E>(
110 operation: string,111 operation: string,
...@@ -180,7 +181,6 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -180,7 +181,6 @@ const makeCodexTextGeneration = Effect.gen(function* () {
180 if (!attachments || attachments.length === 0) {181 if (!attachments || attachments.length === 0) {
181 return [];182 return [];
182 }183 }
183 const serverConfig = yield* Effect.serviceOption(ServerConfig);
184184
185 const imagePaths: string[] = [];185 const imagePaths: string[] = [];
186 for (const [index, attachment] of attachments.entries()) {186 for (const [index, attachment] of attachments.entries()) {
...@@ -209,12 +209,10 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -209,12 +209,10 @@ const makeCodexTextGeneration = Effect.gen(function* () {
209 continue;209 continue;
210 }210 }
211211
212 const persistedPath = Option.isSome(serverConfig)212 const persistedPath = resolveAttachmentRoutePath({
213 ? resolveAttachmentRoutePath({213 stateDir: serverConfig.stateDir,
214 stateDir: serverConfig.value.stateDir,214 dataUrl: attachment.dataUrl,
215 dataUrl: attachment.dataUrl,215 });
216 })
217 : null;
218 const localPath = persistedPath ?? attachment.dataUrl;216 const localPath = persistedPath ?? attachment.dataUrl;
219 if (!path.isAbsolute(localPath)) {217 if (!path.isAbsolute(localPath)) {
220 continue;218 continue;
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+2
...@@ -37,6 +37,7 @@ import {...@@ -37,6 +37,7 @@ import {
37 type ProviderServiceShape,37 type ProviderServiceShape,
38} from "../../provider/Services/ProviderService.ts";38} from "../../provider/Services/ProviderService.ts";
39import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts";39import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts";
40import { ServerConfig } from "../../config.ts";
4041
41const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);42const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
42const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);43const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
...@@ -238,6 +239,7 @@ describe("CheckpointReactor", () => {...@@ -238,6 +239,7 @@ describe("CheckpointReactor", () => {
238 Layer.provideMerge(orchestrationLayer),239 Layer.provideMerge(orchestrationLayer),
239 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),240 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
240 Layer.provideMerge(CheckpointStoreLive),241 Layer.provideMerge(CheckpointStoreLive),
242 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
241 Layer.provideMerge(NodeServices.layer),243 Layer.provideMerge(NodeServices.layer),
242 );244 );
243245
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+6
...@@ -25,6 +25,8 @@ import {...@@ -25,6 +25,8 @@ import {
25 OrchestrationProjectionPipeline,25 OrchestrationProjectionPipeline,
26 type OrchestrationProjectionPipelineShape,26 type OrchestrationProjectionPipelineShape,
27} from "../Services/ProjectionPipeline.ts";27} from "../Services/ProjectionPipeline.ts";
28import { ServerConfig } from "../../config.ts";
29import * as NodeServices from "@effect/platform-node/NodeServices";
2830
29const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);31const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
30const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);32const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);
...@@ -37,6 +39,8 @@ async function createOrchestrationSystem() {...@@ -37,6 +39,8 @@ async function createOrchestrationSystem() {
37 Layer.provide(OrchestrationEventStoreLive),39 Layer.provide(OrchestrationEventStoreLive),
38 Layer.provide(OrchestrationCommandReceiptRepositoryLive),40 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
39 Layer.provide(SqlitePersistenceMemory),41 Layer.provide(SqlitePersistenceMemory),
42 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
43 Layer.provideMerge(NodeServices.layer),
40 );44 );
41 const runtime = ManagedRuntime.make(orchestrationLayer);45 const runtime = ManagedRuntime.make(orchestrationLayer);
42 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));46 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
...@@ -310,6 +314,8 @@ describe("OrchestrationEngine", () => {...@@ -310,6 +314,8 @@ describe("OrchestrationEngine", () => {
310 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),314 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
311 Layer.provide(OrchestrationCommandReceiptRepositoryLive),315 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
312 Layer.provide(SqlitePersistenceMemory),316 Layer.provide(SqlitePersistenceMemory),
317 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
318 Layer.provideMerge(NodeServices.layer),
313 ),319 ),
314 );320 );
315 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));321 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+404-446
...@@ -12,7 +12,7 @@ import {...@@ -12,7 +12,7 @@ import {
12} from "@t3tools/contracts";12} from "@t3tools/contracts";
13import * as NodeServices from "@effect/platform-node/NodeServices";13import * as NodeServices from "@effect/platform-node/NodeServices";
14import { assert, it } from "@effect/vitest";14import { assert, it } from "@effect/vitest";
15import { Effect, Layer } from "effect";15import { Effect, Layer, ManagedRuntime } from "effect";
16import fs from "node:fs";16import fs from "node:fs";
17import os from "node:os";17import os from "node:os";
18import path from "node:path";18import path from "node:path";
...@@ -32,13 +32,32 @@ import {...@@ -32,13 +32,32 @@ import {
32} from "./ProjectionPipeline.ts";32} from "./ProjectionPipeline.ts";
33import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";33import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
34import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";34import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
35import { ServerConfig, type ServerConfigShape } from "../../config.ts";35import { ServerConfig } from "../../config.ts";
3636
37const projectionLayer = it.layer(37const makeProjectionPipelineTestLayer = (stateDir: string) =>
38 OrchestrationProjectionPipelineLive.pipe(38 OrchestrationProjectionPipelineLive.pipe(
39 Layer.provideMerge(OrchestrationEventStoreLive),39 Layer.provideMerge(OrchestrationEventStoreLive),
40 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)),
40 Layer.provideMerge(SqlitePersistenceMemory),41 Layer.provideMerge(SqlitePersistenceMemory),
41 ),42 Layer.provideMerge(NodeServices.layer),
43 );
44
45const runWithProjectionPipelineLayer = <A, E>(
46 stateDir: string,
47 effect: Effect.Effect<
48 A,
49 E,
50 OrchestrationProjectionPipeline | OrchestrationEventStore | SqlClient.SqlClient
51 >,
52) =>
53 Effect.acquireUseRelease(
54 Effect.sync(() => ManagedRuntime.make(makeProjectionPipelineTestLayer(stateDir))),
55 (runtime) => Effect.promise(() => runtime.runPromise(effect)),
56 (runtime) => Effect.promise(() => runtime.dispose()),
57 );
58
59const projectionLayer = it.layer(
60 makeProjectionPipelineTestLayer(process.cwd()),
42);61);
4362
44projectionLayer("OrchestrationProjectionPipeline", (it) => {63projectionLayer("OrchestrationProjectionPipeline", (it) => {
...@@ -160,137 +179,147 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -160,137 +179,147 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
160 );179 );
161180
162 it.effect("materializes message image attachments into stateDir and stores URL references", () =>181 it.effect("materializes message image attachments into stateDir and stores URL references", () =>
163 Effect.gen(function* () {182 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(
164 const projectionPipeline = yield* OrchestrationProjectionPipeline;183 Effect.flatMap((stateDir) =>
165 const eventStore = yield* OrchestrationEventStore;184 Effect.gen(function* () {
166 const sql = yield* SqlClient.SqlClient;185 const projectionPipeline = yield* OrchestrationProjectionPipeline;
167 const now = new Date().toISOString();186 const eventStore = yield* OrchestrationEventStore;
168 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"));187 const sql = yield* SqlClient.SqlClient;
169188 const now = new Date().toISOString();
170 const serverConfig = {189
171 mode: "web",190 yield* eventStore.append({
172 port: 0,191 type: "thread.message-sent",
173 host: undefined,192 eventId: EventId.makeUnsafe("evt-attachments"),
174 cwd: "/tmp/project-attachments",193 aggregateKind: "thread",
175 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),194 aggregateId: ThreadId.makeUnsafe("thread-attachments"),
176 stateDir,195 occurredAt: now,
177 staticDir: undefined,196 commandId: CommandId.makeUnsafe("cmd-attachments"),
178 devUrl: undefined,197 causationEventId: null,
179 noBrowser: true,198 correlationId: CommandId.makeUnsafe("cmd-attachments"),
180 authToken: undefined,199 metadata: {},
181 autoBootstrapProjectFromCwd: false,200 payload: {
182 logWebSocketEvents: false,201 threadId: ThreadId.makeUnsafe("thread-attachments"),
183 } satisfies ServerConfigShape;202 messageId: MessageId.makeUnsafe("message-attachments"),
184203 role: "user",
185 yield* eventStore.append({204 text: "Inspect this",
186 type: "thread.message-sent",205 attachments: [
187 eventId: EventId.makeUnsafe("evt-attachments"),206 {
188 aggregateKind: "thread",207 type: "image",
189 aggregateId: ThreadId.makeUnsafe("thread-attachments"),208 name: "example.png",
190 occurredAt: now,209 mimeType: "image/png",
191 commandId: CommandId.makeUnsafe("cmd-attachments"),210 sizeBytes: 5,
192 causationEventId: null,211 dataUrl: "data:image/png;base64,SGVsbG8=",
193 correlationId: CommandId.makeUnsafe("cmd-attachments"),212 },
194 metadata: {},213 ],
195 payload: {214 turnId: null,
196 threadId: ThreadId.makeUnsafe("thread-attachments"),215 streaming: false,
197 messageId: MessageId.makeUnsafe("message-attachments"),216 createdAt: now,
198 role: "user",217 updatedAt: now,
199 text: "Inspect this",218 },
200 attachments: [219 });
220
221 yield* projectionPipeline.bootstrap;
222
223 const rows = yield* sql<{
224 readonly attachmentsJson: string | null;
225 }>`
226 SELECT
227 attachments_json AS "attachmentsJson"
228 FROM projection_thread_messages
229 WHERE message_id = 'message-attachments'
230 `;
231 assert.equal(rows.length, 1);
232 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [
201 {233 {
202 type: "image",234 type: "image",
203 name: "example.png",235 name: "example.png",
204 mimeType: "image/png",236 mimeType: "image/png",
205 sizeBytes: 5,237 sizeBytes: 5,
206 dataUrl: "data:image/png;base64,SGVsbG8=",238 dataUrl: "/attachments/thread-attachments/message-attachments-0.png",
207 },239 },
208 ],240 ]);
209 turnId: null,241
210 streaming: false,242 const attachmentPath = path.join(
211 createdAt: now,243 stateDir,
212 updatedAt: now,244 "attachments",
213 },245 "thread-attachments",
214 });246 "message-attachments-0.png",
215247 );
216 yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));248 assert.equal(fs.existsSync(attachmentPath), true);
217249 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("SGVsbG8=", "base64"));
218 const rows = yield* sql<{250 }).pipe(
219 readonly attachmentsJson: string | null;251 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
220 }>`252 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
221 SELECT253 ),
222 attachments_json AS "attachmentsJson"254 ),
223 FROM projection_thread_messages255 ),
224 WHERE message_id = 'message-attachments'
225 `;
226 assert.equal(rows.length, 1);
227 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [
228 {
229 type: "image",
230 name: "example.png",
231 mimeType: "image/png",
232 sizeBytes: 5,
233 dataUrl: "/attachments/thread-attachments/message-attachments-0.png",
234 },
235 ]);
236
237 const attachmentPath = path.join(
238 stateDir,
239 "attachments",
240 "thread-attachments",
241 "message-attachments-0.png",
242 );
243 assert.equal(fs.existsSync(attachmentPath), true);
244 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("SGVsbG8=", "base64"));
245 fs.rmSync(stateDir, { recursive: true, force: true });
246 }),
247 );256 );
248257
249 it.effect("materializes only image data URLs and uses a safe extension whitelist", () =>258 it.effect("materializes only image data URLs and uses a safe extension whitelist", () =>
250 Effect.gen(function* () {259 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(
251 const projectionPipeline = yield* OrchestrationProjectionPipeline;260 Effect.flatMap((stateDir) =>
252 const eventStore = yield* OrchestrationEventStore;261 Effect.gen(function* () {
253 const sql = yield* SqlClient.SqlClient;262 const projectionPipeline = yield* OrchestrationProjectionPipeline;
254 const now = new Date().toISOString();263 const eventStore = yield* OrchestrationEventStore;
255 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"));264 const sql = yield* SqlClient.SqlClient;
256265 const now = new Date().toISOString();
257 const serverConfig = {266
258 mode: "web",267 yield* eventStore.append({
259 port: 0,268 type: "thread.message-sent",
260 host: undefined,269 eventId: EventId.makeUnsafe("evt-attachments-safe"),
261 cwd: "/tmp/project-attachments",270 aggregateKind: "thread",
262 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),271 aggregateId: ThreadId.makeUnsafe("thread-attachments-safe"),
263 stateDir,272 occurredAt: now,
264 staticDir: undefined,273 commandId: CommandId.makeUnsafe("cmd-attachments-safe"),
265 devUrl: undefined,274 causationEventId: null,
266 noBrowser: true,275 correlationId: CommandId.makeUnsafe("cmd-attachments-safe"),
267 authToken: undefined,276 metadata: {},
268 autoBootstrapProjectFromCwd: false,277 payload: {
269 logWebSocketEvents: false,278 threadId: ThreadId.makeUnsafe("thread-attachments-safe"),
270 } satisfies ServerConfigShape;279 messageId: MessageId.makeUnsafe("message-attachments-safe"),
271280 role: "user",
272 yield* eventStore.append({281 text: "Inspect this",
273 type: "thread.message-sent",282 attachments: [
274 eventId: EventId.makeUnsafe("evt-attachments-safe"),283 {
275 aggregateKind: "thread",284 type: "image",
276 aggregateId: ThreadId.makeUnsafe("thread-attachments-safe"),285 name: "untrusted.exe",
277 occurredAt: now,286 mimeType: "image/x-unknown",
278 commandId: CommandId.makeUnsafe("cmd-attachments-safe"),287 sizeBytes: 5,
279 causationEventId: null,288 dataUrl: "data:image/x-unknown;base64,SGVsbG8=",
280 correlationId: CommandId.makeUnsafe("cmd-attachments-safe"),289 },
281 metadata: {},290 {
282 payload: {291 type: "image",
283 threadId: ThreadId.makeUnsafe("thread-attachments-safe"),292 name: "not-image.png",
284 messageId: MessageId.makeUnsafe("message-attachments-safe"),293 mimeType: "image/png",
285 role: "user",294 sizeBytes: 5,
286 text: "Inspect this",295 dataUrl: "data:text/plain;base64,SGVsbG8=",
287 attachments: [296 },
297 ],
298 turnId: null,
299 streaming: false,
300 createdAt: now,
301 updatedAt: now,
302 },
303 });
304
305 yield* projectionPipeline.bootstrap;
306
307 const rows = yield* sql<{
308 readonly attachmentsJson: string | null;
309 }>`
310 SELECT
311 attachments_json AS "attachmentsJson"
312 FROM projection_thread_messages
313 WHERE message_id = 'message-attachments-safe'
314 `;
315 assert.equal(rows.length, 1);
316 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [
288 {317 {
289 type: "image",318 type: "image",
290 name: "untrusted.exe",319 name: "untrusted.exe",
291 mimeType: "image/x-unknown",320 mimeType: "image/x-unknown",
292 sizeBytes: 5,321 sizeBytes: 5,
293 dataUrl: "data:image/x-unknown;base64,SGVsbG8=",322 dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin",
294 },323 },
295 {324 {
296 type: "image",325 type: "image",
...@@ -299,60 +328,30 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -299,60 +328,30 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
299 sizeBytes: 5,328 sizeBytes: 5,
300 dataUrl: "data:text/plain;base64,SGVsbG8=",329 dataUrl: "data:text/plain;base64,SGVsbG8=",
301 },330 },
302 ],331 ]);
303 turnId: null,332
304 streaming: false,333 const firstAttachmentPath = path.join(
305 createdAt: now,334 stateDir,
306 updatedAt: now,335 "attachments",
307 },336 "thread-attachments-safe",
308 });337 "message-attachments-safe-0.bin",
309338 );
310 yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));339 assert.equal(fs.existsSync(firstAttachmentPath), true);
311340 assert.deepEqual(fs.readFileSync(firstAttachmentPath), Buffer.from("SGVsbG8=", "base64"));
312 const rows = yield* sql<{341
313 readonly attachmentsJson: string | null;342 const secondAttachmentPath = path.join(
314 }>`343 stateDir,
315 SELECT344 "attachments",
316 attachments_json AS "attachmentsJson"345 "thread-attachments-safe",
317 FROM projection_thread_messages346 "message-attachments-safe-1.png",
318 WHERE message_id = 'message-attachments-safe'347 );
319 `;348 assert.equal(fs.existsSync(secondAttachmentPath), false);
320 assert.equal(rows.length, 1);349 }).pipe(
321 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [350 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
322 {351 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
323 type: "image",352 ),
324 name: "untrusted.exe",353 ),
325 mimeType: "image/x-unknown",354 ),
326 sizeBytes: 5,
327 dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin",
328 },
329 {
330 type: "image",
331 name: "not-image.png",
332 mimeType: "image/png",
333 sizeBytes: 5,
334 dataUrl: "data:text/plain;base64,SGVsbG8=",
335 },
336 ]);
337
338 const firstAttachmentPath = path.join(
339 stateDir,
340 "attachments",
341 "thread-attachments-safe",
342 "message-attachments-safe-0.bin",
343 );
344 assert.equal(fs.existsSync(firstAttachmentPath), true);
345 assert.deepEqual(fs.readFileSync(firstAttachmentPath), Buffer.from("SGVsbG8=", "base64"));
346
347 const secondAttachmentPath = path.join(
348 stateDir,
349 "attachments",
350 "thread-attachments-safe",
351 "message-attachments-safe-1.png",
352 );
353 assert.equal(fs.existsSync(secondAttachmentPath), false);
354 fs.rmSync(stateDir, { recursive: true, force: true });
355 }),
356 );355 );
357356
358 it.effect(357 it.effect(
...@@ -480,31 +479,17 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -480,31 +479,17 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
480 it.effect(479 it.effect(
481 "overwrites attachment file bytes when a message updates the same attachment index",480 "overwrites attachment file bytes when a message updates the same attachment index",
482 () =>481 () =>
483 Effect.gen(function* () {482 Effect.sync(() =>
484 const projectionPipeline = yield* OrchestrationProjectionPipeline;483 fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-")),
485 const eventStore = yield* OrchestrationEventStore;484 ).pipe(
486 const stateDir = fs.mkdtempSync(485 Effect.flatMap((stateDir) =>
487 path.join(os.tmpdir(), "t3-projection-attachments-overwrite-"),486 Effect.gen(function* () {
488 );487 const projectionPipeline = yield* OrchestrationProjectionPipeline;
489 const now = new Date().toISOString();488 const eventStore = yield* OrchestrationEventStore;
490 const later = new Date(Date.now() + 1_000).toISOString();489 const now = new Date().toISOString();
491490 const later = new Date(Date.now() + 1_000).toISOString();
492 const serverConfig = {491
493 mode: "web",492 yield* eventStore.append({
494 port: 0,
495 host: undefined,
496 cwd: "/tmp/project-attachments",
497 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),
498 stateDir,
499 staticDir: undefined,
500 devUrl: undefined,
501 noBrowser: true,
502 authToken: undefined,
503 autoBootstrapProjectFromCwd: false,
504 logWebSocketEvents: false,
505 } satisfies ServerConfigShape;
506
507 yield* eventStore.append({
508 type: "project.created",493 type: "project.created",
509 eventId: EventId.makeUnsafe("evt-overwrite-1"),494 eventId: EventId.makeUnsafe("evt-overwrite-1"),
510 aggregateKind: "project",495 aggregateKind: "project",
...@@ -609,95 +594,86 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -609,95 +594,86 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
609 },594 },
610 });595 });
611596
612 yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));597 yield* projectionPipeline.bootstrap;
613598
614 const attachmentPath = path.join(599 const attachmentPath = path.join(
615 stateDir,600 stateDir,
616 "attachments",601 "attachments",
617 "thread-overwrite",602 "thread-overwrite",
618 "message-overwrite-0.png",603 "message-overwrite-0.png",
619 );604 );
620 assert.equal(fs.existsSync(attachmentPath), true);605 assert.equal(fs.existsSync(attachmentPath), true);
621 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64"));606 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64"));
622 fs.rmSync(stateDir, { recursive: true, force: true });607 }).pipe(
623 }),608 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
609 Effect.ensuring(
610 Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true })),
611 ),
612 ),
613 ),
614 ),
624 );615 );
625616
626 it.effect("does not persist attachment files when projector transaction rolls back", () =>617 it.effect("does not persist attachment files when projector transaction rolls back", () =>
627 Effect.gen(function* () {618 Effect.sync(() =>
628 const projectionPipeline = yield* OrchestrationProjectionPipeline;619 fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-rollback-")),
629 const eventStore = yield* OrchestrationEventStore;620 ).pipe(
630 const sql = yield* SqlClient.SqlClient;621 Effect.flatMap((stateDir) =>
631 const stateDir = fs.mkdtempSync(622 Effect.gen(function* () {
632 path.join(os.tmpdir(), "t3-projection-attachments-rollback-"),623 const projectionPipeline = yield* OrchestrationProjectionPipeline;
633 );624 const eventStore = yield* OrchestrationEventStore;
634 const now = new Date().toISOString();625 const sql = yield* SqlClient.SqlClient;
635626 const now = new Date().toISOString();
636 const serverConfig = {627
637 mode: "web",628 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
638 port: 0,629 eventStore
639 host: undefined,630 .append(event)
640 cwd: "/tmp/project-attachments",631 .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
641 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),632
642 stateDir,633 yield* appendAndProject({
643 staticDir: undefined,634 type: "project.created",
644 devUrl: undefined,635 eventId: EventId.makeUnsafe("evt-rollback-1"),
645 noBrowser: true,636 aggregateKind: "project",
646 authToken: undefined,637 aggregateId: ProjectId.makeUnsafe("project-rollback"),
647 autoBootstrapProjectFromCwd: false,638 occurredAt: now,
648 logWebSocketEvents: false,639 commandId: CommandId.makeUnsafe("cmd-rollback-1"),
649 } satisfies ServerConfigShape;640 causationEventId: null,
650641 correlationId: CorrelationId.makeUnsafe("cmd-rollback-1"),
651 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>642 metadata: {},
652 eventStore.append(event).pipe(643 payload: {
653 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),644 projectId: ProjectId.makeUnsafe("project-rollback"),
654 Effect.provideService(ServerConfig, serverConfig),645 title: "Project Rollback",
655 );646 workspaceRoot: "/tmp/project-rollback",
656647 defaultModel: null,
657 yield* appendAndProject({648 scripts: [],
658 type: "project.created",649 createdAt: now,
659 eventId: EventId.makeUnsafe("evt-rollback-1"),650 updatedAt: now,
660 aggregateKind: "project",651 },
661 aggregateId: ProjectId.makeUnsafe("project-rollback"),652 });
662 occurredAt: now,653
663 commandId: CommandId.makeUnsafe("cmd-rollback-1"),654 yield* appendAndProject({
664 causationEventId: null,655 type: "thread.created",
665 correlationId: CorrelationId.makeUnsafe("cmd-rollback-1"),656 eventId: EventId.makeUnsafe("evt-rollback-2"),
666 metadata: {},657 aggregateKind: "thread",
667 payload: {658 aggregateId: ThreadId.makeUnsafe("thread-rollback"),
668 projectId: ProjectId.makeUnsafe("project-rollback"),659 occurredAt: now,
669 title: "Project Rollback",660 commandId: CommandId.makeUnsafe("cmd-rollback-2"),
670 workspaceRoot: "/tmp/project-rollback",661 causationEventId: null,
671 defaultModel: null,662 correlationId: CorrelationId.makeUnsafe("cmd-rollback-2"),
672 scripts: [],663 metadata: {},
673 createdAt: now,664 payload: {
674 updatedAt: now,665 threadId: ThreadId.makeUnsafe("thread-rollback"),
675 },666 projectId: ProjectId.makeUnsafe("project-rollback"),
676 });667 title: "Thread Rollback",
677668 model: "gpt-5-codex",
678 yield* appendAndProject({669 branch: null,
679 type: "thread.created",670 worktreePath: null,
680 eventId: EventId.makeUnsafe("evt-rollback-2"),671 createdAt: now,
681 aggregateKind: "thread",672 updatedAt: now,
682 aggregateId: ThreadId.makeUnsafe("thread-rollback"),673 },
683 occurredAt: now,674 });
684 commandId: CommandId.makeUnsafe("cmd-rollback-2"),
685 causationEventId: null,
686 correlationId: CorrelationId.makeUnsafe("cmd-rollback-2"),
687 metadata: {},
688 payload: {
689 threadId: ThreadId.makeUnsafe("thread-rollback"),
690 projectId: ProjectId.makeUnsafe("project-rollback"),
691 title: "Thread Rollback",
692 model: "gpt-5-codex",
693 branch: null,
694 worktreePath: null,
695 createdAt: now,
696 updatedAt: now,
697 },
698 });
699675
700 yield* sql`676 yield* sql`
701 CREATE TRIGGER fail_thread_messages_projection_state_update677 CREATE TRIGGER fail_thread_messages_projection_state_update
702 BEFORE UPDATE ON projection_state678 BEFORE UPDATE ON projection_state
703 WHEN NEW.projector = 'projection.thread-messages'679 WHEN NEW.projector = 'projection.thread-messages'
...@@ -706,90 +682,81 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -706,90 +682,81 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
706 END;682 END;
707 `;683 `;
708684
709 const result = yield* Effect.result(685 const result = yield* Effect.result(
710 appendAndProject({686 appendAndProject({
711 type: "thread.message-sent",687 type: "thread.message-sent",
712 eventId: EventId.makeUnsafe("evt-rollback-3"),688 eventId: EventId.makeUnsafe("evt-rollback-3"),
713 aggregateKind: "thread",689 aggregateKind: "thread",
714 aggregateId: ThreadId.makeUnsafe("thread-rollback"),690 aggregateId: ThreadId.makeUnsafe("thread-rollback"),
715 occurredAt: now,691 occurredAt: now,
716 commandId: CommandId.makeUnsafe("cmd-rollback-3"),692 commandId: CommandId.makeUnsafe("cmd-rollback-3"),
717 causationEventId: null,693 causationEventId: null,
718 correlationId: CorrelationId.makeUnsafe("cmd-rollback-3"),694 correlationId: CorrelationId.makeUnsafe("cmd-rollback-3"),
719 metadata: {},695 metadata: {},
720 payload: {696 payload: {
721 threadId: ThreadId.makeUnsafe("thread-rollback"),697 threadId: ThreadId.makeUnsafe("thread-rollback"),
722 messageId: MessageId.makeUnsafe("message-rollback"),698 messageId: MessageId.makeUnsafe("message-rollback"),
723 role: "user",699 role: "user",
724 text: "Rollback me",700 text: "Rollback me",
725 attachments: [701 attachments: [
726 {702 {
727 type: "image",703 type: "image",
728 name: "rollback.png",704 name: "rollback.png",
729 mimeType: "image/png",705 mimeType: "image/png",
730 sizeBytes: 5,706 sizeBytes: 5,
731 dataUrl: "data:image/png;base64,SGVsbG8=",707 dataUrl: "data:image/png;base64,SGVsbG8=",
708 },
709 ],
710 turnId: null,
711 streaming: false,
712 createdAt: now,
713 updatedAt: now,
732 },714 },
733 ],715 }),
734 turnId: null,716 );
735 streaming: false,717 assert.equal(result._tag, "Failure");
736 createdAt: now,
737 updatedAt: now,
738 },
739 }),
740 );
741 assert.equal(result._tag, "Failure");
742718
743 const rows = yield* sql<{719 const rows = yield* sql<{
744 readonly count: number;720 readonly count: number;
745 }>`721 }>`
746 SELECT COUNT(*) AS "count"722 SELECT COUNT(*) AS "count"
747 FROM projection_thread_messages723 FROM projection_thread_messages
748 WHERE message_id = 'message-rollback'724 WHERE message_id = 'message-rollback'
749 `;725 `;
750 assert.equal(rows[0]?.count ?? 0, 0);726 assert.equal(rows[0]?.count ?? 0, 0);
751727
752 const attachmentPath = path.join(728 const attachmentPath = path.join(
753 stateDir,729 stateDir,
754 "attachments",730 "attachments",
755 "thread-rollback",731 "thread-rollback",
756 "message-rollback-0.png",732 "message-rollback-0.png",
757 );733 );
758 assert.equal(fs.existsSync(attachmentPath), false);734 assert.equal(fs.existsSync(attachmentPath), false);
759 yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`;735 yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`;
760 fs.rmSync(stateDir, { recursive: true, force: true });736 }).pipe(
761 }),737 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
738 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
739 ),
740 ),
741 ),
762 );742 );
763743
764 it.effect("removes unreferenced attachment files when a thread is reverted", () =>744 it.effect("removes unreferenced attachment files when a thread is reverted", () =>
765 Effect.gen(function* () {745 Effect.sync(() =>
766 const projectionPipeline = yield* OrchestrationProjectionPipeline;746 fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-revert-")),
767 const eventStore = yield* OrchestrationEventStore;747 ).pipe(
768 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-revert-"));748 Effect.flatMap((stateDir) =>
769 const now = new Date().toISOString();749 Effect.gen(function* () {
770750 const projectionPipeline = yield* OrchestrationProjectionPipeline;
771 const serverConfig = {751 const eventStore = yield* OrchestrationEventStore;
772 mode: "web",752 const now = new Date().toISOString();
773 port: 0,753
774 host: undefined,754 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
775 cwd: "/tmp/project-attachments",755 eventStore
776 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),756 .append(event)
777 stateDir,757 .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
778 staticDir: undefined,758
779 devUrl: undefined,759 yield* appendAndProject({
780 noBrowser: true,
781 authToken: undefined,
782 autoBootstrapProjectFromCwd: false,
783 logWebSocketEvents: false,
784 } satisfies ServerConfigShape;
785
786 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
787 eventStore.append(event).pipe(
788 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),
789 Effect.provideService(ServerConfig, serverConfig),
790 );
791
792 yield* appendAndProject({
793 type: "project.created",760 type: "project.created",
794 eventId: EventId.makeUnsafe("evt-revert-files-1"),761 eventId: EventId.makeUnsafe("evt-revert-files-1"),
795 aggregateKind: "project",762 aggregateKind: "project",
...@@ -969,41 +936,32 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -969,41 +936,32 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
969 },936 },
970 });937 });
971938
972 assert.equal(fs.existsSync(keepPath), true);939 assert.equal(fs.existsSync(keepPath), true);
973 assert.equal(fs.existsSync(removePath), false);940 assert.equal(fs.existsSync(removePath), false);
974 fs.rmSync(stateDir, { recursive: true, force: true });941 }).pipe(
975 }),942 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
943 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
944 ),
945 ),
946 ),
976 );947 );
977948
978 it.effect("removes thread attachment directory when thread is deleted", () =>949 it.effect("removes thread attachment directory when thread is deleted", () =>
979 Effect.gen(function* () {950 Effect.sync(() =>
980 const projectionPipeline = yield* OrchestrationProjectionPipeline;951 fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-delete-")),
981 const eventStore = yield* OrchestrationEventStore;952 ).pipe(
982 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-delete-"));953 Effect.flatMap((stateDir) =>
983 const now = new Date().toISOString();954 Effect.gen(function* () {
984955 const projectionPipeline = yield* OrchestrationProjectionPipeline;
985 const serverConfig = {956 const eventStore = yield* OrchestrationEventStore;
986 mode: "web",957 const now = new Date().toISOString();
987 port: 0,958
988 host: undefined,959 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
989 cwd: "/tmp/project-attachments",960 eventStore
990 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),961 .append(event)
991 stateDir,962 .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
992 staticDir: undefined,963
993 devUrl: undefined,964 yield* appendAndProject({
994 noBrowser: true,
995 authToken: undefined,
996 autoBootstrapProjectFromCwd: false,
997 logWebSocketEvents: false,
998 } satisfies ServerConfigShape;
999
1000 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
1001 eventStore.append(event).pipe(
1002 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),
1003 Effect.provideService(ServerConfig, serverConfig),
1004 );
1005
1006 yield* appendAndProject({
1007 type: "project.created",965 type: "project.created",
1008 eventId: EventId.makeUnsafe("evt-delete-files-1"),966 eventId: EventId.makeUnsafe("evt-delete-files-1"),
1009 aggregateKind: "project",967 aggregateKind: "project",
...@@ -1096,62 +1054,56 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -1096,62 +1054,56 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
1096 },1054 },
1097 });1055 });
10981056
1099 assert.equal(fs.existsSync(threadAttachmentDir), false);1057 assert.equal(fs.existsSync(threadAttachmentDir), false);
1100 fs.rmSync(stateDir, { recursive: true, force: true });1058 }).pipe(
1101 }),1059 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
1060 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
1061 ),
1062 ),
1063 ),
1102 );1064 );
11031065
1104 it.effect("ignores unsafe thread ids for attachment cleanup paths", () =>1066 it.effect("ignores unsafe thread ids for attachment cleanup paths", () =>
1105 Effect.gen(function* () {1067 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-unsafe-"))).pipe(
1106 const projectionPipeline = yield* OrchestrationProjectionPipeline;1068 Effect.flatMap((stateDir) =>
1107 const eventStore = yield* OrchestrationEventStore;1069 Effect.gen(function* () {
1108 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-unsafe-"));1070 const projectionPipeline = yield* OrchestrationProjectionPipeline;
1109 const now = new Date().toISOString();1071 const eventStore = yield* OrchestrationEventStore;
1110 const attachmentsRootDir = path.join(stateDir, "attachments");1072 const now = new Date().toISOString();
1111 const attachmentsSentinelPath = path.join(attachmentsRootDir, "sentinel.txt");1073 const attachmentsRootDir = path.join(stateDir, "attachments");
1112 const stateDirSentinelPath = path.join(stateDir, "state-sentinel.txt");1074 const attachmentsSentinelPath = path.join(attachmentsRootDir, "sentinel.txt");
1113 fs.mkdirSync(attachmentsRootDir, { recursive: true });1075 const stateDirSentinelPath = path.join(stateDir, "state-sentinel.txt");
1114 fs.writeFileSync(attachmentsSentinelPath, "keep-attachments-root", "utf8");1076 fs.mkdirSync(attachmentsRootDir, { recursive: true });
1115 fs.writeFileSync(stateDirSentinelPath, "keep-state-dir", "utf8");1077 fs.writeFileSync(attachmentsSentinelPath, "keep-attachments-root", "utf8");
11161078 fs.writeFileSync(stateDirSentinelPath, "keep-state-dir", "utf8");
1117 const serverConfig = {1079
1118 mode: "web",1080 yield* eventStore.append({
1119 port: 0,1081 type: "thread.deleted",
1120 host: undefined,1082 eventId: EventId.makeUnsafe("evt-unsafe-thread-delete"),
1121 cwd: "/tmp/project-attachments",1083 aggregateKind: "thread",
1122 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),1084 aggregateId: ThreadId.makeUnsafe(".."),
1123 stateDir,1085 occurredAt: now,
1124 staticDir: undefined,1086 commandId: CommandId.makeUnsafe("cmd-unsafe-thread-delete"),
1125 devUrl: undefined,1087 causationEventId: null,
1126 noBrowser: true,1088 correlationId: CorrelationId.makeUnsafe("cmd-unsafe-thread-delete"),
1127 authToken: undefined,1089 metadata: {},
1128 autoBootstrapProjectFromCwd: false,1090 payload: {
1129 logWebSocketEvents: false,1091 threadId: ThreadId.makeUnsafe(".."),
1130 } satisfies ServerConfigShape;1092 deletedAt: now,
11311093 },
1132 yield* eventStore.append({1094 });
1133 type: "thread.deleted",1095
1134 eventId: EventId.makeUnsafe("evt-unsafe-thread-delete"),1096 yield* projectionPipeline.bootstrap;
1135 aggregateKind: "thread",1097
1136 aggregateId: ThreadId.makeUnsafe(".."),1098 assert.equal(fs.existsSync(attachmentsRootDir), true);
1137 occurredAt: now,1099 assert.equal(fs.existsSync(attachmentsSentinelPath), true);
1138 commandId: CommandId.makeUnsafe("cmd-unsafe-thread-delete"),1100 assert.equal(fs.existsSync(stateDirSentinelPath), true);
1139 causationEventId: null,1101 }).pipe(
1140 correlationId: CorrelationId.makeUnsafe("cmd-unsafe-thread-delete"),1102 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
1141 metadata: {},1103 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
1142 payload: {1104 ),
1143 threadId: ThreadId.makeUnsafe(".."),1105 ),
1144 deletedAt: now,1106 ),
1145 },
1146 });
1147
1148 yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));
1149
1150 assert.equal(fs.existsSync(attachmentsRootDir), true);
1151 assert.equal(fs.existsSync(attachmentsSentinelPath), true);
1152 assert.equal(fs.existsSync(stateDirSentinelPath), true);
1153 fs.rmSync(stateDir, { recursive: true, force: true });
1154 }),
1155 );1107 );
11561108
1157 it.effect("resumes from projector last_applied_sequence without replaying older events", () =>1109 it.effect("resumes from projector last_applied_sequence without replaying older events", () =>
...@@ -1868,7 +1820,11 @@ it.effect("restores pending turn-start metadata across projection pipeline resta...@@ -1868,7 +1820,11 @@ it.effect("restores pending turn-start metadata across projection pipeline resta
1868 ]);1820 ]);
18691821
1870 fs.rmSync(tempDir, { recursive: true, force: true });1822 fs.rmSync(tempDir, { recursive: true, force: true });
1871 }).pipe(Effect.provide(NodeServices.layer)),1823 }).pipe(
1824 Effect.provide(
1825 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd()), NodeServices.layer),
1826 ),
1827 ),
1872);1828);
18731829
1874const engineLayer = it.layer(1830const engineLayer = it.layer(
...@@ -1877,6 +1833,8 @@ const engineLayer = it.layer(...@@ -1877,6 +1833,8 @@ const engineLayer = it.layer(
1877 Layer.provide(OrchestrationEventStoreLive),1833 Layer.provide(OrchestrationEventStoreLive),
1878 Layer.provide(OrchestrationCommandReceiptRepositoryLive),1834 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
1879 Layer.provideMerge(SqlitePersistenceMemory),1835 Layer.provideMerge(SqlitePersistenceMemory),
1836 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
1837 Layer.provideMerge(NodeServices.layer),
1880 ),1838 ),
1881);1839);
18821840
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+32-32
...@@ -96,11 +96,10 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: {...@@ -96,11 +96,10 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: {
96}) {96}) {
97 if (input.attachments.length === 0) return [];97 if (input.attachments.length === 0) return [];
9898
99 const serverConfig = yield* Effect.serviceOption(ServerConfig);99 const serverConfig = yield* Effect.service(ServerConfig);
100 if (Option.isNone(serverConfig)) return input.attachments;
101100
102 const path = yield* Path.Path;101 const path = yield* Path.Path;
103 const attachmentsRootDir = path.join(serverConfig.value.stateDir, "attachments");102 const attachmentsRootDir = path.join(serverConfig.stateDir, "attachments");
104 const threadSegment = toSafeThreadAttachmentSegment(input.threadId);103 const threadSegment = toSafeThreadAttachmentSegment(input.threadId);
105 if (!threadSegment) {104 if (!threadSegment) {
106 yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", {105 yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", {
...@@ -289,20 +288,15 @@ function collectThreadAttachmentRelativePaths(...@@ -289,20 +288,15 @@ function collectThreadAttachmentRelativePaths(
289 return relativePaths;288 return relativePaths;
290}289}
291290
292const runAttachmentSideEffects = Effect.fn(function* (input: {291const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSideEffects) {
293 readonly sideEffects: AttachmentSideEffects;292 const serverConfig = yield* Effect.service(ServerConfig);
294 readonly fileSystem: FileSystem.FileSystem;293 const fileSystem = yield* Effect.service(FileSystem.FileSystem);
295 readonly path: Path.Path;294 const path = yield* Effect.service(Path.Path);
296}) {
297 const serverConfig = yield* Effect.serviceOption(ServerConfig);
298 if (Option.isNone(serverConfig)) {
299 return;
300 }
301295
302 const attachmentsRootDir = input.path.join(serverConfig.value.stateDir, "attachments");296 const attachmentsRootDir = path.join(serverConfig.stateDir, "attachments");
303297
304 yield* Effect.forEach(298 yield* Effect.forEach(
305 input.sideEffects.deletedThreadIds,299 sideEffects.deletedThreadIds,
306 (threadId) =>300 (threadId) =>
307 Effect.gen(function* () {301 Effect.gen(function* () {
308 const threadSegment = toSafeThreadAttachmentSegment(threadId);302 const threadSegment = toSafeThreadAttachmentSegment(threadId);
...@@ -312,16 +306,16 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -312,16 +306,16 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
312 });306 });
313 return;307 return;
314 }308 }
315 const threadDir = input.path.join(attachmentsRootDir, threadSegment);309 const threadDir = path.join(attachmentsRootDir, threadSegment);
316 yield* input.fileSystem.remove(threadDir, { recursive: true, force: true });310 yield* fileSystem.remove(threadDir, { recursive: true, force: true });
317 }),311 }),
318 { concurrency: 1 },312 { concurrency: 1 },
319 );313 );
320314
321 yield* Effect.forEach(315 yield* Effect.forEach(
322 input.sideEffects.prunedThreadRelativePaths.entries(),316 sideEffects.prunedThreadRelativePaths.entries(),
323 ([threadId, keptThreadRelativePaths]) => {317 ([threadId, keptThreadRelativePaths]) => {
324 if (input.sideEffects.deletedThreadIds.has(threadId)) {318 if (sideEffects.deletedThreadIds.has(threadId)) {
325 return Effect.void;319 return Effect.void;
326 }320 }
327 return Effect.gen(function* () {321 return Effect.gen(function* () {
...@@ -330,8 +324,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -330,8 +324,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
330 yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });324 yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });
331 return;325 return;
332 }326 }
333 const threadDir = input.path.join(attachmentsRootDir, threadSegment);327 const threadDir = path.join(attachmentsRootDir, threadSegment);
334 const entries = yield* input.fileSystem328 const entries = yield* fileSystem
335 .readDirectory(threadDir, { recursive: true })329 .readDirectory(threadDir, { recursive: true })
336 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));330 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));
337 yield* Effect.forEach(331 yield* Effect.forEach(
...@@ -343,8 +337,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -343,8 +337,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
343 return;337 return;
344 }338 }
345339
346 const absolutePath = input.path.join(threadDir, threadRelativePath);340 const absolutePath = path.join(threadDir, threadRelativePath);
347 const fileInfo = yield* input.fileSystem341 const fileInfo = yield* fileSystem
348 .stat(absolutePath)342 .stat(absolutePath)
349 .pipe(Effect.catch(() => Effect.succeed(null)));343 .pipe(Effect.catch(() => Effect.succeed(null)));
350 if (!fileInfo || fileInfo.type !== "File") {344 if (!fileInfo || fileInfo.type !== "File") {
...@@ -352,7 +346,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -352,7 +346,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
352 }346 }
353347
354 if (!keptThreadRelativePaths.has(threadRelativePath)) {348 if (!keptThreadRelativePaths.has(threadRelativePath)) {
355 yield* input.fileSystem.remove(absolutePath, { force: true });349 yield* fileSystem.remove(absolutePath, { force: true });
356 }350 }
357 }),351 }),
358 { concurrency: 1 },352 { concurrency: 1 },
...@@ -363,7 +357,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -363,7 +357,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
363 );357 );
364358
365 const writesByPath = new Map<string, Uint8Array>();359 const writesByPath = new Map<string, Uint8Array>();
366 for (const pendingWrite of input.sideEffects.writes) {360 for (const pendingWrite of sideEffects.writes) {
367 // Last write wins for a path so updated attachment contents replace stale bytes.361 // Last write wins for a path so updated attachment contents replace stale bytes.
368 writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes);362 writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes);
369 }363 }
...@@ -371,10 +365,10 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -371,10 +365,10 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
371 writesByPath.entries(),365 writesByPath.entries(),
372 ([absolutePath, bytes]) =>366 ([absolutePath, bytes]) =>
373 Effect.gen(function* () {367 Effect.gen(function* () {
374 yield* input.fileSystem.makeDirectory(input.path.dirname(absolutePath), {368 yield* fileSystem.makeDirectory(path.dirname(absolutePath), {
375 recursive: true,369 recursive: true,
376 });370 });
377 yield* input.fileSystem.writeFile(absolutePath, bytes);371 yield* fileSystem.writeFile(absolutePath, bytes);
378 }),372 }),
379 { concurrency: 1 },373 { concurrency: 1 },
380 );374 );
...@@ -394,6 +388,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -394,6 +388,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
394388
395 const fileSystem = yield* FileSystem.FileSystem;389 const fileSystem = yield* FileSystem.FileSystem;
396 const path = yield* Path.Path;390 const path = yield* Path.Path;
391 const serverConfig = yield* ServerConfig;
397392
398 const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) =>393 const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) =>
399 Effect.gen(function* () {394 Effect.gen(function* () {
...@@ -600,7 +595,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -600,7 +595,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
600 stageFileWrite: (pendingWrite) => {595 stageFileWrite: (pendingWrite) => {
601 attachmentSideEffects.writes.push(pendingWrite);596 attachmentSideEffects.writes.push(pendingWrite);
602 },597 },
603 }).pipe(Effect.provideService(Path.Path, path))598 }).pipe(
599 Effect.provideService(ServerConfig, serverConfig),
600 Effect.provideService(Path.Path, path),
601 )
604 : existingMessage?.attachments;602 : existingMessage?.attachments;
605 yield* projectionThreadMessageRepository.upsert({603 yield* projectionThreadMessageRepository.upsert({
606 messageId: event.payload.messageId,604 messageId: event.payload.messageId,
...@@ -1108,11 +1106,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -1108,11 +1106,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
1108 ),1106 ),
1109 );1107 );
11101108
1111 yield* runAttachmentSideEffects({1109 yield* runAttachmentSideEffects(attachmentSideEffects).pipe(
1112 sideEffects: attachmentSideEffects,
1113 fileSystem,
1114 path,
1115 }).pipe(
1116 Effect.catch((cause) =>1110 Effect.catch((cause) =>
1117 Effect.logWarning("failed to apply projected attachment side-effects", {1111 Effect.logWarning("failed to apply projected attachment side-effects", {
1118 projector: projector.name,1112 projector: projector.name,
...@@ -1144,6 +1138,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -1144,6 +1138,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
1144 Effect.forEach(projectors, (projector) => runProjectorForEvent(projector, event), {1138 Effect.forEach(projectors, (projector) => runProjectorForEvent(projector, event), {
1145 concurrency: 1,1139 concurrency: 1,
1146 }).pipe(1140 }).pipe(
1141 Effect.provideService(FileSystem.FileSystem, fileSystem),
1142 Effect.provideService(Path.Path, path),
1143 Effect.provideService(ServerConfig, serverConfig),
1147 Effect.asVoid,1144 Effect.asVoid,
1148 Effect.catchTag("SqlError", (sqlError) =>1145 Effect.catchTag("SqlError", (sqlError) =>
1149 Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)),1146 Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)),
...@@ -1155,6 +1152,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -1155,6 +1152,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
1155 bootstrapProjector,1152 bootstrapProjector,
1156 { concurrency: 1 },1153 { concurrency: 1 },
1157 ).pipe(1154 ).pipe(
1155 Effect.provideService(FileSystem.FileSystem, fileSystem),
1156 Effect.provideService(Path.Path, path),
1157 Effect.provideService(ServerConfig, serverConfig),
1158 Effect.asVoid,1158 Effect.asVoid,
1159 Effect.tap(() =>1159 Effect.tap(() =>
1160 Effect.log("orchestration projection pipeline bootstrapped").pipe(1160 Effect.log("orchestration projection pipeline bootstrapped").pipe(
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+21-27
...@@ -17,7 +17,7 @@ import {...@@ -17,7 +17,7 @@ import {
17import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";17import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";
18import { afterEach, describe, expect, it, vi } from "vitest";18import { afterEach, describe, expect, it, vi } from "vitest";
1919
20import { type ServerConfigShape, ServerConfig } from "../../config.ts";20import { ServerConfig } from "../../config.ts";
21import { TextGenerationError } from "../../git/Errors.ts";21import { TextGenerationError } from "../../git/Errors.ts";
22import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";22import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
23import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";23import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
...@@ -33,6 +33,7 @@ import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";...@@ -33,6 +33,7 @@ import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
33import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";33import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";
34import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";34import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
35import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";35import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";
36import * as NodeServices from "@effect/platform-node/NodeServices";
3637
37const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);38const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
38const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);39const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
...@@ -42,23 +43,6 @@ const asApprovalRequestId = (value: string): ApprovalRequestId =>...@@ -42,23 +43,6 @@ const asApprovalRequestId = (value: string): ApprovalRequestId =>
42const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);43const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);
43const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);44const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);
4445
45function makeTestServerConfig(stateDir: string): ServerConfigShape {
46 return {
47 mode: "web",
48 port: 0,
49 host: undefined,
50 cwd: process.cwd(),
51 keybindingsConfigPath: "",
52 stateDir,
53 staticDir: undefined,
54 devUrl: undefined,
55 noBrowser: true,
56 authToken: undefined,
57 autoBootstrapProjectFromCwd: false,
58 logWebSocketEvents: false,
59 };
60}
61
62async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {46async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
63 const deadline = Date.now() + timeoutMs;47 const deadline = Date.now() + timeoutMs;
64 const poll = async (): Promise<void> => {48 const poll = async (): Promise<void> => {
...@@ -81,6 +65,7 @@ describe("ProviderCommandReactor", () => {...@@ -81,6 +65,7 @@ describe("ProviderCommandReactor", () => {
81 unknown65 unknown
82 > | null = null;66 > | null = null;
83 let scope: Scope.Closeable | null = null;67 let scope: Scope.Closeable | null = null;
68 const createdStateDirs = new Set<string>();
8469
85 afterEach(async () => {70 afterEach(async () => {
86 if (scope) {71 if (scope) {
...@@ -91,10 +76,16 @@ describe("ProviderCommandReactor", () => {...@@ -91,10 +76,16 @@ describe("ProviderCommandReactor", () => {
91 await runtime.dispose();76 await runtime.dispose();
92 }77 }
93 runtime = null;78 runtime = null;
79 for (const stateDir of createdStateDirs) {
80 fs.rmSync(stateDir, { recursive: true, force: true });
81 }
82 createdStateDirs.clear();
94 });83 });
9584
96 async function createHarness(input?: { readonly stateDir?: string }) {85 async function createHarness(input?: { readonly stateDir?: string }) {
97 const now = new Date().toISOString();86 const now = new Date().toISOString();
87 const stateDir = input?.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-"));
88 createdStateDirs.add(stateDir);
98 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());89 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
99 let nextSessionIndex = 1;90 let nextSessionIndex = 1;
100 const startSession = vi.fn((_: unknown, __: unknown) => {91 const startSession = vi.fn((_: unknown, __: unknown) => {
...@@ -147,21 +138,18 @@ describe("ProviderCommandReactor", () => {...@@ -147,21 +138,18 @@ describe("ProviderCommandReactor", () => {
147 Layer.provide(OrchestrationCommandReceiptRepositoryLive),138 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
148 Layer.provide(SqlitePersistenceMemory),139 Layer.provide(SqlitePersistenceMemory),
149 );140 );
150 const liveLayer = ProviderCommandReactorLive.pipe(141 const layer = ProviderCommandReactorLive.pipe(
151 Layer.provideMerge(orchestrationLayer),142 Layer.provideMerge(orchestrationLayer),
152 Layer.provideMerge(Layer.succeed(ProviderService, service)),143 Layer.provideMerge(Layer.succeed(ProviderService, service)),
153 Layer.provideMerge(Layer.succeed(GitCore, { renameBranch } as unknown as GitCoreShape)),144 Layer.provideMerge(Layer.succeed(GitCore, { renameBranch } as unknown as GitCoreShape)),
154 Layer.provideMerge(145 Layer.provideMerge(
155 Layer.succeed(TextGeneration, { generateBranchName } as unknown as TextGenerationShape),146 Layer.succeed(TextGeneration, { generateBranchName } as unknown as TextGenerationShape),
156 ),147 ),
148 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)),
149 Layer.provideMerge(NodeServices.layer),
157 );150 );
158 const layer =151 const runtime = ManagedRuntime.make(layer);
159 input?.stateDir !== undefined152
160 ? liveLayer.pipe(
161 Layer.provideMerge(Layer.succeed(ServerConfig, makeTestServerConfig(input.stateDir))),
162 )
163 : liveLayer;
164 runtime = ManagedRuntime.make(layer);
165 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));153 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
166 const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor));154 const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor));
167 scope = await Effect.runPromise(Scope.make("sequential"));155 scope = await Effect.runPromise(Scope.make("sequential"));
...@@ -202,6 +190,7 @@ describe("ProviderCommandReactor", () => {...@@ -202,6 +190,7 @@ describe("ProviderCommandReactor", () => {
202 stopSession,190 stopSession,
203 renameBranch,191 renameBranch,
204 generateBranchName,192 generateBranchName,
193 stateDir,
205 };194 };
206 }195 }
207196
...@@ -294,7 +283,12 @@ describe("ProviderCommandReactor", () => {...@@ -294,7 +283,12 @@ describe("ProviderCommandReactor", () => {
294 name: "bug.png",283 name: "bug.png",
295 mimeType: "image/png",284 mimeType: "image/png",
296 sizeBytes: 5,285 sizeBytes: 5,
297 dataUrl: "data:image/png;base64,SGVsbG8=",286 dataUrl: path.join(
287 harness.stateDir,
288 "attachments",
289 "thread-1",
290 "user-message-worktree-rename-0.png",
291 ),
298 },292 },
299 ],293 ],
300 });294 });
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+4-9
...@@ -111,7 +111,7 @@ const make = Effect.gen(function* () {...@@ -111,7 +111,7 @@ const make = Effect.gen(function* () {
111 const providerService = yield* ProviderService;111 const providerService = yield* ProviderService;
112 const git = yield* GitCore;112 const git = yield* GitCore;
113 const textGeneration = yield* TextGeneration;113 const textGeneration = yield* TextGeneration;
114 const serverConfig = yield* Effect.serviceOption(ServerConfig);114 const serverConfig = yield* Effect.service(ServerConfig);
115 const handledTurnStartKeys = yield* Cache.make<string, true>({115 const handledTurnStartKeys = yield* Cache.make<string, true>({
116 capacity: HANDLED_TURN_START_KEY_MAX,116 capacity: HANDLED_TURN_START_KEY_MAX,
117 timeToLive: HANDLED_TURN_START_KEY_TTL,117 timeToLive: HANDLED_TURN_START_KEY_TTL,
...@@ -301,20 +301,15 @@ const make = Effect.gen(function* () {...@@ -301,20 +301,15 @@ const make = Effect.gen(function* () {
301 return input.attachments ?? [];301 return input.attachments ?? [];
302 }302 }
303303
304 if (Option.isNone(serverConfig)) {
305 return input.attachments;
306 }
307 const threadSegment = encodeURIComponent(input.threadId);304 const threadSegment = encodeURIComponent(input.threadId);
308 const messageSegment = encodeURIComponent(input.messageId);305 const messageSegment = encodeURIComponent(input.messageId);
309306
310 return yield* Effect.forEach(Array.from(input.attachments.entries()), ([index, attachment]) =>307 return yield* Effect.forEach(Array.from(input.attachments.entries()), ([index, attachment]) =>
311 Effect.gen(function* () {308 Effect.gen(function* () {
312 if (attachment.type !== "image") {309 if (attachment.type !== "image") return attachment;
313 return attachment;
314 }
315310
316 const resolvedRoutePath = resolveAttachmentRoutePath({311 const resolvedRoutePath = resolveAttachmentRoutePath({
317 stateDir: serverConfig.value.stateDir,312 stateDir: serverConfig.stateDir,
318 dataUrl: attachment.dataUrl,313 dataUrl: attachment.dataUrl,
319 });314 });
320 const resolvedMaterializedPath =315 const resolvedMaterializedPath =
...@@ -329,7 +324,7 @@ const make = Effect.gen(function* () {...@@ -329,7 +324,7 @@ const make = Effect.gen(function* () {
329 fileName: attachment.name,324 fileName: attachment.name,
330 });325 });
331 return resolveAttachmentRelativePath({326 return resolveAttachmentRelativePath({
332 stateDir: serverConfig.value.stateDir,327 stateDir: serverConfig.stateDir,
333 relativePath: `${threadSegment}/${messageSegment}-${index}${extension}`,328 relativePath: `${threadSegment}/${messageSegment}-${index}${extension}`,
334 });329 });
335 })();330 })();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+33-30
...@@ -28,11 +28,12 @@ import {...@@ -28,11 +28,12 @@ import {
28 type OrchestrationEngineShape,28 type OrchestrationEngineShape,
29} from "../Services/OrchestrationEngine.ts";29} from "../Services/OrchestrationEngine.ts";
30import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";30import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts";
31import { ServerConfig } from "../../config.ts";
32import * as NodeServices from "@effect/platform-node/NodeServices";
3133
32const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);34const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
33const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);35const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
34const asProviderThreadId = (value: string): ProviderThreadId =>36const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value);
35 ProviderThreadId.makeUnsafe(value);
36const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);37const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
37const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);38const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);
38const asEventId = (value: string): EventId => EventId.makeUnsafe(value);39const asEventId = (value: string): EventId => EventId.makeUnsafe(value);
...@@ -122,6 +123,8 @@ describe("ProviderRuntimeIngestion", () => {...@@ -122,6 +123,8 @@ describe("ProviderRuntimeIngestion", () => {
122 const layer = ProviderRuntimeIngestionLive.pipe(123 const layer = ProviderRuntimeIngestionLive.pipe(
123 Layer.provideMerge(orchestrationLayer),124 Layer.provideMerge(orchestrationLayer),
124 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),125 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
126 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
127 Layer.provideMerge(NodeServices.layer),
125 );128 );
126 runtime = ManagedRuntime.make(layer);129 runtime = ManagedRuntime.make(layer);
127 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));130 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
...@@ -255,7 +258,9 @@ describe("ProviderRuntimeIngestion", () => {...@@ -255,7 +258,9 @@ describe("ProviderRuntimeIngestion", () => {
255258
256 await Effect.runPromise(Effect.sleep("40 millis"));259 await Effect.runPromise(Effect.sleep("40 millis"));
257 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());260 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
258 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));261 const midThread = midReadModel.threads.find(
262 (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),
263 );
259 expect(midThread?.session?.status).toBe("running");264 expect(midThread?.session?.status).toBe("running");
260 expect(midThread?.session?.activeTurnId).toBe("turn-primary");265 expect(midThread?.session?.activeTurnId).toBe("turn-primary");
261266
...@@ -308,7 +313,9 @@ describe("ProviderRuntimeIngestion", () => {...@@ -308,7 +313,9 @@ describe("ProviderRuntimeIngestion", () => {
308313
309 await Effect.runPromise(Effect.sleep("40 millis"));314 await Effect.runPromise(Effect.sleep("40 millis"));
310 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());315 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
311 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));316 const midThread = midReadModel.threads.find(
317 (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),
318 );
312 expect(midThread?.session?.status).toBe("running");319 expect(midThread?.session?.status).toBe("running");
313 expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main");320 expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main");
314321
...@@ -401,7 +408,9 @@ describe("ProviderRuntimeIngestion", () => {...@@ -401,7 +408,9 @@ describe("ProviderRuntimeIngestion", () => {
401408
402 await Effect.runPromise(Effect.sleep("30 millis"));409 await Effect.runPromise(Effect.sleep("30 millis"));
403 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());410 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
404 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));411 const midThread = midReadModel.threads.find(
412 (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),
413 );
405 expect(midThread?.messages.some((message) => message.id === "assistant:item-buffered")).toBe(414 expect(midThread?.messages.some((message) => message.id === "assistant:item-buffered")).toBe(
406 false,415 false,
407 );416 );
...@@ -416,10 +425,10 @@ describe("ProviderRuntimeIngestion", () => {...@@ -416,10 +425,10 @@ describe("ProviderRuntimeIngestion", () => {
416 itemId: asItemId("item-buffered"),425 itemId: asItemId("item-buffered"),
417 });426 });
418427
419 const thread = await waitForThread(428 const thread = await waitForThread(harness.engine, (entry) =>
420 harness.engine,429 entry.messages.some(
421 (entry) =>430 (message) => message.id === "assistant:item-buffered" && !message.streaming,
422 entry.messages.some((message) => message.id === "assistant:item-buffered" && !message.streaming),431 ),
423 );432 );
424 const message = thread.messages.find((entry) => entry.id === "assistant:item-buffered");433 const message = thread.messages.find((entry) => entry.id === "assistant:item-buffered");
425 expect(message?.text).toBe("buffer me");434 expect(message?.text).toBe("buffer me");
...@@ -475,15 +484,13 @@ describe("ProviderRuntimeIngestion", () => {...@@ -475,15 +484,13 @@ describe("ProviderRuntimeIngestion", () => {
475 delta: "hello live",484 delta: "hello live",
476 });485 });
477486
478 const liveThread = await waitForThread(487 const liveThread = await waitForThread(harness.engine, (entry) =>
479 harness.engine,488 entry.messages.some(
480 (entry) =>489 (message) =>
481 entry.messages.some(490 message.id === "assistant:item-streaming-mode" &&
482 (message) =>491 message.streaming &&
483 message.id === "assistant:item-streaming-mode" &&492 message.text === "hello live",
484 message.streaming &&493 ),
485 message.text === "hello live",
486 ),
487 );494 );
488 const liveMessage = liveThread.messages.find(495 const liveMessage = liveThread.messages.find(
489 (entry) => entry.id === "assistant:item-streaming-mode",496 (entry) => entry.id === "assistant:item-streaming-mode",
...@@ -500,12 +507,10 @@ describe("ProviderRuntimeIngestion", () => {...@@ -500,12 +507,10 @@ describe("ProviderRuntimeIngestion", () => {
500 itemId: asItemId("item-streaming-mode"),507 itemId: asItemId("item-streaming-mode"),
501 });508 });
502509
503 const finalThread = await waitForThread(510 const finalThread = await waitForThread(harness.engine, (entry) =>
504 harness.engine,511 entry.messages.some(
505 (entry) =>512 (message) => message.id === "assistant:item-streaming-mode" && !message.streaming,
506 entry.messages.some(513 ),
507 (message) => message.id === "assistant:item-streaming-mode" && !message.streaming,
508 ),
509 );514 );
510 const finalMessage = finalThread.messages.find(515 const finalMessage = finalThread.messages.find(
511 (entry) => entry.id === "assistant:item-streaming-mode",516 (entry) => entry.id === "assistant:item-streaming-mode",
...@@ -554,12 +559,10 @@ describe("ProviderRuntimeIngestion", () => {...@@ -554,12 +559,10 @@ describe("ProviderRuntimeIngestion", () => {
554 itemId: asItemId("item-buffer-spill"),559 itemId: asItemId("item-buffer-spill"),
555 });560 });
556561
557 const thread = await waitForThread(562 const thread = await waitForThread(harness.engine, (entry) =>
558 harness.engine,563 entry.messages.some(
559 (entry) =>564 (message) => message.id === "assistant:item-buffer-spill" && !message.streaming,
560 entry.messages.some(565 ),
561 (message) => message.id === "assistant:item-buffer-spill" && !message.streaming,
562 ),
563 );566 );
564 const message = thread.messages.find((entry) => entry.id === "assistant:item-buffer-spill");567 const message = thread.messages.find((entry) => entry.id === "assistant:item-buffer-spill");
565 expect(message?.text.length).toBe(oversizedText.length);568 expect(message?.text.length).toBe(oversizedText.length);