From dfedee32e24290bcf025ada7c7baa23be84ab0d5 Mon Sep 17 00:00:00 2001 From: Rene Mifsud <50179233+renemifsud@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:05:08 +0200 Subject: [PATCH 1/5] fix(server): an agent can mark its own thread unread when its turn ends An agent that calls t3_thread_organize with mark_unread on its own thread mid-turn was refused, because marking unread rewinds the visit to just before the latest run completed and the latest run is the caller's active one. It now waits for that turn, the way settle does, marks the thread unread if the turn completed, and returns marksUnreadWhenTurnEnds=true. Co-authored-by: Claude Opus 5.5 --- .../src/mcp/toolkits/thread/handlers.ts | 8 +- apps/server/src/mcp/toolkits/thread/tools.ts | 3 +- .../ThreadManagementService.test.ts | 41 ++++++++++ .../ThreadManagementService.ts | 58 +++++++++++++- .../src/orchestration-v2/runtimeLayer.test.ts | 76 +++++++++++++++++++ .../src/relay/AgentAwarenessRelay.test.ts | 2 + 6 files changed, 182 insertions(+), 6 deletions(-) diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 4ab78b4fe07b2..dca65b556319f 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -301,6 +301,11 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { .settleThread({ ...common, byOwnAgent: caller?.id === projection.thread.id }) .pipe(Effect.mapError(dispatchFailure)); } + if (input.action === "mark_unread") { + return yield* threads + .markThreadUnread({ ...common, byOwnAgent: caller?.id === projection.thread.id }) + .pipe(Effect.mapError(dispatchFailure)); + } let command: OrchestrationV2Command; switch (input.action) { case "snooze": @@ -316,9 +321,6 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { case "unsettle": command = { ...common, type: `thread.${input.action}`, reason: "user" }; break; - case "mark_unread": - command = { ...common, type: "thread.mark-unread" }; - break; default: command = { ...common, type: `thread.${input.action}` }; } diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index 21750cbb6eaae..cd9195dd6fb7d 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -30,7 +30,7 @@ import * as McpInvocationContext from "../../McpInvocationContext.ts"; const ThreadOrganizeTool = Tool.make("t3_thread_organize", { description: - "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. Settling this thread takes effect when your turn completes, returning settlesWhenTurnEnds=true; a turn that fails or is interrupted, or a queued message, leaves it active.", + "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. Settling or marking this thread unread takes effect when your turn completes, returning settlesWhenTurnEnds=true or marksUnreadWhenTurnEnds=true; a turn that fails or is interrupted, or a queued message, leaves the thread unchanged.", parameters: Schema.Struct({ threadId: Schema.optional(ThreadId), action: Schema.Literals([ @@ -49,6 +49,7 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", { success: Schema.Union([ OrchestrationV2DispatchCommandResult, Schema.Struct({ settlesWhenTurnEnds: Schema.Literal(true) }), + Schema.Struct({ marksUnreadWhenTurnEnds: Schema.Literal(true) }), ]), failure: OrchestratorMcpFailure, failureMode: "return" as const, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 8d1f613180a3a..4aa88c3b7e310 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -337,3 +337,44 @@ it.effect.each([ expect(dispatched).toEqual(settles ? ["thread.settle"] : []); }), ); + +it.effect.each([ + { status: "completed" as const, marks: true }, + { status: "failed" as const, marks: false }, + { status: "interrupted" as const, marks: false }, +])("markUnreadAfterRun marks unread only when the run $status", ({ status, marks }) => + Effect.gen(function* () { + const projectId = ProjectId.make("project:thread-management:mark-unread-after-run"); + const threadId = ThreadId.make("thread:thread-management:mark-unread-after-run"); + const runId = RunId.make("run:thread-management:mark-unread-after-run"); + const dispatched: Array = []; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(false), + getThreadRecords: () => + Effect.succeed({ + thread: { id: threadId, projectId, deletedAt: null }, + runs: [{ id: runId, status }], + } as unknown as OrchestrationV2ThreadProjection), + dispatch: (command) => + Effect.sync(() => { + dispatched.push(command.type); + return { sequence: 1, storedEvents: [] } as never; + }), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + + yield* service.markUnreadAfterRun({ projectId, threadId, runId }); + + expect(dispatched).toEqual(marks ? ["thread.mark-unread"] : []); + }), +); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index fba2df1de5c24..b029c2f5a88c0 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -415,6 +415,26 @@ export interface ThreadManagementServiceShape { { readonly sequence: number } | { readonly settlesWhenTurnEnds: true }, Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError >; + /** + * Marking unread rewinds the visit to just before the latest run completed, + * so an agent that marks its own thread unread uses this to wait for its + * turn to end. A run that ends any other way leaves the thread as it was, + * as does anything `thread.mark-unread` refuses then, such as a queued + * message. + */ + readonly markUnreadAfterRun: (input: { + readonly projectId: ProjectId; + readonly threadId: ThreadId; + readonly runId: RunId; + }) => Effect.Effect; + readonly markThreadUnread: (input: { + readonly threadId: ThreadId; + readonly commandId: CommandId; + readonly byOwnAgent: boolean; + }) => Effect.Effect< + { readonly sequence: number } | { readonly marksUnreadWhenTurnEnds: true }, + Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError + >; readonly interruptThread: ( input: ThreadManagementInterruptInput, ) => Effect.Effect; @@ -505,7 +525,7 @@ function latestSteerableRun( .toSorted((left, right) => right.ordinal - left.ordinal)[0]; } -const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000; +const AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000; const make = Effect.gen(function* () { const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor; @@ -932,7 +952,7 @@ const make = Effect.gen(function* () { const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn( "orchestrationV2.threadManagement.settleAfterRun", )(function* (input) { - const { run } = yield* waitForThread({ ...input, timeoutMs: SETTLE_AFTER_RUN_WAIT_MS }); + const { run } = yield* waitForThread({ ...input, timeoutMs: AFTER_RUN_WAIT_MS }); if (run?.status !== "completed") return; yield* dispatch({ type: "thread.settle", @@ -961,6 +981,38 @@ const make = Effect.gen(function* () { return { sequence: result.sequence }; }); + const markUnreadAfterRun: ThreadManagementServiceShape["markUnreadAfterRun"] = Effect.fn( + "orchestrationV2.threadManagement.markUnreadAfterRun", + )(function* (input) { + const { run } = yield* waitForThread({ ...input, timeoutMs: AFTER_RUN_WAIT_MS }); + if (run?.status !== "completed") return; + yield* dispatch({ + type: "thread.mark-unread", + commandId: CommandId.make(`server:mark-unread-after-run:${input.runId}`), + threadId: input.threadId, + }); + }); + + const markThreadUnread: ThreadManagementServiceShape["markThreadUnread"] = Effect.fn( + "orchestrationV2.threadManagement.markThreadUnread", + )(function* (input) { + const shell = input.byOwnAgent ? yield* orchestrator.getThreadShell(input.threadId) : null; + if (shell != null && shell.activeRunId !== null) { + yield* markUnreadAfterRun({ + projectId: shell.projectId, + threadId: shell.id, + runId: shell.activeRunId, + }).pipe(Effect.ignoreCause({ log: true }), Effect.forkIn(layerScope)); + return { marksUnreadWhenTurnEnds: true } as const; + } + const result = yield* dispatch({ + type: "thread.mark-unread", + commandId: input.commandId, + threadId: input.threadId, + }); + return { sequence: result.sequence }; + }); + const interruptThread: ThreadManagementServiceShape["interruptThread"] = Effect.fn( "ThreadManagementService.interruptThread", )(function* (input) { @@ -1186,6 +1238,8 @@ const make = Effect.gen(function* () { waitForThread, settleAfterRun, settleThread, + markUnreadAfterRun, + markThreadUnread, interruptThread, stopDelegatedTasks, stopWorkflow, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1c9404da6b920..dd2b3ca956d2d 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3700,6 +3700,82 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("marks a thread its own agent marked unread once the turn completes", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadManagement = yield* ThreadManagementService.ThreadManagementService; + const projectId = ProjectId.make("runtime-layer-mark-unread-after-run-project"); + const threadId = ThreadId.make("runtime-layer-mark-unread-after-run-thread"); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-mark-unread-after-run-create"), + threadId, + projectId, + title: "Mark unread after run", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-mark-unread-after-run", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-mark-unread-after-run-message"), + threadId, + messageId: MessageId.make("runtime-layer-mark-unread-after-run-message"), + text: "Report back and then mark this thread unread.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const started = yield* orchestrator.getThreadProjection(threadId); + const run = started.runs[0]; + if (run === undefined) return yield* Effect.die(new Error("Run missing.")); + + const markedUnread = yield* orchestrator + .streamStoredEventsFrom({ threadId, afterSequence: 0, eventType: "thread.marked-unread" }) + .pipe(Stream.runHead, Effect.forkChild); + const result = yield* threadManagement.markThreadUnread({ + threadId, + commandId: CommandId.make("runtime-layer-mark-unread-after-run-mark"), + byOwnAgent: true, + }); + assert.deepEqual(result, { marksUnreadWhenTurnEnds: true }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread.lastVisitedAt, + started.thread.lastVisitedAt, + ); + yield* TestClock.adjust("1 second"); + const now = yield* DateTime.now; + yield* eventSink.write({ + commandId: CommandId.make("runtime-layer-mark-unread-after-run-completed"), + events: [ + { + id: EventId.make("runtime-layer-mark-unread-after-run-completed"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + yield* Fiber.join(markedUnread); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.thread.lastVisitedAt, + DateTime.subtract(now, { milliseconds: 1 }), + ); + }), + ); + it.effect("settles past held automatic runs but not held user messages", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index d1ad9e7a58a64..d815af98f8302 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -213,6 +213,8 @@ const makeTestRelay = Effect.fnUntraced(function* ( waitForThread: unused, settleAfterRun: unused, settleThread: unused, + markUnreadAfterRun: unused, + markThreadUnread: unused, interruptThread: unused, stopWorkflow: () => Effect.die("unused stopWorkflow"), stopDelegatedTasks: unused, From 7e25929a8fb83d3596930e96deed99c53fcfdd03 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 10:54:48 +0000 Subject: [PATCH 2/5] test(server): reproduce #24136 own agent cannot mark its thread unread mid-run A scheduled run calling t3_thread_organize mark_unread on its own thread is rejected with 'has no completed run to mark unread' because the active run is the latest run, even though an earlier run completed. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.test.ts | 131 ++++++++++++++++++ 1 file changed, 131 insertions(+) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1c9404da6b920..148b295f6c79e 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2,6 +2,7 @@ import { SourceControlProviderKind } from "@t3tools/contracts"; import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { vi } from "vite-plus/test"; @@ -11,6 +12,7 @@ import { CheckpointRef, CommandId, ContextTransferId, + EnvironmentId, EventId, MessageId, NodeId, @@ -42,6 +44,7 @@ import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; +import { McpSchema, McpServer } from "effect/ai"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; @@ -51,6 +54,8 @@ import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ServerSettings from "../serverSettings.ts"; +import * as McpHttpServer from "../mcp/McpHttpServer.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; import * as McpSessionRegistryTestkit from "../mcp/McpSessionRegistry.testkit.ts"; import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; @@ -75,6 +80,7 @@ import * as RuntimeLayer from "./runtimeLayer.ts"; import { shellStreamItemFromThreadShell } from "./ShellStream.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ThreadSearch from "./ThreadSearch.ts"; import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; @@ -3700,6 +3706,131 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("lets a scheduled run mark its own thread unread through t3_thread_organize", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const server = yield* McpServer.McpServer; + const threadId = ThreadId.make("runtime-layer-own-mark-unread-thread"); + const completeRun = (run: OrchestrationV2Run, id: string) => + Effect.gen(function* () { + yield* TestClock.adjust("1 second"); + const now = yield* DateTime.now; + yield* eventSink.write({ + commandId: CommandId.make(id), + events: [ + { + id: EventId.make(id), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + return now; + }); + const sendMessage = (id: string) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(id), + text: id, + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-own-mark-unread-create"), + threadId, + projectId: ProjectId.make("runtime-layer-own-mark-unread-project"), + title: "Own mark unread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-own-mark-unread", + }); + yield* sendMessage("runtime-layer-own-mark-unread-seed"); + const seedRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]; + if (seedRun === undefined) return yield* Effect.die(new Error("Seed run missing.")); + yield* completeRun(seedRun, "runtime-layer-own-mark-unread-seed-completed"); + yield* sendMessage("runtime-layer-own-mark-unread-scheduled"); + const scheduledRun = (yield* orchestrator.getThreadProjection(threadId)).runs[1]; + if (scheduledRun === undefined) return yield* Effect.die(new Error("Scheduled run missing.")); + + const markedUnread = yield* orchestrator + .streamStoredEventsFrom({ threadId, afterSequence: 0, eventType: "thread.marked-unread" }) + .pipe(Stream.runHead, Effect.forkChild); + // The scheduled run's own agent calls the tool while that run is still active. + const result = yield* server + .callTool({ name: "t3_thread_organize", arguments: { action: "mark_unread" } }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + environmentId: EnvironmentId.make("runtime-layer-own-mark-unread-environment"), + requestNamespace: "runtime-layer-own-mark-unread-session", + thread: { + threadId, + providerSessionId: "runtime-layer-own-mark-unread-session", + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + issuedAt: 0, + capabilities: new Set(["orchestration"]), + }), + Effect.provideService( + McpSchema.McpServerClient, + McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + clientCapabilities: {}, + clientInfo: { name: "runtime-layer", version: "1" }, + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "runtime-layer", version: "1" }, + }, + getClient: Effect.die("unused"), + }), + ), + ); + const text = result.content[0]; + assert.deepEqual( + result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined, + undefined, + "mark_unread from the thread's own active run must not be rejected", + ); + + const completedAt = yield* completeRun( + scheduledRun, + "runtime-layer-own-mark-unread-scheduled-completed", + ); + yield* Fiber.join(markedUnread); + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.thread.lastVisitedAt, + DateTime.subtract(completedAt, { milliseconds: 1 }), + ); + }).pipe( + Effect.provide( + McpHttpServer.layerThreadToolkit.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), + Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + ), + ), + ), + ); + it.effect("settles past held automatic runs but not held user messages", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 19235a6884520b5ed2b6dfdaf0ed7e03dee5d5f0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 11:02:40 +0000 Subject: [PATCH 3/5] fix(server): a deferred mark-unread no longer overrides a newer run or visit The own-agent deferred mark now sends thread.mark-unread with the run it waited for. The orchestrator applies it only while that run is the latest completed run and the thread was not visited after it completed, so the check and the write share one command boundary. A wait that times out before a restart continuation settles no longer marks at all. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 19 ++++++++++++++--- .../ThreadManagementService.ts | 8 +++---- .../src/orchestration-v2/runtimeLayer.test.ts | 21 ++++++++++++++++++- packages/contracts/src/orchestrationV2.ts | 5 +++++ 4 files changed, 45 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d2e9333c6999b..5baa1a2aad754 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3060,15 +3060,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio let markUnreadVisitedAt: DateTime.Utc | null = null; if (command.type === "thread.mark-unread") { const projection = yield* loadProjectionForCommand(command, ["runs"]); - const latestRunCompletedAt = projection.runs.at(-1)?.completedAt ?? null; - if (latestRunCompletedAt === null) { + const latestRun = projection.runs.at(-1); + const latestRunCompletedAt = latestRun?.completedAt ?? null; + if (command.runId !== undefined) { + // Checked here, not by the caller, so a newer run or a visit that lands + // while the deferred mark is in flight still wins. + markUnreadVisitedAt = + latestRun?.id === command.runId && + latestRunCompletedAt !== null && + (thread.lastVisitedAt === null || + DateTime.toEpochMillis(thread.lastVisitedAt) < + DateTime.toEpochMillis(latestRunCompletedAt)) + ? DateTime.subtract(latestRunCompletedAt, { milliseconds: 1 }) + : thread.lastVisitedAt; + } else if (latestRunCompletedAt === null) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, cause: `Thread ${command.threadId} has no completed run to mark unread.`, }); + } else { + markUnreadVisitedAt = DateTime.subtract(latestRunCompletedAt, { milliseconds: 1 }); } - markUnreadVisitedAt = DateTime.subtract(latestRunCompletedAt, { milliseconds: 1 }); } const updatedThread: OrchestrationV2AppThread = (() => { switch (command.type) { diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index b029c2f5a88c0..bdaf0ab797418 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -419,8 +419,7 @@ export interface ThreadManagementServiceShape { * Marking unread rewinds the visit to just before the latest run completed, * so an agent that marks its own thread unread uses this to wait for its * turn to end. A run that ends any other way leaves the thread as it was, - * as does anything `thread.mark-unread` refuses then, such as a queued - * message. + * as does a newer run or a visit after the run completed. */ readonly markUnreadAfterRun: (input: { readonly projectId: ProjectId; @@ -984,12 +983,13 @@ const make = Effect.gen(function* () { const markUnreadAfterRun: ThreadManagementServiceShape["markUnreadAfterRun"] = Effect.fn( "orchestrationV2.threadManagement.markUnreadAfterRun", )(function* (input) { - const { run } = yield* waitForThread({ ...input, timeoutMs: AFTER_RUN_WAIT_MS }); - if (run?.status !== "completed") return; + const { run, timedOut } = yield* waitForThread({ ...input, timeoutMs: AFTER_RUN_WAIT_MS }); + if (timedOut || run?.status !== "completed") return; yield* dispatch({ type: "thread.mark-unread", commandId: CommandId.make(`server:mark-unread-after-run:${input.runId}`), threadId: input.threadId, + runId: run.id, }); }); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 148b295f6c79e..45262dd72a63b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3784,7 +3784,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }, client: undefined, issuedAt: 0, - capabilities: new Set(["orchestration"]), + capabilities: new Set(["orchestration"] as const), }), Effect.provideService( McpSchema.McpServerClient, @@ -3819,6 +3819,25 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { projection.thread.lastVisitedAt, DateTime.subtract(completedAt, { milliseconds: 1 }), ); + + // A deferred mark that lands after a newer visit keeps the read state. + const visitedAt = DateTime.add(completedAt, { milliseconds: 500 }); + yield* orchestrator.dispatch({ + type: "thread.visit", + commandId: CommandId.make("runtime-layer-own-mark-unread-visit"), + threadId, + visitedAt: DateTime.formatIso(visitedAt), + }); + yield* orchestrator.dispatch({ + type: "thread.mark-unread", + commandId: CommandId.make("runtime-layer-own-mark-unread-after-visit"), + threadId, + runId: scheduledRun.id, + }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread.lastVisitedAt, + visitedAt, + ); }).pipe( Effect.provide( McpHttpServer.layerThreadToolkit.pipe( diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index ff7c295b8bf91..b1da707d9b897 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2696,6 +2696,11 @@ export const OrchestrationV2Command = Schema.Union([ type: Schema.Literal("thread.mark-unread"), commandId: CommandId, threadId: ThreadId, + /** + * Mark only while this run is the latest and completed, and the thread + * was not visited after it completed; otherwise keep the read state. + */ + runId: Schema.optional(RunId), }), Schema.Struct({ type: Schema.Literal("thread.metadata.update"), From 6831fadd355e1291638d21dea08cd2a195c6c508 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 11:08:30 +0000 Subject: [PATCH 4/5] fix(server): a deferred mark-unread skips a run that was rolled back Rollback rewrites the same run as rolled_back with a later completedAt, so the guard matched it and rewound a visit made after the original completion. Require the run to still be completed. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 1 + .../src/orchestration-v2/runtimeLayer.test.ts | 85 +++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5baa1a2aad754..9a36800ee4f5b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3067,6 +3067,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // while the deferred mark is in flight still wins. markUnreadVisitedAt = latestRun?.id === command.runId && + latestRun.status === "completed" && latestRunCompletedAt !== null && (thread.lastVisitedAt === null || DateTime.toEpochMillis(thread.lastVisitedAt) < diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 45262dd72a63b..7e1309a50633f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3850,6 +3850,91 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ), ); + it.effect("keeps a visit when a deferred mark-unread lands after its run rolled back", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("runtime-layer-rollback-mark-unread-thread"); + const updateRun = ( + run: OrchestrationV2Run, + id: string, + status: "completed" | "rolled_back", + ) => + Effect.gen(function* () { + yield* TestClock.adjust("1 second"); + const now = yield* DateTime.now; + yield* eventSink.write({ + commandId: CommandId.make(id), + events: [ + { + id: EventId.make(id), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status, startedAt: now, completedAt: now }, + }, + ], + }); + return now; + }); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-rollback-mark-unread-create"), + threadId, + projectId: ProjectId.make("runtime-layer-rollback-mark-unread-project"), + title: "Rollback mark unread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-rollback-mark-unread", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-rollback-mark-unread-message"), + threadId, + messageId: MessageId.make("runtime-layer-rollback-mark-unread-message"), + text: "run", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]; + if (run === undefined) return yield* Effect.die(new Error("Run missing.")); + const completedAt = yield* updateRun( + run, + "runtime-layer-rollback-mark-unread-completed", + "completed", + ); + const visitedAt = DateTime.add(completedAt, { milliseconds: 500 }); + yield* orchestrator.dispatch({ + type: "thread.visit", + commandId: CommandId.make("runtime-layer-rollback-mark-unread-visit"), + threadId, + visitedAt: DateTime.formatIso(visitedAt), + }); + // Rollback rewrites the same run with a later completedAt. + yield* updateRun(run, "runtime-layer-rollback-mark-unread-rolled-back", "rolled_back"); + + yield* orchestrator.dispatch({ + type: "thread.mark-unread", + commandId: CommandId.make("runtime-layer-rollback-mark-unread-deferred"), + threadId, + runId: run.id, + }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread.lastVisitedAt, + visitedAt, + ); + }), + ); + it.effect("settles past held automatic runs but not held user messages", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 08b1f88a8bf0d7112eb4bac10969f00307a68db8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 11:19:41 +0000 Subject: [PATCH 5/5] test(server): a deferred mark-unread skips a run that a newer one superseded Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.test.ts | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index fcf83d87ba69d..a299cea951742 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3935,6 +3935,84 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("keeps a visit when a deferred mark-unread lands after a newer run completed", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("runtime-layer-superseded-mark-unread-thread"); + const runMessage = (id: string) => + Effect.gen(function* () { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(id), + text: id, + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const run = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1); + if (run === undefined) return yield* Effect.die(new Error("Run missing.")); + yield* TestClock.adjust("1 second"); + const now = yield* DateTime.now; + yield* eventSink.write({ + commandId: CommandId.make(`${id}-completed`), + events: [ + { + id: EventId.make(`${id}-completed`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + return { run, completedAt: now }; + }); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-superseded-mark-unread-create"), + threadId, + projectId: ProjectId.make("runtime-layer-superseded-mark-unread-project"), + title: "Superseded mark unread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-superseded-mark-unread", + }); + const older = yield* runMessage("runtime-layer-superseded-mark-unread-older"); + const visitedAt = DateTime.add(older.completedAt, { milliseconds: 500 }); + yield* orchestrator.dispatch({ + type: "thread.visit", + commandId: CommandId.make("runtime-layer-superseded-mark-unread-visit"), + threadId, + visitedAt: DateTime.formatIso(visitedAt), + }); + // The newer run completes after the visit, so only the runId guard keeps + // the older run's deferred mark from marking it unread. + yield* runMessage("runtime-layer-superseded-mark-unread-newer"); + + yield* orchestrator.dispatch({ + type: "thread.mark-unread", + commandId: CommandId.make("runtime-layer-superseded-mark-unread-deferred"), + threadId, + runId: older.run.id, + }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread.lastVisitedAt, + visitedAt, + ); + }), + ); + it.effect("settles past held automatic runs but not held user messages", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;