Skip to content

Commit 9dfb8f6

Browse files
PR Batch Testerclaude
andcommitted
fix(server): a Stop that writes no events still stops delegated tasks
Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 8e27cc4 commit 9dfb8f6

3 files changed

Lines changed: 51 additions & 2 deletions

File tree

‎apps/server/src/orchestration-v2/EventSink.ts‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -661,7 +661,12 @@ const layerBase: Layer.Layer<
661661
commandId: input.commandId,
662662
events: normalized,
663663
});
664-
const sequence = storedEvents.at(-1)?.sequence;
664+
// A Stop that changes nothing can still queue the stop of its delegated tasks.
665+
const sequence =
666+
storedEvents.at(-1)?.sequence ??
667+
(input.effects.length === 0
668+
? undefined
669+
: yield* eventStore.latestSequence({ threadId: input.threadId }));
665670
if (sequence === undefined) {
666671
return yield* Effect.die(
667672
new Error(`Command ${input.commandId} produced no orchestration events.`),

‎apps/server/src/orchestration-v2/Orchestrator.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11720,7 +11720,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1172011720
),
1172111721
);
1172211722

11723-
if (plan.events.length === 0) {
11723+
if (plan.events.length === 0 && plan.effects.length === 0) {
1172411724
// A settle that ended nothing still records its receipt: a replayed Stop
1172511725
// effect then finds it instead of settling work that appeared since.
1172611726
const resultSequence = yield* Effect.gen(function* () {

‎apps/server/src/orchestration-v2/ThreadStop.test.ts‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -607,6 +607,50 @@ it.effect.each(["run.interrupt", "thread.stop"] as const)(
607607
}).pipe(Effect.provide(layerTest)),
608608
);
609609

610+
it.effect("a repeated Stop that changes nothing still stops the delegated tasks", () =>
611+
Effect.gen(function* () {
612+
const orchestrator = yield* Orchestrator.OrchestratorV2;
613+
const projections = yield* ProjectionStore.ProjectionStoreV2;
614+
const sql = yield* SqlClient.SqlClient;
615+
const parentThreadId = ThreadId.make("thread:stop-again");
616+
yield* createWatchingThread(parentThreadId, 1);
617+
yield* send(parentThreadId, "first", "start_immediately");
618+
yield* delegate(parentThreadId, "stop-again child");
619+
const now = yield* DateTime.now;
620+
const run = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
621+
yield* projections.apply({
622+
id: EventId.make("event:stop-again:run-completed"),
623+
type: "run.updated",
624+
threadId: parentThreadId,
625+
runId: run.id,
626+
occurredAt: now,
627+
payload: { ...run, status: "completed", completedAt: now },
628+
});
629+
yield* orchestrator.dispatch({
630+
type: "thread.stop",
631+
commandId: CommandId.make("stop-again:first"),
632+
threadId: parentThreadId,
633+
});
634+
635+
// The first Stop already disposed the delivery and ended the watch, so this one writes no
636+
// events, like Stop on a thread an older server stopped without stopping its children.
637+
const again = yield* orchestrator.dispatch({
638+
type: "thread.stop",
639+
commandId: CommandId.make("stop-again:second"),
640+
threadId: parentThreadId,
641+
});
642+
assert.lengthOf(again.storedEvents, 0);
643+
const walks = yield* sql<{ readonly command_id: string }>`
644+
SELECT command_id FROM orchestration_v2_effect_outbox
645+
WHERE effect_type = 'delegated-tasks.stop'
646+
`;
647+
assert.deepEqual(
648+
walks.map((row) => row.command_id),
649+
["stop-again:first", "stop-again:second"],
650+
);
651+
}).pipe(Effect.provide(layerTest)),
652+
);
653+
610654
it.effect.each(["thread.stop", "run.interrupt"] as const)(
611655
"%s cancels buffered wakes and preserves durable queued messages",
612656
(type) =>

0 commit comments

Comments
 (0)