diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index c1dbc833718..6da80a6e77d 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -88,6 +88,7 @@ describe("CheckpointDiffQuery.layer", () => { getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.sync(() => { getThreadCheckpointContextCalls += 1; @@ -195,6 +196,7 @@ describe("CheckpointDiffQuery.layer", () => { getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), @@ -277,6 +279,7 @@ describe("CheckpointDiffQuery.layer", () => { getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), @@ -344,6 +347,7 @@ describe("CheckpointDiffQuery.layer", () => { getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), @@ -396,6 +400,7 @@ describe("CheckpointDiffQuery.layer", () => { getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index b2ef0fed0f9..f073fb48d47 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -196,6 +196,7 @@ describe("OrchestrationEngine", () => { getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f12df850941..e7c933ea314 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -598,6 +598,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti threadId: event.payload.threadId, projectId: event.payload.projectId, title: event.payload.title, + owner: event.payload.owner ?? "user", modelSelection: event.payload.modelSelection, runtimeMode: event.payload.runtimeMode, interactionMode: event.payload.interactionMode, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 9a136b06872..89b841246d1 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -285,6 +285,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { id: ThreadId.make("thread-1"), projectId: asProjectId("project-1"), title: "Thread 1", + owner: "user", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", @@ -696,6 +697,242 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); + it.effect( + "excludes plugin-owned threads from user-facing lists and startup selection while keeping by-id lookups", + () => + Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_turns`; + yield* sql`DELETE FROM projection_state`; + + yield* sql` + INSERT INTO projection_projects ( + project_id, + title, + workspace_root, + default_model_selection_json, + scripts_json, + created_at, + updated_at, + deleted_at + ) + VALUES ( + 'project-owner-filter', + 'Owner Filter', + '/tmp/owner-filter', + '{"provider":"codex","model":"gpt-5-codex"}', + '[]', + '2026-04-07T00:00:00.000Z', + '2026-04-07T00:00:01.000Z', + NULL + ) + `; + + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + owner, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + latest_user_message_at, + pending_approval_count, + pending_user_input_count, + has_actionable_proposed_plan, + created_at, + updated_at, + archived_at, + deleted_at + ) + VALUES + ( + 'thread-plugin-active', + 'project-owner-filter', + 'Plugin Active', + 'plugin:test', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + NULL, + '/tmp/plugin-worktree', + NULL, + NULL, + 0, + 0, + 0, + '2026-04-07T00:00:02.000Z', + '2026-04-07T00:00:03.000Z', + NULL, + NULL + ), + ( + 'thread-user-active', + 'project-owner-filter', + 'User Active', + 'user', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + NULL, + NULL, + NULL, + NULL, + 0, + 0, + 0, + '2026-04-07T00:00:04.000Z', + '2026-04-07T00:00:05.000Z', + NULL, + NULL + ), + ( + 'thread-plugin-archived', + 'project-owner-filter', + 'Plugin Archived', + 'plugin:test', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + NULL, + NULL, + NULL, + NULL, + 0, + 0, + 0, + '2026-04-07T00:00:06.000Z', + '2026-04-07T00:00:07.000Z', + '2026-04-07T00:00:08.000Z', + NULL + ) + `; + + yield* sql` + INSERT INTO projection_turns ( + thread_id, + turn_id, + pending_message_id, + source_proposed_plan_thread_id, + source_proposed_plan_id, + assistant_message_id, + state, + requested_at, + started_at, + completed_at, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json + ) + VALUES ( + 'thread-plugin-active', + 'turn-plugin-1', + NULL, + NULL, + NULL, + NULL, + 'completed', + '2026-04-07T00:00:09.000Z', + '2026-04-07T00:00:09.000Z', + '2026-04-07T00:00:09.000Z', + 1, + 'checkpoint-plugin-1', + 'ready', + '[]' + ) + `; + + yield* sql` + INSERT INTO projection_state (projector, last_applied_sequence, updated_at) + VALUES + (${ORCHESTRATION_PROJECTOR_NAMES.projects}, 6, '2026-04-07T00:00:10.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threads}, 6, '2026-04-07T00:00:10.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadMessages}, 6, '2026-04-07T00:00:10.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans}, 6, '2026-04-07T00:00:10.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadActivities}, 6, '2026-04-07T00:00:10.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.threadSessions}, 6, '2026-04-07T00:00:10.000Z'), + (${ORCHESTRATION_PROJECTOR_NAMES.checkpoints}, 6, '2026-04-07T00:00:10.000Z') + `; + + const counts = yield* snapshotQuery.getCounts(); + assert.deepEqual(counts, { + projectCount: 1, + threadCount: 1, + }); + + // The decider's read model keeps plugin-owned threads: commands and + // events on them must still validate after a restart. + const commandReadModel = yield* snapshotQuery.getCommandReadModel(); + assert.deepEqual( + commandReadModel.threads.map((thread) => thread.id), + [ + ThreadId.make("thread-plugin-active"), + ThreadId.make("thread-user-active"), + ThreadId.make("thread-plugin-archived"), + ], + ); + + const fullSnapshot = yield* snapshotQuery.getSnapshot(); + assert.deepEqual( + fullSnapshot.threads.map((thread) => thread.id), + [ThreadId.make("thread-user-active")], + ); + + const shellSnapshot = yield* snapshotQuery.getShellSnapshot(); + assert.deepEqual( + shellSnapshot.threads.map((thread) => thread.id), + [ThreadId.make("thread-user-active")], + ); + + const archivedShellSnapshot = yield* snapshotQuery.getArchivedShellSnapshot(); + assert.deepEqual( + archivedShellSnapshot.threads.map((thread) => thread.id), + [], + ); + + const firstThreadId = yield* snapshotQuery.getFirstActiveThreadIdByProjectId( + asProjectId("project-owner-filter"), + ); + assert.equal(firstThreadId._tag, "Some"); + if (firstThreadId._tag === "Some") { + assert.equal(firstThreadId.value, ThreadId.make("thread-user-active")); + } + + const pluginShell = yield* snapshotQuery.getThreadShellById( + ThreadId.make("thread-plugin-active"), + ); + assert.equal(pluginShell._tag, "Some"); + + const pluginDetail = yield* snapshotQuery.getThreadDetailById( + ThreadId.make("thread-plugin-active"), + ); + assert.equal(pluginDetail._tag, "Some"); + if (pluginDetail._tag === "Some") { + assert.equal(pluginDetail.value.owner, "plugin:test"); + } + + const checkpointContext = yield* snapshotQuery.getThreadCheckpointContext( + ThreadId.make("thread-plugin-active"), + ); + assert.equal(checkpointContext._tag, "Some"); + + const fullDiffContext = yield* snapshotQuery.getFullThreadDiffContext( + ThreadId.make("thread-plugin-active"), + 1, + ); + assert.equal(fullDiffContext._tag, "Some"); + }), + ); + it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () => Effect.gen(function* () { const snapshotQuery = yield* ProjectionSnapshotQuery; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index e36db35b107..9ba0064ff10 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -1,6 +1,7 @@ import { ChatAttachment, CheckpointRef, + DEFAULT_THREAD_OWNER, IsoDateTime, MessageId, NonNegativeInt, @@ -120,6 +121,9 @@ const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema; const ProjectionThreadIdLookupRowSchema = Schema.Struct({ threadId: ThreadId, }); +const ProjectionThreadOwnerLookupRowSchema = Schema.Struct({ + owner: ProjectionThread.fields.owner, +}); const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({ threadId: ThreadId, projectId: ProjectId, @@ -324,6 +328,41 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { thread_id AS "threadId", project_id AS "projectId", title, + owner, + model_selection_json AS "modelSelection", + runtime_mode AS "runtimeMode", + interaction_mode AS "interactionMode", + branch, + worktree_path AS "worktreePath", + latest_turn_id AS "latestTurnId", + created_at AS "createdAt", + updated_at AS "updatedAt", + archived_at AS "archivedAt", + latest_user_message_at AS "latestUserMessageAt", + pending_approval_count AS "pendingApprovalCount", + pending_user_input_count AS "pendingUserInputCount", + has_actionable_proposed_plan AS "hasActionableProposedPlan", + deleted_at AS "deletedAt" + FROM projection_threads + WHERE owner = ${DEFAULT_THREAD_OWNER} + ORDER BY created_at ASC, thread_id ASC + `, + }); + + // The command read model must see EVERY thread regardless of owner: the + // decider validates commands against it, and events for a thread missing + // from it are rejected. Non-user (plugin-owned) threads are hidden from + // user-facing views only. + const listAllThreadRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadDbRowSchema, + execute: () => + sql` + SELECT + thread_id AS "threadId", + project_id AS "projectId", + title, + owner, model_selection_json AS "modelSelection", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", @@ -352,6 +391,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { thread_id AS "threadId", project_id AS "projectId", title, + owner, model_selection_json AS "modelSelection", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", @@ -369,6 +409,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { FROM projection_threads WHERE deleted_at IS NULL AND archived_at IS NULL + AND owner = ${DEFAULT_THREAD_OWNER} ORDER BY project_id ASC, created_at ASC, thread_id ASC `, }); @@ -382,6 +423,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { thread_id AS "threadId", project_id AS "projectId", title, + owner, model_selection_json AS "modelSelection", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", @@ -399,6 +441,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { FROM projection_threads WHERE deleted_at IS NULL AND archived_at IS NOT NULL + AND owner = ${DEFAULT_THREAD_OWNER} ORDER BY project_id ASC, archived_at DESC, thread_id DESC `, }); @@ -508,6 +551,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ON threads.thread_id = sessions.thread_id WHERE threads.deleted_at IS NULL AND threads.archived_at IS NULL + AND threads.owner = ${DEFAULT_THREAD_OWNER} ORDER BY sessions.thread_id ASC `, }); @@ -533,6 +577,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ON threads.thread_id = sessions.thread_id WHERE threads.deleted_at IS NULL AND threads.archived_at IS NOT NULL + AND threads.owner = ${DEFAULT_THREAD_OWNER} ORDER BY sessions.thread_id ASC `, }); @@ -558,6 +603,36 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { }); const listLatestTurnRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionLatestTurnDbRowSchema, + execute: () => + sql` + SELECT + turns.thread_id AS "threadId", + turns.turn_id AS "turnId", + turns.state, + turns.requested_at AS "requestedAt", + turns.started_at AS "startedAt", + turns.completed_at AS "completedAt", + turns.assistant_message_id AS "assistantMessageId", + turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", + turns.source_proposed_plan_id AS "sourceProposedPlanId" + FROM projection_threads threads + JOIN projection_turns turns + ON turns.thread_id = threads.thread_id + AND turns.turn_id = threads.latest_turn_id + WHERE threads.latest_turn_id IS NOT NULL + AND threads.owner = ${DEFAULT_THREAD_OWNER} + ORDER BY turns.thread_id ASC + `, + }); + + // The command read model must see EVERY thread's latest turn regardless of + // owner (mirroring listAllThreadRows): the decider replays commands/events + // against it, and rehydrating a plugin-owned thread with latestTurn = null + // would diverge from the live in-memory projector, which tracks latest turns + // for all owners. Non-user threads are filtered from user-facing views only. + const listAllLatestTurnRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionLatestTurnDbRowSchema, execute: () => @@ -603,6 +678,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { WHERE threads.deleted_at IS NULL AND threads.archived_at IS NULL AND threads.latest_turn_id IS NOT NULL + AND threads.owner = ${DEFAULT_THREAD_OWNER} ORDER BY turns.thread_id ASC `, }); @@ -629,6 +705,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { WHERE threads.deleted_at IS NULL AND threads.archived_at IS NOT NULL AND threads.latest_turn_id IS NOT NULL + AND threads.owner = ${DEFAULT_THREAD_OWNER} ORDER BY turns.thread_id ASC `, }); @@ -653,7 +730,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sql` SELECT (SELECT COUNT(*) FROM projection_projects) AS "projectCount", - (SELECT COUNT(*) FROM projection_threads) AS "threadCount" + (SELECT COUNT(*) FROM projection_threads WHERE owner = ${DEFAULT_THREAD_OWNER}) AS "threadCount" `, }); @@ -711,11 +788,24 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { WHERE project_id = ${projectId} AND deleted_at IS NULL AND archived_at IS NULL + AND owner = ${DEFAULT_THREAD_OWNER} ORDER BY created_at ASC, thread_id ASC LIMIT 1 `, }); + const getThreadOwnerRowById = SqlSchema.findOneOption({ + Request: ThreadIdLookupInput, + Result: ProjectionThreadOwnerLookupRowSchema, + execute: ({ threadId }) => + sql` + SELECT owner + FROM projection_threads + WHERE thread_id = ${threadId} + LIMIT 1 + `, + }); + const getThreadCheckpointContextThreadRow = SqlSchema.findOneOption({ Request: ThreadIdLookupInput, Result: ProjectionThreadCheckpointContextThreadRowSchema, @@ -744,6 +834,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { thread_id AS "threadId", project_id AS "projectId", title, + owner, model_selection_json AS "modelSelection", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", @@ -1176,6 +1267,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { id: row.threadId, projectId: row.projectId, title: row.title, + owner: row.owner, modelSelection: row.modelSelection, runtimeMode: row.runtimeMode, interactionMode: row.interactionMode, @@ -1227,7 +1319,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), - listThreadRows(undefined).pipe( + listAllThreadRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getCommandReadModel:listThreads:query", @@ -1251,7 +1343,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), - listLatestTurnRows(undefined).pipe( + listAllLatestTurnRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getCommandReadModel:listLatestTurns:query", @@ -1374,6 +1466,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { id: row.threadId, projectId: row.projectId, title: row.title, + owner: row.owner, modelSelection: row.modelSelection, runtimeMode: row.runtimeMode, interactionMode: row.interactionMode, @@ -1767,6 +1860,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { Effect.map(Option.map((row) => row.threadId)), ); + const getThreadOwnerById: ProjectionSnapshotQueryShape["getThreadOwnerById"] = (threadId) => + getThreadOwnerRowById({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadOwnerById:query", + "ProjectionSnapshotQuery.getThreadOwnerById:decodeRow", + ), + ), + Effect.map(Option.map((row) => row.owner)), + ); + const getThreadCheckpointContext: ProjectionSnapshotQueryShape["getThreadCheckpointContext"] = ( threadId, ) => @@ -1971,6 +2075,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { id: threadRow.value.threadId, projectId: threadRow.value.projectId, title: threadRow.value.title, + owner: threadRow.value.owner, modelSelection: threadRow.value.modelSelection, runtimeMode: threadRow.value.runtimeMode, interactionMode: threadRow.value.interactionMode, @@ -2043,6 +2148,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { getActiveProjectByWorkspaceRoot, getProjectShellById, getFirstActiveThreadIdByProjectId, + getThreadOwnerById, getThreadCheckpointContext, getFullThreadDiffContext, getThreadShellById, diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 7d85f0240f7..7ff140f00e4 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -16,6 +16,7 @@ import type { OrchestrationThread, OrchestrationThreadShell, ProjectId, + ThreadOwner, ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -128,6 +129,13 @@ export interface ProjectionSnapshotQueryShape { projectId: ProjectId, ) => Effect.Effect, ProjectionRepositoryError>; + /** + * Read a thread owner by id without applying user-facing visibility filters. + */ + readonly getThreadOwnerById: ( + threadId: ThreadId, + ) => Effect.Effect, ProjectionRepositoryError>; + /** * Read the checkpoint context needed to resolve a single thread diff. */ diff --git a/apps/server/src/orchestration/decider.projectScripts.test.ts b/apps/server/src/orchestration/decider.projectScripts.test.ts index 64ba159c740..e040a614611 100644 --- a/apps/server/src/orchestration/decider.projectScripts.test.ts +++ b/apps/server/src/orchestration/decider.projectScripts.test.ts @@ -94,6 +94,58 @@ it.layer(NodeServices.layer)("decider project scripts", (it) => { }), ); + it.effect("carries thread.create owner into thread.created", () => + Effect.gen(function* () { + const now = "2026-01-01T00:00:00.000Z"; + const readModel = yield* projectEvent(createEmptyReadModel(now), { + sequence: 1, + eventId: asEventId("evt-project-create-owner"), + aggregateKind: "project", + aggregateId: asProjectId("project-owner"), + type: "project.created", + occurredAt: now, + commandId: CommandId.make("cmd-project-create-owner"), + causationEventId: null, + correlationId: CommandId.make("cmd-project-create-owner"), + metadata: {}, + payload: { + projectId: asProjectId("project-owner"), + title: "Project", + workspaceRoot: "/tmp/project-owner", + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-owner"), + threadId: ThreadId.make("thread-plugin"), + projectId: asProjectId("project-owner"), + title: "Plugin thread", + owner: "plugin:test", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + } as never, + readModel, + }); + + const event = Array.isArray(result) ? result[0] : result; + expect(event.type).toBe("thread.created"); + expect((event.payload as { owner?: unknown }).owner).toBe("plugin:test"); + }), + ); + it.effect("emits user message and turn-start-requested events for thread.turn.start", () => Effect.gen(function* () { const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 0d4af771ca8..9c33b473d44 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -234,6 +234,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" threadId: command.threadId, projectId: command.projectId, title: command.title, + owner: "owner" in command ? (command.owner ?? "user") : "user", modelSelection: command.modelSelection, runtimeMode: command.runtimeMode, interactionMode: command.interactionMode, diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index fadd5078026..e633c683638 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -77,6 +77,7 @@ describe("orchestration projector", () => { id: "thread-1", projectId: "project-1", title: "demo", + owner: "user", modelSelection: { instanceId: "codex", model: "gpt-5-codex", @@ -99,6 +100,72 @@ describe("orchestration projector", () => { ]); }); + it("projects thread owner and defaults legacy thread.created events to user", async () => { + const now = "2026-01-01T00:00:00.000Z"; + const model = createEmptyReadModel(now); + + const pluginOwned = await Effect.runPromise( + projectEvent( + model, + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateKind: "thread", + aggregateId: "thread-plugin", + occurredAt: now, + commandId: "cmd-thread-create-plugin", + payload: { + threadId: "thread-plugin", + projectId: "project-1", + title: "plugin", + owner: "plugin:test", + modelSelection: { + provider: ProviderDriverKind.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }), + ), + ); + expect((pluginOwned.threads[0] as { owner?: unknown } | undefined)?.owner).toBe("plugin:test"); + + const legacy = await Effect.runPromise( + projectEvent( + model, + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateKind: "thread", + aggregateId: "thread-legacy", + occurredAt: now, + commandId: "cmd-thread-create-legacy", + payload: { + threadId: "thread-legacy", + projectId: "project-1", + title: "legacy", + modelSelection: { + provider: ProviderDriverKind.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }), + ), + ); + expect((legacy.threads[0] as { owner?: unknown } | undefined)?.owner).toBe("user"); + }); + it("fails when event payload cannot be decoded by runtime schema", async () => { const now = "2026-01-01T00:00:00.000Z"; const model = createEmptyReadModel(now); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index fc6ab8f6fcf..6cb6528d642 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -277,6 +277,7 @@ export function projectEvent( id: payload.threadId, projectId: payload.projectId, title: payload.title, + owner: payload.owner ?? "user", modelSelection: payload.modelSelection, runtimeMode: payload.runtimeMode, interactionMode: payload.interactionMode, diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index a2069e62a14..e5f4f23c83a 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -79,6 +79,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { threadId: ThreadId.make("thread-null-options"), projectId: ProjectId.make("project-null-options"), title: "Null options thread", + owner: "user", modelSelection: { instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-opus-4-6", diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index 1baeb375c15..24018552a7c 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -34,6 +34,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { thread_id, project_id, title, + owner, model_selection_json, runtime_mode, interaction_mode, @@ -53,6 +54,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.threadId}, ${row.projectId}, ${row.title}, + ${row.owner}, ${JSON.stringify(row.modelSelection)}, ${row.runtimeMode}, ${row.interactionMode}, @@ -72,6 +74,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { DO UPDATE SET project_id = excluded.project_id, title = excluded.title, + owner = excluded.owner, model_selection_json = excluded.model_selection_json, runtime_mode = excluded.runtime_mode, interaction_mode = excluded.interaction_mode, @@ -98,6 +101,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { thread_id AS "threadId", project_id AS "projectId", title, + owner, model_selection_json AS "modelSelection", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", @@ -126,6 +130,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { thread_id AS "threadId", project_id AS "projectId", title, + owner, model_selection_json AS "modelSelection", runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index ba1131ee259..782910a5548 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -45,6 +45,7 @@ import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexe import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts"; import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; +import Migration0033 from "./Migrations/033_ThreadOwner.ts"; /** * Migration loader with all migrations defined inline. @@ -89,6 +90,7 @@ export const migrationEntries = [ [30, "ProjectionThreadShellArchiveIndexes", Migration0030], [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], + [33, "ThreadOwner", Migration0033], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/033_ThreadOwner.test.ts b/apps/server/src/persistence/Migrations/033_ThreadOwner.test.ts new file mode 100644 index 00000000000..88dbe60abdf --- /dev/null +++ b/apps/server/src/persistence/Migrations/033_ThreadOwner.test.ts @@ -0,0 +1,78 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("033_ThreadOwner", (it) => { + it.effect("adds a non-null user owner default to projection_threads", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 32 }); + yield* runMigrations({ toMigrationInclusive: 33 }); + + const columns = yield* sql<{ + readonly name: string; + readonly notnull: number; + readonly dflt_value: string | null; + }>` + PRAGMA table_info(projection_threads) + `; + const ownerColumn = columns.find((column) => column.name === "owner"); + assert.ok(ownerColumn); + assert.equal(ownerColumn.notnull, 1); + assert.equal(ownerColumn.dflt_value, "'user'"); + + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + latest_user_message_at, + pending_approval_count, + pending_user_input_count, + has_actionable_proposed_plan, + created_at, + updated_at, + archived_at, + deleted_at + ) + VALUES ( + 'thread-default-owner', + 'project-1', + 'Default owner', + '{"provider":"codex","model":"gpt-5-codex"}', + 'full-access', + 'default', + NULL, + NULL, + NULL, + NULL, + 0, + 0, + 0, + '2026-06-01T00:00:00.000Z', + '2026-06-01T00:00:00.000Z', + NULL, + NULL + ) + `; + + const rows = yield* sql<{ readonly owner: string }>` + SELECT owner FROM projection_threads WHERE thread_id = 'thread-default-owner' + `; + assert.equal(rows[0]?.owner, "user"); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/033_ThreadOwner.ts b/apps/server/src/persistence/Migrations/033_ThreadOwner.ts new file mode 100644 index 00000000000..e7239ebca86 --- /dev/null +++ b/apps/server/src/persistence/Migrations/033_ThreadOwner.ts @@ -0,0 +1,11 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN owner TEXT NOT NULL DEFAULT 'user' + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 44fdc147a4a..3791dd855d4 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -13,6 +13,7 @@ import { ProjectId, ProviderInteractionMode, RuntimeMode, + ThreadOwner, ThreadId, TurnId, } from "@t3tools/contracts"; @@ -27,6 +28,7 @@ export const ProjectionThread = Schema.Struct({ threadId: ThreadId, projectId: ProjectId, title: Schema.String, + owner: ThreadOwner, modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index fdf95df0b99..49b3a9d42c3 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -39,6 +39,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => getProjectShellById: (projectId) => Effect.succeed(projectId === project.id ? Option.some(project) : Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.die("unused"), + getThreadOwnerById: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.die("unused"), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.die("unused"), diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index e976c183a43..3c3a209175a 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -200,6 +200,7 @@ describe("ProviderSessionReaper", () => { getActiveProjectByWorkspaceRoot: () => Effect.die("unused"), getProjectShellById: () => Effect.die("unused"), getFirstActiveThreadIdByProjectId: () => Effect.die("unused"), + getThreadOwnerById: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.die("unused"), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: (threadId) => diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 40ed694723d..75ddc1a2c9d 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -464,6 +464,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { } satisfies OrchestrationEngineShape; const snapshotQuery = { + getThreadOwnerById: () => Effect.succeed(Option.some("user")), getShellSnapshot: () => Effect.succeed({ snapshotSequence: 1, @@ -538,6 +539,125 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { ), ); + it.effect("skips publishing plugin-owned threads to the relay", () => + Effect.scoped( + Effect.gen(function* () { + const originalFetch = globalThis.fetch; + const events = yield* Queue.unbounded(); + let fetchCalls = 0; + const secrets = makeMemorySecretStore(); + const now = "2026-05-25T00:00:00.000Z"; + const projectId = "project-1" as ProjectId; + const threadId = "thread-plugin" as ThreadId; + const environmentId = "env-1" as EnvironmentId; + + const project = { + id: projectId, + title: "T3 Code", + workspaceRoot: "/workspace", + repositoryIdentity: null, + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + } satisfies OrchestrationProjectShell; + + const thread = { + id: threadId, + projectId, + title: "Plugin worker", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: { + turnId: "turn-1" as TurnId, + state: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + assistantMessageId: null, + }, + createdAt: now, + updatedAt: now, + archivedAt: null, + session: null, + latestUserMessageAt: now, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + } satisfies OrchestrationThreadShell; + + globalThis.fetch = (() => { + fetchCalls += 1; + return Promise.resolve(Response.json({ ok: true, deliveries: [] })); + }) as unknown as typeof fetch; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + globalThis.fetch = originalFetch; + }), + ); + + const descriptor = { + environmentId, + label: "Test Desktop", + platform: { + os: "darwin", + arch: "arm64", + }, + serverVersion: "0.0.0-test", + capabilities: { + repositoryIdentity: true, + }, + } satisfies ExecutionEnvironmentDescriptor; + + const layer = Layer.mergeAll( + Layer.succeed(ServerSecretStore.ServerSecretStore, secrets.store), + Layer.succeed(ServerEnvironment.ServerEnvironment, { + getEnvironmentId: Effect.succeed(environmentId), + getDescriptor: Effect.succeed(descriptor), + }), + Layer.succeed(OrchestrationEngineService, { + readEvents: () => Stream.empty, + dispatch: () => Effect.succeed({ sequence: 1 }), + streamDomainEvents: Stream.fromQueue(events), + } satisfies OrchestrationEngineShape), + Layer.succeed(ProjectionSnapshotQuery, { + getThreadOwnerById: () => Effect.succeed(Option.some("plugin:test")), + getShellSnapshot: () => + Effect.succeed({ + snapshotSequence: 1, + projects: [project], + threads: [thread], + updatedAt: now, + } satisfies OrchestrationShellSnapshot), + getThreadShellById: () => Effect.succeed(Option.some(thread)), + getProjectShellById: () => Effect.succeed(Option.some(project)), + } as unknown as ProjectionSnapshotQueryShape), + ); + + yield* Effect.gen(function* () { + const relay = yield* AgentAwarenessRelay.AgentAwarenessRelay; + yield* secrets.setString(RELAY_URL_SECRET, "https://transport.example.test"); + yield* secrets.setString(RELAY_ISSUER_SECRET, "https://issuer.example.test"); + yield* secrets.setString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "relay-credential"); + yield* secrets.setString(PUBLISH_AGENT_ACTIVITY_SECRET, "true"); + yield* relay.publishThread(threadId); + + expect(fetchCalls).toBe(0); + }).pipe( + Effect.provide( + AgentAwarenessRelay.layer.pipe( + Layer.provide(layer), + Layer.provideMerge(NodeServices.layer), + ), + ), + ); + }), + ), + ); + it.effect("publishes agent activity to the relay transport URL, not the relay issuer", () => Effect.scoped( Effect.gen(function* () { @@ -652,6 +772,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { streamDomainEvents: Stream.fromQueue(events), } satisfies OrchestrationEngineShape), Layer.succeed(ProjectionSnapshotQuery, { + getThreadOwnerById: () => Effect.succeed(Option.some("user")), getShellSnapshot: () => Effect.succeed({ snapshotSequence: 1, diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 4e036e3ea0e..fd62c2f8fcc 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -323,6 +323,16 @@ export const make = Effect.gen(function* () { }); return; } + // A missing row (None) proceeds: deleted/unknown threads follow the same + // downstream not-found handling as before this check existed. + const threadOwner = yield* snapshotQuery.getThreadOwnerById(threadId); + if (Option.isSome(threadOwner) && threadOwner.value !== "user") { + yield* Effect.logDebug("agent activity publish skipped; thread is not user-owned", { + threadId, + owner: threadOwner.value, + }); + return; + } const relayClient = yield* makeRelayClient(relayConfig); const environmentId = yield* serverEnvironment.getEnvironmentId; diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 26528c84d34..466e443afd7 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -705,6 +705,7 @@ const buildAppUnderTest = (options?: { getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }), getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.none()), ...options?.layers?.projectionSnapshotQuery, }), diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index e331f0cd4d6..0bdce52916b 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -92,6 +92,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.succeed(Option.none()), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.succeed(Option.none()), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), @@ -154,6 +155,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa ), getProjectShellById: () => Effect.die("unused"), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.some(bootstrapThreadId)), + getThreadOwnerById: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), @@ -196,6 +198,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.die("unused"), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), @@ -244,6 +247,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), getProjectShellById: () => Effect.die("unused"), getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadOwnerById: () => Effect.die("unused"), getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 94eb1c65370..e8bdf7d7d27 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -26,6 +26,7 @@ const baseThread: OrchestrationThread = { projectId: ProjectId.make("project-1"), title: "Test Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + owner: "user", runtimeMode: "full-access", interactionMode: "default", branch: null, @@ -82,6 +83,7 @@ describe("applyThreadDetailEvent", () => { projectId: ProjectId.make("project-1"), title: "New Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + owner: "plugin:test", runtimeMode: "full-access", interactionMode: "default", branch: "main", @@ -95,6 +97,7 @@ describe("applyThreadDetailEvent", () => { if (result.kind === "updated") { expect(result.thread.id).toBe("thread-2"); expect(result.thread.title).toBe("New Thread"); + expect(result.thread.owner).toBe("plugin:test"); expect(result.thread.branch).toBe("main"); expect(result.thread.messages).toEqual([]); expect(result.thread.session).toBeNull(); diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 670540fee70..f1afc7708ce 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -64,6 +64,7 @@ export function applyThreadDetailEvent( projectId: event.payload.projectId, title: event.payload.title, modelSelection: event.payload.modelSelection, + owner: event.payload.owner ?? "user", runtimeMode: event.payload.runtimeMode, interactionMode: event.payload.interactionMode, branch: event.payload.branch, diff --git a/packages/contracts/src/orchestration.test.ts b/packages/contracts/src/orchestration.test.ts index 29a732ca69b..de6c642ef82 100644 --- a/packages/contracts/src/orchestration.test.ts +++ b/packages/contracts/src/orchestration.test.ts @@ -3,6 +3,7 @@ import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; import { + ClientOrchestrationCommand, DEFAULT_PROVIDER_INTERACTION_MODE, DEFAULT_RUNTIME_MODE, ModelSelection, @@ -11,12 +12,14 @@ import { OrchestrationGetFullThreadDiffInput, OrchestrationGetTurnDiffInput, OrchestrationLatestTurn, + OrchestrationThread, ProjectCreatedPayload, ProjectMetaUpdatedPayload, OrchestrationProposedPlan, OrchestrationSession, ProjectCreateCommand, ThreadMetaUpdatedPayload, + ThreadOwner, ThreadTurnStartCommand, ThreadCreatedPayload, ThreadTurnDiff, @@ -34,7 +37,9 @@ const decodeThreadTurnStartCommand = Schema.decodeUnknownEffect(ThreadTurnStartC const decodeThreadTurnStartRequestedPayload = Schema.decodeUnknownEffect( ThreadTurnStartRequestedPayload, ); +const decodeThreadOwner = Schema.decodeUnknownEffect(ThreadOwner); const decodeOrchestrationLatestTurn = Schema.decodeUnknownEffect(OrchestrationLatestTurn); +const decodeOrchestrationThread = Schema.decodeUnknownEffect(OrchestrationThread); const decodeOrchestrationProposedPlan = Schema.decodeUnknownEffect(OrchestrationProposedPlan); const decodeOrchestrationSession = Schema.decodeUnknownEffect(OrchestrationSession); const encodeThreadCreatedPayload = Schema.encodeEffect(ThreadCreatedPayload); @@ -46,6 +51,7 @@ function getOptionValue( return options?.find((option) => option.id === id)?.value; } const decodeThreadCreatedPayload = Schema.decodeUnknownEffect(ThreadCreatedPayload); +const decodeClientOrchestrationCommand = Schema.decodeUnknownEffect(ClientOrchestrationCommand); const decodeOrchestrationCommand = Schema.decodeUnknownEffect(OrchestrationCommand); const decodeOrchestrationEvent = Schema.decodeUnknownEffect(OrchestrationEvent); const decodeThreadMetaUpdatedPayload = Schema.decodeUnknownEffect(ThreadMetaUpdatedPayload); @@ -312,6 +318,130 @@ it.effect("decodes thread.created runtime mode for historical events", () => }), ); +it.effect("decodes thread ownership with legacy user defaults and plugin-owned ids", () => + Effect.gen(function* () { + assert.strictEqual(yield* decodeThreadOwner("user"), "user"); + assert.strictEqual(yield* decodeThreadOwner("plugin:test"), "plugin:test"); + + const invalidOwner = yield* Effect.exit(decodeThreadOwner("plugin:")); + assert.strictEqual(invalidOwner._tag, "Failure"); + + // Owner ids follow the plugin manifest id grammar: lowercase, digits, + // hyphens only. + const uppercaseOwner = yield* Effect.exit(decodeThreadOwner("plugin:Test")); + assert.strictEqual(uppercaseOwner._tag, "Failure"); + const underscoreOwner = yield* Effect.exit(decodeThreadOwner("plugin:my_plugin")); + assert.strictEqual(underscoreOwner._tag, "Failure"); + + // Encode direction: a decoded legacy payload re-encodes WITH an explicit + // owner — new serializations of old events are self-describing. + const encodedLegacy = yield* encodeThreadCreatedPayload( + yield* decodeThreadCreatedPayload({ + threadId: "thread-legacy-encode", + projectId: "project-1", + title: "Legacy encode", + modelSelection: { + provider: "codex", + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }), + ); + assert.strictEqual((encodedLegacy as { owner?: unknown }).owner, "user"); + + const payload = yield* decodeThreadCreatedPayload({ + threadId: "thread-1", + projectId: "project-1", + title: "Thread title", + modelSelection: { + provider: "codex", + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }); + assert.strictEqual(payload.owner, "user"); + + const thread = yield* decodeOrchestrationThread({ + id: "thread-1", + projectId: "project-1", + title: "Thread title", + modelSelection: { + provider: "codex", + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + archivedAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }); + assert.strictEqual(thread.owner, "user"); + + const command = yield* decodeOrchestrationCommand({ + type: "thread.create", + commandId: "cmd-thread-create", + threadId: "thread-plugin", + projectId: "project-1", + title: "Plugin thread", + owner: "plugin:test", + modelSelection: { + provider: "codex", + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:00.000Z", + }); + assert.strictEqual(command.type, "thread.create"); + if (command.type !== "thread.create") { + assert.fail(`Expected thread.create command, received ${command.type}.`); + } + assert.strictEqual("owner" in command, true); + assert.strictEqual((command as { owner?: unknown }).owner, "plugin:test"); + + const clientCommand = yield* decodeClientOrchestrationCommand({ + type: "thread.create", + commandId: "cmd-client-thread-create", + threadId: "thread-client", + projectId: "project-1", + title: "Client thread", + owner: "plugin:test", + modelSelection: { + provider: "codex", + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:00.000Z", + }); + assert.strictEqual(clientCommand.type, "thread.create"); + assert.strictEqual("owner" in clientCommand, false); + }), +); + it.effect("decodes thread.meta-updated payloads with explicit provider", () => Effect.gen(function* () { const parsed = yield* decodeThreadMetaUpdatedPayload({ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 623fed0917b..15cfcffef08 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -124,6 +124,16 @@ export const DEFAULT_RUNTIME_MODE: RuntimeMode = "full-access"; export const ProviderInteractionMode = Schema.Literals(["default", "plan"]); export type ProviderInteractionMode = typeof ProviderInteractionMode.Type; export const DEFAULT_PROVIDER_INTERACTION_MODE: ProviderInteractionMode = "default"; +const THREAD_PLUGIN_OWNER_PATTERN = /^plugin:[a-z][a-z0-9-]{1,40}$/; +export const ThreadOwner = Schema.Union([ + Schema.Literal("user"), + TrimmedNonEmptyString.check( + Schema.isMaxLength(128), + Schema.isPattern(THREAD_PLUGIN_OWNER_PATTERN), + ), +]); +export type ThreadOwner = typeof ThreadOwner.Type; +export const DEFAULT_THREAD_OWNER: ThreadOwner = "user"; export const ProviderRequestKind = Schema.Literals(["command", "file-read", "file-change"]); export type ProviderRequestKind = typeof ProviderRequestKind.Type; export const AssistantDeliveryMode = Schema.Literals(["buffered", "streaming"]); @@ -345,6 +355,9 @@ export const OrchestrationThread = Schema.Struct({ id: ThreadId, projectId: ProjectId, title: TrimmedNonEmptyString, + owner: Schema.optionalKey( + ThreadOwner.pipe(Schema.withDecodingDefault(Effect.succeed(DEFAULT_THREAD_OWNER))), + ), modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode.pipe( @@ -491,6 +504,23 @@ const ProjectDeleteCommand = Schema.Struct({ }); const ThreadCreateCommand = Schema.Struct({ + type: Schema.Literal("thread.create"), + commandId: CommandId, + threadId: ThreadId, + projectId: ProjectId, + title: TrimmedNonEmptyString, + owner: Schema.optional(ThreadOwner), + modelSelection: ModelSelection, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode.pipe( + Schema.withDecodingDefault(Effect.succeed(DEFAULT_PROVIDER_INTERACTION_MODE)), + ), + branch: Schema.NullOr(TrimmedNonEmptyString), + worktreePath: Schema.NullOr(TrimmedNonEmptyString), + createdAt: IsoDateTime, +}); + +const ClientThreadCreateCommand = Schema.Struct({ type: Schema.Literal("thread.create"), commandId: CommandId, threadId: ThreadId, @@ -661,7 +691,7 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ProjectCreateCommand, ProjectMetaUpdateCommand, ProjectDeleteCommand, - ThreadCreateCommand, + ClientThreadCreateCommand, ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, @@ -682,7 +712,7 @@ export const ClientOrchestrationCommand = Schema.Union([ ProjectCreateCommand, ProjectMetaUpdateCommand, ProjectDeleteCommand, - ThreadCreateCommand, + ClientThreadCreateCommand, ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, @@ -775,6 +805,7 @@ const InternalOrchestrationCommand = Schema.Union([ export type InternalOrchestrationCommand = typeof InternalOrchestrationCommand.Type; export const OrchestrationCommand = Schema.Union([ + ThreadCreateCommand, DispatchableClientOrchestrationCommand, InternalOrchestrationCommand, ]); @@ -840,6 +871,9 @@ export const ThreadCreatedPayload = Schema.Struct({ threadId: ThreadId, projectId: ProjectId, title: TrimmedNonEmptyString, + owner: Schema.optionalKey( + ThreadOwner.pipe(Schema.withDecodingDefault(Effect.succeed(DEFAULT_THREAD_OWNER))), + ), modelSelection: ModelSelection, runtimeMode: RuntimeMode.pipe(Schema.withDecodingDefault(Effect.succeed(DEFAULT_RUNTIME_MODE))), interactionMode: ProviderInteractionMode.pipe(