@@ -23,6 +23,7 @@ import {
2323 type ChatAttachment ,
2424 CommandId ,
2525 type GuestMessageId ,
26+ isCompactionCommandMessage ,
2627 isOrchestrationV2WorkActive ,
2728 isProviderNativeSubagentThread ,
2829 MessageId ,
@@ -5774,13 +5775,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
57745775 } ) ;
57755776 }
57765777 if ( goalStopRunId !== undefined ) {
5777- return yield * stopRunForGoalCommand ( {
5778+ return yield * stopRunForQueuedCommand ( {
57785779 command,
57795780 events,
57805781 effects,
57815782 projection : yield * getProjectionWithPendingEvents ( command . threadId , events ) ,
57825783 queuedRunId : runId ,
57835784 targetRunId : goalStopRunId ,
5785+ reason : "Stopped to apply /goal" ,
57845786 now,
57855787 } ) ;
57865788 }
@@ -8355,15 +8357,46 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83558357 ( yield * goalDriverFor ( command , queuedRun . providerInstanceId ) ) === undefined
83568358 ? null
83578359 : parsedGoalCommand ;
8360+ const target = projection . runs . find ( ( run ) => run . id === command . targetRunId ) ;
8361+ const targetMessage = projection . messages . find (
8362+ ( message ) => message . id === target ?. userMessageId ,
8363+ ) ;
8364+ // A compacting turn cannot take steering input, so steering skips the compaction instead.
8365+ if (
8366+ target !== undefined &&
8367+ targetMessage !== undefined &&
8368+ isCompactionCommandMessage ( targetMessage )
8369+ ) {
8370+ if ( goalCommand !== null ) {
8371+ yield * ensureGoalCommandSupported ( command , queuedMessage , queuedRun . providerInstanceId ) ;
8372+ }
8373+ yield * emitEvent ( {
8374+ type : "run.updated" ,
8375+ threadId : command . threadId ,
8376+ runId : queuedRun . id ,
8377+ nodeId : queuedRootNode . id ,
8378+ providerInstanceId : queuedRun . providerInstanceId ,
8379+ occurredAt : now ,
8380+ payload : { ...queuedRun , queuePosition : 1 } ,
8381+ } ) ;
8382+ const stops =
8383+ isLiveRun ( target ) && ! ( yield * stopReachedRun ( command , command . threadId , target . id ) ) ;
8384+ return yield * stopRunForQueuedCommand ( {
8385+ command,
8386+ events,
8387+ effects,
8388+ projection,
8389+ queuedRunId : queuedRun . id ,
8390+ targetRunId : stops ? target . id : undefined ,
8391+ reason : "Skipped compaction to send a queued message" ,
8392+ now,
8393+ } ) ;
8394+ }
83588395 if ( goalCommand !== null ) {
83598396 yield * ensureGoalCommandSupported ( command , queuedMessage , queuedRun . providerInstanceId ) ;
8360- const target = projection . runs . find ( ( run ) => run . id === command . targetRunId ) ;
83618397 const goal = projection . providerThreads . find (
83628398 ( candidate ) => candidate . id === target ?. providerThreadId ,
83638399 ) ?. goal ;
8364- const targetMessage = projection . messages . find (
8365- ( message ) => message . id === target ?. userMessageId ,
8366- ) ;
83678400 yield * emitEvent ( {
83688401 type : "run.updated" ,
83698402 threadId : command . threadId ,
@@ -8379,13 +8412,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83798412 goalCommandStopsActiveRun ( goalCommand , goal ?. status ?? null ) &&
83808413 ( targetMessage === undefined || ! isNativeMaintenanceCommand ( targetMessage ) ) &&
83818414 ! ( yield * stopReachedRun ( command , command . threadId , target . id ) ) ;
8382- return yield * stopRunForGoalCommand ( {
8415+ return yield * stopRunForQueuedCommand ( {
83838416 command,
83848417 events,
83858418 effects,
83868419 projection,
83878420 queuedRunId : queuedRun . id ,
83888421 targetRunId : stops ? target . id : undefined ,
8422+ reason : "Stopped to apply /goal" ,
83898423 now,
83908424 } ) ;
83918425 }
@@ -10496,71 +10530,74 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1049610530 ) ;
1049710531
1049810532 /** Puts a queued goal command first and, given a target, stops the run in its way. */
10499- const stopRunForGoalCommand = Effect . fn ( "Orchestrator.stopRunForGoalCommand" ) ( function * ( input : {
10500- readonly command : Extract <
10501- OrchestrationV2Command ,
10502- { readonly type : "message.dispatch" | "queued-message.promote-to-steer" }
10503- > ;
10504- readonly events : Ref . Ref < Array < OrchestrationV2DomainEvent > > ;
10505- readonly effects : Ref . Ref < Array < PendingOrchestrationEffectV2 > > ;
10506- readonly projection : Pick < OrchestrationV2ThreadProjection , "runs" | "messages" > ;
10507- readonly queuedRunId : RunId ;
10508- readonly targetRunId : RunId | undefined ;
10509- readonly now : DateTime . Utc ;
10510- } ) {
10511- const { command, events, effects, projection, now, targetRunId } = input ;
10512- const others = queuedRunsInDeliveryOrder ( projection ) . filter (
10513- ( run ) => run . id !== input . queuedRunId && ! isAutomaticCompletionRun ( projection , run ) ,
10514- ) ;
10515- for ( const [ index , run ] of others . entries ( ) ) {
10516- const queuePosition = index + 2 ;
10517- if ( run . queuePosition === queuePosition ) continue ;
10518- yield * emit (
10519- events ,
10520- command ,
10521- ) ( {
10522- type : "run.updated" ,
10523- threadId : command . threadId ,
10524- runId : run . id ,
10525- ...( run . rootNodeId === null ? { } : { nodeId : run . rootNodeId } ) ,
10526- providerInstanceId : run . providerInstanceId ,
10527- occurredAt : now ,
10528- payload : { ...run , queuePosition } ,
10529- } ) ;
10530- }
10531- if ( targetRunId === undefined ) return undefined ;
10532- const interruptEvents = yield * Ref . make < Array < OrchestrationV2DomainEvent > > ( [ ] ) ;
10533- const interruptEffects = yield * Ref . make < Array < PendingOrchestrationEffectV2 > > ( [ ] ) ;
10534- const interrupted = yield * Effect . exit (
10535- dispatchRunInterrupt (
10536- {
10537- type : "run.interrupt" ,
10538- commandId : command . commandId ,
10533+ const stopRunForQueuedCommand = Effect . fn ( "Orchestrator.stopRunForQueuedCommand" ) (
10534+ function * ( input : {
10535+ readonly command : Extract <
10536+ OrchestrationV2Command ,
10537+ { readonly type : "message.dispatch" | "queued-message.promote-to-steer" }
10538+ > ;
10539+ readonly events : Ref . Ref < Array < OrchestrationV2DomainEvent > > ;
10540+ readonly effects : Ref . Ref < Array < PendingOrchestrationEffectV2 > > ;
10541+ readonly projection : Pick < OrchestrationV2ThreadProjection , "runs" | "messages" > ;
10542+ readonly queuedRunId : RunId ;
10543+ readonly targetRunId : RunId | undefined ;
10544+ readonly reason : string ;
10545+ readonly now : DateTime . Utc ;
10546+ } ) {
10547+ const { command, events, effects, projection, now, targetRunId } = input ;
10548+ const others = queuedRunsInDeliveryOrder ( projection ) . filter (
10549+ ( run ) => run . id !== input . queuedRunId && ! isAutomaticCompletionRun ( projection , run ) ,
10550+ ) ;
10551+ for ( const [ index , run ] of others . entries ( ) ) {
10552+ const queuePosition = index + 2 ;
10553+ if ( run . queuePosition === queuePosition ) continue ;
10554+ yield * emit (
10555+ events ,
10556+ command ,
10557+ ) ( {
10558+ type : "run.updated" ,
10559+ threadId : command . threadId ,
10560+ runId : run . id ,
10561+ ...( run . rootNodeId === null ? { } : { nodeId : run . rootNodeId } ) ,
10562+ providerInstanceId : run . providerInstanceId ,
10563+ occurredAt : now ,
10564+ payload : { ...run , queuePosition } ,
10565+ } ) ;
10566+ }
10567+ if ( targetRunId === undefined ) return undefined ;
10568+ const interruptEvents = yield * Ref . make < Array < OrchestrationV2DomainEvent > > ( [ ] ) ;
10569+ const interruptEffects = yield * Ref . make < Array < PendingOrchestrationEffectV2 > > ( [ ] ) ;
10570+ const interrupted = yield * Effect . exit (
10571+ dispatchRunInterrupt (
10572+ {
10573+ type : "run.interrupt" ,
10574+ commandId : command . commandId ,
10575+ threadId : command . threadId ,
10576+ runId : targetRunId ,
10577+ reason : input . reason ,
10578+ } ,
10579+ interruptEvents ,
10580+ interruptEffects ,
10581+ ) ,
10582+ ) ;
10583+ // A failed interrupt may have written part of its events, so they are kept apart.
10584+ // The queued command then waits for the turn to end.
10585+ if ( Exit . isFailure ( interrupted ) ) {
10586+ yield * Effect . logWarning ( "queued command could not stop the running turn" , {
1053910587 threadId : command . threadId ,
1054010588 runId : targetRunId ,
10541- reason : "Stopped to apply /goal" ,
10542- } ,
10543- interruptEvents ,
10544- interruptEffects ,
10545- ) ,
10546- ) ;
10547- // A failed interrupt may have written part of its events, so they are kept apart.
10548- // The goal command then waits for the turn to end.
10549- if ( Exit . isFailure ( interrupted ) ) {
10550- yield * Effect . logWarning ( "goal command could not stop the running turn" , {
10551- threadId : command . threadId ,
10552- runId : targetRunId ,
10553- cause : interrupted . cause ,
10554- } ) ;
10555- return undefined ;
10556- }
10557- const written = yield * Ref . get ( interruptEvents ) ;
10558- const enqueued = yield * Ref . get ( interruptEffects ) ;
10559- yield * Ref . update ( events , ( existing ) => [ ...existing , ...written ] ) ;
10560- yield * Ref . update ( effects , ( existing ) => [ ...existing , ...enqueued ] ) ;
10561- // Like Stop: a run stopped before its provider started also abandons that start.
10562- return interrupted . value ;
10563- } ) ;
10589+ cause : interrupted . cause ,
10590+ } ) ;
10591+ return undefined ;
10592+ }
10593+ const written = yield * Ref . get ( interruptEvents ) ;
10594+ const enqueued = yield * Ref . get ( interruptEffects ) ;
10595+ yield * Ref . update ( events , ( existing ) => [ ...existing , ...written ] ) ;
10596+ yield * Ref . update ( effects , ( existing ) => [ ...existing , ...enqueued ] ) ;
10597+ // Like Stop: a run stopped before its provider started also abandons that start.
10598+ return interrupted . value ;
10599+ } ,
10600+ ) ;
1056410601
1056510602 const dispatchThreadStop = Effect . fn ( "Orchestrator.dispatchThreadStop" ) ( function * (
1056610603 command : Extract < OrchestrationV2ServerCommand , { readonly type : "thread.stop" } > ,
0 commit comments