From 9c7fb92be640da39152eed5370bd6e3a65d7ef5a Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 22:01:23 -0700 Subject: [PATCH] feat(mcp): thread standing role and launch presets A thread carries a standing role that T3 sends with provider turns when it is new, changed, in a new session or after a compaction, and t3_preset stores named launch presets that t3_thread_launch and delegate_task accept as `preset`. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/mcp/McpHttpServer.test.ts | 4 + apps/server/src/mcp/McpHttpServer.ts | 17 +- apps/server/src/mcp/McpInvocationContext.ts | 8 + .../src/mcp/OrchestratorMcpService.test.ts | 518 ++++++++++++++++++ apps/server/src/mcp/OrchestratorMcpService.ts | 106 +++- ...OrchestratorMcpToolkit.integration.test.ts | 10 + apps/server/src/mcp/ThreadLaunchMcpService.ts | 70 ++- apps/server/src/mcp/toolkits/core.test.ts | 77 +++ .../src/mcp/toolkits/preset/handlers.ts | 30 + apps/server/src/mcp/toolkits/preset/tools.ts | 25 + .../src/mcp/toolkits/project/handlers.test.ts | 159 ++++++ apps/server/src/mcp/toolkits/project/tools.ts | 4 + .../src/mcp/toolkits/thread/handlers.ts | 59 +- apps/server/src/mcp/toolkits/thread/tools.ts | 8 +- .../ProviderTurnStartService.test.ts | 104 +++- .../ProviderTurnStartService.ts | 57 +- .../src/orchestration-v2/runtimeLayer.ts | 2 + .../standingRoleDelivery.test.ts | 36 ++ .../orchestration-v2/standingRoleDelivery.ts | 25 + .../testkit/ProviderReplayHarness.ts | 2 + apps/server/src/peer/RemoteDelegation.ts | 6 + apps/server/src/persistence/ForkMigrations.ts | 2 + .../016_ThreadRolesAndPresets.ts | 30 + .../src/persistence/LaunchPresets.test.ts | 111 ++++ apps/server/src/persistence/LaunchPresets.ts | 125 +++++ apps/server/src/persistence/ThreadRoles.ts | 118 ++++ .../src/provider/AgentContextBudget.test.ts | 4 + .../src/provider/agentContextBudget.json | 2 +- ...thread-standing-role-and-launch-presets.md | 30 + packages/contracts/src/index.ts | 1 + packages/contracts/src/launchPresets.ts | 83 +++ packages/contracts/src/orchestratorMcp.ts | 6 + 32 files changed, 1803 insertions(+), 36 deletions(-) create mode 100644 apps/server/src/mcp/toolkits/preset/handlers.ts create mode 100644 apps/server/src/mcp/toolkits/preset/tools.ts create mode 100644 apps/server/src/orchestration-v2/standingRoleDelivery.test.ts create mode 100644 apps/server/src/orchestration-v2/standingRoleDelivery.ts create mode 100644 apps/server/src/persistence/ForkMigrations/016_ThreadRolesAndPresets.ts create mode 100644 apps/server/src/persistence/LaunchPresets.test.ts create mode 100644 apps/server/src/persistence/LaunchPresets.ts create mode 100644 apps/server/src/persistence/ThreadRoles.ts create mode 100644 docs/jarvis/decisions/0019-thread-standing-role-and-launch-presets.md create mode 100644 packages/contracts/src/launchPresets.ts diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index c9afb6de87fa6..73dba5c6beb93 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -1,6 +1,8 @@ import * as RequestLog from "../persistence/McpRequestLog.ts"; import * as McpClientThreads from "../persistence/McpClientThreads.ts"; import * as ThreadLaunchMcp from "./ThreadLaunchMcpService.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; +import * as LaunchPresets from "../persistence/LaunchPresets.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; @@ -310,6 +312,8 @@ const layerRecordingTools = (calls: Array) => { Layer.mock(OrchestratorMcpService.OrchestratorMcpService)({}), Layer.mock(McpClientThreads.McpClientThreads)({}), Layer.mock(ThreadLaunchMcp.ThreadLaunchMcpService)({}), + Layer.mock(ThreadRoles.ThreadRoles)({}), + Layer.mock(LaunchPresets.LaunchPresets)({}), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index f9a03e60c037f..cdc68d2eb5254 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -9,6 +9,10 @@ import * as ThreadLaunchMcp from "./ThreadLaunchMcpService.ts"; import * as AgentRegistry from "../persistence/AgentRegistry.ts"; import * as McpClientThreads from "../persistence/McpClientThreads.ts"; import * as WorkLedger from "../persistence/WorkLedger.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; +import * as LaunchPresets from "../persistence/LaunchPresets.ts"; +import { PresetToolkit } from "./toolkits/preset/tools.ts"; +import * as PresetHandlers from "./toolkits/preset/handlers.ts"; import { LedgerToolkit } from "./toolkits/ledger/tools.ts"; import * as LedgerHandlers from "./toolkits/ledger/handlers.ts"; import * as Cause from "effect/Cause"; @@ -1391,8 +1395,13 @@ export const layerOrchestratorToolkit = toolkitRegistration( export const layerLedgerToolkit = toolkitRegistration(LedgerToolkit, LedgerHandlers.layer); +export const layerPresetToolkit = toolkitRegistration(PresetToolkit, PresetHandlers.layer).pipe( + Layer.provide(LaunchPresets.layer), +); + export const layerThreadToolkit = toolkitRegistration(ThreadToolkit, ThreadHandlers.layer).pipe( Layer.provide(PendingApproval.layer), + Layer.provide(ThreadRoles.layer), ); const layerWorktreeToolkitRegistration = toolkitRegistration( @@ -1413,7 +1422,10 @@ const layerEnvironmentToolkit = toolkitRegistration( export const layerProjectRegistration = toolkitRegistration( ProjectToolkit, ProjectHandlers.layer, -).pipe(Layer.provide(ThreadLaunchMcp.layer)); +).pipe( + Layer.provide(ThreadLaunchMcp.layer), + Layer.provide(Layer.mergeAll(ThreadRoles.layer, LaunchPresets.layer)), +); export const layerAttachmentToolkit = toolkitRegistration( AttachmentToolkit, @@ -1488,6 +1500,7 @@ export const layerTools = Layer.mergeAll( Layer.provide(BridgeStatusService.layer), ), layerLedgerToolkit, + layerPresetToolkit, layerPreviewToolkit, layerOrchestratorToolkit, layerThreadToolkit, @@ -1565,6 +1578,8 @@ export const layer = Layer.merge( ).pipe( Layer.provide(layerEvents), Layer.provide(WorkLedger.layer), + Layer.provide(ThreadRoles.layer), + Layer.provide(LaunchPresets.layer), Layer.provide(McpClientThreads.layer), Layer.provide(McpBridgeMetrics.layer), Layer.provide(McpRequestLog.layer), diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index a75eae04db927..377705ea5bd85 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -172,6 +172,14 @@ export const messageSenderFor = (scope: McpInvocationScope) => { }; }; +/** Who changed something, as ADR 0013 names a sender: a thread, or the first 8 characters of a session and its label. */ +export const actorLabel = (scope: McpInvocationScope) => + scope.thread !== undefined + ? `thread ${scope.thread.threadId}` + : scope.client === undefined + ? "mcp" + : `${scope.client.sessionId.slice(0, 8)} (${scope.client.label})`; + /** The command actor for an MCP call: an agent in a thread, or an outside sign-in. */ export const mcpCommandActor = (scope: McpInvocationScope): CommandActor["Service"] => ({ kind: scope.thread === undefined ? "mcp_client" : "agent", diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index db2e6ad290e7b..b3147ecec878d 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -1,5 +1,8 @@ import * as WorkLedger from "../persistence/WorkLedger.ts"; import * as McpClientThreads from "../persistence/McpClientThreads.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; +import * as LaunchPresets from "../persistence/LaunchPresets.ts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as FileSystem from "effect/FileSystem"; @@ -7,6 +10,7 @@ import * as Path from "effect/Path"; import { assert, describe, expect, it } from "@effect/vitest"; import { CommandId, + OrchestratorMcpFailure, ContextTransferId, EnvironmentId, GitCommandError, @@ -2098,6 +2102,520 @@ describe("OrchestratorMcpService provider resolution", () => { }), ); + it.effect("delegation applies a preset and stores the role for the child before dispatch", () => + Effect.gen(function* () { + const altInstanceId = ProviderInstanceId.make("codex-alt"); + const dispatched = yield* Ref.make>([]); + const roles = new Map(); + const rolesAtDispatch: Array = []; + const task = { + id: taskId, + threadId: parentThreadId, + runId: parentRunId, + parentNodeId, + origin: "app_owned", + createdBy: "agent", + driver: ProviderDriverKind.make("codex"), + providerInstanceId: altInstanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Summarize the diff.", + title: null, + model: "gpt-5.4", + status: "running", + result: null, + startedAt: null, + completedAt: null, + }; + let delegated = false; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + getThreadRecords: (threadId) => + Effect.succeed( + threadId === parentThreadId + ? parentProjection(delegated ? [task] : []) + : childProjection, + ), + isProviderInstanceEnabledForProject: ({ instanceId }) => + Effect.succeed(instanceId !== codexInstanceId), + assertProviderInstanceEnabled: ({ projectId, instanceId }) => + instanceId === codexInstanceId + ? Effect.fail( + new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({ + projectId, + instanceId, + }), + ) + : Effect.void, + dispatch: (command) => + Ref.update(dispatched, (commands) => [...commands, command]).pipe( + Effect.tap(() => + Effect.sync(() => { + rolesAtDispatch.push( + roles.get( + IdAllocator.derive.delegatedTaskThread({ + commandId: (command as { commandId: CommandId }).commandId, + }), + ), + ); + }), + ), + Effect.andThen( + Effect.sync(() => { + delegated = true; + }), + ), + Effect.as({ + sequence: 1, + storedEvents: [ + { + sequence: 1, + commandId: null, + event: { type: "subagent.updated", payload: task }, + }, + ], + } as never), + ), + }), + providerRegistryLayer([ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + providerSnapshot({ + instanceId: altInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ]), + adapterRegistryLayer([codexInstanceId, altInstanceId]), + Layer.mock(ThreadRoles.ThreadRoles)({ + set: (threadId, role) => + Effect.sync( + () => void (role === null ? roles.delete(threadId) : roles.set(threadId, role)), + ), + }), + Layer.mock(LaunchPresets.LaunchPresets)({ + get: (name) => + Effect.succeed({ + name, + modelSelection: { instanceId: altInstanceId, model: "gpt-5.4" }, + runtimeMode: "auto-accept-edits", + standingRole: "Be brief.", + doneFormat: "DONE: ", + workspaceStrategy: { type: "root" }, + updatedAt: "2026-01-01T00:00:00Z", + updatedBy: "test", + }), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ServerSettings.layerTest(), + ); + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const result = yield* service.delegateTask(scope, { + task: "Summarize the diff.", + preset: "worker", + mode: "async", + clientRequestId: "delegate-preset-1", + }); + assert.equal(result.status, "running"); + const [request] = (yield* Ref.get(dispatched)) as ReadonlyArray<{ + modelSelection: { instanceId: string }; + runtimeMode: string; + }>; + assert.equal(request?.modelSelection.instanceId, altInstanceId); + assert.equal(request?.runtimeMode, "auto-accept-edits"); + const expected = + "Be brief.\n\nYour final message must follow this format:\nDONE: "; + assert.deepEqual(rolesAtDispatch, [expected]); + assert.deepEqual([...roles.values()], [expected]); + // The parent's own thread names the setter. + assert.deepEqual([...roles.keys()].length, 1); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); + it.effect("delegation refuses a preset whose mode is above the parent's and starts nothing", () => + Effect.gen(function* () { + const altInstanceId = ProviderInstanceId.make("codex-alt"); + const dispatched = yield* Ref.make>([]); + const roles = new Map(); + const rolesAtDispatch: Array = []; + const task = { + id: taskId, + threadId: parentThreadId, + runId: parentRunId, + parentNodeId, + origin: "app_owned", + createdBy: "agent", + driver: ProviderDriverKind.make("codex"), + providerInstanceId: altInstanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Summarize the diff.", + title: null, + model: "gpt-5.4", + status: "running", + result: null, + startedAt: null, + completedAt: null, + }; + let delegated = false; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + getThreadRecords: (threadId) => + Effect.succeed( + threadId === parentThreadId + ? { + ...parentProjection(delegated ? [task] : []), + thread: { + ...parentProjection([]).thread, + runtimeMode: "approval-required" as const, + }, + } + : childProjection, + ), + isProviderInstanceEnabledForProject: ({ instanceId }) => + Effect.succeed(instanceId !== codexInstanceId), + assertProviderInstanceEnabled: ({ projectId, instanceId }) => + instanceId === codexInstanceId + ? Effect.fail( + new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({ + projectId, + instanceId, + }), + ) + : Effect.void, + dispatch: (command) => + Ref.update(dispatched, (commands) => [...commands, command]).pipe( + Effect.tap(() => + Effect.sync(() => { + rolesAtDispatch.push( + roles.get( + IdAllocator.derive.delegatedTaskThread({ + commandId: (command as { commandId: CommandId }).commandId, + }), + ), + ); + }), + ), + Effect.andThen( + Effect.sync(() => { + delegated = true; + }), + ), + Effect.as({ + sequence: 1, + storedEvents: [ + { + sequence: 1, + commandId: null, + event: { type: "subagent.updated", payload: task }, + }, + ], + } as never), + ), + }), + providerRegistryLayer([ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + providerSnapshot({ + instanceId: altInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ]), + adapterRegistryLayer([codexInstanceId, altInstanceId]), + Layer.mock(ThreadRoles.ThreadRoles)({ + set: (threadId, role) => + Effect.sync( + () => void (role === null ? roles.delete(threadId) : roles.set(threadId, role)), + ), + }), + Layer.mock(LaunchPresets.LaunchPresets)({ + get: (name) => + Effect.succeed({ + name, + modelSelection: { instanceId: altInstanceId, model: "gpt-5.4" }, + runtimeMode: "auto-accept-edits", + standingRole: "Be brief.", + doneFormat: "DONE: ", + workspaceStrategy: { type: "root" }, + updatedAt: "2026-01-01T00:00:00Z", + updatedBy: "test", + }), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ServerSettings.layerTest(), + ); + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const error = yield* service + .delegateTask(scope, { + task: "Summarize the diff.", + preset: "worker", + mode: "async", + clientRequestId: "delegate-preset-above-1", + }) + .pipe(Effect.flip); + assert.equal(error.code, "runtime_mode_escalation_denied"); + assert.deepEqual(yield* Ref.get(dispatched), []); + assert.deepEqual([...roles.keys()], []); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); + it.effect("retrying an accepted delegation works after its preset was deleted", () => + Effect.gen(function* () { + const altInstanceId = ProviderInstanceId.make("codex-alt"); + const dispatched = yield* Ref.make>([]); + const roles = new Map(); + const rolesAtDispatch: Array = []; + const task = { + id: taskId, + threadId: parentThreadId, + runId: parentRunId, + parentNodeId, + origin: "app_owned", + createdBy: "agent", + driver: ProviderDriverKind.make("codex"), + providerInstanceId: altInstanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Summarize the diff.", + title: null, + model: "gpt-5.4", + status: "running", + result: null, + startedAt: null, + completedAt: null, + }; + let delegated = false; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(true), + getThreadRecords: (threadId) => + Effect.succeed( + threadId === parentThreadId + ? parentProjection(delegated ? [task] : []) + : childProjection, + ), + isProviderInstanceEnabledForProject: ({ instanceId }) => + Effect.succeed(instanceId !== codexInstanceId), + assertProviderInstanceEnabled: ({ projectId, instanceId }) => + instanceId === codexInstanceId + ? Effect.fail( + new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({ + projectId, + instanceId, + }), + ) + : Effect.void, + dispatch: (command) => + Ref.update(dispatched, (commands) => [...commands, command]).pipe( + Effect.tap(() => + Effect.sync(() => { + rolesAtDispatch.push( + roles.get( + IdAllocator.derive.delegatedTaskThread({ + commandId: (command as { commandId: CommandId }).commandId, + }), + ), + ); + }), + ), + Effect.andThen( + Effect.sync(() => { + delegated = true; + }), + ), + Effect.as({ + sequence: 1, + storedEvents: [ + { + sequence: 1, + commandId: null, + event: { type: "subagent.updated", payload: task }, + }, + ], + } as never), + ), + }), + providerRegistryLayer([ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + providerSnapshot({ + instanceId: altInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ]), + adapterRegistryLayer([codexInstanceId, altInstanceId]), + Layer.mock(ThreadRoles.ThreadRoles)({ + set: (threadId, role) => + Effect.sync( + () => void (role === null ? roles.delete(threadId) : roles.set(threadId, role)), + ), + }), + Layer.mock(LaunchPresets.LaunchPresets)({ + get: (name) => + Effect.fail( + new OrchestratorMcpFailure({ + code: "invalid_request", + message: `Launch preset "${name}" was not found.`, + }), + ), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ServerSettings.layerTest(), + ); + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const result = yield* service.delegateTask(scope, { + task: "Summarize the diff.", + preset: "worker", + mode: "async", + clientRequestId: "delegate-preset-retry-1", + }); + assert.equal(result.status, "running"); + // The replay leaves a role changed since the first call alone. + assert.deepEqual([...roles.keys()], []); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); + it.effect("a delegation that fails before its child exists leaves no role behind", () => + Effect.gen(function* () { + const altInstanceId = ProviderInstanceId.make("codex-alt"); + const roles = new Map(); + const task = { + id: taskId, + threadId: parentThreadId, + runId: parentRunId, + parentNodeId, + origin: "app_owned", + createdBy: "agent", + driver: ProviderDriverKind.make("codex"), + providerInstanceId: altInstanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Summarize the diff.", + title: null, + model: "gpt-5.4", + status: "running", + result: null, + startedAt: null, + completedAt: null, + }; + let delegated = false; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + getThreadRecords: (threadId) => + Effect.succeed( + threadId === parentThreadId + ? parentProjection(delegated ? [task] : []) + : childProjection, + ), + isProviderInstanceEnabledForProject: ({ instanceId }) => + Effect.succeed(instanceId !== codexInstanceId), + assertProviderInstanceEnabled: ({ projectId, instanceId }) => + instanceId === codexInstanceId + ? Effect.fail( + new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({ + projectId, + instanceId, + }), + ) + : Effect.void, + getThreadShell: () => Effect.succeed(null), + dispatch: () => Effect.fail(new Error("refused") as never), + }), + providerRegistryLayer([ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + providerSnapshot({ + instanceId: altInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ]), + adapterRegistryLayer([codexInstanceId, altInstanceId]), + Layer.mock(ThreadRoles.ThreadRoles)({ + set: (threadId, role) => + Effect.sync( + () => void (role === null ? roles.delete(threadId) : roles.set(threadId, role)), + ), + }), + Layer.mock(LaunchPresets.LaunchPresets)({ + get: (name) => + Effect.succeed({ + name, + modelSelection: { instanceId: altInstanceId, model: "gpt-5.4" }, + runtimeMode: "auto-accept-edits", + standingRole: "Be brief.", + doneFormat: "DONE: ", + workspaceStrategy: { type: "root" }, + updatedAt: "2026-01-01T00:00:00Z", + updatedBy: "test", + }), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ServerSettings.layerTest(), + ); + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const error = yield* service + .delegateTask(scope, { + task: "Summarize the diff.", + preset: "worker", + mode: "async", + clientRequestId: "delegate-preset-failed-1", + }) + .pipe(Effect.flip); + assert.equal(error.code, "orchestration_error"); + assert.deepEqual([...roles.keys()], []); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); it.effect( "retrying an accepted driver-only delegation replays even after the project disables every candidate", () => diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 8c1efaba86c06..4b2cfb353d96c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1,5 +1,8 @@ import * as JarvisPlaybook from "./JarvisPlaybook.ts"; import * as WorkLedger from "../persistence/WorkLedger.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; +import * as LaunchPresets from "../persistence/LaunchPresets.ts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import { advertisedModel, requireAdvertisedModel } from "./providerModels.ts"; import { hasGitEntry } from "./gitEntryProbe.ts"; import * as Invocation from "./McpInvocationContext.ts"; @@ -29,6 +32,7 @@ import { type OrchestratorMcpCreateThreadsResult, type OrchestratorMcpCreatedThread, type OrchestratorMcpDelegateTaskInput, + composeStandingRole, type OrchestratorMcpDelegateTaskResult, type OrchestratorMcpDeleteScheduledTaskInput, type OrchestratorMcpRequestSecretInput, @@ -1094,6 +1098,18 @@ const make = Effect.gen(function* () { const gitDriver = yield* GitVcsDriver.GitVcsDriver; const serverSettings = yield* ServerSettings.ServerSettingsService; const remoteDelegation = yield* RemoteDelegation.RemoteDelegation; + // Optional so a build without the fork's role and preset stores still serves the rest. + const threadRoles = yield* Effect.serviceOption(ThreadRoles.ThreadRoles); + const launchPresets = yield* Effect.serviceOption(LaunchPresets.LaunchPresets); + + const loadPreset = Effect.fn("OrchestratorMcpService.loadPreset")(function* ( + name: string | undefined, + ) { + if (name === undefined) return undefined; + if (Option.isNone(launchPresets)) + return yield* failure("orchestration_error", "Launch presets are unavailable."); + return yield* launchPresets.value.get(name); + }); const requireProject = (projectId: ProjectId) => projects.getById(projectId).pipe( @@ -2538,15 +2554,47 @@ const make = Effect.gen(function* () { operation: "delegate-task", }); const isRetryOfAcceptedDelegation = yield* threadManagement.hasReceipt(commandId); + // A retry of an accepted delegation replays it, so a preset deleted since cannot fail it. + const preset = yield* loadPreset(input.preset).pipe( + Effect.catchIf( + () => isRetryOfAcceptedDelegation, + () => Effect.undefined, + ), + ); + // Explicit fields win. A preset names a worktree only as the kind of checkout: a local + // task always branches from the parent's commit, so its baseRef does not apply. + const presetModel = preset?.modelSelection; + const delegateInput: OrchestratorMcpDelegateTaskInput = { + ...input, + ...(input.target === undefined && presetModel !== undefined + ? { + target: { + providerInstanceId: presetModel.instanceId, + model: presetModel.model, + ...(presetModel.options === undefined ? {} : { options: presetModel.options }), + }, + } + : {}), + ...(input.runtimeMode === undefined && preset?.runtimeMode !== undefined + ? { runtimeMode: preset.runtimeMode } + : {}), + ...(input.isolation === undefined && preset?.workspaceStrategy?.type === "worktree" + ? { isolation: "worktree" as const } + : {}), + }; + const standingRole = composeStandingRole({ + standingRole: input.standingRole ?? preset?.standingRole, + doneFormat: preset?.doneFormat, + }); const target = yield* resolveTargetRechecking({ parent, - target: input.target, + target: delegateInput.target, providers, skipProjectEnablementFilter: isRetryOfAcceptedDelegation, }); const runtimeMode = yield* resolveRuntimeMode( parent.thread.runtimeMode, - input.runtimeMode, + delegateInput.runtimeMode, target, ); const interactionMode = yield* resolveInteractionMode( @@ -2554,7 +2602,7 @@ const make = Effect.gen(function* () { input.interactionMode, ); let workspaceStrategy: OrchestrationV2ThreadLaunchWorkspaceStrategy | undefined; - if (input.isolation === "worktree" && !isRetryOfAcceptedDelegation) { + if (delegateInput.isolation === "worktree" && !isRetryOfAcceptedDelegation) { const project = yield* requireProject(parent.thread.projectId); const base = yield* gitDriver .resolveCommit({ @@ -2583,6 +2631,17 @@ const make = Effect.gen(function* () { } workspaceStrategy = { type: "worktree", baseRef: base.commitSha, startFromOrigin: false }; } + // The child's first turn starts as soon as the command commits, so its role is stored + // first, under the thread id the command derives. A retry leaves a role changed since. + if (standingRole !== undefined && !isRetryOfAcceptedDelegation) { + if (Option.isNone(threadRoles)) + return yield* failure("orchestration_error", "Standing roles are unavailable."); + yield* threadRoles.value.set( + IdAllocator.derive.delegatedTaskThread({ commandId }), + standingRole, + Invocation.actorLabel(scope), + ); + } // ThreadManagementService.dispatch is the chokepoint for this command // type's own enablement check, using the same command id and its // receipt-replay exemption, so no separate assert runs here. @@ -2610,6 +2669,18 @@ const make = Effect.gen(function* () { completionWake: input.mode === "wait" ? "settled_only" : "always", }) .pipe( + // A failure after the child committed leaves its role; only a child that never + // came to exist loses the one stored for it. + Effect.tapError(() => + standingRole !== undefined && !isRetryOfAcceptedDelegation && Option.isSome(threadRoles) + ? Effect.gen(function* () { + const childId = IdAllocator.derive.delegatedTaskThread({ commandId }); + const child = yield* threadManagement.getThreadShell(childId); + if (child === null || child.deletedAt !== null) + yield* threadRoles.value.set(childId, null, Invocation.actorLabel(scope)); + }).pipe(Effect.ignore({ log: true })) + : Effect.void, + ), Effect.mapError((error) => error._tag === "ThreadManagementProviderInstanceDisabledError" ? threadManagementFailure(error) @@ -3317,16 +3388,29 @@ const make = Effect.gen(function* () { } } } + const firstPage = input.afterPosition === undefined && input.itemId === undefined; + const standingRole = + firstPage && Option.isSome(threadRoles) + ? yield* threadRoles.value.get(target.thread.id) + : undefined; return { revision, - thread: threadDetail( - scope, - target, - timeline.totalItems, - shell, - yield* Clock.currentTimeMillis, - input.afterPosition === undefined && input.itemId === undefined, - ), + thread: { + ...threadDetail( + scope, + target, + timeline.totalItems, + shell, + yield* Clock.currentTimeMillis, + firstPage, + ), + ...(standingRole === undefined + ? {} + : { + standingRole: standingRole.text, + standingRoleSetBy: `${standingRole.updatedBy}, ${standingRole.updatedAt}`, + }), + }, recentRuns: target.runs .toSorted((left, right) => right.ordinal - left.ordinal) .slice(0, input.runLimit ?? DEFAULT_THREAD_RUN_LIMIT) diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 5a0e5b83ca331..814f84e68ebb7 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -38,6 +38,7 @@ import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; @@ -267,6 +268,7 @@ describe("orchestrator MCP toolkit", () => { ]); const layerTest = layerOrchestratorToolkit.pipe( Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), + Layer.provideMerge(ThreadRoles.layer), Layer.provideMerge(SqlitePersistence.layerMemory), Layer.provideMerge(layerOrchestration), Layer.provide( @@ -718,6 +720,11 @@ describe("orchestrator MCP toolkit", () => { expect( messageReads.filter((query) => !query.includes("$.notification.source.kind")), ).toHaveLength(1); + yield* (yield* ThreadRoles.ThreadRoles).set( + delegated.childThreadId!, + "You are the reviewer.", + "test", + ); const parentlessRead = yield* invoke( "t3_thread_read", { @@ -741,6 +748,9 @@ describe("orchestrator MCP toolkit", () => { maxCharsPerItem: result === "😀" ? 1 : 20_000, }).pipe(Effect.withTracer(tracer)); expect(read.isError).not.toBe(true); + expect(read.structuredContent).toMatchObject({ + thread: { standingRole: "You are the reviewer." }, + }); expect(transferReads.length).toBeGreaterThan(0); expect(read.structuredContent).toEqual(parentlessRead.structuredContent); // The acknowledgement reads the task and its spawning run by id, not the parent's history. diff --git a/apps/server/src/mcp/ThreadLaunchMcpService.ts b/apps/server/src/mcp/ThreadLaunchMcpService.ts index 7fea171e74111..5992c0c7407be 100644 --- a/apps/server/src/mcp/ThreadLaunchMcpService.ts +++ b/apps/server/src/mcp/ThreadLaunchMcpService.ts @@ -37,6 +37,8 @@ import { newCommandId, readCaller, resolveProjectId, unavailable } from "./threa import * as Context from "effect/Context"; import * as Layer from "effect/Layer"; import * as McpClientThreads from "../persistence/McpClientThreads.ts"; +import * as LaunchPresets from "../persistence/LaunchPresets.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; import { McpAttachmentInput } from "./toolkits/attachment/input.ts"; import { OrchestratorMcpClientRequestId, @@ -47,6 +49,9 @@ import { ProviderInteractionMode, OrchestrationV2ThreadLaunchWorkspaceStrategy, OrchestrationV2DelegatedFrom, + LaunchPresetName, + StandingRole, + composeStandingRole, RunId, OrchestrationV2RunStatus, type OrchestrationV2LinkOrigin, @@ -80,6 +85,10 @@ export const launchInput = Schema.Struct({ ), attachments: Schema.optional(Schema.Array(McpAttachmentInput).check(Schema.isMaxLength(8))), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), + preset: Schema.optional( + LaunchPresetName.annotate({ description: "A `t3_preset` name; explicit fields win." }), + ), + standingRole: Schema.optional(StandingRole), /** Set by a linked environment launching its agent's delegated task here; ignored otherwise. */ delegatedFrom: Schema.optional(OrchestrationV2DelegatedFrom), }); @@ -211,6 +220,15 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( const launchedModel = launched === null || launched.deletedAt !== null ? undefined : launched.modelSelection; const replaying = launchedModel !== undefined; + const preset = + input.preset === undefined || replaying + ? undefined + : yield* (yield* LaunchPresets.LaunchPresets).get(input.preset); + // Explicit fields win. A scratch launch has no checkout to choose. + const workspaceStrategy = + input.workspaceStrategy ?? (input.scratch === true ? undefined : preset?.workspaceStrategy); + const requestedModel = input.modelSelection ?? preset?.modelSelection; + const presetMode = input.runtimeMode === undefined ? preset?.runtimeMode : undefined; const project = replaying ? undefined : yield* Project.ProjectService.pipe( @@ -229,16 +247,16 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( }), ), ); - if (project !== undefined && input.workspaceStrategy?.type === "existing_worktree") - yield* assertProjectWorktree(project.workspaceRoot, input.workspaceStrategy.worktreePath); - if (!replaying && input.modelSelection !== undefined) { - const { instanceId, model } = input.modelSelection; + if (project !== undefined && workspaceStrategy?.type === "existing_worktree") + yield* assertProjectWorktree(project.workspaceRoot, workspaceStrategy.worktreePath); + if (!replaying && requestedModel !== undefined) { + const { instanceId, model } = requestedModel; const providers = yield* (yield* ProviderRegistry.ProviderRegistry).getProviders; const provider = providers.find((candidate) => candidate.instanceId === instanceId); if (provider !== undefined) yield* requireAdvertisedModel(provider, model); } // Reject unresolved base refs and blocked branch names before accepting a launch; worktree provisioning happens later. - const strategy = input.workspaceStrategy; + const strategy = workspaceStrategy; if (!replaying && strategy?.type === "worktree") yield* ThreadLaunch.ThreadLaunchService.pipe( Effect.flatMap((launches) => @@ -266,7 +284,7 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( ) : undefined; const modelSelection = - input.modelSelection ?? + requestedModel ?? launchedModel ?? caller?.modelSelection ?? (projectDefaults @@ -289,21 +307,34 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( "Pass modelSelection: the project has no default model. orchestrator_capabilities lists providers and models.", }); let startedRuntimeMode = - projectDefaults !== undefined && input.runtimeMode === undefined + projectDefaults !== undefined && input.runtimeMode === undefined && presetMode === undefined ? yield* resolveRuntimeMode(runtimeMode, projectDefaults.defaultRuntimeMode).pipe( Effect.orElseSucceed(() => runtimeMode), ) : runtimeMode; if (!replaying) { const providers = yield* (yield* ProviderRegistry.ProviderRegistry).getProviders; - startedRuntimeMode = yield* resolveRuntimeMode(startedRuntimeMode, input.runtimeMode, { - modelSelection, - supportedRuntimeModes: providers.find( - (provider) => provider.instanceId === modelSelection.instanceId, - )?.supportedRuntimeModes, - }); + // A preset's mode counts as asked for, so it is held to the caller's own mode. + startedRuntimeMode = yield* resolveRuntimeMode( + presetMode === undefined ? startedRuntimeMode : runtimeMode, + input.runtimeMode ?? presetMode, + { + modelSelection, + supportedRuntimeModes: providers.find( + (provider) => provider.instanceId === modelSelection.instanceId, + )?.supportedRuntimeModes, + }, + ); } const notification = yield* ThreadNotifications.prepare(input.notify); + // The first turn starts when the launch commits, so the role is stored before it. + const standingRole = composeStandingRole({ + standingRole: input.standingRole ?? preset?.standingRole, + doneFormat: preset?.doneFormat, + }); + const roles = standingRole === undefined ? undefined : yield* ThreadRoles.ThreadRoles; + const roleOwner = McpInvocationContext.actorLabel(scope); + if (roles !== undefined && !replaying) yield* roles.set(threadId, standingRole!, roleOwner); const result = yield* ThreadMessageIntake.launchThread({ commandId, threadId, @@ -312,7 +343,7 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( modelSelection, runtimeMode: startedRuntimeMode, interactionMode, - workspaceStrategy: input.workspaceStrategy ?? { type: "root" }, + workspaceStrategy: workspaceStrategy ?? { type: "root" }, ...(input.message === undefined && attachments.length === 0 ? {} : { @@ -333,6 +364,17 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")( // Kept only alongside linkOrigin: thread.create drops it otherwise. ...(input.delegatedFrom === undefined ? {} : { delegatedFrom: input.delegatedFrom }), }).pipe( + // A failure after the thread committed leaves its role; only a launch that never + // created the thread loses the one stored for it. + Effect.tapError(() => + roles !== undefined && !replaying + ? Effect.gen(function* () { + const created = yield* context.threads.getThreadShell(threadId); + if (created === null || created.deletedAt !== null) + yield* roles.set(threadId, null, roleOwner); + }).pipe(Effect.ignore({ log: true })) + : Effect.void, + ), Effect.mapError((error) => { if (error._tag === "AttachmentClaimError") { return new OrchestratorMcpFailure({ diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 7b16c22858402..83ee781c7184c 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -1264,6 +1264,40 @@ it.effect("a read-only client reads threads and is refused every write before it acknowledgeOnly: false, }); expect(dispatched).toHaveLength(1); + // A standing role is a write like any other, stored apart from the thread's events. + const roleArgs = { threadId: "other-project-thread", standingRole: "You are the build agent." }; + expect(declaredFailure(yield* call("t3_thread_configure", roleArgs))).toMatchObject({ + code: "capability_denied", + }); + // Instructions for another agent need full access. + const roleAs = (access: McpInvocationContext.McpClientCaller["access"], args: object) => + server + .callTool({ name: "t3_thread_configure", arguments: args }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, clientScope(access)), + Effect.provideService(McpSchema.McpServerClient, client), + ); + expect(declaredFailure(yield* roleAs("auto", roleArgs))).toMatchObject({ + code: "capability_denied", + }); + expect(dispatched).toHaveLength(1); + const roleSet = yield* roleAs("full-access", roleArgs); + // The record names its setter by the first 8 characters of the session, never the whole id. + const recorded = { + standingRole: "You are the build agent.", + standingRoleSetBy: expect.stringMatching(/^session- \(Claude Code\), \d{4}-/), + }; + expect(roleSet.structuredContent).toMatchObject(recorded); + expect( + (yield* call("t3_thread_configure", { threadId: "other-project-thread" })).structuredContent, + ).toMatchObject(recorded); + expect(dispatched.at(-1)).toBe("note:Standing role set by session- (Claude Code)."); + expect( + declaredFailure( + yield* roleAs("full-access", { ...roleArgs, runtimeMode: "approval-required" }), + ), + ).toMatchObject({ code: "invalid_request" }); + expect(dispatched).toHaveLength(2); }).pipe( Effect.provide( layerThreadToolkit.pipe( @@ -1295,6 +1329,10 @@ it.effect("a read-only client reads threads and is refused every write before it }), dispatch: (command) => Effect.gen(function* () { + if (command.type === "message.dispatch") { + dispatched.push(`note:${command.text}`); + return { sequence: 8 } as never; + } const actor = yield* RuntimeModeChangeActor; dispatched.push( `${command.type}:${command.type === "thread.runtime-mode.set" ? command.runtimeMode : ""}:${actor}`, @@ -1901,3 +1939,42 @@ it.effect( ); }, ); + +it.effect("presets are read by any caller and changed only with full access", () => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const call = (args: object, access: McpInvocationContext.McpClientCaller["access"]) => + server + .callTool({ name: "t3_preset", arguments: args }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, clientScope(access)), + Effect.provideService(McpSchema.McpServerClient, client), + ); + const save = { action: "save", name: "worker", standingRole: "Be brief." }; + expect(declaredFailure(yield* call(save, "read-only"))).toMatchObject({ + code: "capability_denied", + }); + expect(declaredFailure(yield* call(save, "auto"))).toMatchObject({ + code: "capability_denied", + }); + expect((yield* call({ action: "list" }, "read-only")).structuredContent).toEqual({ + presets: [], + }); + expect((yield* call(save, "full-access")).isError).toBe(false); + expect( + (yield* call({ action: "read", name: "worker" }, "read-only")).structuredContent, + ).toMatchObject({ + name: "worker", + standingRole: "Be brief.", + updatedBy: "session- (Claude Code)", + }); + }).pipe( + Effect.provide( + McpHttpServer.layerPresetToolkit.pipe( + Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), + Layer.provide(McpToolAccessTestkit.liveThreadsLayer), + Layer.provide(Sqlite.layerMemory), + ), + ), + ), +); diff --git a/apps/server/src/mcp/toolkits/preset/handlers.ts b/apps/server/src/mcp/toolkits/preset/handlers.ts new file mode 100644 index 0000000000000..cce7e4cdea3f9 --- /dev/null +++ b/apps/server/src/mcp/toolkits/preset/handlers.ts @@ -0,0 +1,30 @@ +import { OrchestratorMcpFailure } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +import * as LaunchPresets from "../../../persistence/LaunchPresets.ts"; +import * as Access from "../../McpToolAccess.ts"; +import { actorLabel } from "../../McpInvocationContext.ts"; +import { readCaller } from "../../threadAccess.ts"; +import { PresetToolkit } from "./tools.ts"; + +const run = Effect.fn("mcp.presetTool")(function* ( + input: Parameters[0], +) { + const { scope } = yield* readCaller(); + if (scope.thread === undefined && scope.client === undefined) + return yield* new OrchestratorMcpFailure({ + code: "capability_denied", + message: "A caller identity is required.", + }); + return yield* (yield* LaunchPresets.LaunchPresets).run(input, actorLabel(scope)); +}); + +export const layer = Access.toLayer(PresetToolkit, { + // A preset is instructions that broader-mode threads later start with, so changing one is + // held to the same access as changing the environment's settings. + t3_preset: Access.either( + (input) => input.action === "list" || input.action === "read", + Access.reads(run), + Access.writesEnvironment((input) => run(input)), + ), +}); diff --git a/apps/server/src/mcp/toolkits/preset/tools.ts b/apps/server/src/mcp/toolkits/preset/tools.ts new file mode 100644 index 0000000000000..91c44ffb70e22 --- /dev/null +++ b/apps/server/src/mcp/toolkits/preset/tools.ts @@ -0,0 +1,25 @@ +import { + LaunchPresetToolInput, + LaunchPresetToolResult, + OrchestratorMcpFailure, +} from "@t3tools/contracts"; +import { Tool, Toolkit } from "effect/ai"; + +import * as LaunchPresets from "../../../persistence/LaunchPresets.ts"; +import * as ThreadManagement from "../../../orchestration-v2/ThreadManagementService.ts"; +import * as Invocation from "../../McpInvocationContext.ts"; + +export const PresetToolkit = Toolkit.make( + Tool.make("t3_preset", { + description: "Launch presets: list, read, save (omitted fields keep, null clears), delete.", + parameters: LaunchPresetToolInput, + success: LaunchPresetToolResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: [ + Invocation.McpInvocationContext, + ThreadManagement.ThreadManagementService, + LaunchPresets.LaunchPresets, + ], + }).annotate(Tool.Destructive, true), +); diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 69de65f6387d8..77e3744a3b7cd 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -1,10 +1,14 @@ import * as McpClientThreads from "../../../persistence/McpClientThreads.ts"; import * as ThreadLaunchMcp from "../../ThreadLaunchMcpService.ts"; +import * as ThreadRoles from "../../../persistence/ThreadRoles.ts"; +import * as LaunchPresets from "../../../persistence/LaunchPresets.ts"; import * as ProviderRegistry from "../../../provider/ProviderRegistry.ts"; import { expect, it } from "@effect/vitest"; import { EnvironmentId, GitCommandError, + OrchestratorMcpFailure, + type LaunchPreset, ProjectId, ProviderInstanceId, ProviderDriverKind, @@ -62,7 +66,13 @@ const clientLaunchHarness = (input: { readonly hostedAppUrl?: string; /** The orchestrator refuses the new thread's mode, as under a managed ceiling. */ readonly refusedByCeiling?: true; + readonly presets?: Readonly>>; + /** The launch fails before it creates the thread. */ + readonly failsBeforeThread?: true; }) => { + const roleStore = new Map(); + // The role as it stood when each launch reached the orchestrator, which starts the first turn. + const rolesAtLaunch: Array = []; const projectId = ProjectId.make("project:client-target"); const modelSelection = { instanceId: ProviderInstanceId.make("claudeAgent"), @@ -131,7 +141,19 @@ const clientLaunchHarness = (input: { Layer.mock(ThreadLaunch.ThreadLaunchService)({ checkWorktreeBase: input.checkWorktreeBase ?? (() => Effect.void), launch: (launch) => { + if (input.failsBeforeThread) + return Effect.fail( + new ThreadLaunch.ThreadLaunchError({ + operation: "create-thread", + commandId: launch.commandId, + projectId: launch.projectId, + cause: "refused", + }), + ); input.launched.push(launch); + rolesAtLaunch.push( + launch.threadId === undefined ? undefined : roleStore.get(launch.threadId), + ); if (input.refusedByCeiling) { return Effect.fail( new ThreadLaunch.ThreadLaunchError({ @@ -178,6 +200,39 @@ const clientLaunchHarness = (input: { }), Layer.mock(PeerForwarding.PeerForwarding)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }), + Layer.mock(ThreadRoles.ThreadRoles)({ + get: (threadId) => { + const text = roleStore.get(threadId); + return Effect.succeed( + text === undefined + ? undefined + : { text, revision: 1, updatedAt: "2026-01-01T00:00:00Z", updatedBy: "test" }, + ); + }, + set: (threadId, role) => + Effect.sync(() => { + if (role === null) roleStore.delete(threadId); + else roleStore.set(threadId, role); + }), + }), + Layer.mock(LaunchPresets.LaunchPresets)({ + get: (name) => { + const preset = input.presets?.[name]; + return preset === undefined + ? Effect.fail( + new OrchestratorMcpFailure({ + code: "invalid_request", + message: `Launch preset "${name}" was not found.`, + }), + ) + : Effect.succeed({ + name, + updatedAt: "2026-01-01T00:00:00Z", + updatedBy: "test", + ...preset, + }); + }, + }), DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer)), ).pipe( Layer.provideMerge(GitVcsDriver.layer), @@ -194,6 +249,8 @@ const clientLaunchHarness = (input: { modelSelection, projectDefaults, provenance, + rolesAtLaunch, + roleStore, dependencies: layerDependencies, }; }; @@ -300,6 +357,108 @@ it.effect( }), ); +it.effect("a launch applies its preset, explicit fields win, and the role is stored first", () => + Effect.gen(function* () { + const presetModel = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-preset" }; + const preset = { + modelSelection: presetModel, + runtimeMode: "auto-accept-edits", + standingRole: "Be brief.", + doneFormat: "DONE: ", + workspaceStrategy: { type: "worktree", baseRef: "origin/main", startFromOrigin: true }, + } as const; + const launchWith = Effect.fn("test.launchWithPreset")(function* ( + ceiling: RuntimeMode, + params: Record, + ) { + const launched: Array = []; + const harness = clientLaunchHarness({ + runtimeModeCeiling: ceiling, + launched, + presets: { worker: preset }, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe( + Layer.provide(harness.dependencies), + ), + ), + ); + const result = yield* toolkit + .handle("t3_thread_launch", { title: "Work", projectId: harness.projectId, ...params }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(harness.dependencies)); + return { result: result.at(-1)?.result, launched, rolesAtLaunch: harness.rolesAtLaunch }; + }); + const doneRole = "Your final message must follow this format:\nDONE: "; + + const fromPreset = yield* launchWith("full-access", { preset: "worker", message: "Go" }); + expect(fromPreset.launched[0]).toMatchObject({ + modelSelection: presetModel, + runtimeMode: "auto-accept-edits", + workspaceStrategy: { type: "worktree", baseRef: "origin/main" }, + }); + expect(fromPreset.rolesAtLaunch).toEqual([`Be brief.\n\n${doneRole}`]); + + const explicit = yield* launchWith("full-access", { + preset: "worker", + runtimeMode: "approval-required", + workspaceStrategy: { type: "root" }, + standingRole: "Explicit.", + }); + expect(explicit.launched[0]).toMatchObject({ + modelSelection: presetModel, + runtimeMode: "approval-required", + workspaceStrategy: { type: "root" }, + }); + expect(explicit.rolesAtLaunch).toEqual([`Explicit.\n\n${doneRole}`]); + + // A preset is never a way past the caller's own mode. + const escalated = yield* launchWith("approval-required", { preset: "worker" }); + expect(escalated.result).toMatchObject({ code: "runtime_mode_escalation_denied" }); + expect(escalated.launched).toHaveLength(0); + + const unknown = yield* launchWith("full-access", { preset: "missing" }); + expect(unknown.result).toMatchObject({ code: "invalid_request" }); + expect(unknown.launched).toHaveLength(0); + }), +); + +it.effect("a failed launch keeps the role of a thread that exists and drops the rest", () => + Effect.gen(function* () { + const launchFailing = Effect.fn("test.launchFailing")(function* ( + failure: { readonly refusedByCeiling: true } | { readonly failsBeforeThread: true }, + ) { + const launched: Array = []; + const harness = clientLaunchHarness({ + runtimeModeCeiling: "full-access", + launched, + presets: { worker: { standingRole: "Be brief." } }, + ...failure, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe( + Layer.provide(harness.dependencies), + ), + ), + ); + yield* toolkit + .handle("t3_thread_launch", { + title: "Work", + projectId: harness.projectId, + preset: "worker", + message: "Go", + }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(harness.dependencies)); + return [...harness.roleStore.values()]; + }); + // The orchestrator refused after the thread existed: its role stays. + expect(yield* launchFailing({ refusedByCeiling: true })).toEqual(["Be brief."]); + // Nothing was created: no role is left for a thread that never was. + expect(yield* launchFailing({ failsBeforeThread: true })).toEqual([]); + }), +); + it.effect("a launch steps an inherited mode down to the provider's, and refuses one it lacks", () => Effect.gen(function* () { // Pi offers no Auto; a mode a provider lacks would run as Supervised. diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index 6febdcdc9f191..540d45a61391a 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -1,5 +1,7 @@ import * as ThreadLaunchMcp from "../../ThreadLaunchMcpService.ts"; import * as McpClientThreads from "../../../persistence/McpClientThreads.ts"; +import * as LaunchPresets from "../../../persistence/LaunchPresets.ts"; +import * as ThreadRoles from "../../../persistence/ThreadRoles.ts"; import { NonNegativeInt, @@ -126,6 +128,8 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { dependencies: [ ...shared.dependencies, ThreadLaunchMcp.ThreadLaunchMcpService, + LaunchPresets.LaunchPresets, + ThreadRoles.ThreadRoles, McpClientThreads.McpClientThreads, ThreadLaunchService.ThreadLaunchService, ManagedProjectFolders.ManagedProjectFolders, diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 2f2bb7f55a3ca..f42aa0b1d2095 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -3,6 +3,7 @@ import { CommandActor } from "../../../orchestration-v2/CommandActor.ts"; import { ProviderRequestKind, type CommandId, + MessageId, type RuntimeRequestId, ThreadId, type OrchestrationV2ThreadProjection, @@ -25,6 +26,7 @@ import { readThread, unavailable, } from "../../threadAccess.ts"; +import * as ThreadRoles from "../../../persistence/ThreadRoles.ts"; import * as ScratchThreadPromotion from "../../../project/ScratchThreadPromotion.ts"; import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts"; import * as ScheduledTasks from "../../../scheduledTasks/ScheduledTaskService.ts"; @@ -50,6 +52,21 @@ function queueEntry( truncated: characters.length > limit, }; } +const configuration = ( + thread: Pick< + OrchestrationV2ThreadProjection["thread"], + "id" | "modelSelection" | "runtimeMode" | "interactionMode" + >, + role: ThreadRoles.ThreadRoleRecord | undefined, +) => ({ + threadId: thread.id, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + ...(role === undefined + ? {} + : { standingRole: role.text, standingRoleSetBy: `${role.updatedBy}, ${role.updatedAt}` }), +}); const withMcpCommandActor = Effect.fn("mcp.withCommandActor")(function* ( effect: Effect.Effect, ) { @@ -258,7 +275,8 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { (input) => input.modelSelection === undefined && input.runtimeMode === undefined && - input.acknowledgeOnly === undefined, + input.acknowledgeOnly === undefined && + input.standingRole === undefined, McpToolAccess.reads((input) => Effect.gen(function* () { const { @@ -266,10 +284,7 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { projection: { thread }, } = yield* readThread(input.threadId); return { - threadId: thread.id, - modelSelection: thread.modelSelection, - runtimeMode: thread.runtimeMode, - interactionMode: thread.interactionMode, + ...configuration(thread, yield* (yield* ThreadRoles.ThreadRoles).get(thread.id)), acknowledgeOnly: yield* threads .acknowledgeOnly(thread.id) .pipe(Effect.mapError(unavailable)), @@ -285,6 +300,40 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { if (input.acknowledgeOnly !== undefined) { return yield* configureAcknowledgeOnly(threads, thread, input); } + if (input.standingRole !== undefined) { + if (input.modelSelection !== undefined || input.runtimeMode !== undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "Set standingRole separately from modelSelection and runtimeMode.", + }); + } + // The role is instructions for another agent, and for the caller's own thread a way + // to drop its own rules, so only full access writes it. An outside client, an agent + // for its own thread and an agent for another all meet the same rule. + yield* McpToolAccess.requireFullAccess("change a thread's standing role"); + const scope = yield* McpInvocationContext.McpInvocationContext; + const roles = yield* ThreadRoles.ThreadRoles; + // The visible record comes first: a role never changes without a note in the thread. + const commandId = yield* newCommandId(); + yield* threads + .dispatch({ + type: "message.dispatch", + postOnly: true, + commandId, + threadId: thread.id, + ...(scope.thread === undefined ? {} : { senderThreadId: scope.thread.threadId }), + ...McpInvocationContext.messageSenderFor(scope), + messageId: MessageId.make(commandId), + text: `Standing role ${input.standingRole.trim() === "" ? "cleared" : "set"} by ${McpInvocationContext.actorLabel(scope)}.`, + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "agent", + creationSource: "mcp", + }) + .pipe(withMcpCommandActor, Effect.mapError(dispatchFailure)); + yield* roles.set(thread.id, input.standingRole, McpInvocationContext.actorLabel(scope)); + return configuration(thread, yield* roles.get(thread.id)); + } if (input.runtimeMode !== undefined) { if (input.modelSelection !== undefined) { return yield* new OrchestratorMcpFailure({ diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index ba79df8fe1e0d..8fea1e20ad33d 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -21,6 +21,7 @@ import { NonNegativeInt, ProjectId, ThreadFreeze, + StandingRole, } from "@t3tools/contracts"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Schema from "effect/Schema"; @@ -32,6 +33,7 @@ import * as ScheduledTaskService from "../../../scheduledTasks/ScheduledTaskServ import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as PendingApproval from "../../PendingApprovalMcpService.ts"; +import * as ThreadRoles from "../../../persistence/ThreadRoles.ts"; const ThreadOrganizeTool = Tool.make("t3_thread_organize", { description: @@ -208,19 +210,23 @@ const threadConfiguration = Schema.Struct({ interactionMode: ProviderInteractionMode, // Optional: a linked environment on another build does not report it. acknowledgeOnly: Schema.optional(Schema.Boolean), + standingRole: Schema.optional(Schema.String), + standingRoleSetBy: Schema.optional(Schema.String), }); const ThreadConfigureTool = Tool.make("t3_thread_configure", { ...commandTool, description: - "Read a thread's provider, model, and modes, or set its provider, model, and options with `modelSelection`. Omit `threadId` for this thread. Set `runtimeMode` alone to lower permissions on an idle thread; raises and active runs are refused. Values from `orchestrator_capabilities`. Full access: `acknowledgeOnly` alone makes the server, not the agent, acknowledge each message.", + 'Read a thread\'s provider, model, and modes, or set its provider, model, and options with `modelSelection`. Omit `threadId` for this thread. Set `runtimeMode` alone to lower permissions on an idle thread; raises and active runs are refused. Values from `orchestrator_capabilities`. Full access: `acknowledgeOnly` alone makes the server, not the agent, acknowledge each message. `standingRole` (alone; "" clears) is sent with every turn.', parameters: Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), modelSelection: Schema.optional(ModelSelection), runtimeMode: Schema.optional(RuntimeMode), acknowledgeOnly: Schema.optional(Schema.Boolean), + standingRole: Schema.optional(StandingRole), }), success: Schema.Union([threadConfiguration, OrchestrationV2DispatchCommandResult]), + dependencies: [...commandTool.dependencies, ThreadRoles.ThreadRoles], }).annotate(Tool.Destructive, true); const ThreadTransferTool = Tool.make("t3_thread_transfer", { diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index f58d83c74fb9d..089624f8a65e2 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -56,6 +56,7 @@ import * as RunExecutionService from "./RunExecutionService.ts"; import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; const isDomainEvent = Schema.is(OrchestrationV2DomainEvent); @@ -196,6 +197,7 @@ function startWithMissingWorktree( }), Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }), Layer.mock(RuntimePolicy.RuntimePolicyV2)({}), + Layer.mock(ThreadRoles.ThreadRoles)({ get: () => Effect.undefined }), ), ), ); @@ -279,6 +281,14 @@ function makeLocalCommandHarness(input: { readonly gitWorkflow?: GitWorkflow.GitWorkflowService["Service"]; readonly fileSystem?: FileSystem.FileSystem; readonly previousNativeSession?: boolean; + /** The thread's standing role, stored after its session started. */ + readonly standingRole?: string; + /** Where the provider last took revision 2 of the role; defaults to never. */ + readonly standingRoleDelivered?: { + readonly revision?: number; + readonly nativeThreadId?: string | null; + readonly runOrdinal?: number; + }; readonly previousMessages?: ReadonlyArray; readonly logoutFailure?: string; readonly openFailure?: unknown; @@ -847,6 +857,11 @@ function makeLocalCommandHarness(input: { lastError: null, }; const startTurn = vi.fn((_turn: ProviderAdapter.ProviderAdapterV2TurnInput) => Effect.void); + const roleDeliveries: Array<{ + revision: number; + nativeThreadId: string | null; + runOrdinal: number; + }> = []; const sideChatThread = { ...providerThread, nativeThreadRef: { @@ -895,13 +910,17 @@ function makeLocalCommandHarness(input: { providerSession, ensureThread: () => Effect.succeed(providerThread), resumeThread: () => Effect.succeed(projection.providerThreads[0]), + startTurn: (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => + startTurn(turn), } as never) : Effect.die("A local command must not open a native session."), ); const startRootRun = vi.fn< (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect >((runInput) => - input.sideChat !== undefined || input.nativeThreadLastRanOn !== undefined + input.sideChat !== undefined || + input.nativeThreadLastRanOn !== undefined || + input.standingRole !== undefined ? runInput.session.startTurn(runInput as never).pipe(Effect.orDie) : input.failReadsAfterRunning === true || input.previousTurn !== undefined ? Effect.suspend(() => @@ -1113,11 +1132,37 @@ function makeLocalCommandHarness(input: { Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy, }), + Layer.mock(ThreadRoles.ThreadRoles)({ + get: () => + Effect.succeed( + input.standingRole === undefined + ? undefined + : { + text: input.standingRole, + revision: 2, + updatedAt: "2026-01-01T00:00:00Z", + updatedBy: "test", + ...(input.standingRoleDelivered === undefined + ? {} + : { + delivered: { + revision: 2, + nativeThreadId: "existing-native-conversation", + runOrdinal: 1, + ...input.standingRoleDelivered, + }, + }), + }, + ), + markDelivered: (_threadId, delivery) => + Effect.sync(() => void roleDeliveries.push(delivery)), + }), ), ), ); return { settledRuns, + roleDeliveries, getThreadRecords, getTurnStartContext, getRuntimeRecoveryProjection, @@ -1466,6 +1511,63 @@ effectIt.effect.each([false, true])( }), ); +effectIt.effect.each([ + // A turn on a native session that began before the role was stored. + { text: "Continue", sent: "ROLE" }, + { text: "/home/user/app fails to build", sent: "ROLE" }, + { text: "/etc is full", sent: "ROLE" }, + // Native commands run verbatim; a prefix would stop them being commands. + { text: "/review", sent: undefined }, + { text: "/goal all tests pass", sent: undefined }, +] as const)("sends the standing role with a turn: %j", ({ text, sent }) => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text, + previousTurn: { runStatus: "completed", turnStatus: "completed" }, + standingRole: "You are the build agent.", + }); + harness.tryHandlePromptCommand.mockReturnValue(Effect.succeed(false)); + + yield* harness.start; + + const heading = + "Standing role for this thread, set in T3 and sent with every turn (not part of the user's message):"; + expect(harness.startTurn.mock.calls.map(([turn]) => turn.message.text)).toEqual([ + sent === undefined ? text : withInlineContext(`${heading}\nYou are the build agent.`, text), + ]); + expect(harness.roleDeliveries).toEqual( + sent === undefined + ? [] + : [{ revision: 2, nativeThreadId: "existing-native-conversation", runOrdinal: 2 }], + ); + }), +); + +// Each copy of the role stays in the provider's history, so it is sent again only when needed. +effectIt.effect.each([ + { delivered: {}, sent: false }, + { delivered: { revision: 1 }, sent: true }, + { delivered: { nativeThreadId: "an-earlier-session" }, sent: true }, + { delivered: { runOrdinal: -23 }, sent: true }, +] as const)("sends the standing role again only when needed: %j", ({ delivered, sent }) => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text: "Continue", + previousTurn: { runStatus: "completed", turnStatus: "completed" }, + standingRole: "You are the build agent.", + standingRoleDelivered: delivered, + }); + harness.tryHandlePromptCommand.mockReturnValue(Effect.succeed(false)); + + yield* harness.start; + + expect( + harness.startTurn.mock.calls.map(([turn]) => turn.message.text.includes("Standing role")), + ).toEqual([sent]); + expect(harness.roleDeliveries).toHaveLength(sent ? 1 : 0); + }), +); + effectIt.effect.each([ { runStatus: "failed", turnStatus: "completed" }, { runStatus: "completed", turnStatus: "completed" }, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index d32cf5b1aff1f..7ba02306d1ec4 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -53,6 +53,8 @@ import { latestNativeContextUsage, } from "@t3tools/provider-core/server/handoffBudget"; import { deliverContextHandoffs, withInlineContext } from "./ContextHandoffDelivery.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; +import { standingRoleDue } from "./standingRoleDelivery.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; @@ -148,6 +150,12 @@ export interface ProviderTurnStartServiceV2Shape { }) => Effect.Effect; } +const STANDING_ROLE_HEADING = + "Standing role for this thread, set in T3 and sent with every turn (not part of the user's message):"; +/** A bare native slash command, such as /compact. A path like /home/x is a message. */ +const NATIVE_COMMAND = + /^\s*\/(?!(?:etc|usr|var|home|tmp|opt|bin|sbin|dev|proc|root|srv|mnt|boot|lib|lib64|sys|run|Users|Library|Applications)(?:\s|\/|$))[A-Za-z][\w-]*(\s|$)/; + export class ProviderTurnStartServiceV2 extends Context.Service< ProviderTurnStartServiceV2, ProviderTurnStartServiceV2Shape @@ -171,6 +179,7 @@ export const layer: Layer.Layer< | RunExecutionService.RunExecutionServiceV2 | RuntimePolicy.RuntimePolicyV2 | TurnItemPositionStore.TurnItemPositionStoreV2 + | ThreadRoles.ThreadRoles > = Layer.effect( ProviderTurnStartServiceV2, Effect.gen(function* () { @@ -188,6 +197,7 @@ export const layer: Layer.Layer< const runExecution = yield* RunExecutionService.RunExecutionServiceV2; const runtimePolicy = yield* RuntimePolicy.RuntimePolicyV2; const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2; + const threadRoles = yield* ThreadRoles.ThreadRoles; const isManagedWorktreePath = yield* makeIsManagedWorktreePath; const serverSettings = yield* ServerSettings.ServerSettingsService; const resolveFailoverTarget = ( @@ -951,6 +961,15 @@ export const layer: Layer.Layer< restartCancelledWork.length === 0 ? "" : restartCancelledBackgroundWorkNote(restartCancelledWork); + // Read at every turn start: providers read AGENTS.md only when a session starts, so a role + // set later, or on a thread whose session predates it, still has to reach the provider. + // Native commands (/compact, /review, goals) go verbatim. + const standingRole = + goalCommand || NATIVE_COMMAND.test(userText) + ? undefined + : yield* threadRoles.get(projection.thread.id); + const standingRoleBlock = + standingRole === undefined ? "" : `${STANDING_ROLE_HEADING}\n${standingRole.text}`; const tokenCap = yield* handoffTokenCapConfig.pipe( Effect.orElseSucceed(() => DEFAULT_HANDOFF_TOKEN_CAP), ); @@ -1096,7 +1115,9 @@ export const layer: Layer.Layer< return handoffBudget({ tokenCap, modelContextWindow, - userText: restartNote === "" ? userText : `${restartNote}\n\n${userText}`, + userText: [standingRoleBlock, restartNote, userText] + .filter((part) => part !== "") + .join("\n\n"), attachments: message.attachments, providerThread: budgetProviderThread, nativeContextEstimate: @@ -1659,6 +1680,24 @@ export const layer: Layer.Layer< createdAt: providerThread.createdAt, updatedAt: now, }; + const roleNativeThreadId = runningProviderThread.nativeThreadRef?.nativeId ?? null; + const compactionMessageIds = new Set( + projection.messages.filter(isCompactionCommandMessage).map((candidate) => candidate.id), + ); + const standingRoleSend = + standingRole !== undefined && + standingRoleDue({ + role: standingRole, + nativeThreadId: roleNativeThreadId, + runOrdinal: run.ordinal, + compactedSinceDelivery: (after) => + projection.runs.some( + (source) => + source.ordinal > after && + source.ordinal < run.ordinal && + compactionMessageIds.has(source.userMessageId), + ), + }); const runningRun: OrchestrationV2Run = { ...run, status: "running", @@ -1831,7 +1870,8 @@ export const layer: Layer.Layer< sideChatHandoff === undefined || compact || goalCommand ? "" : sideChatHandoff.summaryText; - const context = [delivery.context, sideChatContext, restartNote] + const roleContext = compact || goalCommand || !standingRoleSend ? "" : standingRoleBlock; + const context = [roleContext, delivery.context, sideChatContext, restartNote] .filter((part) => part !== "") .join("\n\n"); const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput; @@ -1856,6 +1896,18 @@ export const layer: Layer.Layer< ), ); // After provider acceptance, failed delivery bookkeeping must not stop live ingestion. + if (roleContext !== "" && standingRole !== undefined) + yield* threadRoles + .markDelivered(projection.thread.id, { + revision: standingRole.revision, + nativeThreadId: roleNativeThreadId, + runOrdinal: run.ordinal, + }) + .pipe( + Effect.catchCause(() => + Effect.logWarning("Failed to record standing role delivery", { runId: run.id }), + ), + ); const nativeThreadId = runningProviderThread.nativeThreadRef?.nativeId; yield* Effect.forEach( [ @@ -1899,6 +1951,7 @@ export const layer: Layer.Layer< retryHandoff.length === 0 && sideChatHandoff === undefined && restartNote === "" && + !standingRoleSend && !noteContinuation ? session : makeDeliverySession(session, startWithHandoffs); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index d83f0603d7f7c..465709553a66f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -47,6 +47,7 @@ import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryServic import * as ProviderSwitchService from "./ProviderSwitchService.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; import * as ProviderTurnStartService from "./ProviderTurnStartService.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as ResourceCleanupService from "./ResourceCleanupService.ts"; @@ -177,6 +178,7 @@ const layerProviderTurnStartServiceProvided = ProviderTurnStartService.layer.pip layerRunExecutionServiceProvided, layerRuntimePolicyProvided, TurnItemPositionStore.layer, + ThreadRoles.layer, ), ), ); diff --git a/apps/server/src/orchestration-v2/standingRoleDelivery.test.ts b/apps/server/src/orchestration-v2/standingRoleDelivery.test.ts new file mode 100644 index 0000000000000..d9a3db9d3a0ba --- /dev/null +++ b/apps/server/src/orchestration-v2/standingRoleDelivery.test.ts @@ -0,0 +1,36 @@ +import { expect, it } from "vite-plus/test"; + +import { STANDING_ROLE_REFRESH_RUNS, standingRoleDue } from "./standingRoleDelivery.ts"; + +const role = { + text: "role", + revision: 3, + updatedAt: "2026-01-01T00:00:00Z", + updatedBy: "test", + delivered: { revision: 3, nativeThreadId: "native-1", runOrdinal: 4 }, +} as const; +const due = (overrides: Partial[0]> = {}): boolean => + standingRoleDue({ + role, + nativeThreadId: "native-1", + runOrdinal: 5, + compactedSinceDelivery: () => false, + ...overrides, + }); + +it("is not due again while the provider still holds it", () => { + expect(due()).toBe(false); +}); + +it.each([ + ["never delivered", { role: { ...role, delivered: undefined } }], + ["changed", { role: { ...role, revision: 4 } }], + ["a new native session", { nativeThreadId: "native-2" }], + [ + "enough runs to outlast an automatic compaction", + { runOrdinal: 4 + STANDING_ROLE_REFRESH_RUNS }, + ], + ["a compaction since delivery", { compactedSinceDelivery: (after: number) => after === 4 }], +] as const)("is due after %s", (_name, overrides) => { + expect(due(overrides as never)).toBe(true); +}); diff --git a/apps/server/src/orchestration-v2/standingRoleDelivery.ts b/apps/server/src/orchestration-v2/standingRoleDelivery.ts new file mode 100644 index 0000000000000..8097e687cb6c1 --- /dev/null +++ b/apps/server/src/orchestration-v2/standingRoleDelivery.ts @@ -0,0 +1,25 @@ +import type { ThreadRoleRecord } from "../persistence/ThreadRoles.ts"; + +/** Runs after which a role is sent again, in case the provider compacted it away on its own. */ +export const STANDING_ROLE_REFRESH_RUNS = 25; + +/** + * Whether this turn carries the role. Each copy stays in the provider's history, so a role goes + * once, then again when it changed, the native session is new, a compaction ran, or enough runs + * passed for an automatic compaction to have dropped it. + */ +export function standingRoleDue(input: { + readonly role: ThreadRoleRecord; + readonly nativeThreadId: string | null; + readonly runOrdinal: number; + readonly compactedSinceDelivery: (afterRunOrdinal: number) => boolean; +}): boolean { + const { delivered } = input.role; + return ( + delivered === undefined || + delivered.revision !== input.role.revision || + delivered.nativeThreadId !== input.nativeThreadId || + input.runOrdinal - delivered.runOrdinal >= STANDING_ROLE_REFRESH_RUNS || + input.compactedSinceDelivery(delivered.runOrdinal) + ); +} diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 665a04f754749..adb6fe9212737 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -66,6 +66,7 @@ import * as TurnItemPositionStore from "../TurnItemPositionStore.ts"; import * as RuntimeRequestService from "../RuntimeRequestService.ts"; import * as ThreadCommandExecutor from "../ThreadCommandExecutor.ts"; import * as ThreadForkService from "../ThreadForkService.ts"; +import * as ThreadRoles from "../../persistence/ThreadRoles.ts"; import { runOrchestratorV2Scenario, type OrchestratorV2ScenarioStepError, @@ -454,6 +455,7 @@ export function layerWithRegistry( layerRuntime, layerServerConfig, layerServerSettings, + Layer.mock(ThreadRoles.ThreadRoles)({ get: () => Effect.undefined }), ), ), ); diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index a5790332ebc64..5260b6ebc9997 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -344,6 +344,12 @@ const make = Effect.gen(function* () { ...(runtimeMode === undefined ? {} : { runtimeMode }), ...(interactionMode === undefined ? {} : { interactionMode }), message: task, + // The preset resolves there. Its checkout kind never applies: a remote task runs in + // that project's checkout. + ...(input.preset === undefined + ? {} + : { preset: input.preset, workspaceStrategy: { type: "root" as const } }), + ...(input.standingRole === undefined ? {} : { standingRole: input.standingRole }), clientRequestId: `delegate:${scope.thread.threadId}:${key}`, // The thread there shows it is this thread's subagent, with a way back. delegatedFrom: { diff --git a/apps/server/src/persistence/ForkMigrations.ts b/apps/server/src/persistence/ForkMigrations.ts index 835366ba9f03a..4fcc885c3d568 100644 --- a/apps/server/src/persistence/ForkMigrations.ts +++ b/apps/server/src/persistence/ForkMigrations.ts @@ -17,6 +17,7 @@ import ForkMigration0012 from "./ForkMigrations/012_WorkLedgerApproval.ts"; import ForkMigration0013 from "./ForkMigrations/013_WorkLedgerHygiene.ts"; import ForkMigration0014 from "./ForkMigrations/014_McpRequestLogConnection.ts"; import ForkMigration0015 from "./ForkMigrations/015_ScheduledRunHistoryIndex.ts"; +import ForkMigration0016 from "./ForkMigrations/016_ThreadRolesAndPresets.ts"; // Fork-only migrations, recorded in their own table so upstream keeps every id in // effect_sql_migrations. They run after all upstream migrations. Add new ones here, never in // Migrations.ts, and never rename one: the repair below finds old rows by name. See FORK.md. @@ -36,6 +37,7 @@ const forkMigrationEntries = [ [13, "WorkLedgerHygiene", ForkMigration0013], [14, "McpRequestLogConnection", ForkMigration0014], [15, "ScheduledRunHistoryIndex", ForkMigration0015], + [16, "ThreadRolesAndPresets", ForkMigration0016], ] as const; export const FORK_MIGRATIONS_TABLE = "t3_fork_migrations"; diff --git a/apps/server/src/persistence/ForkMigrations/016_ThreadRolesAndPresets.ts b/apps/server/src/persistence/ForkMigrations/016_ThreadRolesAndPresets.ts new file mode 100644 index 0000000000000..338b8f45e42d7 --- /dev/null +++ b/apps/server/src/persistence/ForkMigrations/016_ThreadRolesAndPresets.ts @@ -0,0 +1,30 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE IF NOT EXISTS thread_standing_roles ( + thread_id TEXT PRIMARY KEY NOT NULL, + role_text TEXT NOT NULL, + revision INTEGER NOT NULL, + updated_at TEXT NOT NULL, + updated_by TEXT NOT NULL, + delivered_revision INTEGER, + delivered_native_thread_id TEXT, + delivered_run_ordinal INTEGER + )`; + yield* sql`CREATE TABLE IF NOT EXISTS launch_presets ( + name TEXT PRIMARY KEY NOT NULL, + preset_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + updated_by TEXT NOT NULL + )`; + // A deleted or purged thread keeps no role. + yield* sql`CREATE TRIGGER IF NOT EXISTS thread_standing_roles_thread_deleted + AFTER UPDATE OF deleted_at ON orchestration_v2_projection_threads + WHEN NEW.deleted_at IS NOT NULL AND OLD.deleted_at IS NULL + BEGIN DELETE FROM thread_standing_roles WHERE thread_id = NEW.thread_id; END`; + yield* sql`CREATE TRIGGER IF NOT EXISTS thread_standing_roles_thread_purged + AFTER DELETE ON orchestration_v2_projection_threads + BEGIN DELETE FROM thread_standing_roles WHERE thread_id = OLD.thread_id; END`; +}); diff --git a/apps/server/src/persistence/LaunchPresets.test.ts b/apps/server/src/persistence/LaunchPresets.test.ts new file mode 100644 index 0000000000000..2852a56461efa --- /dev/null +++ b/apps/server/src/persistence/LaunchPresets.test.ts @@ -0,0 +1,111 @@ +import { ProviderInstanceId, type LaunchPresetToolInput } from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as LaunchPresets from "./LaunchPresets.ts"; +import * as Sqlite from "./Sqlite.ts"; +import * as ThreadRoles from "./ThreadRoles.ts"; + +const layer = Layer.mergeAll(LaunchPresets.layer, ThreadRoles.layer).pipe( + Layer.provideMerge(Sqlite.layerMemory), +); +const model = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }; + +it.effect("saves, merges, clears, lists, reads and deletes presets, and records who", () => + Effect.gen(function* () { + const presets = yield* LaunchPresets.LaunchPresets; + const run = (input: LaunchPresetToolInput, by = "thread-a") => presets.run(input, by); + + yield* run({ + action: "save", + name: "worker", + modelSelection: model, + runtimeMode: "auto", + standingRole: "Be brief.", + doneFormat: "DONE: ", + }); + // Omitted fields keep their value; null clears one. + const updated = yield* run( + { action: "save", name: "worker", runtimeMode: null, standingRole: "Be thorough." }, + "thread-b", + ); + expect(updated).toMatchObject({ + name: "worker", + modelSelection: model, + standingRole: "Be thorough.", + doneFormat: "DONE: ", + updatedBy: "thread-b", + }); + expect(updated).not.toHaveProperty("runtimeMode"); + expect(yield* run({ action: "read", name: "worker" })).toEqual(updated); + expect(yield* run({ action: "list" })).toEqual({ presets: ["worker"] }); + + // The tool advertises modelSelection loosely, so the service is what refuses a bad one. + const badModel = yield* run({ action: "save", name: "bad", modelSelection: { model: 1 } }).pipe( + Effect.flip, + ); + expect(badModel.code).toBe("invalid_request"); + expect(yield* run({ action: "list" })).toEqual({ presets: ["worker"] }); + + expect(yield* run({ action: "delete", name: "worker" })).toEqual({ + deleted: true, + name: "worker", + }); + expect((yield* presets.get("worker").pipe(Effect.flip)).code).toBe("invalid_request"); + expect((yield* run({ action: "save" }).pipe(Effect.flip)).code).toBe("invalid_request"); + }).pipe(Effect.provide(layer)), +); + +it.effect("tracks a thread's role revisions and deliveries, and drops it with the thread", () => + Effect.gen(function* () { + const roles = yield* ThreadRoles.ThreadRoles; + const sql = yield* SqlClient.SqlClient; + expect(yield* roles.get("thread-1")).toBeUndefined(); + yield* roles.set("thread-1", " You are the build agent. ", "thread-a"); + expect(yield* roles.get("thread-1")).toMatchObject({ + text: "You are the build agent.", + revision: 1, + updatedBy: "thread-a", + }); + yield* roles.markDelivered("thread-1", { + revision: 1, + nativeThreadId: "native", + runOrdinal: 3, + }); + expect((yield* roles.get("thread-1"))?.delivered).toEqual({ + revision: 1, + nativeThreadId: "native", + runOrdinal: 3, + }); + // A change bumps the revision, and a delivery of the old one no longer counts. + yield* roles.set("thread-1", "Now the review agent.", "thread-b"); + yield* roles.markDelivered("thread-1", { + revision: 1, + nativeThreadId: "native", + runOrdinal: 4, + }); + expect(yield* roles.get("thread-1")).toMatchObject({ + text: "Now the review agent.", + revision: 2, + updatedBy: "thread-b", + delivered: { revision: 1, runOrdinal: 3 }, + }); + const refused = yield* roles + .set("thread-1", "x".repeat(ThreadRoles.THREAD_ROLE_STORED_MAX_CHARS + 1), "thread-a") + .pipe(Effect.flip); + expect(refused.code).toBe("invalid_request"); + + yield* sql`INSERT INTO orchestration_v2_projection_threads (thread_id, project_id, title, + default_provider, runtime_mode, interaction_mode, created_at, updated_at, payload_json) + VALUES ('thread-1', 'project', 'one', 'codex', 'full-access', 'default', 'now', 'now', '{}')`; + yield* sql`UPDATE orchestration_v2_projection_threads + SET deleted_at = '1970-01-01T00:00:03.000Z' WHERE thread_id = 'thread-1'`; + expect(yield* roles.get("thread-1")).toBeUndefined(); + + yield* roles.set("thread-2", "Another.", "thread-a"); + yield* roles.set("thread-2", "", "thread-a"); + expect(yield* roles.get("thread-2")).toBeUndefined(); + }).pipe(Effect.provide(layer)), +); diff --git a/apps/server/src/persistence/LaunchPresets.ts b/apps/server/src/persistence/LaunchPresets.ts new file mode 100644 index 0000000000000..2ccca6911dc10 --- /dev/null +++ b/apps/server/src/persistence/LaunchPresets.ts @@ -0,0 +1,125 @@ +import { + LaunchPreset, + ModelSelection, + OrchestratorMcpFailure, + type LaunchPresetToolInput, + type LaunchPresetToolResult, +} from "@t3tools/contracts"; +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/sql/SqlClient"; + +export class LaunchPresets extends Context.Service< + LaunchPresets, + { + readonly get: (name: string) => Effect.Effect; + readonly run: ( + input: LaunchPresetToolInput, + updatedBy: string, + ) => Effect.Effect; + } +>()("t3/persistence/LaunchPresets") {} + +const presetJson = Schema.fromJsonString(LaunchPreset); +const decodeModel = Schema.decodeUnknownEffect(ModelSelection); +const decodePreset = Schema.decodeUnknownEffect(presetJson); +const encodePreset = Schema.encodeEffect(presetJson); +const storage = () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "Could not access launch presets.", + }); +const invalid = (message: string) => + new OrchestratorMcpFailure({ code: "invalid_request", message }); + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const find = (name: string) => + sql<{ preset_json: string }>`SELECT preset_json FROM launch_presets WHERE name = ${name}`.pipe( + Effect.flatMap((rows) => + rows[0] === undefined ? Effect.undefined : decodePreset(rows[0].preset_json), + ), + Effect.mapError(storage), + ); + const get = Effect.fn("LaunchPresets.get")(function* (name: string) { + return ( + (yield* find(name)) ?? + (yield* invalid(`Launch preset "${name}" was not found. Use t3_preset list.`)) + ); + }); + const run = Effect.fn("LaunchPresets.run")(function* ( + input: LaunchPresetToolInput, + updatedBy: string, + ) { + if (input.action === "list") { + const rows = yield* sql<{ name: string }>`SELECT name FROM launch_presets ORDER BY name`.pipe( + Effect.mapError(storage), + ); + return { presets: rows.map((row) => row.name) }; + } + const name = input.name; + if (name === undefined) return yield* invalid("This action needs `name`."); + if (input.action === "read") return yield* get(name); + if (input.action === "delete") { + yield* get(name); + yield* sql`DELETE FROM launch_presets WHERE name = ${name}`.pipe(Effect.mapError(storage)); + return { deleted: true as const, name }; + } + const existing = yield* find(name); + const model = + input.modelSelection === undefined || input.modelSelection === null + ? input.modelSelection + : yield* decodeModel(input.modelSelection).pipe( + Effect.mapError(() => invalid("modelSelection must be {instanceId, model, options?}.")), + ); + const given = { ...input, modelSelection: model }; + // Omitted fields keep their value; null clears one. + const pick = < + K extends + | "modelSelection" + | "runtimeMode" + | "standingRole" + | "workspaceStrategy" + | "doneFormat", + >( + key: K, + ) => { + const value = given[key]; + return value === undefined ? existing?.[key] : (value ?? undefined); + }; + const merged = { + name, + modelSelection: pick("modelSelection"), + runtimeMode: pick("runtimeMode"), + standingRole: pick("standingRole"), + workspaceStrategy: pick("workspaceStrategy"), + doneFormat: pick("doneFormat"), + }; + const preset: LaunchPreset = { + name, + ...(merged.modelSelection === undefined ? {} : { modelSelection: merged.modelSelection }), + ...(merged.runtimeMode === undefined ? {} : { runtimeMode: merged.runtimeMode }), + ...(merged.standingRole === undefined ? {} : { standingRole: merged.standingRole }), + ...(merged.workspaceStrategy === undefined + ? {} + : { workspaceStrategy: merged.workspaceStrategy }), + ...(merged.doneFormat === undefined ? {} : { doneFormat: merged.doneFormat }), + updatedAt: DateTime.formatIso(yield* DateTime.now), + updatedBy, + }; + const json = yield* encodePreset(preset).pipe(Effect.mapError(storage)); + yield* sql`INSERT INTO launch_presets (name, preset_json, updated_at, updated_by) + VALUES (${name}, ${json}, ${preset.updatedAt}, ${updatedBy}) + ON CONFLICT (name) DO UPDATE SET preset_json = excluded.preset_json, + updated_at = excluded.updated_at, updated_by = excluded.updated_by`.pipe( + Effect.mapError(storage), + ); + return preset; + }); + return LaunchPresets.of({ get, run }); +}); + +export const layer = Layer.effect(LaunchPresets, make); diff --git a/apps/server/src/persistence/ThreadRoles.ts b/apps/server/src/persistence/ThreadRoles.ts new file mode 100644 index 0000000000000..f9ebf943ee03f --- /dev/null +++ b/apps/server/src/persistence/ThreadRoles.ts @@ -0,0 +1,118 @@ +import { OrchestratorMcpFailure } from "@t3tools/contracts"; +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 SqlClient from "effect/sql/SqlClient"; + +/** Room for a 4000-character role, a 1000-character done format and the line that introduces it. */ +export const THREAD_ROLE_STORED_MAX_CHARS = 5200; + +export interface ThreadRoleRecord { + readonly text: string; + /** Counts each set, so a changed role is told apart from one already delivered. */ + readonly revision: number; + readonly updatedAt: string; + readonly updatedBy: string; + /** Where the provider last accepted this role, if it did. */ + readonly delivered?: { + readonly revision: number; + readonly nativeThreadId: string | null; + readonly runOrdinal: number; + }; +} + +export class ThreadRoles extends Context.Service< + ThreadRoles, + { + /** Read at each provider turn start, so a role set after the session began still applies. */ + readonly get: ( + threadId: string, + ) => Effect.Effect; + /** An empty or null `role` clears it. */ + readonly set: ( + threadId: string, + role: string | null, + updatedBy: string, + ) => Effect.Effect; + /** Recorded after the provider accepts a turn that carried revision `revision`. */ + readonly markDelivered: ( + threadId: string, + delivery: { revision: number; nativeThreadId: string | null; runOrdinal: number }, + ) => Effect.Effect; + } +>()("t3/persistence/ThreadRoles") {} + +const storage = () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "Could not access the thread's standing role.", + }); + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + return ThreadRoles.of({ + get: Effect.fn("ThreadRoles.get")(function* (threadId) { + const [row] = yield* sql<{ + role_text: string; + revision: number; + updated_at: string; + updated_by: string; + delivered_revision: number | null; + delivered_native_thread_id: string | null; + delivered_run_ordinal: number | null; + }>`SELECT role_text, revision, updated_at, updated_by, delivered_revision, + delivered_native_thread_id, delivered_run_ordinal + FROM thread_standing_roles WHERE thread_id = ${threadId}`.pipe(Effect.mapError(storage)); + if (row === undefined) return undefined; + return { + text: row.role_text, + revision: row.revision, + updatedAt: row.updated_at, + updatedBy: row.updated_by, + ...(row.delivered_revision === null || row.delivered_run_ordinal === null + ? {} + : { + delivered: { + revision: row.delivered_revision, + nativeThreadId: row.delivered_native_thread_id, + runOrdinal: row.delivered_run_ordinal, + }, + }), + }; + }), + set: Effect.fn("ThreadRoles.set")(function* (threadId, role, updatedBy) { + const text = role?.trim() ?? ""; + if (text === "") { + yield* sql`DELETE FROM thread_standing_roles WHERE thread_id = ${threadId}`.pipe( + Effect.mapError(storage), + ); + return; + } + if (text.length > THREAD_ROLE_STORED_MAX_CHARS) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: `A standing role is at most ${THREAD_ROLE_STORED_MAX_CHARS} characters with its done format.`, + }); + const now = DateTime.formatIso(yield* DateTime.now); + yield* sql`INSERT INTO thread_standing_roles (thread_id, role_text, revision, updated_at, updated_by) + VALUES (${threadId}, ${text}, 1, ${now}, ${updatedBy}) + ON CONFLICT (thread_id) DO UPDATE SET role_text = excluded.role_text, + revision = thread_standing_roles.revision + 1, + updated_at = excluded.updated_at, updated_by = excluded.updated_by`.pipe( + Effect.mapError(storage), + ); + }), + markDelivered: Effect.fn("ThreadRoles.markDelivered")(function* (threadId, delivery) { + // A role changed while the turn started keeps its newer revision, so it is sent again. + yield* sql`UPDATE thread_standing_roles SET delivered_revision = ${delivery.revision}, + delivered_native_thread_id = ${delivery.nativeThreadId}, + delivered_run_ordinal = ${delivery.runOrdinal} + WHERE thread_id = ${threadId} AND revision = ${delivery.revision}`.pipe( + Effect.mapError(storage), + ); + }), + }); +}); + +export const layer = Layer.effect(ThreadRoles, make); diff --git a/apps/server/src/provider/AgentContextBudget.test.ts b/apps/server/src/provider/AgentContextBudget.test.ts index 0f7f98e867247..23fac5e4e07d9 100644 --- a/apps/server/src/provider/AgentContextBudget.test.ts +++ b/apps/server/src/provider/AgentContextBudget.test.ts @@ -5,6 +5,8 @@ import * as McpClientThreads from "../persistence/McpClientThreads.ts"; import * as AgentRegistry from "../persistence/AgentRegistry.ts"; import * as McpRequestLog from "../persistence/McpRequestLog.ts"; import * as WorkLedger from "../persistence/WorkLedger.ts"; +import * as ThreadRoles from "../persistence/ThreadRoles.ts"; +import * as LaunchPresets from "../persistence/LaunchPresets.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { expect, it } from "@effect/vitest"; @@ -62,6 +64,8 @@ const toolHandlerServices = Layer.mergeAll( Layer.mock(McpBridgeMetrics.McpBridgeMetrics)({}), Layer.mock(McpClientThreads.McpClientThreads)({}), Layer.mock(WorkLedger.WorkLedger)({}), + Layer.mock(ThreadRoles.ThreadRoles)({}), + Layer.mock(LaunchPresets.LaunchPresets)({}), Layer.mock(McpRequestLog.McpRequestLog)({}), Layer.mock(AgentRegistry.AgentRegistry)({}), Layer.mock(AssetSigner)({}), diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index b373520153d4b..0b247644b163c 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5597, "codexAdditionalContextBytes": 7327, - "mcpToolListJsonBytes": 58061 + "mcpToolListJsonBytes": 59122 } diff --git a/docs/jarvis/decisions/0019-thread-standing-role-and-launch-presets.md b/docs/jarvis/decisions/0019-thread-standing-role-and-launch-presets.md new file mode 100644 index 0000000000000..d949b5843d99b --- /dev/null +++ b/docs/jarvis/decisions/0019-thread-standing-role-and-launch-presets.md @@ -0,0 +1,30 @@ +# Thread standing role and launch presets + +Accepted 2026-11-13. + +## Decision + +- **A thread carries a standing role** of up to 4,000 characters. T3 sends it ahead of the user's text, not only when a session starts: with the first turn, after a change, in a new native session, after a compaction, and every 25 runs in case the provider compacted it away itself. Set it with `t3_thread_configure` (`standingRole`, alone; empty text clears), with `t3_thread_launch` or `delegate_task`, or through a preset. `t3_thread_configure` and the first page of `t3_thread_read` show it, with who set it and when. +- **Changing a role needs full access.** A role is instructions for another agent, and for an agent's own thread a way to drop its own rules. Setting or clearing one through `t3_thread_configure` posts a note in the thread naming the setter, with the sender record of [0013](0013-server-recorded-message-senders.md), and the role keeps its setter as the first 8 characters of a session and its label, or a thread id. A launch or delegation sets the role of the thread it creates, within the caller's own modes. +- **Launch presets** are named, server-side bundles of model, runtime mode, standing role, checkout kind (`root` or `worktree`) and an optional done-message format. `t3_preset` lists, reads, saves and deletes them. `t3_thread_launch` and `delegate_task` take `preset: ""`. Explicit fields override the preset. +- **The done format is stored text.** It is appended to the role as "Your final message must follow this format:". T3 does not check the reply. +- **A preset never raises authority.** Its runtime mode is held to the caller's own mode like an explicit one, and is refused when above it. Saving or deleting one needs full access, because a thread with broader modes later starts with its text, which `t3_thread_send` would refuse a narrower caller. Listing and reading are open. +- **Presets are named by the environment that runs the thread.** A launch or delegation to a linked environment forwards the name, and that environment resolves it; `t3_preset` has no `environmentId`, so a remote preset is made there. A linked build that predates this change ignores `preset` and `standingRole`. A preset's model then does not apply to `delegate_task` there (name `target.model`), and its checkout kind never applies: a remote task runs in that project's checkout. A local `delegate_task` maps a `worktree` preset to `isolation: "worktree"`, branching from the parent's commit, not the preset's `baseRef`. + +## Why + +On 2026-11-13 the Jarvis inbox thread's Codex agent ignored its role file for hours. Providers read `AGENTS.md` only when a session starts, and the file was written after the session began. In the same round the coordinator pasted the same rules into nine worker briefs, and one worker woke snoozed threads because its brief lacked a rule. A role that T3 sends each turn cannot be missed by a running session, and a preset keeps a rule set in one place. + +## Constraints + +- **Inline prefix, not a provider field.** The turn input has no per-turn instruction field for every provider. `appContext` reaches only Codex, and Claude takes a system prompt only at session start. The role is a prefix on the user text, like the restart note and side-chat context in `ProviderTurnStartService`. A 4,000-character role is about 1,000 tokens, and each copy stays in the session history, so it is sent only when due. Delivery is recorded after the provider accepts the turn; the table keeps the revision, native session and run it reached. Native commands (`/compact`, `/review`, `/goal`) go without it and are not counted as delivery. +- **Stored outside thread events.** The role and presets live in fork tables (`thread_standing_roles`, `launch_presets`), read at each turn start. Changing a role takes effect on the next turn, and a deleted thread loses its role. A launch or delegation that fails before its thread exists removes the role it stored; one that fails after keeps it. +- **Stored before the thread starts.** A launch or delegation writes the role first, under the thread id it derives, because the first turn begins when the command commits. +- **Not the handoff's `standingItemId`.** That names the first user message in a provider-switch handoff. It is unrelated and unchanged. +- **Context budget.** `t3_preset` and three optional fields grow the advertised tool list by about 1 KB. The preset tool checks `modelSelection` on save instead of advertising its full schema. + +## Alternatives + +- Write the role into `AGENTS.md` and restart sessions. It still depends on a restart. +- Add the role to thread events and projections. That edits upstream-owned tables and the wire contract for a fork-only need. +- Check the done message against its format. It adds machinery; a reviewer already checks it. diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 3d37729ec3d2f..6245c6a0d2e81 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -77,3 +77,4 @@ export * from "./clientRpcPermissions.ts"; export * from "./threadFreeze.ts"; export * from "./workLedger.ts"; +export * from "./launchPresets.ts"; diff --git a/packages/contracts/src/launchPresets.ts b/packages/contracts/src/launchPresets.ts new file mode 100644 index 0000000000000..e9209791ec4eb --- /dev/null +++ b/packages/contracts/src/launchPresets.ts @@ -0,0 +1,83 @@ +import * as Schema from "effect/Schema"; +import * as Struct from "effect/Struct"; + +import { IsoDateTime, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { ModelSelection } from "./modelSelection.ts"; +import { RuntimeMode } from "./providerPolicy.ts"; + +const STANDING_ROLE_MAX_CHARS = 4000; +const DONE_FORMAT_MAX_CHARS = 1000; + +export const StandingRole = Schema.String.check(Schema.isMaxLength(STANDING_ROLE_MAX_CHARS)); +export const DoneFormat = Schema.String.check(Schema.isMaxLength(DONE_FORMAT_MAX_CHARS)); + +export const LaunchPresetName = TrimmedNonEmptyString.check( + Schema.isMaxLength(64), + Schema.isPattern(/^[a-z0-9][a-z0-9._-]*$/), +); + +/** A preset names a kind of checkout, never a path or branch, which belong to one launch. */ +export const LaunchPresetWorkspaceStrategy = Schema.Union([ + Schema.Struct({ type: Schema.Literal("root") }), + Schema.Struct({ + type: Schema.Literal("worktree"), + baseRef: TrimmedNonEmptyString, + startFromOrigin: Schema.optional(Schema.Boolean), + }), +]); + +export const LaunchPreset = Schema.Struct({ + name: LaunchPresetName, + modelSelection: Schema.optional(ModelSelection), + runtimeMode: Schema.optional(RuntimeMode), + standingRole: Schema.optional(StandingRole), + workspaceStrategy: Schema.optional(LaunchPresetWorkspaceStrategy), + doneFormat: Schema.optional(DoneFormat), + updatedAt: IsoDateTime, + updatedBy: Schema.String, +}); +export type LaunchPreset = typeof LaunchPreset.Type; + +/** Fields the caller may set. `null` clears one on an existing preset. */ +const presetFields = { + // Checked against ModelSelection on save: the full schema is large in the tool list. + modelSelection: Schema.optional( + Schema.Unknown.annotate({ description: "{instanceId, model, options?}; null clears." }), + ), + runtimeMode: Schema.optional(Schema.NullOr(RuntimeMode)), + standingRole: Schema.optional(Schema.NullOr(StandingRole)), + workspaceStrategy: Schema.optional(Schema.NullOr(LaunchPresetWorkspaceStrategy)), + doneFormat: Schema.optional(Schema.NullOr(DoneFormat)), +}; + +export const LaunchPresetToolInput = Schema.Struct({ + action: Schema.Literals(["list", "read", "save", "delete"]), + name: Schema.optional(LaunchPresetName), + ...presetFields, +}); +export type LaunchPresetToolInput = typeof LaunchPresetToolInput.Type; + +export const LaunchPresetToolResult = Schema.Union([ + Schema.Struct({ presets: Schema.Array(Schema.String) }), + // Loose here: the tool list carries this schema, and the stored preset is already decoded. + LaunchPreset.mapFields( + Struct.assign({ + modelSelection: Schema.optional(Schema.Unknown), + workspaceStrategy: Schema.optional(Schema.Unknown), + }), + ), + Schema.Struct({ deleted: Schema.Literal(true), name: LaunchPresetName }), +]); +export type LaunchPresetToolResult = typeof LaunchPresetToolResult.Type; + +/** The standing role a launch stores: the role text, then the required done-message format. */ +export function composeStandingRole(input: { + readonly standingRole?: string | undefined; + readonly doneFormat?: string | undefined; +}): string | undefined { + const role = input.standingRole?.trim() ?? ""; + const done = input.doneFormat?.trim() ?? ""; + const parts = [role, done === "" ? "" : `Your final message must follow this format:\n${done}`]; + const text = parts.filter((part) => part !== "").join("\n\n"); + return text === "" ? undefined : text; +} diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 1c7bcdbc09328..3898867087aba 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -47,6 +47,7 @@ import { } from "./orchestrationV2.ts"; import { ProviderOptionSelection, ProviderOptionSelectionValue } from "./model.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; +import { LaunchPresetName, StandingRole } from "./launchPresets.ts"; const OrchestratorMcpPrompt = TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)); const OrchestratorMcpTitle = TrimmedNonEmptyString.check(Schema.isMaxLength(512)); @@ -198,6 +199,8 @@ export const OrchestratorMcpDelegateTaskInput = Schema.Struct({ }), ), interactionMode: Schema.optional(OrchestratorMcpInteractionMode), + preset: Schema.optional(LaunchPresetName), + standingRole: Schema.optional(StandingRole), }); export type OrchestratorMcpDelegateTaskInput = typeof OrchestratorMcpDelegateTaskInput.Type; @@ -462,6 +465,9 @@ export const OrchestratorMcpThreadDetail = Schema.Struct({ pullRequests: Schema.optional(Schema.Array(OrchestratorMcpThreadPullRequest)), pullRequestCount: Schema.optional(NonNegativeInt), titleRegeneration: Schema.NullOr(ThreadTitleRegeneration), + /** The thread's standing role, on the first page only. Optional: older peers omit it. */ + standingRole: Schema.optional(Schema.String), + standingRoleSetBy: Schema.optional(Schema.String), branch: Schema.NullOr(Schema.String), worktreePath: Schema.NullOr(Schema.String), parentThreadId: Schema.NullOr(ThreadId),