diff --git a/apps/server/package.json b/apps/server/package.json index d0903c77d75..8e5f35f4b89 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -30,6 +30,7 @@ "@ff-labs/fff-node": "0.9.4", "@opencode-ai/sdk": "^1.3.15", "@pierre/diffs": "catalog:", + "@t3tools/plugin-sdk": "workspace:*", "effect": "catalog:", "node-pty": "^1.1.0" }, 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/config.ts b/apps/server/src/config.ts index 2608ccc16ae..dde1feeb349 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -28,6 +28,7 @@ export type StartupPresentation = typeof StartupPresentation.Type; export interface ServerDerivedPaths { readonly stateDir: string; readonly dbPath: string; + readonly pluginsDir: string; readonly keybindingsConfigPath: string; readonly settingsPath: string; readonly providerStatusCacheDir: string; @@ -95,6 +96,7 @@ export const deriveServerPaths = Effect.fn(function* ( const { join } = yield* Path.Path; const stateDir = join(baseDir, devUrl !== undefined ? "dev" : "userdata"); const dbPath = join(stateDir, "state.sqlite"); + const pluginsDir = join(stateDir, "plugins"); const attachmentsDir = join(stateDir, "attachments"); const logsDir = join(stateDir, "logs"); const providerLogsDir = join(logsDir, "provider"); @@ -102,6 +104,7 @@ export const deriveServerPaths = Effect.fn(function* ( return { stateDir, dbPath, + pluginsDir, keybindingsConfigPath: join(stateDir, "keybindings.json"), settingsPath: join(stateDir, "settings.json"), providerStatusCacheDir, 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..c17105e2f8b 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -45,6 +45,8 @@ 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"; +import Migration0034 from "./Migrations/034_PluginMigrations.ts"; /** * Migration loader with all migrations defined inline. @@ -89,6 +91,8 @@ export const migrationEntries = [ [30, "ProjectionThreadShellArchiveIndexes", Migration0030], [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], + [33, "ThreadOwner", Migration0033], + [34, "PluginMigrations", Migration0034], ] 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/Migrations/034_PluginMigrations.test.ts b/apps/server/src/persistence/Migrations/034_PluginMigrations.test.ts new file mode 100644 index 00000000000..a67a02b0579 --- /dev/null +++ b/apps/server/src/persistence/Migrations/034_PluginMigrations.test.ts @@ -0,0 +1,37 @@ +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("034_PluginMigrations", (it) => { + it.effect("creates plugin migration tracking table", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 34 }); + + const tables = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master + WHERE type = 'table' AND name = 'plugin_migrations' + `; + assert.equal(tables.length, 1); + + yield* sql` + INSERT INTO plugin_migrations (plugin_id, version, name, applied_at) + VALUES ('test-plugin', 1, 'Init', '2026-07-03T00:00:00.000Z') + `; + + const rows = yield* sql<{ readonly version: number; readonly name: string }>` + SELECT version, name + FROM plugin_migrations + WHERE plugin_id = 'test-plugin' + `; + assert.deepEqual(rows, [{ version: 1, name: "Init" }]); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/034_PluginMigrations.ts b/apps/server/src/persistence/Migrations/034_PluginMigrations.ts new file mode 100644 index 00000000000..d09c074c2af --- /dev/null +++ b/apps/server/src/persistence/Migrations/034_PluginMigrations.ts @@ -0,0 +1,16 @@ +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` + CREATE TABLE plugin_migrations ( + plugin_id TEXT NOT NULL, + version INTEGER NOT NULL, + name TEXT NOT NULL, + applied_at TEXT NOT NULL, + PRIMARY KEY (plugin_id, version) + ) + `; +}); 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/plugins/PluginHost.test.ts b/apps/server/src/plugins/PluginHost.test.ts new file mode 100644 index 00000000000..36ffc847804 --- /dev/null +++ b/apps/server/src/plugins/PluginHost.test.ts @@ -0,0 +1,299 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { PluginId, PluginManifest, type PluginLockfilePlugin } from "@t3tools/contracts/plugin"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import { pathToFileURL } from "node:url"; + +import * as ServerConfig from "../config.ts"; +import { runMigrations } from "../persistence/Migrations.ts"; +import * as NodeSqliteClient from "../persistence/NodeSqliteClient.ts"; +import * as PluginHostModule from "./PluginHost.ts"; +import * as PluginLockfileStoreLayer from "./PluginLockfileStore.ts"; +import * as PluginMigrator from "./PluginMigrator.ts"; +import * as PluginModuleLoaderLayer from "./PluginModuleLoader.ts"; +import { pluginDataDir, pluginVersionDir } from "./PluginPaths.ts"; +import * as PluginRuntimeRegistryLayer from "./PluginRuntimeRegistry.ts"; + +const encodeManifestJson = Schema.encodeEffect(Schema.fromJsonString(PluginManifest)); + +const testLayer = PluginHostModule.layer.pipe( + Layer.provideMerge(PluginLockfileStoreLayer.layer), + Layer.provideMerge(PluginModuleLoaderLayer.layer), + Layer.provideMerge(PluginMigrator.layer), + Layer.provideMerge(PluginRuntimeRegistryLayer.layer), + Layer.provideMerge(NodeSqliteClient.layerMemory()), + Layer.provideMerge( + Layer.fresh(ServerConfig.layerTest(process.cwd(), { prefix: "t3-plugin-host-" })), + ), + Layer.provideMerge(NodeServices.layer), +); + +const layer = it.layer(testLayer); + +const now = "2026-07-03T00:00:00.000Z"; + +const makeLockEntry = (overrides: Partial = {}): PluginLockfilePlugin => ({ + version: "1.0.0", + sha256: "sha", + sourceId: "local", + enabled: true, + state: "active", + activation: { activatingSince: null, crashCount: 0 }, + installedAt: now, + lastError: null, + ...overrides, +}); + +const pluginEntrySource = () => ` +import { createRequire } from "node:module"; +const require = createRequire(${JSON.stringify(pathToFileURL(import.meta.url).href)}); +const Effect = require("effect/Effect"); +const SqlClient = require("effect/unstable/sql/SqlClient"); +const NodeFs = require("node:fs"); + +export default { + register(hostApi) { + return { + migrations: [ + { + version: 1, + name: "Init", + up: Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql\`CREATE TABLE p_test_plugin_items (id TEXT PRIMARY KEY)\`; + }), + }, + ], + services: [ + { + name: "marker", + run: () => + Effect.sync(() => { + NodeFs.writeFileSync(hostApi.config.dataDir + "/service-ran", "1"); + }).pipe(Effect.andThen(Effect.never)), + }, + ], + }; + }, +}; +`; + +const installPlugin = (input: { + readonly pluginId: PluginId; + readonly manifestHostApi?: string; + readonly entrySource?: string; + readonly lockEntry?: Partial; +}) => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const store = yield* PluginLockfileStoreLayer.PluginLockfileStore; + const entry = makeLockEntry(input.lockEntry); + const pluginDir = pluginVersionDir(config.pluginsDir, input.pluginId, entry.version, path.join); + + yield* fs.makeDirectory(pluginDir, { recursive: true }); + const encodedManifest = yield* encodeManifestJson({ + id: input.pluginId, + name: "Test Plugin", + version: entry.version, + hostApi: input.manifestHostApi ?? "^1.0.0", + capabilities: [], + entries: { server: "server.js" }, + }); + yield* fs.writeFileString(path.join(pluginDir, "manifest.json"), encodedManifest); + yield* fs.writeFileString( + path.join(pluginDir, "server.js"), + input.entrySource ?? pluginEntrySource(), + ); + yield* store.updatePlugin(input.pluginId, () => Effect.succeed(entry)); + return { pluginDir, entry }; + }); + +layer("PluginModuleLoader", (it) => { + it.effect("loads a definePlugin-shaped default export from inside the plugin dir", () => + Effect.gen(function* () { + const pluginId = PluginId.make("loader-plugin"); + const loader = yield* PluginModuleLoaderLayer.PluginModuleLoader; + const { pluginDir } = yield* installPlugin({ + pluginId, + entrySource: "export default { register() { return {}; } };", + }); + + const definition = yield* loader.loadServerEntry(pluginDir, "server.js"); + + assert.equal(typeof definition.register, "function"); + }), + ); + + it.effect("rejects entries that resolve outside the plugin dir", () => + Effect.gen(function* () { + const pluginId = PluginId.make("loader-escape"); + const loader = yield* PluginModuleLoaderLayer.PluginModuleLoader; + const { pluginDir } = yield* installPlugin({ + pluginId, + entrySource: "export default { register() { return {}; } };", + }); + + const result = yield* Effect.result(loader.loadServerEntry(pluginDir, "../server.js")); + + assert.isTrue(Result.isFailure(result)); + }), + ); +}); + +layer("PluginHost", (it) => { + it.effect("activates a plugin, records migrations, starts services, and clears activation", () => + Effect.gen(function* () { + const pluginId = PluginId.make("test-plugin"); + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const sql = yield* SqlClient.SqlClient; + const host = yield* PluginHostModule.PluginHost; + const registry = yield* PluginRuntimeRegistryLayer.PluginRuntimeRegistry; + const store = yield* PluginLockfileStoreLayer.PluginLockfileStore; + const previousHealthyDelay = process.env.T3_PLUGIN_HOST_HEALTHY_DELAY_MS; + + yield* runMigrations({ toMigrationInclusive: 34 }); + yield* installPlugin({ pluginId }); + + process.env.T3_PLUGIN_HOST_HEALTHY_DELAY_MS = "0"; + try { + yield* host.start; + yield* Effect.yieldNow; + } finally { + if (previousHealthyDelay === undefined) { + delete process.env.T3_PLUGIN_HOST_HEALTHY_DELAY_MS; + } else { + process.env.T3_PLUGIN_HOST_HEALTHY_DELAY_MS = previousHealthyDelay; + } + } + + const runtimes = yield* registry.list; + assert.equal(runtimes.length, 1); + const migrationRows = yield* sql<{ readonly version: number }>` + SELECT version FROM plugin_migrations WHERE plugin_id = ${pluginId} + `; + assert.deepEqual(migrationRows, [{ version: 1 }]); + assert.isTrue( + yield* fs.exists( + path.join(pluginDataDir(config.pluginsDir, pluginId, path.join), "service-ran"), + ), + ); + + let lockfile = yield* store.readLockfile; + for (let attempt = 0; attempt < 5; attempt++) { + if (lockfile.plugins[pluginId]?.activation.activatingSince === null) break; + yield* Effect.yieldNow; + lockfile = yield* store.readLockfile; + } + assert.equal(lockfile.plugins[pluginId]?.activation.activatingSince, null); + assert.equal(lockfile.plugins[pluginId]?.activation.crashCount, 0); + }), + ); + + it.effect("marks failed imports without failing host startup", () => + Effect.gen(function* () { + const pluginId = PluginId.make("failed-plugin"); + const host = yield* PluginHostModule.PluginHost; + const registry = yield* PluginRuntimeRegistryLayer.PluginRuntimeRegistry; + const store = yield* PluginLockfileStoreLayer.PluginLockfileStore; + + yield* runMigrations({ toMigrationInclusive: 34 }); + yield* installPlugin({ pluginId, entrySource: "throw new Error('boom');" }); + + yield* host.start; + + const runtimes = yield* registry.list; + const lockfile = yield* store.readLockfile; + assert.isFalse(runtimes.some((runtime) => runtime.manifest.id === pluginId)); + assert.equal(lockfile.plugins[pluginId]?.state, "failed"); + }), + ); + + it.effect("disables crash-looping plugins before import", () => + Effect.gen(function* () { + const pluginId = PluginId.make("crash-plugin"); + const host = yield* PluginHostModule.PluginHost; + const store = yield* PluginLockfileStoreLayer.PluginLockfileStore; + + yield* installPlugin({ + pluginId, + lockEntry: { + activation: { activatingSince: now, crashCount: 1 }, + }, + }); + + yield* host.start; + + const lockfile = yield* store.readLockfile; + assert.equal(lockfile.plugins[pluginId]?.state, "failed"); + assert.equal(lockfile.plugins[pluginId]?.lastError, "disabled after repeated crashes"); + }), + ); + + it.effect("does not load anything when T3_NO_PLUGINS is set", () => + Effect.gen(function* () { + const pluginId = PluginId.make("disabled-env"); + const host = yield* PluginHostModule.PluginHost; + const registry = yield* PluginRuntimeRegistryLayer.PluginRuntimeRegistry; + const previous = process.env.T3_NO_PLUGINS; + + yield* installPlugin({ pluginId }); + process.env.T3_NO_PLUGINS = "1"; + try { + yield* host.start; + } finally { + if (previous === undefined) { + delete process.env.T3_NO_PLUGINS; + } else { + process.env.T3_NO_PLUGINS = previous; + } + } + + const runtimes = yield* registry.list; + assert.isFalse(runtimes.some((runtime) => runtime.manifest.id === pluginId)); + }), + ); + + it.effect("sets disabled-by-host when hostApi range is not satisfied", () => + Effect.gen(function* () { + const pluginId = PluginId.make("host-mismatch"); + const host = yield* PluginHostModule.PluginHost; + const store = yield* PluginLockfileStoreLayer.PluginLockfileStore; + + yield* installPlugin({ pluginId, manifestHostApi: "^2.0.0" }); + + yield* host.start; + + const lockfile = yield* store.readLockfile; + assert.equal(lockfile.plugins[pluginId]?.state, "disabled-by-host"); + }), + ); + + it.effect("applies pending-remove before loading plugins", () => + Effect.gen(function* () { + const pluginId = PluginId.make("remove-plugin"); + const host = yield* PluginHostModule.PluginHost; + const fs = yield* FileSystem.FileSystem; + const store = yield* PluginLockfileStoreLayer.PluginLockfileStore; + const { pluginDir } = yield* installPlugin({ + pluginId, + lockEntry: { state: "pending-remove" }, + }); + + yield* host.start; + + const lockfile = yield* store.readLockfile; + assert.isUndefined(lockfile.plugins[pluginId]); + assert.isFalse(yield* fs.exists(pluginDir)); + }), + ); +}); diff --git a/apps/server/src/plugins/PluginHost.ts b/apps/server/src/plugins/PluginHost.ts new file mode 100644 index 00000000000..7ad9f32edbd --- /dev/null +++ b/apps/server/src/plugins/PluginHost.ts @@ -0,0 +1,434 @@ +import { + HOST_API_VERSION, + PluginManifest, + hostApiSatisfies, + type PluginId, + type PluginLockfile, + type PluginLockfilePlugin, +} from "@t3tools/contracts/plugin"; +import type { + PluginDefinition, + PluginHostApi, + PluginLogger, + PluginRegistration, + PluginServiceDescriptor, +} from "@t3tools/plugin-sdk"; +import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; + +import packageJson from "../../package.json" with { type: "json" }; +import * as ServerConfig from "../config.ts"; +import { PluginLockfileStore } from "./PluginLockfileStore.ts"; +import { PluginMigrator } from "./PluginMigrator.ts"; +import { PluginModuleLoader } from "./PluginModuleLoader.ts"; +import { pluginDataDir, pluginManifestPath, pluginVersionDir } from "./PluginPaths.ts"; +import { PluginRuntimeRegistry } from "./PluginRuntimeRegistry.ts"; + +const APP_VERSION = packageJson.version; +const decodeManifest = Schema.decodeUnknownEffect(Schema.fromJsonString(PluginManifest)); + +const healthyActivationDelay = () => { + const overrideMs = Number.parseInt(process.env.T3_PLUGIN_HOST_HEALTHY_DELAY_MS ?? "", 10); + return Number.isFinite(overrideMs) && overrideMs >= 0 + ? Duration.millis(overrideMs) + : Duration.seconds(30); +}; + +export class PluginRegistrationError extends Schema.TaggedErrorClass()( + "PluginRegistrationError", + { pluginId: Schema.String, detail: Schema.String }, +) { + override get message(): string { + return `Plugin ${this.pluginId} returned an invalid registration: ${this.detail}`; + } +} + +export class PluginCapabilityUnavailable extends Schema.TaggedErrorClass()( + "PluginCapabilityUnavailable", + { capability: Schema.String }, +) { + override get message(): string { + return `Capability ${this.capability} is not available in this host build.`; + } +} + +export class PluginHost extends Context.Service< + PluginHost, + { + readonly start: Effect.Effect; + } +>()("t3/plugins/PluginHost") {} + +function isPromiseLike(value: unknown): value is Promise { + return typeof value === "object" && value !== null && "then" in value; +} + +const resolveRegistration = ( + pluginId: PluginId, + definition: PluginDefinition, + hostApi: PluginHostApi, +) => + Effect.suspend(() => { + const value = definition.register(hostApi); + if (Effect.isEffect(value)) return value; + if (isPromiseLike(value)) return Effect.promise(() => value as Promise); + return Effect.succeed(value); + }).pipe( + Effect.catchCause((cause) => + Effect.fail( + new PluginRegistrationError({ + pluginId, + detail: Cause.pretty(cause), + }), + ), + ), + ); + +function validateRegistration( + pluginId: PluginId, + registration: PluginRegistration, +): Effect.Effect { + const methods = new Set(); + for (const rpc of registration.rpc ?? []) { + if (rpc.scope !== "read" && rpc.scope !== "operate") { + return Effect.fail( + new PluginRegistrationError({ pluginId, detail: `invalid RPC scope ${rpc.scope}` }), + ); + } + if (methods.has(rpc.method)) { + return Effect.fail( + new PluginRegistrationError({ pluginId, detail: `duplicate RPC method ${rpc.method}` }), + ); + } + methods.add(rpc.method); + } + return Effect.void; +} + +const makeLogger = (pluginId: PluginId): PluginLogger => ({ + debug: (message, attributes) => Effect.logDebug(message, { ...attributes, pluginId }), + info: (message, attributes) => Effect.logInfo(message, { ...attributes, pluginId }), + warn: (message, attributes) => Effect.logWarning(message, { ...attributes, pluginId }), + error: (message, attributes) => Effect.logError(message, { ...attributes, pluginId }), +}); + +const unavailable = (capability: string) => + Effect.die(new PluginCapabilityUnavailable({ capability })); + +const makeHostApi = (input: { + readonly pluginId: PluginId; + readonly dataDir: string; + readonly logger: PluginLogger; +}): PluginHostApi => ({ + hostApiVersion: HOST_API_VERSION, + config: { + appVersion: APP_VERSION, + hostApiVersion: HOST_API_VERSION, + dataDir: input.dataDir, + logger: input.logger, + }, + agents: unavailable("agents"), + vcs: unavailable("vcs"), + terminals: unavailable("terminals"), + database: unavailable("database"), + projectionsRead: unavailable("projections.read"), + environmentsRead: unavailable("environments.read"), + secrets: unavailable("secrets"), + http: unavailable("http"), + sourceControl: unavailable("sourceControl"), + textGeneration: unavailable("textGeneration"), +}); + +const upgradeLockfileEntry = ( + entry: PluginLockfilePlugin, + staged: NonNullable, +): PluginLockfilePlugin => ({ + version: staged.version, + sha256: staged.sha256, + sourceId: entry.sourceId, + enabled: entry.enabled, + state: "active", + activation: entry.activation, + installedAt: entry.installedAt, + lastError: entry.lastError, +}); + +const getLockfilePlugin = (lockfile: PluginLockfile, pluginId: PluginId) => + (lockfile.plugins as Readonly>)[pluginId]; + +const updateFailure = ( + store: PluginLockfileStore["Service"], + pluginId: PluginId, + message: string, +) => + store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + state: "failed", + lastError: message, + activation: { + ...current.activation, + activatingSince: null, + }, + } + : undefined, + ), + ); + +const startService = (input: { + readonly pluginId: PluginId; + readonly logger: PluginLogger; + readonly service: PluginServiceDescriptor; +}) => + input.service.run({ pluginId: input.pluginId, logger: input.logger }).pipe( + Effect.catchCause((cause) => + input.logger.error("plugin service failed; restarting", { + service: input.service.name, + cause: Cause.pretty(cause), + }), + ), + // Exponential backoff capped at 30s so a flapping service keeps + // retrying at a bounded cadence instead of backing off forever. + Effect.repeat( + Schedule.either(Schedule.exponential("250 millis"), Schedule.spaced("30 seconds")), + ), + ); + +export const make = Effect.fn("PluginHost.make")(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const store = yield* PluginLockfileStore; + const loader = yield* PluginModuleLoader; + const migrator = yield* PluginMigrator; + const registry = yield* PluginRuntimeRegistry; + const clock = yield* Clock.Clock; + + const readManifest = (pluginDir: string) => + fs + .readFileString(pluginManifestPath(pluginDir, path.join)) + .pipe(Effect.flatMap(decodeManifest)); + + const loadPlugin = (pluginId: PluginId, entry: PluginLockfilePlugin) => + Effect.gen(function* () { + const pluginDir = pluginVersionDir(config.pluginsDir, pluginId, entry.version, path.join); + const manifest = yield* readManifest(pluginDir); + if (manifest.id !== pluginId) { + return yield* new PluginRegistrationError({ + pluginId, + detail: `manifest id ${manifest.id} does not match lockfile id`, + }); + } + if (!hostApiSatisfies(manifest.hostApi, HOST_API_VERSION)) { + yield* store.updatePlugin(pluginId, ({ current }) => + Effect.succeed(current ? { ...current, state: "disabled-by-host" } : undefined), + ); + yield* Effect.logWarning("Plugin disabled by host API version mismatch", { + pluginId, + requested: manifest.hostApi, + hostApiVersion: HOST_API_VERSION, + }); + return; + } + if (!manifest.entries.server) { + yield* Effect.logDebug("Skipping web-only plugin in server plugin host", { pluginId }); + return; + } + + const serverEntry = manifest.entries.server; + const serverEntryPath = path.join(pluginDir, serverEntry); + if (!(yield* fs.exists(pluginDir)) || !(yield* fs.exists(serverEntryPath))) { + yield* updateFailure(store, pluginId, "plugin directory or server entry is missing"); + return; + } + + const activatingSince = DateTime.formatIso(yield* DateTime.now); + yield* store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + activation: { + ...current.activation, + activatingSince, + }, + } + : undefined, + ), + ); + + const scope = yield* Scope.make("sequential"); + const readiness = yield* Deferred.make(); + const logger = makeLogger(pluginId); + const dataDir = pluginDataDir(config.pluginsDir, pluginId, path.join); + const hostApi = makeHostApi({ pluginId, dataDir, logger }); + + const activation = Effect.gen(function* () { + yield* fs.makeDirectory(dataDir, { recursive: true }); + const definition = yield* loader.loadServerEntry(pluginDir, serverEntry); + const registration = yield* resolveRegistration(pluginId, definition, hostApi); + yield* validateRegistration(pluginId, registration); + yield* migrator.run(pluginId, registration.migrations ?? []); + if (registration.recover) { + yield* registration.recover(); + } + yield* registry.put(pluginId, { manifest, registration, readiness, scope }); + for (const service of registration.services ?? []) { + yield* startService({ pluginId, logger, service }).pipe( + Effect.forkScoped, + Scope.provide(scope), + ); + } + yield* Deferred.succeed(readiness, undefined).pipe(Effect.orDie); + const clearHealthyActivation = store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + activation: { activatingSince: null, crashCount: 0 }, + lastError: null, + } + : undefined, + ), + ); + const healthyDelay = healthyActivationDelay(); + if (Duration.toMillis(healthyDelay) === 0) { + yield* clearHealthyActivation; + } else { + yield* clock.sleep(healthyDelay).pipe( + Effect.flatMap(() => clearHealthyActivation), + Effect.ignoreCause({ log: true }), + Effect.forkScoped, + Scope.provide(scope), + ); + } + }); + + const exit = yield* activation.pipe(Scope.provide(scope), Effect.exit); + if (Exit.isFailure(exit)) { + yield* Scope.close(scope, exit); + const message = Cause.pretty(exit.cause); + yield* updateFailure(store, pluginId, message); + yield* Effect.logWarning("Plugin activation failed", { pluginId, cause: message }); + } + }); + + const reconcilePendingState = (pluginId: PluginId, entry: PluginLockfilePlugin) => + Effect.gen(function* () { + if (entry.state === "pending-remove") { + yield* fs.remove(path.join(config.pluginsDir, pluginId), { recursive: true, force: true }); + yield* store.removePlugin(pluginId); + return false; + } + if (entry.state === "pending-upgrade") { + if (!entry.staged) { + yield* updateFailure( + store, + pluginId, + "pending upgrade is missing staged plugin metadata", + ); + return false; + } + const staged = entry.staged; + yield* store.updatePlugin(pluginId, ({ current }) => + Effect.succeed(current ? upgradeLockfileEntry(current, staged) : undefined), + ); + return true; + } + if (entry.activation.activatingSince !== null) { + const crashCount = entry.activation.crashCount + 1; + if (crashCount >= 2) { + yield* store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + state: "failed", + lastError: "disabled after repeated crashes", + activation: { activatingSince: null, crashCount }, + } + : undefined, + ), + ); + return false; + } + yield* store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + activation: { activatingSince: null, crashCount }, + } + : undefined, + ), + ); + } + return true; + }); + + const start = Effect.gen(function* () { + if (process.env.T3_NO_PLUGINS === "1") { + yield* Effect.logInfo("Plugin host disabled by T3_NO_PLUGINS"); + return; + } + if (!(yield* fs.exists(store.lockfilePath).pipe(Effect.orElseSucceed(() => false)))) { + return; + } + yield* loader.ensureHostSingletonResolution; + const lockfile = yield* store.readLockfile.pipe( + Effect.catch((error) => + Effect.logWarning("Plugin host could not read lockfile", { + path: store.lockfilePath, + error: error.message, + }).pipe(Effect.as({ plugins: {}, sources: [] })), + ), + ); + + for (const [rawPluginId, entry] of Object.entries(lockfile.plugins)) { + const pluginId = rawPluginId as PluginId; + const shouldContinue = yield* reconcilePendingState(pluginId, entry).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Plugin pending-state reconciliation failed", { + pluginId, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), + ); + if (!shouldContinue || !entry.enabled) continue; + const currentLockfile = yield* store.readLockfile.pipe(Effect.orElseSucceed(() => lockfile)); + const currentEntry = getLockfilePlugin(currentLockfile, pluginId); + if (!currentEntry?.enabled || currentEntry.state !== "active") continue; + yield* loadPlugin(pluginId, currentEntry).pipe( + Effect.catchCause((cause) => + updateFailure(store, pluginId, Cause.pretty(cause)).pipe( + Effect.andThen( + Effect.logWarning("Plugin activation failed before scope acquisition", { + pluginId, + cause: Cause.pretty(cause), + }), + ), + Effect.ignore, + ), + ), + ); + } + }).pipe(Effect.ignoreCause({ log: true })); + + return PluginHost.of({ start }); +}); + +export const layer = Layer.effect(PluginHost, make()); diff --git a/apps/server/src/plugins/PluginLockfileStore.test.ts b/apps/server/src/plugins/PluginLockfileStore.test.ts new file mode 100644 index 00000000000..8650835776d --- /dev/null +++ b/apps/server/src/plugins/PluginLockfileStore.test.ts @@ -0,0 +1,146 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { PluginId, type PluginLockfilePlugin } from "@t3tools/contracts/plugin"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Result from "effect/Result"; +import * as TestClock from "effect/testing/TestClock"; + +import * as ServerConfig from "../config.ts"; +import * as PluginLockfileStoreModule from "./PluginLockfileStore.ts"; +import { + PluginLockfileCorruptError, + PluginLockfileTransitionError, +} from "./PluginLockfileStore.ts"; + +const layer = it.layer( + PluginLockfileStoreModule.layer.pipe( + Layer.provideMerge( + Layer.fresh(ServerConfig.layerTest(process.cwd(), { prefix: "t3-plugin-lockfile-" })), + ), + Layer.provideMerge(NodeServices.layer), + Layer.provideMerge(TestClock.layer()), + ), +); + +const pluginId = PluginId.make("test-plugin"); + +const makePlugin = (overrides: Partial = {}): PluginLockfilePlugin => ({ + version: "1.0.0", + sha256: "sha", + sourceId: "local", + enabled: true, + state: "active", + activation: { activatingSince: null, crashCount: 0 }, + installedAt: "2026-07-03T00:00:00.000Z", + lastError: null, + ...overrides, +}); + +layer("PluginLockfileStore", (it) => { + it.effect("returns an empty lockfile when plugins.json is missing", () => + Effect.gen(function* () { + const store = yield* PluginLockfileStoreModule.PluginLockfileStore; + + const lockfile = yield* store.readLockfile; + + assert.deepEqual(lockfile, { sources: [], plugins: {} }); + }), + ); + + it.effect("returns a typed error for corrupt lockfile JSON", () => + Effect.gen(function* () { + const store = yield* PluginLockfileStoreModule.PluginLockfileStore; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + + yield* fs.makeDirectory(path.dirname(store.lockfilePath), { recursive: true }); + yield* fs.writeFileString(store.lockfilePath, "{not-json"); + + const result = yield* Effect.result(store.readLockfile); + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginLockfileCorruptError); + } + yield* fs.remove(store.lockfilePath, { force: true }); + }), + ); + + it.effect("serializes concurrent mutations so both updates apply", () => + Effect.gen(function* () { + const store = yield* PluginLockfileStoreModule.PluginLockfileStore; + + yield* store.updatePlugin(pluginId, () => Effect.succeed(makePlugin())); + yield* Effect.all( + [ + store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + activation: { + ...current.activation, + crashCount: current.activation.crashCount + 1, + }, + } + : undefined, + ), + ), + store.updatePlugin(pluginId, ({ current }) => + Effect.succeed( + current + ? { + ...current, + activation: { + ...current.activation, + crashCount: current.activation.crashCount + 1, + }, + } + : undefined, + ), + ), + ], + { concurrency: "unbounded" }, + ); + + const lockfile = yield* store.readLockfile; + assert.equal(lockfile.plugins[pluginId]?.activation.crashCount, 2); + }), + ); + + it.effect("reclaims stale advisory locks", () => + Effect.gen(function* () { + const store = yield* PluginLockfileStoreModule.PluginLockfileStore; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + + yield* fs.makeDirectory(path.dirname(store.advisoryLockPath), { recursive: true }); + yield* fs.writeFileString(store.advisoryLockPath, "stale"); + const stale = DateTime.toDateUtc(DateTime.makeUnsafe("1970-01-01T00:00:00.000Z")); + yield* fs.utimes(store.advisoryLockPath, stale, stale); + yield* TestClock.setTime(120_000); + + yield* store.updatePlugin(pluginId, () => Effect.succeed(makePlugin())); + + const lockfile = yield* store.readLockfile; + assert.equal(lockfile.plugins[pluginId]?.version, "1.0.0"); + }), + ); + + it.effect("rejects invalid state transitions", () => + Effect.gen(function* () { + const store = yield* PluginLockfileStoreModule.PluginLockfileStore; + + yield* store.updatePlugin(pluginId, () => Effect.succeed(makePlugin({ state: "disabled" }))); + const result = yield* Effect.result(store.transition(pluginId, ["active"], "failed")); + + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginLockfileTransitionError); + } + }), + ); +}); diff --git a/apps/server/src/plugins/PluginLockfileStore.ts b/apps/server/src/plugins/PluginLockfileStore.ts new file mode 100644 index 00000000000..4649599aa30 --- /dev/null +++ b/apps/server/src/plugins/PluginLockfileStore.ts @@ -0,0 +1,327 @@ +import { + EMPTY_PLUGIN_LOCKFILE, + PluginId, + PluginLockfile, + PluginState, + type PluginLockfilePlugin, +} from "@t3tools/contracts/plugin"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; + +import * as ServerConfig from "../config.ts"; +import { pluginAdvisoryLockPath, pluginLockfilePath } from "./PluginPaths.ts"; + +const STALE_LOCK_MS = 60_000; +const PluginLockfileJson = Schema.fromJsonString(PluginLockfile); +const decodePluginLockfileJson = Schema.decodeUnknownEffect(PluginLockfileJson); +const encodePluginLockfileJson = Schema.encodeEffect(PluginLockfileJson); + +export class PluginLockfileReadError extends Schema.TaggedErrorClass()( + "PluginLockfileReadError", + { path: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Could not read plugin lockfile at ${this.path}.`; + } +} + +export class PluginLockfileCorruptError extends Schema.TaggedErrorClass()( + "PluginLockfileCorruptError", + { path: Schema.String, detail: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Plugin lockfile at ${this.path} is corrupt: ${this.detail}`; + } +} + +export class PluginLockfileWriteError extends Schema.TaggedErrorClass()( + "PluginLockfileWriteError", + { path: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Could not write plugin lockfile at ${this.path}.`; + } +} + +export class PluginLockfileLockError extends Schema.TaggedErrorClass()( + "PluginLockfileLockError", + { path: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Could not acquire plugin lockfile advisory lock at ${this.path}.`; + } +} + +export class PluginLockfileTransitionError extends Schema.TaggedErrorClass()( + "PluginLockfileTransitionError", + { + pluginId: PluginId, + from: Schema.Array(PluginState), + to: PluginState, + actual: Schema.NullOr(PluginState), + }, +) { + override get message(): string { + return `Cannot transition plugin ${this.pluginId} from ${this.actual ?? "missing"} to ${this.to}.`; + } +} + +export type PluginLockfileStoreError = + | PluginLockfileReadError + | PluginLockfileCorruptError + | PluginLockfileWriteError + | PluginLockfileLockError + | PluginLockfileTransitionError; + +export interface PluginLockfileMutationContext { + readonly lockfile: PluginLockfile; + readonly current: PluginLockfilePlugin | undefined; +} + +export class PluginLockfileStore extends Context.Service< + PluginLockfileStore, + { + readonly lockfilePath: string; + readonly advisoryLockPath: string; + readonly readLockfile: Effect.Effect< + PluginLockfile, + PluginLockfileReadError | PluginLockfileCorruptError + >; + readonly updatePlugin: ( + id: PluginId, + fn: ( + context: PluginLockfileMutationContext, + ) => Effect.Effect, + ) => Effect.Effect; + readonly removePlugin: ( + id: PluginId, + ) => Effect.Effect; + readonly transition: ( + id: PluginId, + from: ReadonlyArray, + to: PluginState, + ) => Effect.Effect; + } +>()("t3/plugins/PluginLockfileStore") {} + +const isNotFound = (cause: { readonly reason?: { readonly _tag?: string } }) => + cause.reason?._tag === "NotFound"; + +const readLockfileFromPath = (lockfilePath: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const raw = yield* fs + .readFileString(lockfilePath) + .pipe( + Effect.catch((cause) => + isNotFound(cause) + ? Effect.succeed(null) + : Effect.fail(new PluginLockfileReadError({ path: lockfilePath, cause })), + ), + ); + if (raw === null) return EMPTY_PLUGIN_LOCKFILE; + return yield* decodePluginLockfileJson(raw).pipe( + Effect.mapError( + (cause) => + new PluginLockfileCorruptError({ + path: lockfilePath, + detail: String(cause), + cause, + }), + ), + ); + }); + +const writeLockfileToPath = (input: { + readonly pluginsDir: string; + readonly lockfilePath: string; + readonly lockfile: PluginLockfile; +}) => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const encoded = yield* encodePluginLockfileJson(input.lockfile); + const bytes = new TextEncoder().encode(`${encoded}\n`); + + yield* fs.makeDirectory(input.pluginsDir, { recursive: true }); + const tempDir = yield* fs.makeTempDirectoryScoped({ + directory: input.pluginsDir, + prefix: `${path.basename(input.lockfilePath)}.`, + }); + const tempPath = path.join(tempDir, "contents.tmp"); + const file = yield* fs.open(tempPath, { flag: "w", mode: 0o600 }); + yield* file.writeAll(bytes); + yield* file.sync; + yield* fs.rename(tempPath, input.lockfilePath); + yield* fs.open(input.pluginsDir, { flag: "r" }).pipe( + Effect.flatMap((directory) => directory.sync), + Effect.ignore, + ); + }), + ).pipe( + Effect.mapError((cause) => new PluginLockfileWriteError({ path: input.lockfilePath, cause })), + ); + +const acquireAdvisoryLock = (input: { + readonly pluginsDir: string; + readonly advisoryLockPath: string; +}) => + Effect.acquireRelease( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + yield* fs + .makeDirectory(input.pluginsDir, { recursive: true }) + .pipe( + Effect.mapError( + (cause) => new PluginLockfileLockError({ path: input.advisoryLockPath, cause }), + ), + ); + + const openLock = Effect.scoped( + Effect.gen(function* () { + const file = yield* fs.open(input.advisoryLockPath, { flag: "wx", mode: 0o600 }); + yield* file.writeAll( + new TextEncoder().encode(`${process.pid}:${yield* Clock.currentTimeMillis}\n`), + ); + yield* file.sync; + }), + ); + + const opened = yield* openLock.pipe(Effect.result); + if (Result.isSuccess(opened)) return input.advisoryLockPath; + + const stat = yield* fs + .stat(input.advisoryLockPath) + .pipe( + Effect.mapError( + (cause) => new PluginLockfileLockError({ path: input.advisoryLockPath, cause }), + ), + ); + const mtime = Option.getOrUndefined(stat.mtime); + const ageMs = mtime ? (yield* Clock.currentTimeMillis) - mtime.getTime() : 0; + if (ageMs <= STALE_LOCK_MS) { + return yield* new PluginLockfileLockError({ + path: input.advisoryLockPath, + cause: opened.failure, + }); + } + + yield* Effect.logWarning("Reclaiming stale plugin lockfile advisory lock", { + path: input.advisoryLockPath, + ageMs, + }); + yield* fs + .remove(input.advisoryLockPath, { force: true }) + .pipe( + Effect.mapError( + (cause) => new PluginLockfileLockError({ path: input.advisoryLockPath, cause }), + ), + ); + yield* openLock.pipe( + Effect.mapError( + (cause) => new PluginLockfileLockError({ path: input.advisoryLockPath, cause }), + ), + ); + return input.advisoryLockPath; + }), + (lockPath) => + FileSystem.FileSystem.pipe( + Effect.flatMap((fs) => fs.remove(lockPath, { force: true })), + Effect.ignore, + ), + ); + +export const make = Effect.fn("PluginLockfileStore.make")(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const semaphore = yield* Semaphore.make(1); + const lockfilePath = pluginLockfilePath(config.pluginsDir, path.join); + const advisoryLockPath = pluginAdvisoryLockPath(config.pluginsDir, path.join); + const provideLocalServices = ( + effect: Effect.Effect, + ) => + effect.pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + ); + + const readLockfile = provideLocalServices(readLockfileFromPath(lockfilePath)); + + const mutate = ( + update: (lockfile: PluginLockfile) => Effect.Effect, + ) => + provideLocalServices( + semaphore.withPermits(1)( + Effect.scoped( + acquireAdvisoryLock({ pluginsDir: config.pluginsDir, advisoryLockPath }).pipe( + Effect.flatMap(() => + Effect.gen(function* () { + const current = yield* readLockfile; + const next = yield* update(current); + yield* writeLockfileToPath({ + pluginsDir: config.pluginsDir, + lockfilePath, + lockfile: next, + }); + return next; + }), + ), + ), + ), + ), + ); + + const updatePlugin: PluginLockfileStore["Service"]["updatePlugin"] = (id, fn) => + mutate((lockfile) => + Effect.gen(function* () { + const current = lockfile.plugins[id]; + const nextPlugin = yield* fn({ lockfile, current }); + const plugins = { ...lockfile.plugins }; + if (nextPlugin === undefined) { + delete plugins[id]; + } else { + plugins[id] = nextPlugin; + } + return { ...lockfile, plugins }; + }), + ); + + const removePlugin: PluginLockfileStore["Service"]["removePlugin"] = (id) => + updatePlugin(id, () => Effect.succeed(undefined as PluginLockfilePlugin | undefined)); + + const transition: PluginLockfileStore["Service"]["transition"] = (id, from, to) => + updatePlugin(id, ({ current }) => { + if (!current || !from.includes(current.state)) { + return Effect.fail( + new PluginLockfileTransitionError({ + pluginId: id, + from: Array.from(from), + to, + actual: current?.state ?? null, + }), + ); + } + return Effect.succeed({ ...current, state: to }); + }); + + return PluginLockfileStore.of({ + lockfilePath, + advisoryLockPath, + readLockfile, + updatePlugin, + removePlugin, + transition, + }); +}); + +export const layer = Layer.effect(PluginLockfileStore, make()); diff --git a/apps/server/src/plugins/PluginMigrator.test.ts b/apps/server/src/plugins/PluginMigrator.test.ts new file mode 100644 index 00000000000..3caaf501fdf --- /dev/null +++ b/apps/server/src/plugins/PluginMigrator.test.ts @@ -0,0 +1,265 @@ +import { assert, it } from "@effect/vitest"; +import { PluginId } from "@t3tools/contracts/plugin"; +import type { PluginMigration } from "@t3tools/plugin-sdk"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Result from "effect/Result"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../persistence/Migrations.ts"; +import * as NodeSqliteClient from "../persistence/NodeSqliteClient.ts"; +import * as PluginMigratorModule from "./PluginMigrator.ts"; +import { PluginMigrationDowngradeError, PluginMigrationViolation } from "./PluginMigrator.ts"; + +const layer = it.layer( + PluginMigratorModule.layer.pipe(Layer.provideMerge(NodeSqliteClient.layerMemory())), +); + +const pluginPrefix = (pluginId: PluginId) => `p_${pluginId.replaceAll("-", "_")}_`; + +const migration = ( + version: number, + name: string, + statements: string | ReadonlyArray, +): PluginMigration => ({ + version, + name, + up: Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + for (const statement of Array.isArray(statements) ? statements : [statements]) { + yield* sql.unsafe(statement).unprepared; + } + }), +}); + +const setup = Effect.gen(function* () { + yield* runMigrations({ toMigrationInclusive: 34 }); + return yield* PluginMigratorModule.PluginMigrator; +}); + +layer("PluginMigrator", (it) => { + it.effect("runs migrations once and records applied rows", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const migrator = yield* setup; + const pluginId = PluginId.make("test-plugin"); + const prefix = pluginPrefix(pluginId); + const migrations = [ + migration(1, "Init", `CREATE TABLE ${prefix}items (id TEXT PRIMARY KEY)`), + ]; + + yield* migrator.run(pluginId, migrations); + yield* migrator.run(pluginId, migrations); + + const rows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM plugin_migrations WHERE plugin_id = ${pluginId} + `; + assert.equal(rows[0]?.count, 1); + }), + ); + + it.effect("refuses downgrades when recorded head exceeds provided migrations", () => + Effect.gen(function* () { + const migrator = yield* setup; + const pluginId = PluginId.make("downgrade-plugin"); + const prefix = pluginPrefix(pluginId); + + yield* migrator.run(pluginId, [ + migration(1, "Init", `CREATE TABLE ${prefix}items (id TEXT PRIMARY KEY)`), + migration(2, "Next", `CREATE TABLE ${prefix}more (id TEXT PRIMARY KEY)`), + ]); + + const result = yield* Effect.result( + migrator.run(pluginId, [ + migration(1, "Init", `CREATE TABLE ${prefix}items (id TEXT PRIMARY KEY)`), + ]), + ); + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationDowngradeError); + } + }), + ); + + it.effect("rolls back non-prefixed tables and does not record the row", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const migrator = yield* setup; + const pluginId = PluginId.make("bad-table-plugin"); + + const result = yield* Effect.result( + migrator.run(pluginId, [migration(1, "Bad", "CREATE TABLE bad_items (id TEXT)")]), + ); + + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationViolation); + } + const tables = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master WHERE name = 'bad_items' + `; + const rows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM plugin_migrations WHERE plugin_id = ${pluginId} + `; + assert.deepEqual(tables, []); + assert.equal(rows[0]?.count, 0); + }), + ); + + it.effect("rejects triggers that reference pre-existing core tables", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const migrator = yield* setup; + const pluginId = PluginId.make("trigger-plugin"); + const prefix = pluginPrefix(pluginId); + yield* sql`CREATE TABLE core_items (id TEXT PRIMARY KEY)`; + + const result = yield* Effect.result( + migrator.run(pluginId, [ + migration(1, "Trigger", [ + `CREATE TABLE ${prefix}items (id TEXT PRIMARY KEY)`, + ` + CREATE TRIGGER ${prefix}items_ai + AFTER INSERT ON ${prefix}items + BEGIN + INSERT INTO core_items (id) VALUES (NEW.id); + END + `, + ]), + ]), + ); + + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationViolation); + } + }), + ); + + it.effect("rejects migrations that drop objects outside the plugin namespace", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const migrator = yield* setup; + const pluginId = PluginId.make("drop-plugin"); + yield* sql`CREATE TABLE core_victim (id TEXT PRIMARY KEY)`; + + const result = yield* Effect.result( + migrator.run(pluginId, [migration(1, "Drop", "DROP TABLE core_victim")]), + ); + + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationViolation); + } + const tables = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master WHERE name = 'core_victim' + `; + assert.equal(tables.length, 1); + }), + ); + + it.effect("rejects migrations that rename a core table into the plugin namespace", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const migrator = yield* setup; + const pluginId = PluginId.make("rename-plugin"); + const prefix = pluginPrefix(pluginId); + yield* sql`CREATE TABLE core_renamed (id TEXT PRIMARY KEY)`; + + const result = yield* Effect.result( + migrator.run(pluginId, [ + migration(1, "Rename", `ALTER TABLE core_renamed RENAME TO ${prefix}stolen`), + ]), + ); + + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationViolation); + } + const tables = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master WHERE name = 'core_renamed' + `; + assert.equal(tables.length, 1); + }), + ); + + it.effect("rejects ATTACH DATABASE in plugin migrations", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const migrator = yield* setup; + const pluginId = PluginId.make("attach-plugin"); + + // SQLite itself refuses ATTACH inside the migration transaction, so + // this surfaces as a migration failure; the PRAGMA database_list gate + // additionally covers a migration that breaks out of the transaction. + const result = yield* Effect.result( + migrator.run(pluginId, [ + migration(1, "Attach", [ + "ATTACH DATABASE ':memory:' AS escape_hatch", + "CREATE TABLE escape_hatch.evil (id TEXT)", + ]), + ]), + ); + + assert.isTrue(Result.isFailure(result)); + const rows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM plugin_migrations WHERE plugin_id = ${pluginId} + `; + assert.equal(rows[0]?.count, 0); + }), + ); + + it.effect("rejects TEMP objects in plugin migrations", () => + Effect.gen(function* () { + const migrator = yield* setup; + const pluginId = PluginId.make("temp-plugin"); + + const result = yield* Effect.result( + migrator.run(pluginId, [migration(1, "Temp", "CREATE TEMP TABLE sneaky (id TEXT)")]), + ); + + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationViolation); + } + }), + ); + + it.effect("treats an empty migration list as a no-op even with recorded history", () => + Effect.gen(function* () { + const migrator = yield* setup; + const pluginId = PluginId.make("empty-plugin"); + const prefix = pluginPrefix(pluginId); + + yield* migrator.run(pluginId, [ + migration(1, "Init", `CREATE TABLE ${prefix}items (id TEXT PRIMARY KEY)`), + ]); + yield* migrator.run(pluginId, []); + }), + ); + + it.effect("enforces prefixes for indexes and views", () => + Effect.gen(function* () { + const migrator = yield* setup; + const pluginId = PluginId.make("view-plugin"); + const prefix = pluginPrefix(pluginId); + + yield* migrator.run(pluginId, [ + migration(1, "Index", [ + `CREATE TABLE ${prefix}items (id TEXT PRIMARY KEY)`, + `CREATE INDEX ${prefix}items_id_idx ON ${prefix}items (id)`, + ]), + ]); + + const result = yield* Effect.result( + migrator.run(pluginId, [ + migration(2, "BadView", `CREATE VIEW bad_items_view AS SELECT id FROM ${prefix}items`), + ]), + ); + assert.isTrue(Result.isFailure(result)); + if (Result.isFailure(result)) { + assert.instanceOf(result.failure, PluginMigrationViolation); + } + }), + ); +}); diff --git a/apps/server/src/plugins/PluginMigrator.ts b/apps/server/src/plugins/PluginMigrator.ts new file mode 100644 index 00000000000..5ef03ccbe5e --- /dev/null +++ b/apps/server/src/plugins/PluginMigrator.ts @@ -0,0 +1,289 @@ +import type { PluginId } from "@t3tools/contracts/plugin"; +import type { PluginMigration } from "@t3tools/plugin-sdk"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import type { SqlError } from "effect/unstable/sql/SqlError"; + +export class PluginMigrationDowngradeError extends Schema.TaggedErrorClass()( + "PluginMigrationDowngradeError", + { + pluginId: Schema.String, + recordedHead: Schema.Number, + providedHead: Schema.Number, + }, +) { + override get message(): string { + return `Plugin ${this.pluginId} has migration head ${this.recordedHead}, but only migrations through ${this.providedHead} were provided.`; + } +} + +export class PluginMigrationViolation extends Schema.TaggedErrorClass()( + "PluginMigrationViolation", + { + pluginId: Schema.String, + version: Schema.Number, + objectName: Schema.String, + detail: Schema.String, + }, +) { + override get message(): string { + return `Plugin ${this.pluginId} migration ${this.version} violated database namespace rules for ${this.objectName}: ${this.detail}`; + } +} + +export class PluginMigrationOrderError extends Schema.TaggedErrorClass()( + "PluginMigrationOrderError", + { pluginId: Schema.String, detail: Schema.String }, +) { + override get message(): string { + return `Plugin ${this.pluginId} migration list is invalid: ${this.detail}`; + } +} + +export class PluginMigrationExecutionError extends Schema.TaggedErrorClass()( + "PluginMigrationExecutionError", + { + pluginId: Schema.String, + version: Schema.Number, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Plugin ${this.pluginId} migration ${this.version} failed.`; + } +} + +export type PluginMigratorError = + | PluginMigrationDowngradeError + | PluginMigrationViolation + | PluginMigrationOrderError + | PluginMigrationExecutionError + | SqlError; + +interface SqliteMasterObject { + readonly name: string; + readonly type: string; + readonly sql: string | null; +} + +export class PluginMigrator extends Context.Service< + PluginMigrator, + { + readonly run: ( + pluginId: PluginId, + migrations: ReadonlyArray, + ) => Effect.Effect; + } +>()("t3/plugins/PluginMigrator") {} + +const pluginSqlPrefix = (pluginId: string) => `p_${pluginId.replaceAll("-", "_")}_`; + +const sqliteMasterSnapshot = (sql: SqlClient.SqlClient) => + sql` + SELECT name, type, sql + FROM sqlite_master + WHERE type IN ('table', 'index', 'trigger', 'view') + AND name NOT LIKE 'sqlite_%' + ORDER BY type, name + `; + +const objectKey = (entry: SqliteMasterObject) => `${entry.type}:${entry.name}`; + +const changedObjects = ( + before: ReadonlyArray, + after: ReadonlyArray, +) => { + const beforeByKey = new Map(before.map((entry) => [objectKey(entry), entry])); + return after.filter((entry) => beforeByKey.get(objectKey(entry))?.sql !== entry.sql); +}; + +const removedObjects = ( + before: ReadonlyArray, + after: ReadonlyArray, +) => { + const afterKeys = new Set(after.map(objectKey)); + return before.filter((entry) => !afterKeys.has(objectKey(entry))); +}; + +const validateMigrationObjects = (input: { + readonly pluginId: PluginId; + readonly version: number; + readonly prefix: string; + readonly before: ReadonlyArray; + readonly after: ReadonlyArray; +}) => + Effect.gen(function* () { + const preMigrationCoreTables = input.before + .filter((entry) => entry.type === "table" && !entry.name.startsWith(input.prefix)) + .map((entry) => entry.name); + + // Dropping (or renaming away) an object the plugin does not own is a + // violation: a dropped object never appears in the after-snapshot, so it + // must be detected from the before side. + for (const entry of removedObjects(input.before, input.after)) { + if (!entry.name.startsWith(input.prefix)) { + return yield* new PluginMigrationViolation({ + pluginId: input.pluginId, + version: input.version, + objectName: entry.name, + detail: "migration removed an object outside the plugin namespace", + }); + } + } + + for (const entry of changedObjects(input.before, input.after)) { + if (!entry.name.startsWith(input.prefix)) { + return yield* new PluginMigrationViolation({ + pluginId: input.pluginId, + version: input.version, + objectName: entry.name, + detail: `object name must start with ${input.prefix}`, + }); + } + if (entry.type !== "trigger" && entry.type !== "view") continue; + const body = entry.sql ?? ""; + for (const tableName of preMigrationCoreTables) { + if ( + new RegExp(`\\b${tableName.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}\\b`, "i").test(body) + ) { + return yield* new PluginMigrationViolation({ + pluginId: input.pluginId, + version: input.version, + objectName: entry.name, + detail: `trigger/view body references core table ${tableName}`, + }); + } + } + } + }); + +const validateMigrationList = (pluginId: PluginId, migrations: ReadonlyArray) => + Effect.gen(function* () { + const versions = new Set(); + for (const migration of migrations) { + if (!Number.isInteger(migration.version) || migration.version <= 0) { + return yield* new PluginMigrationOrderError({ + pluginId, + detail: `version ${migration.version} must be a positive integer`, + }); + } + if (versions.has(migration.version)) { + return yield* new PluginMigrationOrderError({ + pluginId, + detail: `duplicate version ${migration.version}`, + }); + } + versions.add(migration.version); + } + }); + +export const make = Effect.fn("PluginMigrator.make")(function* () { + const sql = yield* SqlClient.SqlClient; + + const run: PluginMigrator["Service"]["run"] = (pluginId, migrations) => + Effect.gen(function* () { + // No migrations means nothing to run — not a downgrade (a plugin may + // legitimately ship no migrations even after earlier versions did). + if (migrations.length === 0) return; + yield* validateMigrationList(pluginId, migrations); + const sorted = Array.from(migrations).sort((left, right) => left.version - right.version); + const providedHead = sorted.at(-1)?.version ?? 0; + const rows = yield* sql<{ readonly version: number | null }>` + SELECT MAX(version) AS version + FROM plugin_migrations + WHERE plugin_id = ${pluginId} + `; + const recordedHead = rows[0]?.version ?? 0; + if (recordedHead > providedHead) { + return yield* new PluginMigrationDowngradeError({ + pluginId, + recordedHead, + providedHead, + }); + } + + const prefix = pluginSqlPrefix(pluginId); + for (const migration of sorted) { + if (migration.version <= recordedHead) continue; + yield* sql.withTransaction( + Effect.gen(function* () { + const before = yield* sqliteMasterSnapshot(sql); + const tempBefore = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_temp_master + `.pipe(Effect.orElseSucceed(() => [])); + yield* migration.up.pipe( + Effect.provideService(SqlClient.SqlClient, sql), + Effect.mapError( + (cause) => + new PluginMigrationExecutionError({ + pluginId, + version: migration.version, + cause, + }), + ), + ); + const after = yield* sqliteMasterSnapshot(sql); + // ATTACH and TEMP objects live outside the main sqlite_master + // snapshot, so the diff gate cannot see them — forbid them + // outright rather than pretend they are covered. + // database_list always reports "main" (and "temp" once the temp + // schema exists); anything else is an ATTACHed database. + const databases = yield* sql<{ readonly name: string }>`PRAGMA database_list`; + const attached = databases.find( + (database) => database.name !== "main" && database.name !== "temp", + ); + if (attached) { + // Best-effort DETACH so a rogue attach cannot persist on the + // shared connection past this violation. + yield* sql + .unsafe(`DETACH DATABASE "${attached.name.replaceAll('"', '""')}"`) + .unprepared.pipe(Effect.ignore); + return yield* new PluginMigrationViolation({ + pluginId, + version: migration.version, + objectName: attached.name, + detail: "ATTACH DATABASE is not permitted in plugin migrations", + }); + } + const tempAfter = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_temp_master + `.pipe(Effect.orElseSucceed(() => [])); + const tempBeforeNames = new Set(tempBefore.map((row) => row.name)); + const newTempObject = tempAfter.find((row) => !tempBeforeNames.has(row.name)); + if (newTempObject) { + return yield* new PluginMigrationViolation({ + pluginId, + version: migration.version, + objectName: newTempObject.name, + detail: "TEMP objects are not permitted in plugin migrations", + }); + } + yield* validateMigrationObjects({ + pluginId, + version: migration.version, + prefix, + before, + after, + }); + yield* sql` + INSERT INTO plugin_migrations (plugin_id, version, name, applied_at) + VALUES ( + ${pluginId}, + ${migration.version}, + ${migration.name}, + ${DateTime.formatIso(yield* DateTime.now)} + ) + `; + }), + ); + } + }); + + return PluginMigrator.of({ run }); +}); + +export const layer = Layer.effect(PluginMigrator, make()); diff --git a/apps/server/src/plugins/PluginModuleLoader.ts b/apps/server/src/plugins/PluginModuleLoader.ts new file mode 100644 index 00000000000..3ae58d521dc --- /dev/null +++ b/apps/server/src/plugins/PluginModuleLoader.ts @@ -0,0 +1,144 @@ +import type { PluginDefinition } from "@t3tools/plugin-sdk"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import { pathToFileURL } from "node:url"; + +import * as ServerConfig from "../config.ts"; + +export class PluginModuleLoadError extends Schema.TaggedErrorClass()( + "PluginModuleLoadError", + { pluginDir: Schema.String, entry: Schema.String, cause: Schema.Defect() }, +) { + override get message(): string { + return `Could not load plugin server entry ${this.entry} from ${this.pluginDir}.`; + } +} + +export class PluginModulePathError extends Schema.TaggedErrorClass()( + "PluginModulePathError", + { pluginDir: Schema.String, entry: Schema.String, resolvedPath: Schema.String }, +) { + override get message(): string { + return `Plugin server entry ${this.entry} resolves outside ${this.pluginDir}.`; + } +} + +export class PluginModuleShapeError extends Schema.TaggedErrorClass()( + "PluginModuleShapeError", + { pluginDir: Schema.String, entry: Schema.String }, +) { + override get message(): string { + return `Plugin server entry ${this.entry} does not default-export a definePlugin-shaped object.`; + } +} + +export type PluginModuleLoaderError = + | PluginModuleLoadError + | PluginModulePathError + | PluginModuleShapeError; + +export class PluginModuleLoader extends Context.Service< + PluginModuleLoader, + { + readonly ensureHostSingletonResolution: Effect.Effect; + readonly loadServerEntry: ( + pluginDir: string, + entryRelPath: string, + ) => Effect.Effect; + } +>()("t3/plugins/PluginModuleLoader") {} + +let hostResolutionHookRegistered = false; + +function isPluginDefinition(value: unknown): value is PluginDefinition { + return ( + typeof value === "object" && + value !== null && + "register" in value && + typeof (value as { readonly register?: unknown }).register === "function" + ); +} + +function isInside(parent: string, child: string, separator: string): boolean { + return ( + child === parent || + child.startsWith(parent.endsWith(separator) ? parent : `${parent}${separator}`) + ); +} + +export const make = Effect.fn("PluginModuleLoader.make")(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + + const ensureHostSingletonResolution = Effect.gen(function* () { + if (hostResolutionHookRegistered) return; + hostResolutionHookRegistered = true; + const nodeModule = yield* Effect.promise(() => import("node:module")); + if (typeof nodeModule.register !== "function") { + yield* Effect.logWarning( + "Node module.register is unavailable; plugin host singleton resolution is disabled", + ); + return; + } + yield* Effect.sync(() => + nodeModule.register(new URL("./pluginResolveHooks.ts", import.meta.url), { + parentURL: import.meta.url, + data: { + pluginsRootUrl: pathToFileURL(config.pluginsDir).href, + }, + }), + ).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to register plugin host singleton resolution hook", { + cause, + }), + ), + ); + }); + + const loadServerEntry: PluginModuleLoader["Service"]["loadServerEntry"] = ( + pluginDir, + entryRelPath, + ) => + Effect.gen(function* () { + const realPluginDir = yield* fs + .realPath(pluginDir) + .pipe( + Effect.mapError( + (cause) => new PluginModuleLoadError({ pluginDir, entry: entryRelPath, cause }), + ), + ); + const resolvedEntry = path.resolve(realPluginDir, entryRelPath); + const realEntry = yield* fs + .realPath(resolvedEntry) + .pipe( + Effect.mapError( + (cause) => new PluginModuleLoadError({ pluginDir, entry: entryRelPath, cause }), + ), + ); + if (!isInside(realPluginDir, realEntry, path.sep)) { + return yield* new PluginModulePathError({ + pluginDir, + entry: entryRelPath, + resolvedPath: realEntry, + }); + } + const imported = yield* Effect.tryPromise({ + try: () => import(pathToFileURL(realEntry).href), + catch: (cause) => new PluginModuleLoadError({ pluginDir, entry: entryRelPath, cause }), + }); + if (!isPluginDefinition(imported.default)) { + return yield* new PluginModuleShapeError({ pluginDir, entry: entryRelPath }); + } + return imported.default; + }); + + return PluginModuleLoader.of({ ensureHostSingletonResolution, loadServerEntry }); +}); + +export const layer = Layer.effect(PluginModuleLoader, make()); diff --git a/apps/server/src/plugins/PluginPaths.ts b/apps/server/src/plugins/PluginPaths.ts new file mode 100644 index 00000000000..7f8c29b7ae3 --- /dev/null +++ b/apps/server/src/plugins/PluginPaths.ts @@ -0,0 +1,34 @@ +import type { PluginId } from "@t3tools/contracts/plugin"; + +export const pluginsRoot = ( + stateDir: string, + join: (...segments: ReadonlyArray) => string, +) => join(stateDir, "plugins"); + +export const pluginVersionDir = ( + root: string, + id: PluginId | string, + version: string, + join: (...segments: ReadonlyArray) => string, +) => join(root, id, version); + +export const pluginDataDir = ( + root: string, + id: PluginId | string, + join: (...segments: ReadonlyArray) => string, +) => join(root, id, "data"); + +export const pluginManifestPath = ( + pluginDir: string, + join: (...segments: ReadonlyArray) => string, +) => join(pluginDir, "manifest.json"); + +export const pluginLockfilePath = ( + root: string, + join: (...segments: ReadonlyArray) => string, +) => join(root, "plugins.json"); + +export const pluginAdvisoryLockPath = ( + root: string, + join: (...segments: ReadonlyArray) => string, +) => join(root, "plugins.json.lock"); diff --git a/apps/server/src/plugins/PluginRuntimeRegistry.ts b/apps/server/src/plugins/PluginRuntimeRegistry.ts new file mode 100644 index 00000000000..8abcccbd340 --- /dev/null +++ b/apps/server/src/plugins/PluginRuntimeRegistry.ts @@ -0,0 +1,52 @@ +import type { PluginId, PluginManifest } from "@t3tools/contracts/plugin"; +import type { PluginRegistration } from "@t3tools/plugin-sdk"; +import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import type * as Scope from "effect/Scope"; + +export interface ActivePluginRuntime { + readonly manifest: PluginManifest; + readonly registration: PluginRegistration; + readonly readiness: Deferred.Deferred; + readonly scope: Scope.Scope; +} + +export class PluginRuntimeRegistry extends Context.Service< + PluginRuntimeRegistry, + { + readonly put: (pluginId: PluginId, runtime: ActivePluginRuntime) => Effect.Effect; + readonly remove: (pluginId: PluginId) => Effect.Effect; + readonly list: Effect.Effect>; + readonly get: (pluginId: PluginId) => Effect.Effect>; + } +>()("t3/plugins/PluginRuntimeRegistry") {} + +export const make = Effect.fn("PluginRuntimeRegistry.make")(function* () { + const runtimes = yield* Ref.make(new Map()); + + return PluginRuntimeRegistry.of({ + put: (pluginId, runtime) => + Ref.update(runtimes, (current) => { + const next = new Map(current); + next.set(pluginId, runtime); + return next; + }), + remove: (pluginId) => + Ref.update(runtimes, (current) => { + const next = new Map(current); + next.delete(pluginId); + return next; + }), + list: Ref.get(runtimes).pipe(Effect.map((current) => Array.from(current.values()))), + get: (pluginId) => + Ref.get(runtimes).pipe( + Effect.map((current) => Option.fromUndefinedOr(current.get(pluginId))), + ), + }); +}); + +export const layer = Layer.effect(PluginRuntimeRegistry, make()); diff --git a/apps/server/src/plugins/pluginResolveHooks.ts b/apps/server/src/plugins/pluginResolveHooks.ts new file mode 100644 index 00000000000..1bada6a1094 --- /dev/null +++ b/apps/server/src/plugins/pluginResolveHooks.ts @@ -0,0 +1,34 @@ +let pluginsRootUrl = ""; + +export function initialize(data: unknown) { + if (data && typeof data === "object" && "pluginsRootUrl" in data) { + const value = (data as { readonly pluginsRootUrl?: unknown }).pluginsRootUrl; + if (typeof value === "string") { + pluginsRootUrl = value.endsWith("/") ? value : `${value}/`; + } + } +} + +function shouldResolveFromHost(specifier: string, parentURL: string | undefined): boolean { + if (!parentURL || !pluginsRootUrl || !parentURL.startsWith(pluginsRootUrl)) return false; + return ( + specifier === "effect" || specifier.startsWith("effect/") || specifier === "@t3tools/plugin-sdk" + ); +} + +export async function resolve( + specifier: string, + context: { readonly parentURL?: string | undefined }, + nextResolve: ( + specifier: string, + context: { readonly parentURL?: string | undefined }, + ) => Promise, +) { + if (shouldResolveFromHost(specifier, context.parentURL)) { + return { + shortCircuit: true, + url: import.meta.resolve(specifier), + }; + } + return nextResolve(specifier, context); +} 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/server.ts b/apps/server/src/server.ts index 0c632d8486c..a20d859c09a 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -32,6 +32,11 @@ import * as BitbucketApi from "./sourceControl/BitbucketApi.ts"; import * as GitHubCli from "./sourceControl/GitHubCli.ts"; import * as GitLabCli from "./sourceControl/GitLabCli.ts"; import * as TextGeneration from "./textGeneration/TextGeneration.ts"; +import * as PluginHost from "./plugins/PluginHost.ts"; +import * as PluginLockfileStore from "./plugins/PluginLockfileStore.ts"; +import * as PluginMigrator from "./plugins/PluginMigrator.ts"; +import * as PluginModuleLoader from "./plugins/PluginModuleLoader.ts"; +import * as PluginRuntimeRegistry from "./plugins/PluginRuntimeRegistry.ts"; import { ProviderInstanceRegistryHydrationLive } from "./provider/Layers/ProviderInstanceRegistryHydration.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; @@ -183,6 +188,13 @@ const ProviderLayerLive = ProviderServiceLive.pipe( const PersistenceLayerLive = Layer.empty.pipe(Layer.provideMerge(SqlitePersistenceLayerLive)); +const PluginLayerLive = PluginHost.layer.pipe( + Layer.provideMerge(PluginLockfileStore.layer), + Layer.provideMerge(PluginModuleLoader.layer), + Layer.provideMerge(PluginMigrator.layer), + Layer.provideMerge(PluginRuntimeRegistry.layer), +); + const VcsDriverRegistryLayerLive = VcsDriverRegistry.layer.pipe( Layer.provide(VcsProjectConfig.layer), ); @@ -284,7 +296,7 @@ const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( Layer.provideMerge(OrchestrationLayerLive), ); -const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( +const RuntimeCoreBaseDependenciesLive = ReactorLayerLive.pipe( // Core Services Layer.provideMerge(CheckpointingLayerLive), Layer.provideMerge(SourceControlProviderRegistryLayerLive), @@ -293,6 +305,10 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(ProviderRuntimeLayerLive), Layer.provideMerge(Layer.mergeAll(TerminalLayerLive, PreviewLayerLive)), Layer.provideMerge(PersistenceLayerLive), + Layer.provideMerge(PluginLayerLive), +); + +const RuntimeCoreDependenciesLive = RuntimeCoreBaseDependenciesLive.pipe( Layer.provideMerge(Keybindings.layer), Layer.provideMerge(ProviderRegistryLive), // The instance registry is the new routing keystone — text generation, 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/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index b52b577c5b5..0f4b8dee939 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -34,6 +34,7 @@ import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts"; +import * as PluginHost from "./plugins/PluginHost.ts"; import { formatHeadlessServeOutput, formatHostForUrl, @@ -293,6 +294,7 @@ export const make = Effect.gen(function* () { const keybindings = yield* Keybindings.Keybindings; const orchestrationReactor = yield* OrchestrationReactor.OrchestrationReactor; const providerSessionReaper = yield* ProviderSessionReaper.ProviderSessionReaper; + const pluginHost = yield* PluginHost.PluginHost; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; const serverSettings = yield* ServerSettings.ServerSettingsService; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; @@ -430,6 +432,8 @@ export const make = Effect.gen(function* () { yield* Effect.logDebug("Accepting commands"); yield* commandGate.signalCommandReady; + yield* Effect.logDebug("startup phase: starting plugin host"); + yield* runStartupPhase("plugins.start", pluginHost.start); yield* Effect.logDebug("startup phase: waiting for http listener"); yield* runStartupPhase("http.wait", Deferred.await(httpListening)); yield* Effect.logDebug("startup phase: publishing ready event"); diff --git a/apps/server/vite.config.ts b/apps/server/vite.config.ts index 473df069ed7..38fc4a66b44 100644 --- a/apps/server/vite.config.ts +++ b/apps/server/vite.config.ts @@ -30,7 +30,11 @@ export default mergeConfig( }, }, pack: { - entry: ["src/bin.ts"], + // The ESM plugin-host resolution hook must ship as its OWN module file: + // it is loaded by URL in a loader-hook worker via `module.register` (see + // PluginModuleLoader), so it cannot be inlined into bin.mjs. Emitting it as + // a second entry produces dist/pluginResolveHooks.mjs alongside bin.mjs. + entry: ["src/bin.ts", "src/plugins/pluginResolveHooks.ts"], outDir: "dist", sourcemap: true, clean: true, 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/package.json b/packages/contracts/package.json index e1acf1e948e..0d8850a2674 100644 --- a/packages/contracts/package.json +++ b/packages/contracts/package.json @@ -18,6 +18,10 @@ "./relay": { "types": "./src/relay.ts", "import": "./src/relay.ts" + }, + "./plugin": { + "types": "./src/plugin.ts", + "import": "./src/plugin.ts" } }, "scripts": { diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 43270efdec7..b08debc2559 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -26,3 +26,4 @@ export * from "./review.ts"; export * from "./preview.ts"; export * from "./previewAutomation.ts"; export * from "./rpc.ts"; +export * from "./plugin.ts"; 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( diff --git a/packages/contracts/src/plugin.test.ts b/packages/contracts/src/plugin.test.ts new file mode 100644 index 00000000000..889c5cbf226 --- /dev/null +++ b/packages/contracts/src/plugin.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as Schema from "effect/Schema"; + +import { HOST_API_VERSION, PluginLockfile, PluginManifest, hostApiSatisfies } from "./plugin.ts"; + +const decodeManifest = Schema.decodeUnknownSync(PluginManifest); +const decodeLockfile = Schema.decodeUnknownSync(PluginLockfile); +const encodeLockfile = Schema.encodeSync(PluginLockfile); + +const minimalManifest = { + id: "test-plugin", + name: "Test Plugin", + version: "1.2.3", + hostApi: "^1.0.0", + entries: { server: "server.js" }, +}; + +describe("PluginManifest", () => { + it("decodes a minimal server manifest and defaults capabilities", () => { + const decoded = decodeManifest(minimalManifest); + expect(decoded.id).toBe("test-plugin"); + expect(decoded.capabilities).toEqual([]); + }); + + it("decodes a full manifest", () => { + const decoded = decodeManifest({ + ...minimalManifest, + name: " Test Plugin ", + description: "Adds test plugin behavior.", + author: { name: "T3", url: "https://example.test" }, + homepage: "https://example.test/plugin", + license: "MIT", + minAppVersion: "1.0.0", + capabilities: ["agents", "database"], + entries: { server: "dist/server.js", web: "dist/web.js" }, + }); + expect(decoded.name).toBe("Test Plugin"); + expect(decoded.capabilities).toEqual(["agents", "database"]); + }); + + it.each(["x", "1test-plugin", "test_plugin", "Test-Plugin", "a".repeat(42)])( + "rejects invalid plugin id %s", + (id) => { + expect(() => decodeManifest({ ...minimalManifest, id })).toThrow(); + }, + ); + + it("rejects unknown capabilities", () => { + expect(() => decodeManifest({ ...minimalManifest, capabilities: ["not-real"] })).toThrow(); + }); + + it("rejects duplicate capabilities", () => { + expect(() => + decodeManifest({ ...minimalManifest, capabilities: ["agents", "agents"] }), + ).toThrow(); + }); + + it("rejects unknown top-level fields", () => { + expect(() => decodeManifest({ ...minimalManifest, surprise: true })).toThrow(); + }); + + it("rejects manifests without server or web entries", () => { + expect(() => decodeManifest({ ...minimalManifest, entries: {} })).toThrow(); + }); + + it("rejects web-only manifests with server capabilities", () => { + expect(() => + decodeManifest({ + ...minimalManifest, + capabilities: ["agents"], + entries: { web: "web.js" }, + }), + ).toThrow(); + }); + + it("rejects unsafe entry paths", () => { + expect(() => + decodeManifest({ ...minimalManifest, entries: { server: "../server.js" } }), + ).toThrow(); + expect(() => + decodeManifest({ ...minimalManifest, entries: { server: "/server.js" } }), + ).toThrow(); + }); + + it("rejects bad hostApi ranges", () => { + expect(() => decodeManifest({ ...minimalManifest, hostApi: ">=1.0.0" })).toThrow(); + }); +}); + +describe("hostApiSatisfies", () => { + it("matches exact, caret, and tilde ranges", () => { + expect(hostApiSatisfies("1.0.0", HOST_API_VERSION)).toBe(true); + expect(hostApiSatisfies("^1.0.0", "1.9.9")).toBe(true); + expect(hostApiSatisfies("~1.0.0", "1.0.5")).toBe(true); + }); + + it("rejects versions outside the supported range", () => { + expect(hostApiSatisfies("1.0.0", "1.0.1")).toBe(false); + expect(hostApiSatisfies("^1.0.0", "2.0.0")).toBe(false); + expect(hostApiSatisfies("~1.0.0", "1.1.0")).toBe(false); + expect(hostApiSatisfies("^1.2.0", "1.1.9")).toBe(false); + }); +}); + +describe("PluginLockfile", () => { + it("round-trips a decoded lockfile", () => { + const decoded = decodeLockfile({ + sources: [{ id: "local", url: "file:///plugins", addedAt: "2026-07-03T00:00:00.000Z" }], + plugins: { + "test-plugin": { + version: "1.2.3", + sha256: "abc123", + sourceId: "local", + enabled: true, + state: "active", + activation: { activatingSince: null, crashCount: 0 }, + installedAt: "2026-07-03T00:00:00.000Z", + lastError: null, + }, + }, + }); + + expect(encodeLockfile(decoded)).toEqual(decoded); + }); +}); diff --git a/packages/contracts/src/plugin.ts b/packages/contracts/src/plugin.ts new file mode 100644 index 00000000000..234604c4cb3 --- /dev/null +++ b/packages/contracts/src/plugin.ts @@ -0,0 +1,203 @@ +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import { IsoDateTime, TrimmedNonEmptyString, TrimmedString } from "./baseSchemas.ts"; + +const SEMVER_PATTERN = /^\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$/; +const HOST_API_RANGE_PATTERN = /^[~^]?\d+\.\d+\.\d+$/; +const PLUGIN_ID_PATTERN = /^[a-z][a-z0-9-]{1,40}$/; + +export const HOST_API_VERSION = "1.0.0"; + +export const PluginId = TrimmedString.check(Schema.isPattern(PLUGIN_ID_PATTERN)).pipe( + Schema.brand("PluginId"), +); +export type PluginId = typeof PluginId.Type; + +export const PluginCapability = Schema.Literals([ + "agents", + "vcs", + "terminals", + "database", + "projections.read", + "environments.read", + "secrets", + "http", + "sourceControl", + "textGeneration", +]); +export type PluginCapability = typeof PluginCapability.Type; + +const SemverString = TrimmedNonEmptyString.check(Schema.isPattern(SEMVER_PATTERN)); +const HostApiRange = TrimmedNonEmptyString.check(Schema.isPattern(HOST_API_RANGE_PATTERN)); +const OptionalUrl = Schema.optionalKey(TrimmedNonEmptyString.check(Schema.isMaxLength(2048))); + +const RelativeEntryPath = TrimmedNonEmptyString.check( + Schema.makeFilter((entryPath) => { + if (entryPath.startsWith("/") || entryPath.startsWith("\\")) { + return "entry paths must be relative"; + } + if (entryPath.split(/[\\/]/).includes("..")) { + return "entry paths may not contain '..' segments"; + } + return true; + }), +); + +const ManifestEntries = Schema.Struct({ + server: Schema.optionalKey(RelativeEntryPath), + web: Schema.optionalKey(RelativeEntryPath), +}).check( + Schema.makeFilter<{ readonly server?: string; readonly web?: string }>((entries) => + entries.server || entries.web ? true : "manifest entries must include server or web", + ), +); +export type PluginManifestEntries = typeof ManifestEntries.Type; + +const PluginAuthor = Schema.Struct({ + name: TrimmedNonEmptyString.check(Schema.isMaxLength(100)), + url: OptionalUrl, +}); +export type PluginAuthor = typeof PluginAuthor.Type; + +const PluginCapabilities = Schema.Array(PluginCapability) + .check( + Schema.makeFilter>((capabilities) => + new Set(capabilities).size === capabilities.length ? true : "capabilities must be unique", + ), + ) + .pipe(Schema.withDecodingDefault(Effect.succeed([]))); + +interface PluginManifestShape { + readonly id: PluginId; + readonly name: string; + readonly version: string; + readonly description?: string | undefined; + readonly author?: PluginAuthor | undefined; + readonly homepage?: string | undefined; + readonly license?: string | undefined; + readonly hostApi: string; + readonly minAppVersion?: string | undefined; + readonly capabilities: ReadonlyArray; + readonly entries: PluginManifestEntries; +} + +export const PluginManifest = Schema.Struct({ + id: PluginId, + name: TrimmedNonEmptyString.check(Schema.isMaxLength(100)), + version: SemverString, + description: Schema.optionalKey(TrimmedString.check(Schema.isMaxLength(500))), + author: Schema.optionalKey(PluginAuthor), + homepage: OptionalUrl, + license: Schema.optionalKey(TrimmedNonEmptyString.check(Schema.isMaxLength(128))), + hostApi: HostApiRange, + minAppVersion: Schema.optionalKey(SemverString), + capabilities: PluginCapabilities, + entries: ManifestEntries, +}) + .check( + Schema.makeFilter((manifest) => { + if (!manifest.entries.server && manifest.capabilities.length > 0) { + return { + path: ["capabilities"], + issue: "web-only plugins may not declare server capabilities", + }; + } + return true; + }), + ) + .annotate({ parseOptions: { onExcessProperty: "error" } }); +export type PluginManifest = typeof PluginManifest.Type; + +export const PluginState = Schema.Literals([ + "active", + "pending-remove", + "pending-upgrade", + "failed", + "disabled", + "disabled-by-host", +]); +export type PluginState = typeof PluginState.Type; + +const LockfileSource = Schema.Struct({ + id: TrimmedNonEmptyString, + url: TrimmedNonEmptyString, + addedAt: IsoDateTime, +}); +export type PluginLockfileSource = typeof LockfileSource.Type; + +const LockfilePlugin = Schema.Struct({ + version: SemverString, + sha256: TrimmedNonEmptyString, + sourceId: TrimmedNonEmptyString, + enabled: Schema.Boolean, + state: PluginState, + staged: Schema.optionalKey( + Schema.Struct({ + version: SemverString, + sha256: TrimmedNonEmptyString, + stagedAt: IsoDateTime, + }), + ), + activation: Schema.Struct({ + activatingSince: Schema.NullOr(IsoDateTime), + crashCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), + }), + installedAt: IsoDateTime, + lastError: Schema.NullOr(Schema.String), +}); +export type PluginLockfilePlugin = typeof LockfilePlugin.Type; + +export const PluginLockfile = Schema.Struct({ + sources: Schema.Array(LockfileSource), + plugins: Schema.Record(PluginId, LockfilePlugin), +}); +export type PluginLockfile = typeof PluginLockfile.Type; + +export const EMPTY_PLUGIN_LOCKFILE: PluginLockfile = { + sources: [], + plugins: {}, +}; + +interface ParsedSemver { + readonly major: number; + readonly minor: number; + readonly patch: number; +} + +function parseStrictSemver(value: string): ParsedSemver | null { + const match = value.trim().match(/^(\d+)\.(\d+)\.(\d+)$/); + if (!match) return null; + return { + major: Number(match[1]), + minor: Number(match[2]), + patch: Number(match[3]), + }; +} + +function compareSemver(left: ParsedSemver, right: ParsedSemver): number { + if (left.major !== right.major) return left.major - right.major; + if (left.minor !== right.minor) return left.minor - right.minor; + return left.patch - right.patch; +} + +export function hostApiSatisfies(range: string, version: string): boolean { + const trimmedRange = range.trim(); + const operator = + trimmedRange.startsWith("^") || trimmedRange.startsWith("~") ? trimmedRange[0] : ""; + const target = parseStrictSemver(operator ? trimmedRange.slice(1) : trimmedRange); + const actual = parseStrictSemver(version); + if (!target || !actual) return false; + + const compared = compareSemver(actual, target); + if (operator === "") return compared === 0; + if (compared < 0) return false; + + if (operator === "^") { + if (target.major > 0) return actual.major === target.major; + if (target.minor > 0) return actual.major === 0 && actual.minor === target.minor; + return actual.major === 0 && actual.minor === 0 && actual.patch === target.patch; + } + + return actual.major === target.major && actual.minor === target.minor; +} diff --git a/packages/plugin-sdk/package.json b/packages/plugin-sdk/package.json new file mode 100644 index 00000000000..a6d915a09f0 --- /dev/null +++ b/packages/plugin-sdk/package.json @@ -0,0 +1,19 @@ +{ + "name": "@t3tools/plugin-sdk", + "private": true, + "type": "module", + "exports": { + ".": { + "types": "./src/index.ts", + "import": "./src/index.ts" + } + }, + "scripts": { + "typecheck": "tsgo --noEmit", + "test": "vp test run" + }, + "dependencies": { + "@t3tools/contracts": "workspace:*", + "effect": "catalog:" + } +} diff --git a/packages/plugin-sdk/src/index.test.ts b/packages/plugin-sdk/src/index.test.ts new file mode 100644 index 00000000000..36521a15e0e --- /dev/null +++ b/packages/plugin-sdk/src/index.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as Effect from "effect/Effect"; + +import { definePlugin, HOST_API_VERSION } from "./index.ts"; + +describe("definePlugin", () => { + it("preserves the plugin definition shape", () => { + const definition = definePlugin({ + register: () => Effect.succeed({ rpc: [] }), + }); + + expect(typeof definition.register).toBe("function"); + expect(HOST_API_VERSION).toBe("1.0.0"); + }); +}); diff --git a/packages/plugin-sdk/src/index.ts b/packages/plugin-sdk/src/index.ts new file mode 100644 index 00000000000..cc41cc23265 --- /dev/null +++ b/packages/plugin-sdk/src/index.ts @@ -0,0 +1,157 @@ +import type * as Effect from "effect/Effect"; +import type * as SqlClient from "effect/unstable/sql/SqlClient"; + +export type { + PluginCapability, + PluginId, + PluginLockfile, + PluginLockfilePlugin, + PluginLockfileSource, + PluginManifest, + PluginManifestEntries, + PluginState, +} from "@t3tools/contracts/plugin"; +export { HOST_API_VERSION, hostApiSatisfies } from "@t3tools/contracts/plugin"; + +export type PluginRpcScope = "read" | "operate"; +export type PluginReadiness = "requires-ready" | "always"; + +export interface PluginLogger { + readonly debug: (message: string, attributes?: Record) => Effect.Effect; + readonly info: (message: string, attributes?: Record) => Effect.Effect; + readonly warn: (message: string, attributes?: Record) => Effect.Effect; + readonly error: (message: string, attributes?: Record) => Effect.Effect; +} + +export interface PluginHostConfig { + readonly appVersion: string; + readonly hostApiVersion: string; + readonly dataDir: string; + readonly logger: PluginLogger; +} + +export interface PluginCapabilityUnavailable { + readonly _tag: "PluginCapabilityUnavailable"; + readonly capability: string; + readonly message: string; +} + +export interface AgentsCapability { + readonly list: Effect.Effect>; +} + +export interface VcsCapability { + readonly status: (input: { readonly cwd: string }) => Effect.Effect; +} + +export interface TerminalsCapability { + readonly open: (input: unknown) => Effect.Effect; +} + +export interface DatabaseCapability { + readonly sql: SqlClient.SqlClient; +} + +export interface ProjectionsReadCapability { + readonly getSnapshot: (input: unknown) => Effect.Effect; +} + +export interface EnvironmentsReadCapability { + readonly list: Effect.Effect>; +} + +export interface SecretsCapability { + readonly get: (name: string) => Effect.Effect; + readonly set: (name: string, value: Uint8Array) => Effect.Effect; +} + +export interface HttpCapability { + readonly baseUrl: string | null; +} + +export interface SourceControlCapability { + readonly listPullRequests: (input: unknown) => Effect.Effect>; +} + +export interface TextGenerationCapability { + readonly generateText: (input: unknown) => Effect.Effect; +} + +export interface PluginHostApi { + readonly hostApiVersion: string; + readonly config: PluginHostConfig; + readonly agents: Effect.Effect; + readonly vcs: Effect.Effect; + readonly terminals: Effect.Effect; + readonly database: Effect.Effect; + readonly projectionsRead: Effect.Effect; + readonly environmentsRead: Effect.Effect; + readonly secrets: Effect.Effect; + readonly http: Effect.Effect; + readonly sourceControl: Effect.Effect; + readonly textGeneration: Effect.Effect; +} + +export interface PluginRpcContext { + readonly pluginId: string; + readonly logger: PluginLogger; +} + +export interface PluginRpcDescriptor { + readonly method: string; + readonly scope: PluginRpcScope; + readonly readiness?: PluginReadiness | undefined; + readonly handler: (payload: unknown, ctx: PluginRpcContext) => Effect.Effect; +} + +export interface PluginStreamDescriptor { + readonly method: string; + readonly scope: PluginRpcScope; + readonly readiness?: PluginReadiness | undefined; + readonly handler: (payload: unknown, ctx: PluginRpcContext) => Effect.Effect; +} + +export interface PluginHttpDescriptor { + readonly method: string; + readonly path: string; + readonly auth: "public" | "token"; + readonly handler: (request: unknown, ctx: PluginRpcContext) => Effect.Effect; +} + +export interface PluginServiceContext { + readonly pluginId: string; + readonly logger: PluginLogger; +} + +export interface PluginServiceDescriptor { + readonly name: string; + readonly run: (ctx: PluginServiceContext) => Effect.Effect; +} + +export interface PluginMigration { + readonly version: number; + readonly name: string; + readonly up: Effect.Effect; +} + +export interface PluginRegistration { + readonly migrations?: ReadonlyArray | undefined; + readonly recover?: (() => Effect.Effect) | undefined; + readonly rpc?: ReadonlyArray | undefined; + readonly streams?: ReadonlyArray | undefined; + readonly http?: ReadonlyArray | undefined; + readonly services?: ReadonlyArray | undefined; +} + +export interface PluginDefinition { + readonly register: + | ((hostApi: PluginHostApi) => Effect.Effect) + | ((hostApi: PluginHostApi) => Promise) + | ((hostApi: PluginHostApi) => PluginRegistration); +} + +export function definePlugin( + definition: Definition, +): Definition { + return definition; +} diff --git a/packages/plugin-sdk/tsconfig.json b/packages/plugin-sdk/tsconfig.json new file mode 100644 index 00000000000..73a306f847a --- /dev/null +++ b/packages/plugin-sdk/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": {}, + "include": ["src"] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ef7146e194a..2a929f4a688 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -462,6 +462,9 @@ importers: '@pierre/diffs': specifier: 'catalog:' version: 1.3.0-beta.5(patch_hash=7cb6da88544119adda056b2f46f43956f99326227732da0b345081e285a6c53a)(@shikijs/themes@4.3.0)(react-dom@19.2.6(react@19.2.6))(react@19.2.6) + '@t3tools/plugin-sdk': + specifier: workspace:* + version: link:../../packages/plugin-sdk effect: specifier: 4.0.0-beta.78 version: 4.0.0-beta.78(patch_hash=c502bc684210b707dfceb87d8fe6ad6843395af6e19cfc02cd65854898bde2c5) @@ -801,6 +804,15 @@ importers: specifier: 'catalog:' version: 0.2.2(@types/node@24.12.4)(bufferutil@4.1.0)(esbuild@0.28.1)(jiti@2.7.0)(msw@2.12.11(@types/node@24.12.4)(typescript@6.0.3))(terser@5.48.0)(typescript@6.0.3)(unrun@0.2.39)(utf-8-validate@6.0.6)(yaml@2.9.0) + packages/plugin-sdk: + dependencies: + '@t3tools/contracts': + specifier: workspace:* + version: link:../contracts + effect: + specifier: 4.0.0-beta.78 + version: 4.0.0-beta.78(patch_hash=c502bc684210b707dfceb87d8fe6ad6843395af6e19cfc02cd65854898bde2c5) + packages/shared: dependencies: '@noble/curves':