diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index e426eece00654..78bf5dcce18d7 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -976,24 +976,27 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); it.effect.each([ - { before: "running", after: "failed", capture: false }, - { before: "waiting", after: "completed", capture: true }, + { command: "acknowledge", before: "running", after: "failed", capture: false }, + { command: "acknowledge", before: "waiting", after: "completed", capture: true }, + { command: "accept", before: "waiting", after: "completed", capture: true }, ] as const)( - "keeps a $after run that settled between task_status's acknowledge read and commit", - ({ before, after, capture }) => + "keeps a $after run that settled between $command's read and commit", + ({ command, before, after, capture }) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; const now = yield* DateTime.now; - const threadId = ThreadId.make(`thread:ack-settle-race-${after}`); - const runId = RunId.make(`run:ack-settle-race-${after}`); - const taskId = NodeId.make(`node:ack-settle-race-${after}-task`); - const checkpointId = CheckpointId.make(`checkpoint:ack-settle-race-${after}`); + const race = `${command}-settle-race-${after}`; + const threadId = ThreadId.make(`thread:${race}`); + const runId = RunId.make(`run:${race}`); + const taskId = NodeId.make(`node:${race}-task`); + const checkpointId = CheckpointId.make(`checkpoint:${race}`); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); yield* seedParentWithTerminalTask({ threadId, - projectId: ProjectId.make(`project:ack-settle-race-${after}`), + projectId: ProjectId.make(`project:${race}`), runId, - rootNodeId: NodeId.make(`node:ack-settle-race-${after}-root`), + rootNodeId: NodeId.make(`node:${race}-root`), taskId, deliveryState: "claimed", deliveryTaskIds: [taskId], @@ -1003,13 +1006,35 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { yield* sink.write({ events: [ { - id: EventId.make(`event:ack-settle-race-${after}-before`), + id: EventId.make(`event:${race}-before`), type: "run.updated", threadId, runId, occurredAt: now, payload: { ...parent, status: before }, }, + { + id: EventId.make(`event:${race}-message`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: parent.rootNodeId, + role: "user", + text: "Background task finished", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, + }, + }, ], }); const { delegatedCompletion: _cohort, ...settledRun } = parent; @@ -1021,19 +1046,22 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { Effect.gen(function* () { if ( !settled && - input.commandType === "delegated_task.completion-delivery.acknowledge" + input.commandType === + (command === "accept" + ? "notification.delivery.accept" + : "delegated_task.completion-delivery.acknowledge") ) { settled = true; - // The provider turn ends (or its checkpoint lands) after acknowledge read the run. + // The provider turn ends (or its checkpoint lands) after the command read the run. yield* commit({ - commandId: CommandId.make(`command:ack-settle-race-${after}`), + commandId: CommandId.make(`command:${race}`), threadId, commandType: capture ? "checkpoint.capture" : "provider.event", acceptedAt: now, effects: [], events: [ { - id: EventId.make(`event:ack-settle-race-${after}-settled`), + id: EventId.make(`event:${race}-settled`), type: "run.updated", threadId, runId, @@ -1053,19 +1081,31 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ), ), () => - orchestrator.dispatch({ - type: "delegated_task.completion-delivery.acknowledge", - commandId: CommandId.make(`command:ack-settle-race-${after}:task-status`), - parentThreadId: threadId, - taskId, - observedByRunId: runId, - }), + orchestrator.dispatch( + command === "accept" + ? { + type: "notification.delivery.accept", + commandId: CommandId.make(`command:${race}:accept`), + threadId, + messageId, + } + : { + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:${race}:task-status`), + parentThreadId: threadId, + taskId, + observedByRunId: runId, + }, + ), (spy) => Effect.sync(() => spy.mockRestore()), ); assert.isTrue(settled); - const run = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (candidate) => candidate.id === runId, - )!; + const finished = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + command === "accept" ? "delivered" : "acknowledged", + ); + const run = finished.runs.find((candidate) => candidate.id === runId)!; assert.deepEqual( { status: run.status, completedAt: run.completedAt, checkpointId: run.checkpointId }, { status: after, completedAt: now, checkpointId: capture ? checkpointId : null }, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ee8ab83b952aa..716bdc5f9df1c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11733,6 +11733,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, }, }); + return parentRun.id; }); const dispatchUnsupported = (command: OrchestrationV2ServerCommand) => @@ -11907,7 +11908,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio break; } case "notification.delivery.accept": - yield* dispatchNotificationAccepted(command, events); + updateDelegatedCompletionOnlyRunId = yield* dispatchNotificationAccepted(command, events); break; case "prepared-run.release": yield* dispatchPreparedRunRelease(command, events, effects);