Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 55 additions & 0 deletions apps/server/src/mcp/McpEvents.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1085,6 +1085,61 @@ it.each([
);
});

it("says why a cancelled run stopped, so a restart cancel does not read as a failure", () => {
const threadId = ThreadId.make("restart-thread");
const instanceId = ProviderInstanceId.make("fixture");
const run = {
id: RunId.make("restart-run"),
threadId,
ordinal: 75,
providerInstanceId: instanceId,
modelSelection: { instanceId, model: "fixture" },
providerThreadId: null,
userMessageId: MessageId.make("restart-message"),
rootNodeId: null,
activeAttemptId: null,
status: "cancelled" as const,
requestedAt: DateTime.makeUnsafe(0),
startedAt: DateTime.makeUnsafe(0),
completedAt: DateTime.makeUnsafe(0),
checkpointId: null,
contextHandoffId: null,
};
const observe = (
payload: Omit<typeof run, "status"> & {
readonly status: "cancelled" | "rolled_back";
readonly stop?: { reason: string; actor: { kind: string } };
},
) =>
McpEvents.observeEvent(
{
sequence: 10,
commandId: null,
event: {
id: EventId.make("restart-event"),
threadId,
runId: run.id,
occurredAt: DateTime.makeUnsafe(0),
type: "run.updated",
payload,
},
},
{ lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null } },
"https://app.example.com/thread",
)?.data;
expect(
observe({ ...run, stop: { reason: "server_shutdown", actor: { kind: "server" } } }),
).toMatchObject({ state: "cancelled", stopReason: "server_shutdown" });
expect(observe(run)).not.toHaveProperty("stopReason");
expect(
observe({
...run,
status: "rolled_back",
stop: { reason: "server_shutdown", actor: { kind: "server" } },
}),
).not.toHaveProperty("stopReason");
});

