From 39226f4213c3eb6b673beee387009c04a0d92bc0 Mon Sep 17 00:00:00 2001 From: ScottN-PV Date: Wed, 14 Oct 2026 21:29:30 -0400 Subject: [PATCH] fix(server): reject thread.create for an existing id --- .../Orchestrator.control-reads.test.ts | 85 +++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 24 ++++++ .../ThreadLaunchService.test.ts | 14 +++ .../orchestration-v2/ThreadMessageIntake.ts | 1 + 4 files changed, 124 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index de616556b4eb..8002ef84196b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -21,6 +21,8 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Result from "effect/Result"; +import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; @@ -960,3 +962,86 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }); }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); + +it.effect.each([false, true])("rejects thread.create for an existing id (deleted: %s)", (deleted) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make(`thread:duplicate-create:${deleted}`); + const create = { + type: "thread.create", + commandId: CommandId.make(`create:${threadId}`), + threadId, + projectId: ProjectId.make("project:duplicate-create"), + title: "Original", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + importedNativeThread: { + ref: { driver: adapter.driver, nativeId: "native-original", strength: "strong" }, + }, + } as const; + yield* orchestrator.dispatch(create); + if (deleted) { + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make(`delete:${threadId}`), + threadId, + }); + } + const before = yield* projections.getThread(threadId); + assert.isNotNull(before.activeProviderThreadId); + assert.equal(before.deletedAt !== null, deleted); + const sequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* TestClock.adjust("1 second"); + const duplicate = { + ...create, + commandId: CommandId.make(`duplicate:${threadId}`), + title: "Replacement", + importedNativeThread: undefined, + }; + const outcome = yield* orchestrator.dispatch(duplicate).pipe(Effect.result); + assert.ok(Result.isFailure(outcome), "duplicate create must be rejected"); + const refused = outcome.failure; + assert.equal(refused._tag, "OrchestratorThreadAlreadyExistsError"); + assert.equal(refused.message, `Thread ${threadId} already exists. Use a new thread id.`); + assert.deepEqual(yield* projections.getThread(threadId), before); + assert.equal(yield* orchestrator.getThreadEventSequence(threadId), sequence); + const retried = yield* orchestrator.dispatch(duplicate).pipe(Effect.flip); + assert.equal(retried._tag, "OrchestratorCommandPreviouslyRejectedError"); + assert.include(retried.message, refused.message); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("replays a thread.create receipt without changing the thread", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread:create-receipt-replay"); + const create = { + type: "thread.create", + commandId: CommandId.make("create:receipt-replay"), + threadId, + projectId: ProjectId.make("project:duplicate-create"), + title: "Original", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + } as const; + const original = yield* orchestrator.dispatch(create); + const before = yield* projections.getThread(threadId); + const replayed = yield* orchestrator.dispatch(create); + assert.equal(replayed.sequence, original.sequence); + assert.deepEqual(replayed, original); + assert.deepEqual(yield* projections.getThread(threadId), before); + assert.equal(yield* orchestrator.getThreadEventSequence(threadId), original.sequence); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c7b322f2c492..08146615cdc9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -150,6 +150,15 @@ export class OrchestratorDispatchError extends Schema.TaggedError()( + "OrchestratorThreadAlreadyExistsError", + { commandId: CommandId, threadId: ThreadId }, +) { + override get message(): string { + return `Thread ${this.threadId} already exists. Use a new thread id.`; + } +} + export class OrchestratorCommandRejectedError extends Schema.TaggedError()( "OrchestratorCommandRejectedError", { commandId: CommandId, commandType: Schema.String, cause: Schema.optional(Schema.Defect()) }, @@ -262,6 +271,7 @@ export function canReplayCommandReceipt( export const OrchestratorV2Error = Schema.Union([ OrchestratorDispatchError, + OrchestratorThreadAlreadyExistsError, OrchestratorCommandRejectedError, OrchestratorProjectionError, OrchestratorDomainEventStreamError, @@ -2250,6 +2260,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "orchestration_v2.driver": command.modelSelection.instanceId, }); + // Tombstones reserve their ids too; recreating one would overwrite its history. + const existing = yield* projectionStore.getThread(command.threadId).pipe( + Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + if (existing !== null) { + return yield* new OrchestratorThreadAlreadyExistsError({ + commandId: command.commandId, + threadId: command.threadId, + }); + } + const now = yield* DateTime.now; const emitEvent = emit(events, command); const thread: OrchestrationV2AppThread = { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 26e9b8c8d78d..ad028212fdf3 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -36,6 +36,7 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; import * as TestClock from "effect/testing/TestClock"; @@ -2308,6 +2309,19 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch ); assert.equal((yield* claimedFiles).length, 1); + const duplicate = yield* ThreadMessageIntake.launchThread({ + ...input, + commandId: CommandId.make("intake-duplicate-create"), + initialMessage: { ...input.initialMessage, messageId: MessageId.make("intake-duplicate") }, + }).pipe(Effect.result); + assert.ok(Result.isFailure(duplicate)); + assert.ok(duplicate.failure._tag === "ThreadLaunchError"); + assert.equal(duplicate.failure.operation, "create-thread"); + assert.propertyVal(duplicate.failure.cause, "_tag", "OrchestratorThreadAlreadyExistsError"); + assert.equal((yield* claimedFiles).length, 1); + assert.deepEqual(yield* fs.readFile(storedPath), new Uint8Array([1, 2, 3, 4])); + assert.deepEqual(yield* fs.readFile(pendingPath), new Uint8Array([1, 2, 3, 4])); + const missingProject = yield* ThreadMessageIntake.launchThread({ ...input, commandId: CommandId.make("intake-no-project"), diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts index 21540703ffbc..511388d5558c 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts @@ -20,6 +20,7 @@ function dispatchWasNotAccepted( ) { switch (error._tag) { case "OrchestratorCommandRejectedError": + case "OrchestratorThreadAlreadyExistsError": case "OrchestratorProjectionError": case "OrchestratorProviderAdapterError": case "OrchestratorCommandPreviouslyRejectedError":