From 6cb51fa8070ff0d58f8a26552e7ce1daf1a7598d Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 20:07:06 -0700 Subject: [PATCH 1/2] feat(server): acknowledge-only threads answer messages without a provider A thread listed in the new acknowledgeOnlyThreadIds server setting answers each message with a server-written "Queued for Jarvis: " reply. The run starts, runs and completes with the same run and assistant-message events a provider turn emits, so inbox subscriptions and the notify relay see an ordinary run result, but no provider session opens and no model is called. Full-access MCP callers turn the mode on and off with t3_thread_configure acknowledgeOnly; reading the thread's configuration shows it. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/toolkits/core.test.ts | 44 ++- .../src/mcp/toolkits/thread/handlers.ts | 122 +++++--- apps/server/src/mcp/toolkits/thread/tools.ts | 5 +- .../AcknowledgeOnly.integration.test.ts | 273 ++++++++++++++++++ .../src/orchestration-v2/AcknowledgeOnly.ts | 171 +++++++++++ .../ProviderTurnStartService.ts | 20 ++ .../ThreadManagementService.ts | 18 ++ .../testkit/ProviderReplayHarness.ts | 30 +- apps/server/src/peer/PeerForwarding.test.ts | 2 + .../src/provider/agentContextBudget.json | 2 +- .../src/relay/AgentAwarenessRelay.test.ts | 2 + docs/jarvis/GLOSSARY.md | 3 + docs/jarvis/decisions/0009-jarvis-inbox.md | 17 +- docs/user/outside-agents.md | 6 +- packages/contracts/src/settings.test.ts | 2 + packages/contracts/src/settings.ts | 14 + 16 files changed, 669 insertions(+), 62 deletions(-) create mode 100644 apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts create mode 100644 apps/server/src/orchestration-v2/AcknowledgeOnly.ts diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 0d2a231e66d14..7b16c22858402 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -1189,7 +1189,10 @@ it.effect("a read-only client reads threads and is refused every write before it threadId: "other-project-thread", }); expect(configuration.isError).toBe(false); - expect(configuration.structuredContent).toMatchObject({ runtimeMode: "auto" }); + expect(configuration.structuredContent).toMatchObject({ + runtimeMode: "auto", + acknowledgeOnly: false, + }); const pinned = yield* call("t3_thread_organize", { action: "pin", @@ -1228,6 +1231,39 @@ it.effect("a read-only client reads threads and is refused every write before it expect(dispatched).toEqual([ 'thread.runtime-mode.set:approval-required:MCP client "Claude Code" (session-)', ]); + + // Who answers a thread's messages is environment policy: full access only, both ways. + const acknowledgeAs = ( + access: "read-only" | "auto" | "full-access", + acknowledgeOnly: boolean, + ) => + server + .callTool({ + name: "t3_thread_configure", + arguments: { threadId: "other-project-thread", acknowledgeOnly }, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, clientScope(access)), + Effect.provideService(McpSchema.McpServerClient, client), + ); + for (const access of ["read-only", "auto"] as const) { + expect(declaredFailure(yield* acknowledgeAs(access, true))).toMatchObject({ + code: "capability_denied", + }); + } + expect((yield* acknowledgeAs("full-access", true)).structuredContent).toMatchObject({ + acknowledgeOnly: true, + }); + expect( + (yield* call("t3_thread_configure", { threadId: "other-project-thread" })).structuredContent, + ).toMatchObject({ acknowledgeOnly: true }); + expect(declaredFailure(yield* acknowledgeAs("auto", false))).toMatchObject({ + code: "capability_denied", + }); + expect((yield* acknowledgeAs("full-access", false)).structuredContent).toMatchObject({ + acknowledgeOnly: false, + }); + expect(dispatched).toHaveLength(1); }).pipe( Effect.provide( layerThreadToolkit.pipe( @@ -1252,6 +1288,11 @@ it.effect("a read-only client reads threads and is refused every write before it deletedAt: null, }, } as never), + acknowledgeOnly: () => Effect.succeed(acknowledgeOnly), + setAcknowledgeOnly: (_, enabled) => + Effect.sync(() => { + acknowledgeOnly = enabled; + }), dispatch: (command) => Effect.gen(function* () { const actor = yield* RuntimeModeChangeActor; @@ -1267,6 +1308,7 @@ it.effect("a read-only client reads threads and is refused every write before it ), ); const dispatched: Array = []; +let acknowledgeOnly = false; it.effect("refuses act-as-caller tools to a client caller", () => Effect.gen(function* () { diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index be06287c5aa3b..44ac90f68a6a7 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -106,6 +106,21 @@ const required = (value: A | undefined, name: string) => }), ) : Effect.succeed(value); +const readThreadConfiguration = Effect.fn("mcp.readThreadConfiguration")(function* ( + threadId: ThreadId | undefined, +) { + const { + threads, + projection: { thread }, + } = yield* readThread(threadId); + return { + threadId: thread.id, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + acknowledgeOnly: yield* threads.acknowledgeOnly(thread.id).pipe(Effect.mapError(unavailable)), + }; +}); const writesThread =

( handle: (params: P, authorize: McpToolAccess.AuthorizeThread) => Effect.Effect, ) => McpToolAccess.writesThreads((params: P) => [params.threadId], handle); @@ -224,60 +239,79 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { ), ), t3_thread_configure: McpToolAccess.either( - (input) => input.modelSelection === undefined && input.runtimeMode === undefined, - McpToolAccess.reads((input) => - Effect.gen(function* () { - const { - projection: { thread }, - } = yield* readThread(input.threadId); - return { - threadId: thread.id, - modelSelection: thread.modelSelection, - runtimeMode: thread.runtimeMode, - interactionMode: thread.interactionMode, - }; - }), - ), - writesThread((input) => - Effect.gen(function* () { - const { - threads, - projection: { thread }, - } = yield* readThread(input.threadId); - if (input.runtimeMode !== undefined) { - if (input.modelSelection !== undefined) { + (input) => + input.modelSelection === undefined && + input.runtimeMode === undefined && + input.acknowledgeOnly === undefined, + McpToolAccess.reads((input) => readThreadConfiguration(input.threadId)), + McpToolAccess.either( + (input) => input.acknowledgeOnly !== undefined, + // Who answers the thread's messages is environment policy, so it needs full access. + McpToolAccess.writesEnvironment((input) => + Effect.gen(function* () { + const acknowledgeOnly = yield* required(input.acknowledgeOnly, "acknowledgeOnly"); + if (input.modelSelection !== undefined || input.runtimeMode !== undefined) { return yield* new OrchestratorMcpFailure({ code: "invalid_request", - message: "Set runtimeMode separately from modelSelection.", + message: "Set acknowledgeOnly separately from modelSelection and runtimeMode.", }); } - const scope = yield* McpInvocationContext.McpInvocationContext; - const actor = - scope.thread !== undefined - ? `MCP thread ${scope.thread.threadId}` - : scope.client !== undefined - ? `MCP client "${scope.client.label}" (${scope.client.sessionId.slice(0, 8)})` - : "MCP caller"; + const { + threads, + projection: { thread }, + } = yield* readThread(input.threadId); + yield* threads + .setAcknowledgeOnly(thread.id, acknowledgeOnly) + .pipe(Effect.mapError(unavailable)); + return yield* readThreadConfiguration(thread.id); + }), + ), + writesThread((input) => + Effect.gen(function* () { + const { + threads, + projection: { thread }, + } = yield* readThread(input.threadId); + if (input.runtimeMode !== undefined) { + if (input.modelSelection !== undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "Set runtimeMode separately from modelSelection.", + }); + } + const scope = yield* McpInvocationContext.McpInvocationContext; + const actor = + scope.thread !== undefined + ? `MCP thread ${scope.thread.threadId}` + : scope.client !== undefined + ? `MCP client "${scope.client.label}" (${scope.client.sessionId.slice(0, 8)})` + : "MCP caller"; + const result = yield* threads + .dispatch({ + type: "thread.runtime-mode.set", + threadId: thread.id, + commandId: yield* newCommandId(), + runtimeMode: input.runtimeMode, + }) + .pipe( + Effect.provideService(RuntimeModeChangeActor, actor), + Effect.mapError(dispatchFailure), + ); + return { sequence: result.sequence }; + } + const modelSelection = yield* required(input.modelSelection, "modelSelection"); + const type = modelSelectionCommandType(thread.providerInstanceId, modelSelection); const result = yield* threads .dispatch({ - type: "thread.runtime-mode.set", + type, threadId: thread.id, commandId: yield* newCommandId(), - runtimeMode: input.runtimeMode, + modelSelection, }) - .pipe( - Effect.provideService(RuntimeModeChangeActor, actor), - Effect.mapError(dispatchFailure), - ); + .pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence }; - } - const modelSelection = yield* required(input.modelSelection, "modelSelection"); - const type = modelSelectionCommandType(thread.providerInstanceId, modelSelection); - const result = yield* threads - .dispatch({ type, threadId: thread.id, commandId: yield* newCommandId(), modelSelection }) - .pipe(Effect.mapError(dispatchFailure)); - return { sequence: result.sequence }; - }), + }), + ), ), ), t3_pending_request: McpToolAccess.either( diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index a8b11064d21a5..ba79df8fe1e0d 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -206,16 +206,19 @@ const threadConfiguration = Schema.Struct({ modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, + // Optional: a linked environment on another build does not report it. + acknowledgeOnly: Schema.optional(Schema.Boolean), }); 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`.", + "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.", parameters: Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), modelSelection: Schema.optional(ModelSelection), runtimeMode: Schema.optional(RuntimeMode), + acknowledgeOnly: Schema.optional(Schema.Boolean), }), success: Schema.Union([threadConfiguration, OrchestrationV2DispatchCommandResult]), }).annotate(Tool.Destructive, true); diff --git a/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts b/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts new file mode 100644 index 0000000000000..acc75ee23095e --- /dev/null +++ b/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts @@ -0,0 +1,273 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, expect, it } from "@effect/vitest"; +import { + CommandId, + MessageId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + ThreadId, +} from "@t3tools/contracts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import * as McpEvents from "../mcp/McpEvents.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as AcknowledgeOnly from "./AcknowledgeOnly.ts"; +import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as EventSink from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; + +const driver = ProviderDriverKind.make("claudeAgent"); +const instanceId = ProviderInstanceId.make("claudeAgent"); +const modelSelection = { instanceId, model: "claude-test" }; +const threadId = ThreadId.make("thread:acknowledge-only-inbox"); +const otherInbox = ThreadId.make("thread:another-inbox"); + +const messages = [ + "First request for Jarvis", + "Second request", + `Third request\nwith a second line and ${"padding ".repeat(20)}to pass the excerpt limit`, +]; + +// The Jarvis inbox: each message ends in a run whose result wakes Jarvis, without a model call. +it.effect("an acknowledge-only thread answers queued messages in order without a provider", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("acknowledge-only-inbox"); + let sessionsOpened = 0; + const turns: Array = []; + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + sessionsOpened += 1; + const now = yield* DateTime.now; + const events = yield* Queue.unbounded(); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd, + model: modelSelection.model, + capabilities: ClaudeProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: ({ threadId: appThreadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:${appThreadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: input.providerSessionId, + appThreadId, + ownerNodeId: null, + nativeThreadRef: { + driver, + nativeId: `native:${appThreadId}`, + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + turns.push(turn); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: null, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: () => Effect.die("unused"), + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), + }; + const settingsLayer = ServerSettings.layerTest({ + acknowledgeOnlyThreadIds: [otherInbox], + }).pipe(Layer.orDie); + const layerTest = Layer.merge( + ProviderReplayHarness.layerWithRegistry( + { name: "acknowledge-only-inbox" }, + ProviderAdapterRegistry.layerSingle(adapter), + { settingsLayer, runEffectWorker: false }, + ), + settingsLayer, + ).pipe(Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))); + + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const settings = yield* ServerSettings.ServerSettingsService; + const send = (index: number, text: string) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`send-${index}`), + threadId, + messageId: MessageId.make(`message:inbox-${index}`), + text, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "mcp", + }); + + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-inbox"), + threadId, + projectId: ProjectId.make("project:acknowledge-only"), + title: "Jarvis inbox", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + yield* AcknowledgeOnly.setAcknowledgeOnly(threadId, true); + assert.deepEqual((yield* settings.getSettings).acknowledgeOnlyThreadIds, [ + otherInbox, + threadId, + ]); + const settled = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence(), + eventType: "run.updated", + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && stored.event.payload.status === "completed", + ), + Stream.take(messages.length), + Stream.runCollect, + Effect.forkScoped, + ); + yield* Effect.forEach(messages, (text, index) => send(index, text), { discard: true }); + // Started after the sends, so the later messages wait in the queue. + yield* EffectWorker.runDaemon.pipe(Effect.forkScoped); + const results = Array.from(yield* Fiber.join(settled)); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal(sessionsOpened, 0); + assert.deepEqual(turns, []); + assert.deepEqual(projection.providerSessions, []); + assert.deepEqual(projection.providerTurns, []); + const runs = projection.runs.toSorted((left, right) => left.ordinal - right.ordinal); + const replies = runs.map((run) => ({ + request: projection.messages.find((message) => message.id === run.userMessageId)?.text, + // What the inbox relay quotes: the run's last assistant message. + reply: projection.messages + .filter((message) => message.runId === run.id && message.role === "assistant") + .at(-1)?.text, + })); + assert.deepEqual(replies, [ + { request: messages[0], reply: "Queued for Jarvis: First request for Jarvis" }, + { request: messages[1], reply: "Queued for Jarvis: Second request" }, + { + request: messages[2], + reply: `Queued for Jarvis: ${Array.from(messages[2]!.replace(/\s+/g, " ")) + .slice(0, 120) + .join("")}…`, + }, + ]); + // Completions arrive in send order, each the event that wakes an inbox subscription. + expect( + results.map((stored) => McpEvents.observeEvent(stored, projection.thread, "t3://inbox")), + ).toEqual( + runs.map((run) => + expect.objectContaining({ + name: "thread.run.result", + data: expect.objectContaining({ runId: run.id, state: "completed" }), + }), + ), + ); + const lifecycle = yield* eventSink.stream({ threadId, afterSequence: 0 }).pipe( + Stream.takeUntil((stored) => stored.sequence >= results.at(-1)!.sequence), + Stream.runCollect, + ); + expect( + runs.map((run) => + Array.from(lifecycle).flatMap(({ event }) => + (event.type === "run.created" || event.type === "run.updated") && + event.payload.id === run.id + ? [`${event.type}:${event.payload.status}`] + : [], + ), + ), + ).toEqual([ + ["run.created:starting", "run.updated:running", "run.updated:completed"], + ...[1, 2].map(() => [ + "run.created:queued", + "run.updated:starting", + "run.updated:running", + "run.updated:completed", + ]), + ]); + + // Off again, the next message is an ordinary provider turn. + yield* AcknowledgeOnly.setAcknowledgeOnly(threadId, false); + assert.deepEqual((yield* settings.getSettings).acknowledgeOnlyThreadIds, [otherInbox]); + const turnStarted = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence(), + eventType: "provider-turn.updated", + }) + .pipe(Stream.take(1), Stream.runDrain, Effect.forkScoped); + yield* send(3, "Answer this one yourself"); + yield* Fiber.join(turnStarted); + assert.equal(sessionsOpened, 1); + assert.equal(turns.length, 1); + assert.include(JSON.stringify(turns[0]), "Answer this one yourself"); + }).pipe(Effect.provide(layerTest)); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/AcknowledgeOnly.ts b/apps/server/src/orchestration-v2/AcknowledgeOnly.ts new file mode 100644 index 0000000000000..55151a7bd16e5 --- /dev/null +++ b/apps/server/src/orchestration-v2/AcknowledgeOnly.ts @@ -0,0 +1,171 @@ +import type { + OrchestrationV2ConversationMessage, + OrchestrationV2DomainEvent, + OrchestrationV2ExecutionNode, + OrchestrationV2ProviderThread, + OrchestrationV2Run, + OrchestrationV2RunAttempt, + OrchestrationV2TurnItem, + ServerSettings, + ThreadId, +} from "@t3tools/contracts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; +import * as Settings from "../serverSettings.ts"; +import type * as EventSink from "./EventSink.ts"; + +const EXCERPT_LENGTH = 120; + +export const isAcknowledgeOnlyThread = ( + settings: Pick, + threadId: ThreadId, +): boolean => settings.acknowledgeOnlyThreadIds.includes(threadId); + +export const setAcknowledgeOnly = Effect.fn("AcknowledgeOnly.set")(function* ( + threadId: ThreadId, + enabled: boolean, +) { + const settings = yield* Settings.ServerSettingsService; + const current = (yield* settings.getSettings).acknowledgeOnlyThreadIds; + if (current.includes(threadId) === enabled) return; + yield* settings.updateSettings({ + acknowledgeOnlyThreadIds: enabled + ? [...current, threadId] + : current.filter((id) => id !== threadId), + }); +}); + +const acknowledgementText = (message: string): string => { + const line = Array.from(message.replace(/\s+/g, " ").trim()); + const excerpt = line.slice(0, EXCERPT_LENGTH).join(""); + return `Queued for Jarvis: ${line.length > EXCERPT_LENGTH ? `${excerpt}…` : excerpt}`; +}; + +/** + * Completes a starting run with a written acknowledgement and no provider turn. It emits the + * run lifecycle a provider turn would (running, then an assistant message, then completed), so + * event subscriptions and relays see an ordinary run. No checkpoint is captured. + */ +export const settleAcknowledgeOnlyRun = Effect.fn("AcknowledgeOnly.settleRun")(function* (input: { + readonly eventSink: EventSink.EventSinkV2["Service"]; + readonly idAllocator: IdAllocator.IdAllocatorV2["Service"]; + readonly run: OrchestrationV2Run; + readonly attempt: OrchestrationV2RunAttempt; + readonly rootNode: OrchestrationV2ExecutionNode; + readonly providerThread: OrchestrationV2ProviderThread; + readonly message: Pick; + readonly nextTurnItemOrdinal: number; +}) { + const { run, attempt, rootNode, providerThread } = input; + const now = yield* DateTime.now; + const nativeItemId = `acknowledgement:${run.id}`; + const itemIds = { driver: providerThread.driver, nativeItemId }; + const nodeId = input.idAllocator.derive.nodeFromProviderItem(itemIds); + const messageId = input.idAllocator.derive.messageFromProviderItem(itemIds); + const text = acknowledgementText(input.message.text); + const assistantNode: OrchestrationV2ExecutionNode = { + id: nodeId, + threadId: run.threadId, + runId: run.id, + parentNodeId: rootNode.id, + rootNodeId: rootNode.id, + kind: "assistant_message", + status: "completed", + countsForRun: false, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: now, + }; + const message: OrchestrationV2ConversationMessage = { + createdBy: "agent", + creationSource: "provider", + id: messageId, + threadId: run.threadId, + runId: run.id, + nodeId, + role: "assistant", + text, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }; + const turnItem: OrchestrationV2TurnItem = { + id: input.idAllocator.derive.turnItemFromProviderItem(itemIds), + threadId: run.threadId, + runId: run.id, + nodeId, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: input.nextTurnItemOrdinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "assistant_message", + messageId, + text, + streaming: false, + }; + const running = { status: "running" as const, startedAt: now }; + const completed = { status: "completed" as const, startedAt: now, completedAt: now }; + const payloads = [ + { type: "run.updated", nodeId: rootNode.id, payload: { ...run, ...running } }, + { type: "run-attempt.updated", nodeId: rootNode.id, payload: { ...attempt, ...running } }, + { type: "node.updated", nodeId: rootNode.id, payload: { ...rootNode, ...running } }, + { type: "node.updated", nodeId, payload: assistantNode }, + { type: "message.updated", nodeId, payload: message }, + { type: "turn-item.updated", nodeId, payload: turnItem }, + { type: "run-attempt.updated", nodeId: rootNode.id, payload: { ...attempt, ...completed } }, + { type: "run.updated", nodeId: rootNode.id, payload: { ...run, ...completed } }, + { type: "node.updated", nodeId: rootNode.id, payload: { ...rootNode, ...completed } }, + { + type: "provider-thread.updated", + payload: { + ...providerThread, + status: providerThread.nativeThreadRef === null ? "not_loaded" : "idle", + updatedAt: now, + }, + }, + ] as const; + const events = yield* Effect.forEach(payloads, (event) => + input.idAllocator.allocate.event({ threadId: run.threadId }).pipe( + Effect.map( + (id) => + ({ + ...event, + id, + threadId: run.threadId, + runId: run.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + }) satisfies OrchestrationV2DomainEvent, + ), + ), + ); + const written = yield* input.eventSink.writeIfRunCurrent({ + threadId: run.threadId, + runId: run.id, + activeAttemptId: attempt.id, + expectedStatus: "starting", + events, + }); + if (written.committed) { + yield* (yield* ProviderUsageAnalytics.ProviderTurnAnalytics).runSettled({ + runId: run.id, + status: "completed", + completedAt: now, + childrenRunning: false, + }); + } +}); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 8cff9d01687c6..241011f1cc77b 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -34,6 +34,7 @@ import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; +import * as AcknowledgeOnly from "./AcknowledgeOnly.ts"; import * as ProviderFailover from "./ProviderFailover.ts"; import * as ServerSettings from "../serverSettings.ts"; import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts"; @@ -454,6 +455,25 @@ export const layer: Layer.Layer< cause: `Run ${runId} is missing its execution projection state.`, }); } + const settings = yield* serverSettings.getSettings.pipe( + Effect.mapError((cause) => new ProviderTurnStartError({ runId, cause })), + ); + if (AcknowledgeOnly.isAcknowledgeOnlyThread(settings, projection.thread.id)) { + return yield* AcknowledgeOnly.settleAcknowledgeOnlyRun({ + eventSink, + idAllocator, + run, + attempt, + rootNode, + providerThread, + message, + nextTurnItemOrdinal: + Math.max( + 0, + ...projection.turnItems.filter((item) => item.runId === runId).map((i) => i.ordinal), + ) + 1, + }); + } const settleRunBeforeStart = Effect.fn("orchestrationV2.providerTurnStart.settleBeforeStart")( function* (input: { readonly signal: string; diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d8118f13454f3..520c013a30f0c 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1,6 +1,7 @@ import type { OrchestrationV2SearchThreadInput, OrchestrationV2SearchThreadResult, + ServerSettingsError, } from "@t3tools/contracts"; import type { ProjectionRecordField, @@ -58,6 +59,7 @@ import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; +import * as AcknowledgeOnly from "./AcknowledgeOnly.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -377,6 +379,12 @@ export type ThreadManagementError = typeof ThreadManagementError.Type; type ThreadManagementFailure = ThreadManagementError | Orchestrator.OrchestratorV2Error; export interface ThreadManagementServiceShape { + /** Fork: whether the server, not the agent, answers the thread's messages. */ + readonly acknowledgeOnly: (threadId: ThreadId) => Effect.Effect; + readonly setAcknowledgeOnly: ( + threadId: ThreadId, + enabled: boolean, + ) => Effect.Effect; readonly searchThreadStream: ( input: OrchestrationV2SearchThreadInput, ) => Stream.Stream; @@ -1625,6 +1633,16 @@ const make = Effect.gen(function* () { }); return ThreadManagementService.of({ + acknowledgeOnly: Effect.fn("ThreadManagementService.acknowledgeOnly")(function* (threadId) { + return AcknowledgeOnly.isAcknowledgeOnlyThread(yield* serverSettings.getSettings, threadId); + }), + setAcknowledgeOnly: Effect.fn("ThreadManagementService.setAcknowledgeOnly")( + function* (threadId, enabled) { + yield* AcknowledgeOnly.setAcknowledgeOnly(threadId, enabled).pipe( + Effect.provideService(ServerSettings.ServerSettingsService, serverSettings), + ); + }, + ), ensureLegacyTranscript, searchThreadStream: (input) => Stream.unwrap( diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 8bbf31a067a23..665a04f754749 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -289,6 +289,8 @@ export function layerWithRegistry( ) => ProjectionStore.ProjectionStoreV2Shape; readonly managedPolicy?: ManagedPolicyDocument; readonly settings?: Parameters[0]; + /** Replaces `settings` with a layer the test also provides, so it can change settings live. */ + readonly settingsLayer?: Layer.Layer; } = {}, ): Layer.Layer< | Orchestrator.OrchestratorV2 @@ -316,20 +318,22 @@ export function layerWithRegistry( const layerProvidedRegistry = registryLayer.pipe( Layer.provide(Layer.merge(layerContinuationRequests, McpProviderSessions.layer)), ); - const layerServerSettings = Layer.unwrap( - Effect.map(ManagedPolicy.fromDocument("/test/policy.json", options.managedPolicy), (policy) => - ServerSettings.layerTest( - { - responseStreamingMode: "turn", - ...options.settings, - ...(options.continueThreadsAfterServerUpdate === undefined - ? {} - : { continueThreadsAfterServerUpdate: options.continueThreadsAfterServerUpdate }), - }, - policy.summary, + const layerServerSettings = + options.settingsLayer ?? + Layer.unwrap( + Effect.map(ManagedPolicy.fromDocument("/test/policy.json", options.managedPolicy), (policy) => + ServerSettings.layerTest( + { + responseStreamingMode: "turn", + ...options.settings, + ...(options.continueThreadsAfterServerUpdate === undefined + ? {} + : { continueThreadsAfterServerUpdate: options.continueThreadsAfterServerUpdate }), + }, + policy.summary, + ), ), - ), - ).pipe(Layer.orDie); + ).pipe(Layer.orDie); const layerEffectOutboxProvided = EffectOutbox.layer.pipe(Layer.provide(layerDatabase)); const wrapProjectionStore = options.wrapProjectionStore; const layerProjectionStore = diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index e9ab121dbbf90..3acea412f9bdd 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -148,6 +148,7 @@ const boxThreads = (seen: Ref.Ref, linkSession: Ref.Ref) => { ), ); return Layer.mock(ThreadManagement.ThreadManagementService)({ + acknowledgeOnly: () => Effect.succeed(false), getThreadShell: shellOf, getProjectThreadRecords: (input) => projectionOf(input.threadId), getThreadRecords: (threadId) => projectionOf(threadId), @@ -371,6 +372,7 @@ const makeLaptop = Effect.gen(function* () { Layer.provide(NodeCrypto.layer), Layer.provideMerge( Layer.mock(ThreadManagement.ThreadManagementService)({ + acknowledgeOnly: () => Effect.succeed(false), getThreadShell: (threadId) => Effect.succeed( threadId === ThreadId.make("thread:laptop-from-desk") diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index a31ea110412d0..b373520153d4b 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5597, "codexAdditionalContextBytes": 7327, - "mcpToolListJsonBytes": 57928 + "mcpToolListJsonBytes": 58061 } diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 6bd994d72a8eb..b8dcc0addf3d0 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -208,6 +208,8 @@ const makeTestRelay = Effect.fnUntraced(function* ( const shellReads: ThreadId[] = []; const catchUp = { activitySnapshotReads: 0 }; const threads = ThreadManagementService.ThreadManagementService.of({ + acknowledgeOnly: unused, + setAcknowledgeOnly: unused, getThreadShell: unused, getShellSnapshot: unused, getAgentAwarenessSnapshot: (threadId) => diff --git a/docs/jarvis/GLOSSARY.md b/docs/jarvis/GLOSSARY.md index a6a9f54da6575..129aa17e6ee19 100644 --- a/docs/jarvis/GLOSSARY.md +++ b/docs/jarvis/GLOSSARY.md @@ -25,6 +25,9 @@ One outcome tracked in the work ledger under a stable key. **inbox**: The T3 thread where agents leave requests for Jarvis. +**acknowledge-only mode**: +A thread setting under which T3 answers each message with a fixed acknowledgement instead of running an agent. The run still ends normally, so its result still wakes Jarvis. + **event subscription**: An arrangement to receive notice when a selected kind of change happens in T3. diff --git a/docs/jarvis/decisions/0009-jarvis-inbox.md b/docs/jarvis/decisions/0009-jarvis-inbox.md index 09a4a8493738c..4a98b1d70663b 100644 --- a/docs/jarvis/decisions/0009-jarvis-inbox.md +++ b/docs/jarvis/decisions/0009-jarvis-inbox.md @@ -1,15 +1,30 @@ # Reach Jarvis through a T3 inbox and an event -Accepted 2026-11-12. +Accepted 2026-11-12. Acknowledge-only mode added 2026-11-13. ## Decision T3 agents send requests to Jarvis's T3 inbox thread. An event wakes Jarvis, which reads the request and answers in the requesting thread. +The inbox acknowledges each message with a run of its own; that run's result event is the wake signal (0012, 0014). Acknowledge-only mode lets the server write that acknowledgement instead of a model. The coordinator turns it on for the inbox after deploy, once it has verified that Jarvis still wakes. + +## Acknowledge-only mode + +- **What it does.** Each message to an acknowledge-only thread starts a run as usual. The server, not a provider, completes it with the assistant reply `Queued for Jarvis: `. No provider session opens, no tools run and no tokens are spent. Messages are answered one at a time in queue order. +- **Events.** The run emits the lifecycle a provider turn does: `run.created` (queued or starting), `run.updated` starting and running, the assistant message, then `run.updated` completed. Subscriptions see the same `thread.run.result` event with the same source run id, and the relay quotes the same assistant message as its excerpt. It deliberately has no provider turn, no `waiting` step and no end-of-turn checkpoint. +- **What gets acknowledged.** Every run on the thread, whatever started it. Relay messages are meant for Jarvis, so acknowledging them is right. A restart continuation follows only an interrupted provider turn, which acknowledged runs never have. Idle compaction targets only Claude sessions and never compacts a compaction, so a Claude thread in this mode acknowledges at most one `/compact` per idle spell. +- **Who can change it.** `t3_thread_configure` with `acknowledgeOnly` alone, both to turn it on and to turn it off, needs a full-access caller: a live calling thread in full-access mode, or an MCP client signed in with full access, and not a linked environment's caller. Anyone who can edit server settings can also change `acknowledgeOnlyThreadIds` in `settings.json`. Reading the thread's configuration shows the mode. +- **Turning it off.** The next message runs the thread's agent again. The agent's native conversation never saw the acknowledged messages. +- **Known limits.** The thread still needs a configured, enabled provider: starting a queued run resolves the provider adapter, and a disabled instance holds the queue. A queue held by an earlier failed or usage-limited run stays held after the mode is turned on, until someone resumes the queue in the app. A run that finishes in this mode never holds the queue. Turning the mode on does not detach a provider session already attached; it stays attached but receives no turns. + ## Alternatives Type into Jarvis's ChatGPT chat through browser automation, or introduce a separate direct messaging path. +For the acknowledgement: keep a model as acknowledger with acknowledge-only instructions, or coalesce queued messages into one turn. On 2026-11-12 a long-lived acknowledger session that predated its instructions acted on messages meant for Jarvis: it sent 19 messages as Jarvis, wrote the ledger 8 times and used a one-run approval. In the 48 hours before, the 169 acknowledgement runs had a p90 time to reply of 285 s, with bursts of 57 messages an hour. Coalescing would change the one-run-per-message wake that 0014 relies on. + ## Why The inbox keeps requests and replies in T3's durable conversation history and uses the event channel already proven with Jarvis. An inbox acknowledgment means the request was queued, not that Jarvis answered. + +A written acknowledgement cannot act on a message, costs nothing and finishes in milliseconds, while the run and its events stay the same. diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index 767ef72a7e779..f59eba7ee6236 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -255,7 +255,11 @@ with `event_inbox_set: { threadId: "" }`. Then include `notify: {}` i `t3_thread_launch` or `t3_thread_send`. T3 queues a message to that inbox when the source run produces a matching event. This creates no event subscription or ChatGPT watch. Your existing inbox watch receives the inbox run's normal event. -Keep the inbox's run-per-message acknowledger enabled. +Keep the inbox's run-per-message acknowledger enabled. To acknowledge without a +model, a full-access caller can call `t3_thread_configure` with +`acknowledgeOnly: true` on the inbox thread. The server then answers each message +with `Queued for Jarvis: ` and ends its run, so the inbox +watch still wakes. `acknowledgeOnly: false` returns the thread to its agent. The response reports `bound` with the inbox thread id, or `unavailable` or `failed` with a reason. Without a pin it reports `no_pinned_inbox`; the launch or send still diff --git a/packages/contracts/src/settings.test.ts b/packages/contracts/src/settings.test.ts index 4c194f473c7b0..d96c166f4afdf 100644 --- a/packages/contracts/src/settings.test.ts +++ b/packages/contracts/src/settings.test.ts @@ -144,9 +144,11 @@ it("drops a hand-edited non-string hidden tag and keeps the rest of the settings const settings = decodeServerSettings({ hiddenThreadTags: ["private", 123], hideTaggedThreads: true, + acknowledgeOnlyThreadIds: ["thread:inbox", 7, ""], }); expect(settings.hiddenThreadTags).toEqual(["private"]); expect(settings.hideTaggedThreads).toBe(true); + expect(settings.acknowledgeOnlyThreadIds).toEqual(["thread:inbox"]); }); describe("ServerSettings secret providers", () => { diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 2844c8181324a..67a88a11acd83 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1574,6 +1574,18 @@ export const ServerSettings = Schema.Struct({ /** Sent by the Usage tab with each summary request to this environment; edit in settings.json. Invalid values are ignored. */ usageModelFilter: ForwardCompatibleOptional(UsageModelFilter), + + /** Fork: the server, not the agent, acknowledges these threads' messages. Bad entries drop alone. */ + acknowledgeOnlyThreadIds: Schema.Array(Schema.Unknown).pipe( + Schema.decodeTo( + Schema.Array(ThreadId), + SchemaTransformation.transform, ReadonlyArray>({ + decode: (values) => values.filter(Schema.is(ThreadId)), + encode: (values) => values, + }), + ), + Schema.withDecodingDefault(Effect.succeed([])), + ), }); export type ServerSettings = typeof ServerSettings.Type; @@ -1953,6 +1965,8 @@ export const ServerSettingsPatch = Schema.Struct({ usageModelAliases: Schema.optionalKey( Schema.Record(TrimmedNonEmptyString, Schema.NullOr(TrimmedNonEmptyString)), ), + + acknowledgeOnlyThreadIds: Schema.optionalKey(Schema.Array(ThreadId)), }); export type ServerSettingsPatch = typeof ServerSettingsPatch.Type; From c94302d35ccd1f15f2a70ef26a9aa876335da539 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 22:05:21 -0700 Subject: [PATCH 2/2] fix(server): acknowledge-only threads keep moving after a restart Review fixes for the acknowledge-only mode: - Restart recovery no longer holds an acknowledge-only thread's queue, and releases a hold left from before, so the inbox keeps waking Jarvis after a deploy. - The reply is neutral ("Queued: ``"); the excerpt sits in a code span so no link or image renders, and token-like strings are redacted. - Toggles are serialized, so concurrent toggles cannot drop one. - Context handoffs addressed to an acknowledged run are superseded explicitly. - t3_thread_configure branches inside the existing write handler, keeping upstream lines unindented. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/McpToolAccess.ts | 7 + .../src/mcp/toolkits/thread/handlers.ts | 139 +++++++++--------- .../AcknowledgeOnly.integration.test.ts | 15 +- .../src/orchestration-v2/AcknowledgeOnly.ts | 29 +++- .../ProviderRuntimeRecoveryService.test.ts | 19 ++- .../ProviderRuntimeRecoveryService.ts | 12 +- .../ProviderTurnStartService.ts | 1 + .../ThreadManagementService.ts | 3 + docs/jarvis/decisions/0009-jarvis-inbox.md | 9 +- docs/user/outside-agents.md | 2 +- 10 files changed, 152 insertions(+), 84 deletions(-) diff --git a/apps/server/src/mcp/McpToolAccess.ts b/apps/server/src/mcp/McpToolAccess.ts index 5b4e2dbaa182f..31468cb704da7 100644 --- a/apps/server/src/mcp/McpToolAccess.ts +++ b/apps/server/src/mcp/McpToolAccess.ts @@ -340,6 +340,13 @@ export const afterForwarding = ( const fullAccessRequired = "Changing projects or environment settings needs a live full-access/default calling thread or a full-access client."; +/** Fork: the `writesEnvironment` check, for one branch of a tool that otherwise writes a thread. */ +export const requireFullAccess = (what: string) => + orchestratingCaller.pipe( + Effect.tap((caller) => assertNotLinked(caller, what)), + Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)), + ); + /** Recheck after waits before writing; caller modes may have changed. */ export const writesEnvironment = ( handle: ( diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 44ac90f68a6a7..2f2bb7f55a3ca 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -30,6 +30,7 @@ import * as ThreadSearch from "../../../orchestration-v2/ThreadSearch.ts"; import * as ScheduledTasks from "../../../scheduledTasks/ScheduledTaskService.ts"; import { queuedRunsInDeliveryOrder } from "../../../orchestration-v2/QueuedRunOrder.ts"; import { freezeThread, unfreezeThread } from "../../threadFreeze.ts"; +import type * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; import { ThreadToolkit } from "./tools.ts"; const isApprovalKind = Schema.is(ProviderRequestKind); @@ -106,19 +107,34 @@ const required = (value: A | undefined, name: string) => }), ) : Effect.succeed(value); -const readThreadConfiguration = Effect.fn("mcp.readThreadConfiguration")(function* ( - threadId: ThreadId | undefined, +// Who answers a thread's messages is environment policy, so it needs full access both ways. +const configureAcknowledgeOnly = Effect.fn("mcp.configureAcknowledgeOnly")(function* ( + threads: ThreadManagementService.ThreadManagementService["Service"], + thread: Pick< + OrchestrationV2ThreadProjection["thread"], + "id" | "modelSelection" | "runtimeMode" | "interactionMode" + >, + input: { + readonly acknowledgeOnly?: boolean | undefined; + readonly modelSelection?: unknown; + readonly runtimeMode?: unknown; + }, ) { - const { - threads, - projection: { thread }, - } = yield* readThread(threadId); + yield* McpToolAccess.requireFullAccess("change who answers a thread"); + if (input.modelSelection !== undefined || input.runtimeMode !== undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "Set acknowledgeOnly separately from modelSelection and runtimeMode.", + }); + } + const acknowledgeOnly = yield* required(input.acknowledgeOnly, "acknowledgeOnly"); + yield* threads.setAcknowledgeOnly(thread.id, acknowledgeOnly).pipe(Effect.mapError(unavailable)); return { threadId: thread.id, modelSelection: thread.modelSelection, runtimeMode: thread.runtimeMode, interactionMode: thread.interactionMode, - acknowledgeOnly: yield* threads.acknowledgeOnly(thread.id).pipe(Effect.mapError(unavailable)), + acknowledgeOnly, }; }); const writesThread =

( @@ -243,75 +259,66 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { input.modelSelection === undefined && input.runtimeMode === undefined && input.acknowledgeOnly === undefined, - McpToolAccess.reads((input) => readThreadConfiguration(input.threadId)), - McpToolAccess.either( - (input) => input.acknowledgeOnly !== undefined, - // Who answers the thread's messages is environment policy, so it needs full access. - McpToolAccess.writesEnvironment((input) => - Effect.gen(function* () { - const acknowledgeOnly = yield* required(input.acknowledgeOnly, "acknowledgeOnly"); - if (input.modelSelection !== undefined || input.runtimeMode !== undefined) { + McpToolAccess.reads((input) => + Effect.gen(function* () { + const { + threads, + projection: { thread }, + } = yield* readThread(input.threadId); + return { + threadId: thread.id, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + acknowledgeOnly: yield* threads + .acknowledgeOnly(thread.id) + .pipe(Effect.mapError(unavailable)), + }; + }), + ), + writesThread((input) => + Effect.gen(function* () { + const { + threads, + projection: { thread }, + } = yield* readThread(input.threadId); + if (input.acknowledgeOnly !== undefined) { + return yield* configureAcknowledgeOnly(threads, thread, input); + } + if (input.runtimeMode !== undefined) { + if (input.modelSelection !== undefined) { return yield* new OrchestratorMcpFailure({ code: "invalid_request", - message: "Set acknowledgeOnly separately from modelSelection and runtimeMode.", + message: "Set runtimeMode separately from modelSelection.", }); } - const { - threads, - projection: { thread }, - } = yield* readThread(input.threadId); - yield* threads - .setAcknowledgeOnly(thread.id, acknowledgeOnly) - .pipe(Effect.mapError(unavailable)); - return yield* readThreadConfiguration(thread.id); - }), - ), - writesThread((input) => - Effect.gen(function* () { - const { - threads, - projection: { thread }, - } = yield* readThread(input.threadId); - if (input.runtimeMode !== undefined) { - if (input.modelSelection !== undefined) { - return yield* new OrchestratorMcpFailure({ - code: "invalid_request", - message: "Set runtimeMode separately from modelSelection.", - }); - } - const scope = yield* McpInvocationContext.McpInvocationContext; - const actor = - scope.thread !== undefined - ? `MCP thread ${scope.thread.threadId}` - : scope.client !== undefined - ? `MCP client "${scope.client.label}" (${scope.client.sessionId.slice(0, 8)})` - : "MCP caller"; - const result = yield* threads - .dispatch({ - type: "thread.runtime-mode.set", - threadId: thread.id, - commandId: yield* newCommandId(), - runtimeMode: input.runtimeMode, - }) - .pipe( - Effect.provideService(RuntimeModeChangeActor, actor), - Effect.mapError(dispatchFailure), - ); - return { sequence: result.sequence }; - } - const modelSelection = yield* required(input.modelSelection, "modelSelection"); - const type = modelSelectionCommandType(thread.providerInstanceId, modelSelection); + const scope = yield* McpInvocationContext.McpInvocationContext; + const actor = + scope.thread !== undefined + ? `MCP thread ${scope.thread.threadId}` + : scope.client !== undefined + ? `MCP client "${scope.client.label}" (${scope.client.sessionId.slice(0, 8)})` + : "MCP caller"; const result = yield* threads .dispatch({ - type, + type: "thread.runtime-mode.set", threadId: thread.id, commandId: yield* newCommandId(), - modelSelection, + runtimeMode: input.runtimeMode, }) - .pipe(Effect.mapError(dispatchFailure)); + .pipe( + Effect.provideService(RuntimeModeChangeActor, actor), + Effect.mapError(dispatchFailure), + ); return { sequence: result.sequence }; - }), - ), + } + const modelSelection = yield* required(input.modelSelection, "modelSelection"); + const type = modelSelectionCommandType(thread.providerInstanceId, modelSelection); + const result = yield* threads + .dispatch({ type, threadId: thread.id, commandId: yield* newCommandId(), modelSelection }) + .pipe(Effect.mapError(dispatchFailure)); + return { sequence: result.sequence }; + }), ), ), t3_pending_request: McpToolAccess.either( diff --git a/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts b/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts index acc75ee23095e..ba5061961aa52 100644 --- a/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts +++ b/apps/server/src/orchestration-v2/AcknowledgeOnly.integration.test.ts @@ -38,7 +38,8 @@ const otherInbox = ThreadId.make("thread:another-inbox"); const messages = [ "First request for Jarvis", - "Second request", + // Sender text becomes agent output: no rendered image or link, no token. + "See ![chart](https://example.com/x.png) and `code`, key ghp_abcdef0123456789 here", `Third request\nwith a second line and ${"padding ".repeat(20)}to pass the excerpt limit`, ]; @@ -209,13 +210,17 @@ it.effect("an acknowledge-only thread answers queued messages in order without a .at(-1)?.text, })); assert.deepEqual(replies, [ - { request: messages[0], reply: "Queued for Jarvis: First request for Jarvis" }, - { request: messages[1], reply: "Queued for Jarvis: Second request" }, + { request: messages[0], reply: "Queued: `First request for Jarvis`" }, + { + request: messages[1], + reply: + "Queued: `See ![chart](https://example.com/x.png) and 'code', key [redacted] here`", + }, { request: messages[2], - reply: `Queued for Jarvis: ${Array.from(messages[2]!.replace(/\s+/g, " ")) + reply: `Queued: \`${Array.from(messages[2]!.replace(/\s+/g, " ")) .slice(0, 120) - .join("")}…`, + .join("")}…\``, }, ]); // Completions arrive in send order, each the event that wakes an inbox subscription. diff --git a/apps/server/src/orchestration-v2/AcknowledgeOnly.ts b/apps/server/src/orchestration-v2/AcknowledgeOnly.ts index 55151a7bd16e5..84d283f9c34a9 100644 --- a/apps/server/src/orchestration-v2/AcknowledgeOnly.ts +++ b/apps/server/src/orchestration-v2/AcknowledgeOnly.ts @@ -1,4 +1,5 @@ import type { + OrchestrationV2ContextHandoff, OrchestrationV2ConversationMessage, OrchestrationV2DomainEvent, OrchestrationV2ExecutionNode, @@ -38,10 +39,22 @@ export const setAcknowledgeOnly = Effect.fn("AcknowledgeOnly.set")(function* ( }); }); +// Token prefixes and long opaque runs, as the update guard redacts them. +const SECRET_PATTERNS = [ + /(?:glpat-|gh[pousr]_|github_pat_|xox[abposr]-|sk-|AKIA|ASIA|AIza)[A-Za-z0-9_-]*/g, + /[A-Za-z0-9+/=_-]{32,}/g, +]; + +/** Sender text becomes agent-authored output, so it is redacted and kept in a code span. */ const acknowledgementText = (message: string): string => { - const line = Array.from(message.replace(/\s+/g, " ").trim()); - const excerpt = line.slice(0, EXCERPT_LENGTH).join(""); - return `Queued for Jarvis: ${line.length > EXCERPT_LENGTH ? `${excerpt}…` : excerpt}`; + const line = SECRET_PATTERNS.reduce( + (text, pattern) => text.replace(pattern, "[redacted]"), + message.replace(/\s+/g, " ").trim(), + ).replaceAll("`", "'"); + const characters = Array.from(line); + if (characters.length === 0) return "Queued."; + const excerpt = characters.slice(0, EXCERPT_LENGTH).join(""); + return `Queued: \`${characters.length > EXCERPT_LENGTH ? `${excerpt}…` : excerpt}\``; }; /** @@ -58,6 +71,8 @@ export const settleAcknowledgeOnlyRun = Effect.fn("AcknowledgeOnly.settleRun")(f readonly providerThread: OrchestrationV2ProviderThread; readonly message: Pick; readonly nextTurnItemOrdinal: number; + /** Handoffs addressed to this run; no provider reads them, so they are superseded. */ + readonly handoffs: ReadonlyArray; }) { const { run, attempt, rootNode, providerThread } = input; const now = yield* DateTime.now; @@ -126,6 +141,14 @@ export const settleAcknowledgeOnlyRun = Effect.fn("AcknowledgeOnly.settleRun")(f { type: "node.updated", nodeId, payload: assistantNode }, { type: "message.updated", nodeId, payload: message }, { type: "turn-item.updated", nodeId, payload: turnItem }, + ...input.handoffs.map( + (handoff) => + ({ + type: "context-handoff.updated", + nodeId: rootNode.id, + payload: { ...handoff, status: "superseded", updatedAt: now }, + }) as const, + ), { type: "run-attempt.updated", nodeId: rootNode.id, payload: { ...attempt, ...completed } }, { type: "run.updated", nodeId: rootNode.id, payload: { ...run, ...completed } }, { type: "node.updated", nodeId: rootNode.id, payload: { ...rootNode, ...completed } }, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 8c9f6e282f340..85a113c895575 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -512,13 +512,18 @@ it.effect("recovery closes a secret card's form but leaves a link card open", () it.effect.each([ "queued", + "queued-acknowledge-only", "preparing", "starting", "starting-other-request", "starting-own-request", "starting-scheduled", ] as const)("recovers $0 work after restart", (scenario) => { - const status = scenario.startsWith("starting") ? "starting" : scenario; + const status = scenario.startsWith("starting") + ? "starting" + : scenario.startsWith("queued") + ? "queued" + : scenario; const terminalized = status === "preparing" || scenario === "starting-own-request" || @@ -604,7 +609,12 @@ it.effect.each([ turnItems: [], } as unknown as OrchestrationV2ThreadProjection; const layer = ProviderRuntimeRecovery.layer.pipe( - Layer.provide(ServerSettings.layerTest()), + Layer.provide( + ServerSettings.layerTest( + // The server answers these runs itself, so a restart leaves their queue moving. + scenario === "queued-acknowledge-only" ? { acknowledgeOnlyThreadIds: [threadId] } : {}, + ), + ), Layer.provide( Layer.mergeAll( Layer.mock(ProjectionStore.ProjectionStoreV2)({ @@ -665,7 +675,10 @@ it.effect.each([ ); } else { assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : null, 1); - assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : false, true); + assert.equal( + runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : null, + scenario !== "queued-acknowledge-only", + ); assert.isUndefined(attemptEvent); assert.isUndefined(nodeEvent); } diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 46b739bec5732..201235cd5ed4d 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -23,6 +23,7 @@ import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; +import * as AcknowledgeOnly from "./AcknowledgeOnly.ts"; import { restartContinuationRun } from "./RestartContinuation.ts"; import { isBufferedProviderNotification } from "./NotificationMailbox.ts"; import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; @@ -421,12 +422,19 @@ export const make = Effect.gen(function* () { ...retryableRuns, ...queuedRunsInDeliveryOrder({ runs: queuedRuns, messages: queuedMessages }), ]; + // Fork: acknowledge-only runs have no provider work to protect, so their queue keeps moving. + const holdQueue = !AcknowledgeOnly.isAcknowledgeOnlyThread( + yield* settings.getSettings.pipe( + Effect.orElseSucceed(() => ({ acknowledgeOnlyThreadIds: [] })), + ), + projection.thread.id, + ); for (const [index, run] of heldRuns.entries()) { if (bufferedMessageIds.has(run.userMessageId)) { runs.push(run); continue; } - if (run.queueHeld === true && retryableRuns.length === 0) continue; + if (run.queueHeld === true && holdQueue && retryableRuns.length === 0) continue; events.push({ id: yield* allocateEventId(), type: "run.updated", @@ -437,7 +445,7 @@ export const make = Effect.gen(function* () { payload: { ...run, status: "queued", - queueHeld: true, + queueHeld: holdQueue, ...(retryableRuns.length === 0 ? {} : { queuePosition: index + 1 }), }, }); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 241011f1cc77b..d32cf5b1aff1f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -467,6 +467,7 @@ export const layer: Layer.Layer< rootNode, providerThread, message, + handoffs: handoffs.filter((handoff) => handoff.targetRunId === run.id), nextTurnItemOrdinal: Math.max( 0, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 520c013a30f0c..3c27cba119aa7 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1632,6 +1632,8 @@ const make = Effect.gen(function* () { if (failures[0] !== undefined) return yield* failures[0]; }); + // Fork: toggles read and rewrite one settings list; one at a time so none is lost. + const acknowledgeOnlyLock = yield* Semaphore.make(1); return ThreadManagementService.of({ acknowledgeOnly: Effect.fn("ThreadManagementService.acknowledgeOnly")(function* (threadId) { return AcknowledgeOnly.isAcknowledgeOnlyThread(yield* serverSettings.getSettings, threadId); @@ -1640,6 +1642,7 @@ const make = Effect.gen(function* () { function* (threadId, enabled) { yield* AcknowledgeOnly.setAcknowledgeOnly(threadId, enabled).pipe( Effect.provideService(ServerSettings.ServerSettingsService, serverSettings), + acknowledgeOnlyLock.withPermits(1), ); }, ), diff --git a/docs/jarvis/decisions/0009-jarvis-inbox.md b/docs/jarvis/decisions/0009-jarvis-inbox.md index 4a98b1d70663b..74fdd1bf83bb8 100644 --- a/docs/jarvis/decisions/0009-jarvis-inbox.md +++ b/docs/jarvis/decisions/0009-jarvis-inbox.md @@ -10,12 +10,13 @@ The inbox acknowledges each message with a run of its own; that run's result eve ## Acknowledge-only mode -- **What it does.** Each message to an acknowledge-only thread starts a run as usual. The server, not a provider, completes it with the assistant reply `Queued for Jarvis: `. No provider session opens, no tools run and no tokens are spent. Messages are answered one at a time in queue order. -- **Events.** The run emits the lifecycle a provider turn does: `run.created` (queued or starting), `run.updated` starting and running, the assistant message, then `run.updated` completed. Subscriptions see the same `thread.run.result` event with the same source run id, and the relay quotes the same assistant message as its excerpt. It deliberately has no provider turn, no `waiting` step and no end-of-turn checkpoint. +- **What it does.** Each message to an acknowledge-only thread starts a run as usual. The server, not a provider, completes it with the assistant reply ``Queued: `` `` (`Queued.` for a message with no text). The prefix is neutral because any thread can use the mode. The excerpt is sender text published as agent output, so it sits in a code span, where no link or image renders, and token-like strings are redacted as the update guard redacts them. No provider session opens, no tools run and no tokens are spent. Messages are answered one at a time in queue order. +- **Restarts.** A restart holds every other thread's queue until a person sends a message, to protect interrupted provider work. Acknowledge-only threads have none, so their queue stays unheld and keeps moving after a deploy, and a hold left over from before the mode was turned on is released. +- **Events.** The run emits the lifecycle a provider turn does: `run.created` (queued or starting), `run.updated` starting and running, the assistant message, then `run.updated` completed. Subscriptions see the same `thread.run.result` event with the same source run id, and the relay quotes the same assistant message as its excerpt. It deliberately has no provider turn, no `waiting` step and no end-of-turn checkpoint. A context handoff addressed to the acknowledged run is marked superseded in the same commit, since no provider reads it; handoffs carried forward to a later provider run are left alone. - **What gets acknowledged.** Every run on the thread, whatever started it. Relay messages are meant for Jarvis, so acknowledging them is right. A restart continuation follows only an interrupted provider turn, which acknowledged runs never have. Idle compaction targets only Claude sessions and never compacts a compaction, so a Claude thread in this mode acknowledges at most one `/compact` per idle spell. -- **Who can change it.** `t3_thread_configure` with `acknowledgeOnly` alone, both to turn it on and to turn it off, needs a full-access caller: a live calling thread in full-access mode, or an MCP client signed in with full access, and not a linked environment's caller. Anyone who can edit server settings can also change `acknowledgeOnlyThreadIds` in `settings.json`. Reading the thread's configuration shows the mode. +- **Who can change it.** `t3_thread_configure` with `acknowledgeOnly` alone, both to turn it on and to turn it off, needs a full-access caller: a live calling thread in full-access mode, or an MCP client signed in with full access, and not a linked environment's caller. Toggles are applied one at a time, so concurrent toggles for different threads are all kept. Anyone who can edit server settings can also change `acknowledgeOnlyThreadIds` in `settings.json`. Reading the thread's configuration shows the mode. - **Turning it off.** The next message runs the thread's agent again. The agent's native conversation never saw the acknowledged messages. -- **Known limits.** The thread still needs a configured, enabled provider: starting a queued run resolves the provider adapter, and a disabled instance holds the queue. A queue held by an earlier failed or usage-limited run stays held after the mode is turned on, until someone resumes the queue in the app. A run that finishes in this mode never holds the queue. Turning the mode on does not detach a provider session already attached; it stays attached but receives no turns. +- **Known limits.** The thread still needs a configured, enabled provider: starting a queued run resolves the provider adapter, and a disabled instance holds the queue. A queue held by an earlier failed or usage-limited run stays held after the mode is turned on, until the next restart or until someone resumes the queue in the app. A run that finishes in this mode never holds the queue. Turning the mode on does not detach a provider session already attached; it stays attached but receives no turns. ## Alternatives diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index f59eba7ee6236..9571aefc37569 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -258,7 +258,7 @@ ChatGPT watch. Your existing inbox watch receives the inbox run's normal event. Keep the inbox's run-per-message acknowledger enabled. To acknowledge without a model, a full-access caller can call `t3_thread_configure` with `acknowledgeOnly: true` on the inbox thread. The server then answers each message -with `Queued for Jarvis: ` and ends its run, so the inbox +with ``Queued: `` `` and ends its run, so the inbox watch still wakes. `acknowledgeOnly: false` returns the thread to its agent. The response reports `bound` with the inbox thread id, or `unavailable` or `failed`