it.effect(
"emits a minimal pending-decision observation without approval execution or resolved-request prose",
() =>
Expand Down
6 changes: 6 additions & 0 deletions apps/server/src/mcp/McpEvents.ts
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ export const observeEvent = (
let name: McpEventName;
let state: string;
let resourceId: string | undefined;
let stopReason: string | undefined;
switch (event.type) {
case "run.updated":
case "subagent.updated":
Expand All @@ -95,6 +96,10 @@ export const observeEvent = (
return;
name = state === "failed" ? "thread.blocker" : "thread.run.result";
resourceId = event.payload.id;
// A rolled-back run can still carry an earlier cancel's stop.
if (event.type === "run.updated" && (state === "cancelled" || state === "interrupted")) {
stopReason = event.payload.stop?.reason;
}
break;
case "runtime-request.updated":
if (event.payload.status !== "pending") return;
Expand Down Expand Up @@ -122,6 +127,7 @@ export const observeEvent = (
state,
...(event.runId === undefined ? {} : { runId: event.runId }),
...(resourceId === undefined ? {} : { resourceId }),
...(stopReason === undefined ? {} : { stopReason }),
parentThreadId: thread.lineage.parentThreadId,
relationshipToParent: thread.lineage.relationshipToParent,
},
Expand Down
6 changes: 3 additions & 3 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ import type {
} from "../orchestration-v2/ProjectionStore.ts";
import { delegatedTaskResultRun, runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts";
import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
import { runStopDetail } from "../orchestration-v2/RunStop.ts";
import { mayHaveRestartContinuation, runStopDetail } from "../orchestration-v2/RunStop.ts";
import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
import { isSnoozed } from "../orchestration-v2/ThreadSettlementService.ts";
Expand Down Expand Up @@ -3172,9 +3172,9 @@ const make = Effect.gen(function* () {
excludeSilentMaintenance: true,
})
.pipe(Effect.mapError(threadManagementFailure));
// A stopped run may have been continued as a later run; only then are the others read.
// Only a run a restart can continue needs the others read; t3_thread_read uses the same rule.
const continuationRuns =
run === undefined || !["cancelled", "interrupted", "failed"].includes(run.status)
run === undefined || !mayHaveRestartContinuation(run)
? []
: (yield* threadManagement
.getThreadRecords(input.threadId, ["runs"])
Expand Down
8 changes: 6 additions & 2 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6882,7 +6882,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
: run.id === continuedSourceRunId,
)?.delegatedTaskId;

if (projection.thread.settledOverride !== null) {
// A notice the server posts for the record is not a message someone sent: it reopens nothing.
const serverNotice = command.postOnly === true && command.creationSource === "server";
if (projection.thread.settledOverride !== null && !serverNotice) {
const now = yield* DateTime.now;
const thread: OrchestrationV2AppThread = {
...projection.thread,
Expand All @@ -6909,7 +6911,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
// Background wakes and delegated completions continue work the user already snoozed, so
// only a message someone actually sent ends the snooze.
const automaticWake =
command.notification !== undefined || command.delegatedCompletion !== undefined;
command.notification !== undefined ||
command.delegatedCompletion !== undefined ||
serverNotice;
if (projection.thread.snoozedUntil != null && !automaticWake) {
const now = yield* DateTime.now;
const thread: OrchestrationV2AppThread = {
Expand Down
3 changes: 2 additions & 1 deletion apps/server/src/orchestration-v2/QueuedSendCancelNotice.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,8 @@ import { runStopLabel } from "./RunStop.ts";

/**
* Tells an agent when a message it sent to another thread was cancelled before it ran, so a
* sender never assumes delivery. The notice is posted, not queued: it informs without a turn.
* sender never assumes delivery. The notice is posted, not queued: it informs without a turn, and
* a settled or snoozed sender stays that way (server notices reopen nothing; see Orchestrator).
*/
export class QueuedSendCancelNotice extends Context.Service<
QueuedSendCancelNotice,
Expand Down
40 changes: 40 additions & 0 deletions apps/server/src/orchestration-v2/RunStop.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,25 @@ it.effect(
});
assert.deepEqual(stopOf(rewritten), [ownerStop]);

// A message that replaces the running turn asks for the interrupt; its reason is held too.
const replaced = run(5);
stored.set(replaced.id, replaced);
yield* stamp([interruptRequest(replaced.id)], { commandType: "message.dispatch" }).pipe(
Effect.provideService(CommandActor, owner),
);
assert.deepEqual(
stopOf(yield* stamp([runUpdated({ ...replaced, status: "interrupted" })], {})),
[{ ...ownerStop, reason: "replaced", commandType: "message.dispatch" }],
);

// An interrupt no Stop asked for, written outside any command, is the provider's.
const dropped = run(4);
stored.set(dropped.id, dropped);
assert.deepEqual(
stopOf(yield* stamp([runUpdated({ ...dropped, status: "interrupted" })], {})),
[{ reason: "provider_interrupted", actor: { kind: "server" } }],
);

// A run the queue holds again is no longer stopped.
const held = run(2, {
status: "queued",
Expand Down Expand Up @@ -192,4 +211,25 @@ it("labels a restart cancel with the run that continued it", () => {
"removed from the queue before it started by an agent (session ab3aa7f5) from thread mcp:8270",
);
assert.isUndefined(runStopDetail(run(77, { status: "completed" }), []));

// A finished run lost only the background work it left running; one with none is not continued.
const finished = run(80, {
status: "completed",
restartCancelledBackgroundWork: [{ kind: "shell", label: "vp test" }],
});
const resumed = run(81, { restartContinuationOfRunId: finished.id });
assert.equal(
runStopDetail(finished, [finished, resumed])?.label,
"completed; its background work was cut short by restart, continued as run 81",
);
const { restartCancelledBackgroundWork: _work, ...plain } = finished;
assert.isUndefined(runStopDetail(plain, [plain, resumed]));

// A continuation that failed before its prompt was taken reads as retried.
const failedStart = run(82, { status: "failed", restartContinuationOfRunId: finished.id });
const retry = run(83, { restartContinuationOfRunId: failedStart.id });
assert.equal(
runStopDetail(failedStart, [failedStart, retry])?.label,
"failed to start; retried as run 83",
);
});
57 changes: 45 additions & 12 deletions apps/server/src/orchestration-v2/RunStop.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ const REVIVED: ReadonlySet<OrchestrationV2Run["status"]> = new Set([
"preparing",
"starting",
]);
const STOP_COMMANDS = new Set(["thread.stop", "run.interrupt", "subagent.stop"]);
const SUPERSEDED_COMMAND_ID = /^(command:system:stale-question:|server:delegated-task-question)/;

export function runStopReason(context: RunStopContext): OrchestrationV2RunStopReason {
Expand Down Expand Up @@ -114,7 +113,8 @@ export const makeRunStopStamper = <E, R>(
events: ReadonlyArray<OrchestrationV2DomainEvent>,
context: RunStopContext,
) {
if (STOP_COMMANDS.has(context.commandType ?? "")) {
// Any command can ask for an interrupt: a Stop, or a message or steer that replaces the turn.
if (context.commandType !== undefined) {
for (const event of events) {
if (event.type === "turn-item.updated" && event.payload.type === "run_interrupt_request") {
if (event.payload.runId !== null) {
Expand Down Expand Up @@ -147,10 +147,14 @@ export const makeRunStopStamper = <E, R>(
);
continue;
}
// The provider's confirmation, or the effect worker's settle after a Stop, has no actor.
const reason = runStopReason(context);
// An interrupt written outside any command came from the provider or its runtime, unless a
// Stop asked for it: the provider's confirmation and the worker's settle carry no actor.
const reason =
context.commandType === undefined && run.status === "interrupted"
? "provider_interrupted"
: runStopReason(context);
const pending =
(reason === "stopped" || reason === "other") &&
(reason === "stopped" || reason === "provider_interrupted" || reason === "other") &&
Option.isNone(yield* Effect.serviceOption(CommandActor))
? requested.get(run.id)
: undefined;
Expand All @@ -166,6 +170,7 @@ const REASON_TEXT: Record<OrchestrationV2RunStopReason, string> = {
server_shutdown: "cancelled by restart",
server_restart: "cancelled by restart",
stopped: "stopped",
provider_interrupted: "interrupted by the provider",
queue_cancelled: "removed from the queue before it started",
steered: "delivered into the running turn instead",
replaced: "replaced by a newer message",
Expand All @@ -189,16 +194,24 @@ const actorText = new Map<string, string>(Object.entries(ACTOR_TEXT));

/** One line that says why a run stopped and where its work went, for thread reads. */
export function runStopLabel(input: {
readonly status?: OrchestrationV2Run["status"];
readonly stop: OrchestrationV2RunStop | undefined;
readonly continuedAs: Pick<OrchestrationV2Run, "id" | "ordinal"> | undefined;
}): string {
const { stop, continuedAs } = input;
// A continuation that failed before its prompt was taken is retried as another run.
if (input.status === "failed" && continuedAs !== undefined) {
return `failed to start; retried as run ${continuedAs.ordinal}`;
}
// A finished or waiting run a restart continued lost only background work it left running.
const reason =
stop === undefined
? continuedAs === undefined
? "stopped; no reason was recorded"
: "cut short by restart"
: (reasonText.get(stop.reason) ?? stop.reason.replaceAll("_", " "));
input.status !== undefined && !STOPPED.has(input.status)
? `${input.status}; its background work was cut short by restart`
: stop === undefined
? continuedAs === undefined
? "stopped; no reason was recorded"
: "cut short by restart"
: (reasonText.get(stop.reason) ?? stop.reason.replaceAll("_", " "));
const actor =
stop === undefined ||
stop.reason === "server_shutdown" ||
Expand All @@ -211,14 +224,34 @@ export function runStopLabel(input: {
: `${reason}${actor}, continued as run ${continuedAs.ordinal}`;
}

/**
* Runs a restart can continue: cancelled ones, finished ones whose background work it cut, and a
* continuation that failed before its prompt was taken (retried). Reads look for a continuation
* only for these, so every read reports the same thing.
*/
export function mayHaveRestartContinuation(
run: Pick<
OrchestrationV2Run,
"status" | "restartCancelledBackgroundWork" | "restartContinuationOfRunId"
>,
): boolean {
return (
run.status === "cancelled" ||
(run.status === "failed" && run.restartContinuationOfRunId !== undefined) ||
(run.restartCancelledBackgroundWork?.length ?? 0) > 0
);
}

/** Shown on a cancelled or interrupted run, or one a restart continued. */
export function runStopDetail(run: OrchestrationV2Run, runs: ReadonlyArray<OrchestrationV2Run>) {
const continuedAs = runs.find((candidate) => candidate.restartContinuationOfRunId === run.id);
const continuedAs = mayHaveRestartContinuation(run)
? runs.find((candidate) => candidate.restartContinuationOfRunId === run.id)
: undefined;
if (!STOPPED.has(run.status) && continuedAs === undefined) return undefined;
return {
...run.stop,
...(continuedAs === undefined ? {} : { continuedAsRunId: continuedAs.id }),
label: runStopLabel({ stop: run.stop, continuedAs }),
label: runStopLabel({ status: run.status, stop: run.stop, continuedAs }),
};
}

Expand Down
25 changes: 25 additions & 0 deletions apps/server/src/orchestration-v2/runtimeLayer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1101,6 +1101,31 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => {
expect(projection.turnItems.filter((item) => item.type === "user_message")).toMatchObject([
{ messageId: command.messageId, postOnly: true, inputIntent: "post" },
]);

// A notice the server posts for the record leaves a settled, snoozed thread as it was.
yield* orchestrator.dispatch({
type: "thread.settle",
commandId: CommandId.make("post-settle-again"),
threadId,
});
yield* orchestrator.dispatch({
type: "thread.snooze",
commandId: CommandId.make("post-snooze-again"),
threadId,
snoozedUntil: "2099-01-01T00:00:00.000Z",
});
yield* orchestrator.dispatch({
...command,
createdBy: "agent",
creationSource: "server",
commandId: CommandId.make("post-server-notice"),
messageId: MessageId.make("post-server-notice"),
text: "Your message was not delivered.",
});
const noticed = yield* orchestrator.getThreadProjection(threadId);
expect(noticed.thread.settledOverride).toBe("settled");
expect(noticed.thread.snoozedUntil).not.toBeNull();
expect(noticed.messages.map((message) => message.id)).toContain("post-server-notice");
}),
);

Expand Down
2 changes: 2 additions & 0 deletions packages/contracts/src/mcpEvents.ts
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,8 @@ export const McpEventData = Schema.Struct({
state: Schema.String,
runId: Schema.optionalKey(Schema.String),
resourceId: Schema.optionalKey(Schema.String),
/** Why a cancelled or interrupted run stopped, e.g. `server_shutdown` for a restart. */
stopReason: Schema.optionalKey(Schema.String),
parentThreadId: Schema.NullOr(ThreadId),
relationshipToParent: Schema.NullOr(Schema.Literals(["fork", "subagent"])),
});
Expand Down
1 change: 1 addition & 0 deletions packages/contracts/src/orchestrationV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -720,6 +720,7 @@ export const OrchestrationV2RunStopReason = Schema.Literals([
"server_shutdown",
"server_restart",
"stopped",
"provider_interrupted",
"queue_cancelled",
"steered",
"replaced",
Expand Down
Loading