feat(composer): sync drafts across web and desktop

apps/server/src/auth/RpcAuthorization.ts

@@ -66,6 +66,8 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.shellOpenInEditor]: AuthOrchestrationOperateScope,
   [WS_METHODS.filesystemBrowse]: AuthOrchestrationReadScope,
   [WS_METHODS.assetsCreateUrl]: AuthOrchestrationReadScope,
+  [WS_METHODS.composerDraftUpdate]: AuthOrchestrationOperateScope,
+  [WS_METHODS.subscribeComposerDraft]: AuthOrchestrationReadScope,
   [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope,
   [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope,
   [WS_METHODS.vcsRefreshStatus]: AuthOrchestrationReadScope,

apps/server/src/environment/ServerEnvironment.ts

@@ -148,6 +148,7 @@ export const make = Effect.gen(function* () {
       threadPinning: true,
       threadPinReorder: true,
       threadTitleRegeneration: true,
+      composerDraftSync: true,
       ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
       ...(serverSelfUpdate === "boot-service" ? { serverSelfUpdateProgress: true } : {}),
     },

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -239,6 +239,50 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
       assert.deepEqual(unsettledRows, [{ settledOverride: "active", settledAt: null }]);
     }),
   );
+
+  it.effect("removes the composer draft when its thread is deleted", () =>
+    Effect.gen(function* () {
+      const projectionPipeline = yield* OrchestrationProjectionPipeline;
+      const eventStore = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const threadId = ThreadId.make("thread-with-composer-draft");
+      const now = "2026-01-01T00:00:00.000Z";
+      const commonJson =
+        '{"text":"delete me","modelSelection":null,"runtimeMode":null,"interactionMode":null}';
+
+      yield* sql`
+        INSERT INTO composer_drafts (
+          thread_id, revision, common_json, updated_at, client_mutation_id
+        ) VALUES (
+          ${threadId}, 1, ${commonJson}, ${now}, 'test:delete-composer-draft'
+        )
+      `;
+      yield* eventStore.append({
+        type: "thread.deleted",
+        eventId: EventId.make("evt-delete-composer-draft"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        occurredAt: now,
+        commandId: CommandId.make("cmd-delete-composer-draft"),
+        causationEventId: null,
+        correlationId: CommandId.make("cmd-delete-composer-draft"),
+        metadata: {},
+        payload: {
+          threadId,
+          deletedAt: now,
+        },
+      });
+
+      yield* projectionPipeline.bootstrap;
+
+      const remaining = yield* sql<{ readonly count: number }>`
+        SELECT COUNT(*) AS count
+        FROM composer_drafts
+        WHERE thread_id = ${threadId}
+      `;
+      assert.deepEqual(remaining, [{ count: 0 }]);
+    }),
+  );
 });
 
 it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-base-")))(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -828,6 +828,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
         case "thread.deleted": {
           attachmentSideEffects.deletedThreadIds.add(event.payload.threadId);
+          yield* sql`
+            DELETE FROM composer_drafts
+            WHERE thread_id = ${event.payload.threadId}
+          `.pipe(Effect.mapError(toPersistenceSqlError("delete thread composer draft")));
           const existingRow = yield* projectionThreadRepository.getById({
             threadId: event.payload.threadId,
           });

apps/server/src/persistence/ComposerDrafts.test.ts

@@ -0,0 +1,119 @@
+import { ProviderInstanceId, ThreadId } from "@t3tools/contracts";
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import * as ComposerDrafts from "./ComposerDrafts.ts";
+import { runMigrations } from "./Migrations.ts";
+import * as NodeSqliteClient from "./NodeSqliteClient.ts";
+
+const layer = it.layer(
+  ComposerDrafts.layer.pipe(Layer.provideMerge(NodeSqliteClient.layerMemory())),
+);
+
+layer("ComposerDraftRepository", (it) => {
+  it.effect("uses revision compare-and-swap and preserves the winning snapshot", () =>
+    Effect.gen(function* () {
+      yield* runMigrations({ toMigrationInclusive: 39 });
+      const repository = yield* ComposerDrafts.ComposerDraftRepository;
+      const threadId = ThreadId.make("draft-cas-thread");
+      const common = {
+        text: "hello from device one",
+        modelSelection: {
+          instanceId: ProviderInstanceId.make("codex"),
+          model: "gpt-5.4",
+        },
+        runtimeMode: "full-access" as const,
+        interactionMode: "default" as const,
+      };
+
+      const accepted = yield* repository.update({
+        threadId,
+        baseRevision: 0,
+        common,
+        clientMutationId: "device-one-1",
+      });
+      assert.equal(accepted._tag, "accepted");
+      assert.equal(accepted.snapshot.revision, 1);
+
+      const conflict = yield* repository.update({
+        threadId,
+        baseRevision: 0,
+        common: { ...common, text: "stale device" },
+        clientMutationId: "device-two-1",
+      });
+      assert.equal(conflict._tag, "conflict");
+      assert.deepEqual(conflict.snapshot.common, common);
+      assert.equal(conflict.snapshot.revision, 1);
+    }),
+  );
+
+  it.effect("keeps clears as revisioned tombstones", () =>
+    Effect.gen(function* () {
+      yield* runMigrations({ toMigrationInclusive: 39 });
+      const repository = yield* ComposerDrafts.ComposerDraftRepository;
+      const threadId = ThreadId.make("draft-tombstone-thread");
+
+      yield* repository.update({
+        threadId,
+        baseRevision: 0,
+        common: {
+          text: "sent later",
+          modelSelection: null,
+          runtimeMode: null,
+          interactionMode: null,
+        },
+        clientMutationId: "write-1",
+      });
+      const cleared = yield* repository.update({
+        threadId,
+        baseRevision: 1,
+        common: null,
+        clientMutationId: "clear-2",
+      });
+
+      assert.equal(cleared._tag, "accepted");
+      assert.equal(cleared.snapshot.revision, 2);
+      assert.isNull(cleared.snapshot.common);
+      assert.deepEqual(yield* repository.get({ threadId }), cleared.snapshot);
+    }),
+  );
+
+  it.effect("does not let a delayed send clear a newer device revision", () =>
+    Effect.gen(function* () {
+      yield* runMigrations({ toMigrationInclusive: 39 });
+      const repository = yield* ComposerDrafts.ComposerDraftRepository;
+      const threadId = ThreadId.make("draft-delayed-send-thread");
+      const first = {
+        text: "message being sent",
+        modelSelection: null,
+        runtimeMode: null,
+        interactionMode: null,
+      };
+      const newer = { ...first, text: "new text from another device" };
+
+      yield* repository.update({
+        threadId,
+        baseRevision: 0,
+        common: first,
+        clientMutationId: "first-device",
+      });
+      yield* repository.update({
+        threadId,
+        baseRevision: 1,
+        common: newer,
+        clientMutationId: "second-device",
+      });
+      const delayedClear = yield* repository.update({
+        threadId,
+        baseRevision: 1,
+        common: null,
+        clientMutationId: "delayed-send",
+      });
+
+      assert.equal(delayedClear._tag, "conflict");
+      assert.equal(delayedClear.snapshot.revision, 2);
+      assert.deepEqual(delayedClear.snapshot.common, newer);
+    }),
+  );
+});