authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-27 21:27:19-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-27 21:30:46-08:00
logfb45896588afb89b3151e2432682f39e98bfb51f
treea2cdd86e52f03f636fc80db89ef5c1178229977d
parentb52518f85fa3f11a6a405e3bebaf908ebdc5add7

simplify layers

8 files changed, 276 insertions(+), 187 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+1-2
...@@ -215,14 +215,13 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {...@@ -215,14 +215,13 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {
215 Layer.provide(providerSessionDirectoryLayer),215 Layer.provide(providerSessionDirectoryLayer),
216 Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),216 Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),
217 );217 );
218 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));
219218
220 const runtimeServicesLayer = Layer.mergeAll(219 const runtimeServicesLayer = Layer.mergeAll(
221 orchestrationLayer,220 orchestrationLayer,
222 OrchestrationProjectionSnapshotQueryLive,221 OrchestrationProjectionSnapshotQueryLive,
223 ProjectionCheckpointRepositoryLive,222 ProjectionCheckpointRepositoryLive,
224 ProjectionPendingApprovalRepositoryLive,223 ProjectionPendingApprovalRepositoryLive,
225 checkpointStoreLayer,224 CheckpointStoreLive,
226 providerLayer,225 providerLayer,
227 );226 );
228 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(227 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+6-3
...@@ -233,11 +233,12 @@ describe("CheckpointReactor", () => {...@@ -233,11 +233,12 @@ describe("CheckpointReactor", () => {
233 Layer.provide(OrchestrationCommandReceiptRepositoryLive),233 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
234 Layer.provide(SqlitePersistenceMemory),234 Layer.provide(SqlitePersistenceMemory),
235 );235 );
236 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));236
237 const layer = CheckpointReactorLive.pipe(237 const layer = CheckpointReactorLive.pipe(
238 Layer.provideMerge(orchestrationLayer),238 Layer.provideMerge(orchestrationLayer),
239 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),239 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
240 Layer.provideMerge(checkpointStoreLayer),240 Layer.provideMerge(CheckpointStoreLive),
241 Layer.provideMerge(NodeServices.layer),
241 );242 );
242243
243 runtime = ManagedRuntime.make(layer);244 runtime = ManagedRuntime.make(layer);
...@@ -437,7 +438,9 @@ describe("CheckpointReactor", () => {...@@ -437,7 +438,9 @@ describe("CheckpointReactor", () => {
437438
438 await Effect.runPromise(Effect.sleep("40 millis"));439 await Effect.runPromise(Effect.sleep("40 millis"));
439 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());440 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
440 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));441 const midThread = midReadModel.threads.find(
442 (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),
443 );
441 expect(midThread?.checkpoints).toHaveLength(0);444 expect(midThread?.checkpoints).toHaveLength(0);
442445
443 harness.provider.emit({446 harness.provider.emit({
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+206-153
...@@ -477,146 +477,150 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -477,146 +477,150 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
477 }),477 }),
478 );478 );
479479
480 it.effect("overwrites attachment file bytes when a message updates the same attachment index", () =>480 it.effect(
481 Effect.gen(function* () {481 "overwrites attachment file bytes when a message updates the same attachment index",
482 const projectionPipeline = yield* OrchestrationProjectionPipeline;482 () =>
483 const eventStore = yield* OrchestrationEventStore;483 Effect.gen(function* () {
484 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-"));484 const projectionPipeline = yield* OrchestrationProjectionPipeline;
485 const now = new Date().toISOString();485 const eventStore = yield* OrchestrationEventStore;
486 const later = new Date(Date.now() + 1_000).toISOString();486 const stateDir = fs.mkdtempSync(
487 path.join(os.tmpdir(), "t3-projection-attachments-overwrite-"),
488 );
489 const now = new Date().toISOString();
490 const later = new Date(Date.now() + 1_000).toISOString();
487491
488 const serverConfig = {492 const serverConfig = {
489 mode: "web",493 mode: "web",
490 port: 0,494 port: 0,
491 host: undefined,495 host: undefined,
492 cwd: "/tmp/project-attachments",496 cwd: "/tmp/project-attachments",
493 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),497 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),
494 stateDir,498 stateDir,
495 staticDir: undefined,499 staticDir: undefined,
496 devUrl: undefined,500 devUrl: undefined,
497 noBrowser: true,501 noBrowser: true,
498 authToken: undefined,502 authToken: undefined,
499 autoBootstrapProjectFromCwd: false,503 autoBootstrapProjectFromCwd: false,
500 logWebSocketEvents: false,504 logWebSocketEvents: false,
501 } satisfies ServerConfigShape;505 } satisfies ServerConfigShape;
502506
503 yield* eventStore.append({507 yield* eventStore.append({
504 type: "project.created",508 type: "project.created",
505 eventId: EventId.makeUnsafe("evt-overwrite-1"),509 eventId: EventId.makeUnsafe("evt-overwrite-1"),
506 aggregateKind: "project",510 aggregateKind: "project",
507 aggregateId: ProjectId.makeUnsafe("project-overwrite"),511 aggregateId: ProjectId.makeUnsafe("project-overwrite"),
508 occurredAt: now,512 occurredAt: now,
509 commandId: CommandId.makeUnsafe("cmd-overwrite-1"),513 commandId: CommandId.makeUnsafe("cmd-overwrite-1"),
510 causationEventId: null,514 causationEventId: null,
511 correlationId: CommandId.makeUnsafe("cmd-overwrite-1"),515 correlationId: CommandId.makeUnsafe("cmd-overwrite-1"),
512 metadata: {},516 metadata: {},
513 payload: {517 payload: {
514 projectId: ProjectId.makeUnsafe("project-overwrite"),518 projectId: ProjectId.makeUnsafe("project-overwrite"),
515 title: "Project Overwrite",519 title: "Project Overwrite",
516 workspaceRoot: "/tmp/project-overwrite",520 workspaceRoot: "/tmp/project-overwrite",
517 defaultModel: null,521 defaultModel: null,
518 scripts: [],522 scripts: [],
519 createdAt: now,523 createdAt: now,
520 updatedAt: now,524 updatedAt: now,
521 },525 },
522 });526 });
523
524 yield* eventStore.append({
525 type: "thread.created",
526 eventId: EventId.makeUnsafe("evt-overwrite-2"),
527 aggregateKind: "thread",
528 aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
529 occurredAt: now,
530 commandId: CommandId.makeUnsafe("cmd-overwrite-2"),
531 causationEventId: null,
532 correlationId: CommandId.makeUnsafe("cmd-overwrite-2"),
533 metadata: {},
534 payload: {
535 threadId: ThreadId.makeUnsafe("thread-overwrite"),
536 projectId: ProjectId.makeUnsafe("project-overwrite"),
537 title: "Thread Overwrite",
538 model: "gpt-5-codex",
539 branch: null,
540 worktreePath: null,
541 createdAt: now,
542 updatedAt: now,
543 },
544 });
545527
546 yield* eventStore.append({528 yield* eventStore.append({
547 type: "thread.message-sent",529 type: "thread.created",
548 eventId: EventId.makeUnsafe("evt-overwrite-3"),530 eventId: EventId.makeUnsafe("evt-overwrite-2"),
549 aggregateKind: "thread",531 aggregateKind: "thread",
550 aggregateId: ThreadId.makeUnsafe("thread-overwrite"),532 aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
551 occurredAt: now,533 occurredAt: now,
552 commandId: CommandId.makeUnsafe("cmd-overwrite-3"),534 commandId: CommandId.makeUnsafe("cmd-overwrite-2"),
553 causationEventId: null,535 causationEventId: null,
554 correlationId: CommandId.makeUnsafe("cmd-overwrite-3"),536 correlationId: CommandId.makeUnsafe("cmd-overwrite-2"),
555 metadata: {},537 metadata: {},
556 payload: {538 payload: {
557 threadId: ThreadId.makeUnsafe("thread-overwrite"),539 threadId: ThreadId.makeUnsafe("thread-overwrite"),
558 messageId: MessageId.makeUnsafe("message-overwrite"),540 projectId: ProjectId.makeUnsafe("project-overwrite"),
559 role: "user",541 title: "Thread Overwrite",
560 text: "first image",542 model: "gpt-5-codex",
561 attachments: [543 branch: null,
562 {544 worktreePath: null,
563 type: "image",545 createdAt: now,
564 name: "file.png",546 updatedAt: now,
565 mimeType: "image/png",547 },
566 sizeBytes: 5,548 });
567 dataUrl: "data:image/png;base64,SGVsbG8=",
568 },
569 ],
570 turnId: null,
571 streaming: false,
572 createdAt: now,
573 updatedAt: now,
574 },
575 });
576549
577 yield* eventStore.append({550 yield* eventStore.append({
578 type: "thread.message-sent",551 type: "thread.message-sent",
579 eventId: EventId.makeUnsafe("evt-overwrite-4"),552 eventId: EventId.makeUnsafe("evt-overwrite-3"),
580 aggregateKind: "thread",553 aggregateKind: "thread",
581 aggregateId: ThreadId.makeUnsafe("thread-overwrite"),554 aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
582 occurredAt: later,555 occurredAt: now,
583 commandId: CommandId.makeUnsafe("cmd-overwrite-4"),556 commandId: CommandId.makeUnsafe("cmd-overwrite-3"),
584 causationEventId: null,557 causationEventId: null,
585 correlationId: CommandId.makeUnsafe("cmd-overwrite-4"),558 correlationId: CommandId.makeUnsafe("cmd-overwrite-3"),
586 metadata: {},559 metadata: {},
587 payload: {560 payload: {
588 threadId: ThreadId.makeUnsafe("thread-overwrite"),561 threadId: ThreadId.makeUnsafe("thread-overwrite"),
589 messageId: MessageId.makeUnsafe("message-overwrite"),562 messageId: MessageId.makeUnsafe("message-overwrite"),
590 role: "user",563 role: "user",
591 text: "",564 text: "first image",
592 attachments: [565 attachments: [
593 {566 {
594 type: "image",567 type: "image",
595 name: "file.png",568 name: "file.png",
596 mimeType: "image/png",569 mimeType: "image/png",
597 sizeBytes: 5,570 sizeBytes: 5,
598 dataUrl: "data:image/png;base64,V29ybGQ=",571 dataUrl: "data:image/png;base64,SGVsbG8=",
599 },572 },
600 ],573 ],
601 turnId: null,574 turnId: null,
602 streaming: false,575 streaming: false,
603 createdAt: now,576 createdAt: now,
604 updatedAt: later,577 updatedAt: now,
605 },578 },
606 });579 });
607580
608 yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));581 yield* eventStore.append({
582 type: "thread.message-sent",
583 eventId: EventId.makeUnsafe("evt-overwrite-4"),
584 aggregateKind: "thread",
585 aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
586 occurredAt: later,
587 commandId: CommandId.makeUnsafe("cmd-overwrite-4"),
588 causationEventId: null,
589 correlationId: CommandId.makeUnsafe("cmd-overwrite-4"),
590 metadata: {},
591 payload: {
592 threadId: ThreadId.makeUnsafe("thread-overwrite"),
593 messageId: MessageId.makeUnsafe("message-overwrite"),
594 role: "user",
595 text: "",
596 attachments: [
597 {
598 type: "image",
599 name: "file.png",
600 mimeType: "image/png",
601 sizeBytes: 5,
602 dataUrl: "data:image/png;base64,V29ybGQ=",
603 },
604 ],
605 turnId: null,
606 streaming: false,
607 createdAt: now,
608 updatedAt: later,
609 },
610 });
609611
610 const attachmentPath = path.join(612 yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));
611 stateDir,613
612 "attachments",614 const attachmentPath = path.join(
613 "thread-overwrite",615 stateDir,
614 "message-overwrite-0.png",616 "attachments",
615 );617 "thread-overwrite",
616 assert.equal(fs.existsSync(attachmentPath), true);618 "message-overwrite-0.png",
617 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64"));619 );
618 fs.rmSync(stateDir, { recursive: true, force: true });620 assert.equal(fs.existsSync(attachmentPath), true);
619 }),621 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64"));
622 fs.rmSync(stateDir, { recursive: true, force: true });
623 }),
620 );624 );
621625
622 it.effect("does not persist attachment files when projector transaction rolls back", () =>626 it.effect("does not persist attachment files when projector transaction rolls back", () =>
...@@ -624,7 +628,9 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -624,7 +628,9 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
624 const projectionPipeline = yield* OrchestrationProjectionPipeline;628 const projectionPipeline = yield* OrchestrationProjectionPipeline;
625 const eventStore = yield* OrchestrationEventStore;629 const eventStore = yield* OrchestrationEventStore;
626 const sql = yield* SqlClient.SqlClient;630 const sql = yield* SqlClient.SqlClient;
627 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-rollback-"));631 const stateDir = fs.mkdtempSync(
632 path.join(os.tmpdir(), "t3-projection-attachments-rollback-"),
633 );
628 const now = new Date().toISOString();634 const now = new Date().toISOString();
629635
630 const serverConfig = {636 const serverConfig = {
...@@ -643,12 +649,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -643,12 +649,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
643 } satisfies ServerConfigShape;649 } satisfies ServerConfigShape;
644650
645 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>651 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
646 eventStore652 eventStore.append(event).pipe(
647 .append(event)653 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),
648 .pipe(654 Effect.provideService(ServerConfig, serverConfig),
649 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),655 );
650 Effect.provideService(ServerConfig, serverConfig),
651 );
652656
653 yield* appendAndProject({657 yield* appendAndProject({
654 type: "project.created",658 type: "project.created",
...@@ -780,12 +784,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -780,12 +784,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
780 } satisfies ServerConfigShape;784 } satisfies ServerConfigShape;
781785
782 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>786 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
783 eventStore787 eventStore.append(event).pipe(
784 .append(event)788 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),
785 .pipe(789 Effect.provideService(ServerConfig, serverConfig),
786 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),790 );
787 Effect.provideService(ServerConfig, serverConfig),
788 );
789791
790 yield* appendAndProject({792 yield* appendAndProject({
791 type: "project.created",793 type: "project.created",
...@@ -996,12 +998,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -996,12 +998,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
996 } satisfies ServerConfigShape;998 } satisfies ServerConfigShape;
997999
998 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>1000 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
999 eventStore1001 eventStore.append(event).pipe(
1000 .append(event)1002 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),
1001 .pipe(1003 Effect.provideService(ServerConfig, serverConfig),
1002 Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)),1004 );
1003 Effect.provideService(ServerConfig, serverConfig),
1004 );
10051005
1006 yield* appendAndProject({1006 yield* appendAndProject({
1007 type: "project.created",1007 type: "project.created",
...@@ -1101,6 +1101,59 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -1101,6 +1101,59 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
1101 }),1101 }),
1102 );1102 );
11031103
1104 it.effect("ignores unsafe thread ids for attachment cleanup paths", () =>
1105 Effect.gen(function* () {
1106 const projectionPipeline = yield* OrchestrationProjectionPipeline;
1107 const eventStore = yield* OrchestrationEventStore;
1108 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-unsafe-"));
1109 const now = new Date().toISOString();
1110 const attachmentsRootDir = path.join(stateDir, "attachments");
1111 const attachmentsSentinelPath = path.join(attachmentsRootDir, "sentinel.txt");
1112 const stateDirSentinelPath = path.join(stateDir, "state-sentinel.txt");
1113 fs.mkdirSync(attachmentsRootDir, { recursive: true });
1114 fs.writeFileSync(attachmentsSentinelPath, "keep-attachments-root", "utf8");
1115 fs.writeFileSync(stateDirSentinelPath, "keep-state-dir", "utf8");
1116
1117 const serverConfig = {
1118 mode: "web",
1119 port: 0,
1120 host: undefined,
1121 cwd: "/tmp/project-attachments",
1122 keybindingsConfigPath: path.join(stateDir, "keybindings.json"),
1123 stateDir,
1124 staticDir: undefined,
1125 devUrl: undefined,
1126 noBrowser: true,
1127 authToken: undefined,
1128 autoBootstrapProjectFromCwd: false,
1129 logWebSocketEvents: false,
1130 } satisfies ServerConfigShape;
1131
1132 yield* eventStore.append({
1133 type: "thread.deleted",
1134 eventId: EventId.makeUnsafe("evt-unsafe-thread-delete"),
1135 aggregateKind: "thread",
1136 aggregateId: ThreadId.makeUnsafe(".."),
1137 occurredAt: now,
1138 commandId: CommandId.makeUnsafe("cmd-unsafe-thread-delete"),
1139 causationEventId: null,
1140 correlationId: CorrelationId.makeUnsafe("cmd-unsafe-thread-delete"),
1141 metadata: {},
1142 payload: {
1143 threadId: ThreadId.makeUnsafe(".."),
1144 deletedAt: now,
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 );
1156
1104 it.effect("resumes from projector last_applied_sequence without replaying older events", () =>1157 it.effect("resumes from projector last_applied_sequence without replaying older events", () =>
1105 Effect.gen(function* () {1158 Effect.gen(function* () {
1106 const projectionPipeline = yield* OrchestrationProjectionPipeline;1159 const projectionPipeline = yield* OrchestrationProjectionPipeline;
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+47-17
...@@ -77,6 +77,21 @@ interface AttachmentSideEffects {...@@ -77,6 +77,21 @@ interface AttachmentSideEffects {
77 readonly prunedThreadRelativePaths: Map<string, Set<string>>;77 readonly prunedThreadRelativePaths: Map<string, Set<string>>;
78}78}
7979
80function toSafeThreadAttachmentSegment(threadId: string): string | null {
81 const segment = encodeURIComponent(threadId);
82 if (
83 segment.length === 0 ||
84 segment === "." ||
85 segment === ".." ||
86 segment.includes("/") ||
87 segment.includes("\\") ||
88 segment.includes("\0")
89 ) {
90 return null;
91 }
92 return segment;
93}
94
80function parseBase64DataUrl(95function parseBase64DataUrl(
81 dataUrl: string,96 dataUrl: string,
82): { readonly mimeType: string; readonly base64: string } | null {97): { readonly mimeType: string; readonly base64: string } | null {
...@@ -114,7 +129,14 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: {...@@ -114,7 +129,14 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: {
114129
115 const path = yield* Path.Path;130 const path = yield* Path.Path;
116 const attachmentsRootDir = path.join(serverConfig.value.stateDir, "attachments");131 const attachmentsRootDir = path.join(serverConfig.value.stateDir, "attachments");
117 const threadSegment = encodeURIComponent(input.threadId);132 const threadSegment = toSafeThreadAttachmentSegment(input.threadId);
133 if (!threadSegment) {
134 yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", {
135 threadId: input.threadId,
136 messageId: input.messageId,
137 });
138 return input.attachments;
139 }
118 const messageSegment = encodeURIComponent(input.messageId);140 const messageSegment = encodeURIComponent(input.messageId);
119141
120 return yield* Effect.forEach(142 return yield* Effect.forEach(
...@@ -282,7 +304,11 @@ function collectThreadAttachmentRelativePaths(...@@ -282,7 +304,11 @@ function collectThreadAttachmentRelativePaths(
282 threadId: string,304 threadId: string,
283 messages: ReadonlyArray<ProjectionThreadMessage>,305 messages: ReadonlyArray<ProjectionThreadMessage>,
284): Set<string> {306): Set<string> {
285 const threadPrefix = `${encodeURIComponent(threadId)}/`;307 const threadSegment = toSafeThreadAttachmentSegment(threadId);
308 if (!threadSegment) {
309 return new Set();
310 }
311 const threadPrefix = `${threadSegment}/`;
286 const relativePaths = new Set<string>();312 const relativePaths = new Set<string>();
287 for (const message of messages) {313 for (const message of messages) {
288 for (const attachment of message.attachments ?? []) {314 for (const attachment of message.attachments ?? []) {
...@@ -317,10 +343,18 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -317,10 +343,18 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
317343
318 yield* Effect.forEach(344 yield* Effect.forEach(
319 input.sideEffects.deletedThreadIds,345 input.sideEffects.deletedThreadIds,
320 (threadId) => {346 (threadId) =>
321 const threadDir = input.path.join(attachmentsRootDir, encodeURIComponent(threadId));347 Effect.gen(function* () {
322 return input.fileSystem.remove(threadDir, { recursive: true, force: true });348 const threadSegment = toSafeThreadAttachmentSegment(threadId);
323 },349 if (!threadSegment) {
350 yield* Effect.logWarning("skipping attachment cleanup for unsafe thread id", {
351 threadId,
352 });
353 return;
354 }
355 const threadDir = input.path.join(attachmentsRootDir, threadSegment);
356 yield* input.fileSystem.remove(threadDir, { recursive: true, force: true });
357 }),
324 { concurrency: 1 },358 { concurrency: 1 },
325 );359 );
326360
...@@ -330,8 +364,13 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {...@@ -330,8 +364,13 @@ const runAttachmentSideEffects = Effect.fn(function* (input: {
330 if (input.sideEffects.deletedThreadIds.has(threadId)) {364 if (input.sideEffects.deletedThreadIds.has(threadId)) {
331 return Effect.void;365 return Effect.void;
332 }366 }
333 const threadDir = input.path.join(attachmentsRootDir, encodeURIComponent(threadId));
334 return Effect.gen(function* () {367 return Effect.gen(function* () {
368 const threadSegment = toSafeThreadAttachmentSegment(threadId);
369 if (!threadSegment) {
370 yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });
371 return;
372 }
373 const threadDir = input.path.join(attachmentsRootDir, threadSegment);
335 const entries = yield* input.fileSystem374 const entries = yield* input.fileSystem
336 .readDirectory(threadDir, { recursive: true })375 .readDirectory(threadDir, { recursive: true })
337 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));376 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));
...@@ -601,16 +640,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -601,16 +640,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
601 stageFileWrite: (pendingWrite) => {640 stageFileWrite: (pendingWrite) => {
602 attachmentSideEffects.writes.push(pendingWrite);641 attachmentSideEffects.writes.push(pendingWrite);
603 },642 },
604 }).pipe(643 }).pipe(Effect.provideService(Path.Path, path))
605 Effect.provideService(Path.Path, path),
606 Effect.catch((cause) =>
607 Effect.logWarning("failed to persist message attachments", {
608 threadId: event.payload.threadId,
609 messageId: event.payload.messageId,
610 cause,
611 }).pipe(Effect.as(event.payload.attachments)),
612 ),
613 )
614 : existingMessage?.attachments;644 : existingMessage?.attachments;
615 yield* projectionThreadMessageRepository.upsert({645 yield* projectionThreadMessageRepository.upsert({
616 messageId: event.payload.messageId,646 messageId: event.payload.messageId,
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+2-1
...@@ -26,6 +26,7 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli...@@ -26,6 +26,7 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli
26 return Layer.mergeAll(26 return Layer.mergeAll(
27 runtimeRepositoryLayer,27 runtimeRepositoryLayer,
28 ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)),28 ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)),
29 NodeServices.layer,
29 );30 );
30}31}
3132
...@@ -223,5 +224,5 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL...@@ -223,5 +224,5 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL
223 }).pipe(Effect.provide(directoryLayer));224 }).pipe(Effect.provide(directoryLayer));
224225
225 fs.rmSync(tempDir, { recursive: true, force: true });226 fs.rmSync(tempDir, { recursive: true, force: true });
226 }).pipe(Effect.provide(NodeServices.layer)));227 }));
227});228});
apps/server/src/serverLayers.ts+3-3
...@@ -61,16 +61,16 @@ export function makeServerRuntimeServicesLayer() {...@@ -61,16 +61,16 @@ export function makeServerRuntimeServicesLayer() {
61 Layer.provide(OrchestrationEventStoreLive),61 Layer.provide(OrchestrationEventStoreLive),
62 Layer.provide(OrchestrationCommandReceiptRepositoryLive),62 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
63 );63 );
64 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));64
65 const checkpointDiffQueryLayer = CheckpointDiffQueryLive.pipe(65 const checkpointDiffQueryLayer = CheckpointDiffQueryLive.pipe(
66 Layer.provideMerge(OrchestrationProjectionSnapshotQueryLive),66 Layer.provideMerge(OrchestrationProjectionSnapshotQueryLive),
67 Layer.provideMerge(checkpointStoreLayer),67 Layer.provideMerge(CheckpointStoreLive),
68 );68 );
6969
70 const runtimeServicesLayer = Layer.mergeAll(70 const runtimeServicesLayer = Layer.mergeAll(
71 orchestrationLayer,71 orchestrationLayer,
72 OrchestrationProjectionSnapshotQueryLive,72 OrchestrationProjectionSnapshotQueryLive,
73 checkpointStoreLayer,73 CheckpointStoreLive,
74 checkpointDiffQueryLayer,74 checkpointDiffQueryLayer,
75 );75 );
76 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(76 const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
apps/server/src/wsServer.test.ts+10-6
...@@ -4,7 +4,7 @@ import os from "node:os";...@@ -4,7 +4,7 @@ import os from "node:os";
4import path from "node:path";4import path from "node:path";
55
6import * as NodeServices from "@effect/platform-node/NodeServices";6import * as NodeServices from "@effect/platform-node/NodeServices";
7import { Effect, Exit, Layer, PubSub, Scope, Stream } from "effect";7import { Effect, Exit, Layer, PlatformError, PubSub, Scope, Stream } from "effect";
8import { describe, expect, it, afterEach, vi } from "vitest";8import { describe, expect, it, afterEach, vi } from "vitest";
9import { createServer } from "./wsServer";9import { createServer } from "./wsServer";
10import WebSocket from "ws";10import WebSocket from "ws";
...@@ -40,13 +40,14 @@ import type {...@@ -40,13 +40,14 @@ import type {
40} from "@t3tools/contracts";40} from "@t3tools/contracts";
41import { TerminalManager, type TerminalManagerShape } from "./terminal/Services/Manager";41import { TerminalManager, type TerminalManagerShape } from "./terminal/Services/Manager";
42import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./persistence/Layers/Sqlite";42import { makeSqlitePersistenceLive, SqlitePersistenceMemory } from "./persistence/Layers/Sqlite";
43import { SqlClient } from "effect/unstable/sql";43import { SqlClient, SqlError } from "effect/unstable/sql";
44import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";44import { ProviderService, type ProviderServiceShape } from "./provider/Services/ProviderService";
45import { Open, type OpenShape } from "./open";45import { Open, type OpenShape } from "./open";
46import { GitManager, type GitManagerShape } from "./git/Services/GitManager.ts";46import { GitManager, type GitManagerShape } from "./git/Services/GitManager.ts";
47import type { GitCoreShape } from "./git/Services/GitCore.ts";47import type { GitCoreShape } from "./git/Services/GitCore.ts";
48import { GitCore } from "./git/Services/GitCore.ts";48import { GitCore } from "./git/Services/GitCore.ts";
49import { GitCommandError, GitManagerError } from "./git/Errors.ts";49import { GitCommandError, GitManagerError } from "./git/Errors.ts";
50import { MigrationError } from "@effect/sql-sqlite-bun/SqliteMigrator";
5051
51interface PendingMessages {52interface PendingMessages {
52 queue: unknown[];53 queue: unknown[];
...@@ -354,7 +355,10 @@ describe("WebSocket Server", () => {...@@ -354,7 +355,10 @@ describe("WebSocket Server", () => {
354355
355 async function createTestServer(356 async function createTestServer(
356 options: {357 options: {
357 persistenceLayer?: Layer.Layer<SqlClient.SqlClient, never>;358 persistenceLayer?: Layer.Layer<
359 SqlClient.SqlClient,
360 SqlError.SqlError | MigrationError | PlatformError.PlatformError
361 >;
358 cwd?: string;362 cwd?: string;
359 autoBootstrapProjectFromCwd?: boolean;363 autoBootstrapProjectFromCwd?: boolean;
360 logWebSocketEvents?: boolean;364 logWebSocketEvents?: boolean;
...@@ -614,9 +618,9 @@ describe("WebSocket Server", () => {...@@ -614,9 +618,9 @@ describe("WebSocket Server", () => {
614618
615 it("includes bootstrap ids in welcome when cwd project and thread already exist", async () => {619 it("includes bootstrap ids in welcome when cwd project and thread already exist", async () => {
616 const stateDir = makeTempDir("t3code-state-bootstrap-existing-");620 const stateDir = makeTempDir("t3code-state-bootstrap-existing-");
617 const persistenceLayer = makeSqlitePersistenceLive(621 const persistenceLayer = makeSqlitePersistenceLive(path.join(stateDir, "state.sqlite")).pipe(
618 path.join(stateDir, "state.sqlite"),622 Layer.provide(NodeServices.layer),
619 ).pipe(Layer.provide(NodeServices.layer)) as unknown as Layer.Layer<SqlClient.SqlClient, never>;623 );
620 const cwd = "/test/bootstrap-existing";624 const cwd = "/test/bootstrap-existing";
621625
622 server = await createTestServer({626 server = await createTestServer({
apps/server/src/wsServer.ts+1-2
...@@ -23,7 +23,6 @@ import {...@@ -23,7 +23,6 @@ import {
23 WsPush,23 WsPush,
24 WsResponse,24 WsResponse,
25} from "@t3tools/contracts";25} from "@t3tools/contracts";
26import * as NodeServices from "@effect/platform-node/NodeServices";
27import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";26import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
28import {27import {
29 Cause,28 Cause,
...@@ -787,6 +786,6 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -787,6 +786,6 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
787});786});
788787
789export const ServerLive = Layer.succeed(Server, {788export const ServerLive = Layer.succeed(Server, {
790 start: createServer().pipe(Effect.provide(NodeServices.layer)),789 start: createServer(),
791 stopSignal: Effect.never,790 stopSignal: Effect.never,
792} satisfies ServerShape);791} satisfies ServerShape);