@@ -53,6 +53,7 @@ import { chooseBackend, type BackendPath } from "../core/backend.js";
5353import { OllamaBrain } from "../core/brain_ollama.js" ;
5454import { isLocalModelId , localModelId , ollamaTagFromId , resolveHostedModel , resolveLocalModel } from "../core/local_ollama.js" ;
5555import type { Brain } from "../core/brain.js" ;
56+ import { SteerChannel , formatSteerAck } from "../core/steer_channel.js" ;
5657import type { RunOptions , ToolResult } from "../core/tool_executor.js" ;
5758import { ToolExecutor } from "../core/tool_executor.js" ;
5859import { ConsoleShell , classifyConsoleInput , type ConsoleInput } from "./console_input.js" ;
@@ -113,8 +114,13 @@ export interface TurnSkillOptions {
113114 exec ?: ToolExecutor ;
114115 /** Host-observed results only; no model text or shell output. */
115116 onToolResult ?: ( tool : ObservedTool ) => void ;
117+ /** Console /steer routing for this turn (#283). Never serialized to Cloud. */
118+ steer ?: SteerChannel ;
116119}
117120
121+ /** Why a hosted chat turn only takes steering for the next turn (#283). */
122+ export const HOSTED_STEER_DEFERRED = "this hosted chat route has no live control acknowledgement" ;
123+
118124export const DEFAULT_CHAT_TURN_DEADLINE_MS = 30 * 60_000 ;
119125/** A positive override is useful for tests and operators; 0 cannot disable it. */
120126export function chatTurnDeadlineMs ( env : Readonly < Record < string , string | undefined > > = process . env ) : number {
@@ -519,8 +525,11 @@ export async function runTurn(
519525 getRegistry ( ) . markLocalUnmetered ( ) ;
520526 // The signal used to be dropped here, so the REPL Ctrl+C controller could
521527 // not reach a local turn at all: the abort fired and nothing observed it.
522- return await runLocalTurn ( ctx , brief , boundedSignal . signal , { lifecycle, onPulsePaint, deadlineAt, ...( skillOpts . exec ? { exec : skillOpts . exec } : { } ) , ...( skillOpts . onToolResult ? { onToolResult : skillOpts . onToolResult } : { } ) } , run . guard ) ;
528+ return await runLocalTurn ( ctx , brief , boundedSignal . signal , { lifecycle, onPulsePaint, deadlineAt, ...( skillOpts . exec ? { exec : skillOpts . exec } : { } ) , ...( skillOpts . onToolResult ? { onToolResult : skillOpts . onToolResult } : { } ) , ... ( skillOpts . steer ? { steer : skillOpts . steer } : { } ) } , run . guard ) ;
523529 }
530+ // /agent/chat/stream exposes no control acknowledgement, so a steer typed
531+ // during this turn is kept for the next one rather than reported as live.
532+ skillOpts . steer ?. markNextTurnOnly ( HOSTED_STEER_DEFERRED ) ;
524533 // The cloud REPL turn streams from /agent/chat/stream, where the SERVER runs
525534 // the tools. This host executes nothing on that path, so it can enforce
526535 // nothing on it either. Say so rather than let the Policy line above read as
@@ -807,6 +816,13 @@ export interface LocalTurnDeps {
807816 * workspace. `false` disables it for a deliberate embed. */
808817 failureBudget ?: ToolFailureBudget | false ;
809818 onToolResult ?: ( tool : ObservedTool ) => void ;
819+ /** Live /steer routing: attached once the brain runs (#283). */
820+ steer ?: SteerChannel ;
821+ }
822+
823+ /** What the brain is told when the host refuses a call selected before a steer. */
824+ export function staleAfterSteerResult ( name : string ) : ToolResult {
825+ return { output : `[tool ${ name } not executed: superseded by operator steering accepted before it ran]` , exitCode : 1 } ;
810826}
811827
812828export async function runLocalTurn (
@@ -879,6 +895,7 @@ export async function runLocalTurn(
879895 try {
880896 pulse . start ( ) ;
881897 iterator = brain . run ( task ) [ Symbol . asyncIterator ] ( ) ;
898+ deps . steer ?. attach ( brain ) ;
882899 for ( ; ; ) {
883900 const next = await boundedLocalOperation (
884901 ( ) => iterator ! . next ( ) ,
@@ -936,6 +953,22 @@ export async function runLocalTurn(
936953 const note = origin ? failures ?. repeatNote ( key ) : null ;
937954 brain . sendToolResult ( ev . id , note && result . exitCode !== 0 ? { ...result , output : `${ result . output } \n${ note } ` } : result ) ;
938955 } ;
956+ // A steer the brain accepted but has not yet put in front of the
957+ // model means this call was selected before it. It is answered
958+ // without prompting or executing, and counts as no failure (#283).
959+ const supersededBySteer = ( ) : boolean => {
960+ if ( ! deps . steer ?. hasPendingSteer ( ) ) return false ;
961+ deps . steer . noteToolSkipped ( ev . name ) ;
962+ deliver ( staleAfterSteerResult ( ev . name ) , null ) ;
963+ return true ;
964+ } ;
965+ if ( supersededBySteer ( ) ) {
966+ lastMeaningfulAt = Date . now ( ) ;
967+ modelOutput . reset ( ) ;
968+ noteStreamingActivity ( lifecycle ) ;
969+ pulse . start ( ) ;
970+ continue ;
971+ }
939972 // Same repeated-failure budget as hostLoop, checked before any prompt
940973 // or execution: a spent operation is refused once, then stops the turn.
941974 failures ?. noteModelRound ( brain . modelRound ?.( ) ) ;
@@ -1002,13 +1035,16 @@ export async function runLocalTurn(
10021035 } ;
10031036 if ( ! approved ) {
10041037 deliver ( { output : `[tool ${ ev . name } blocked: permission denied]` , exitCode : 1 } , "approval" ) ;
1005- } else {
1038+ } else if ( ! supersededBySteer ( ) ) {
1039+ // Approval can take a while; a steer accepted during the prompt is
1040+ // checked above, so an approved-but-stale write still never runs.
10061041 let result = await execute ( ) ;
10071042 // Read-only transient failures only; a mutation is never replayed.
10081043 for ( let retry = 0 ; retry < TRANSIENT_READ_AUTO_RETRIES ; retry += 1 ) {
1009- if ( controller . signal . aborted || ! hostMayRetry ( ev . name , classifyToolFailure ( ev . name , result ) ) ) break ;
1044+ if ( controller . signal . aborted || deps . steer ?. hasPendingSteer ( ) || ! hostMayRetry ( ev . name , classifyToolFailure ( ev . name , result ) ) ) break ;
10101045 result = await execute ( ) ;
10111046 }
1047+ deps . steer ?. noteToolFinished ( ev . name ) ;
10121048 deliver ( result , "execution" ) ;
10131049 }
10141050 }
@@ -1043,6 +1079,13 @@ export async function runLocalTurn(
10431079 if ( terminalError ) throw terminalError ;
10441080 const settled = lifecycle . outcome ;
10451081 if ( settled ) return settled ;
1082+ // Ctrl+C while the model is generating closes the brain, which ends its
1083+ // stream without a done event. That is a cancellation, not a stream the
1084+ // server cut short, and must settle the same way a cancelled tool wait does.
1085+ const cancelReason : unknown = signal ?. aborted ? signal . reason ?? new DOMException ( "turn cancelled" , "AbortError" ) : null ;
1086+ if ( cancelReason && isAbortError ( cancelReason ) ) {
1087+ return finalizeThrownTurn ( lifecycle , cancelReason , ctx . cfg . baseUrl , partialOutput ) ;
1088+ }
10461089 const incomplete = new StreamIncompleteError ( ) ;
10471090 lifecycle . finalize ( "incomplete" , {
10481091 message : incomplete . message ,
@@ -1096,15 +1139,17 @@ function switchDisposition(target: ModelTarget, brief: string, queueCount: numbe
10961139}
10971140
10981141/** Build a prompt with optional steering and btw context prepended.
1099- * Clears steering and btwNotes in the returned result so callers
1100- * can use single-shot semantics. Exported for testing. */
1142+ * Several steering notes keep their order, one line each. Clears steering
1143+ * and btwNotes in the returned result so callers can use single-shot
1144+ * semantics. Exported for testing. */
11011145export function buildPromptContext (
11021146 base : string ,
1103- steering : string | null ,
1147+ steering : string | readonly string [ ] | null ,
11041148 btwNotes : string [ ] ,
11051149) : { prompt : string ; steering : string | null ; btwNotes : string [ ] } {
11061150 const ctxParts : string [ ] = [ ] ;
1107- if ( steering ) ctxParts . push ( `STEERING: ${ steering } ` ) ;
1151+ const steers = steering === null ? [ ] : typeof steering === "string" ? [ steering ] : steering ;
1152+ for ( const steer of steers ) if ( steer ) ctxParts . push ( `STEERING: ${ steer } ` ) ;
11081153 if ( btwNotes . length ) ctxParts . push ( `NOTE: ${ btwNotes . join ( "; " ) } ` ) ;
11091154 const prompt = ctxParts . length ? ctxParts . join ( "\n" ) + "\n\n" + base : base ;
11101155 return { prompt, steering : null , btwNotes : [ ] } ;
@@ -1310,7 +1355,12 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
13101355 let pasteAcc = "" ;
13111356 let carry = "" ; // partial escape sequence held across chunk boundaries
13121357 const queue : ConsoleInput [ ] = [ ] ;
1313- let steering : string | null = null ;
1358+ // /steer during a turn goes to that turn's brain when it can acknowledge
1359+ // it; everything else waits for the next turn and is reported as such.
1360+ const steerChannel = new SteerChannel ( ( ack ) => {
1361+ process . stdout . write ( ( busy ? "\n" : "" ) + formatSteerAck ( ack ) + "\n" ) ;
1362+ if ( ! busy ) repaint ( ) ;
1363+ } ) ;
13141364 const btwNotes : string [ ] = [ ] ;
13151365 let turnAbort : AbortController | null = null ; // live while a local/cloud turn runs
13161366 // Live while a slash command (e.g. /audit, /doctor) is in flight — kept
@@ -1366,10 +1416,16 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
13661416 } finally { turnAbort = null ; }
13671417 }
13681418 const text = input . text ;
1369- const built = authContinuation ? { prompt : text , steering, btwNotes } : buildPromptContext ( text , steering , btwNotes ) ;
1419+ // An auth continuation replays the saved task verbatim; kept notes wait.
1420+ const carriedSteers = authContinuation ? [ ] : steerChannel . takeNextTurnNotes ( ) ;
1421+ const built = authContinuation ? { prompt : text } : buildPromptContext ( text , carriedSteers , btwNotes ) ;
13701422 const beforeChanges = observedWorkspaceChanges ( ctx . flags . cwd ) ;
13711423 const toolResults : ObservedTool [ ] = [ ] ;
1372- if ( ! authContinuation ) { steering = built . steering ; btwNotes . length = 0 ; }
1424+ if ( ! authContinuation ) btwNotes . length = 0 ;
1425+ const steerTurn = steerChannel . beginTurn ( ) ;
1426+ // How the turn ended, for its steering: a cancel drops unapplied notes
1427+ // (reported); a failure the operator may retry keeps the carried ones.
1428+ let steerEnd : "ended" | "cancelled" | "retryable" = "ended" ;
13731429 viewerState = createViewerState ( ) ;
13741430 viewerOpen = false ;
13751431 viewerLastLines = 0 ;
@@ -1381,6 +1437,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
13811437 if ( authRepair . submissionBlocked ) {
13821438 process . stdout . write ( "Account changed or could not be verified. Use /auth new before sending another hosted task.\n" ) ;
13831439 if ( ! buf . value ) buf . insert ( text ) ;
1440+ steerEnd = "retryable" ;
13841441 return "failed" ;
13851442 }
13861443 await authRepair . captureAccount ( ) ;
@@ -1443,13 +1500,14 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
14431500 viewerOpen = false ;
14441501 break ;
14451502 }
1446- } , redrawInput , { ...skillOpts , onToolResult : tool => toolResults . push ( tool ) } ) ;
1503+ } , redrawInput , { ...skillOpts , onToolResult : tool => toolResults . push ( tool ) , steer : steerChannel } ) ;
14471504 const afterChanges = observedWorkspaceChanges ( ctx . flags . cwd ) ;
14481505 continuation . recordTurn ( text , toolResults , afterChanges . filter ( change => ! beforeChanges . includes ( change ) ) , outcome . state , ! sharedShellResult && ! authContinuation ) ;
14491506 if ( ctx . flags . json ) process . stdout . write ( turnOutcomeJson ( outcome ) + "\n" ) ;
14501507 if ( outcome . state === "cancelled" ) {
14511508 queue . length = 0 ;
14521509 if ( ! ctx . flags . json ) process . stdout . write ( "\n" + theme . dim ( "✗ turn aborted" ) + "\n" ) ;
1510+ steerEnd = "cancelled" ;
14531511 return "aborted" ;
14541512 }
14551513 return "completed" ;
@@ -1462,6 +1520,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
14621520 const outcome = turnOutcomeForError ( err ) ;
14631521 if ( ctx . flags . json && outcome ) process . stdout . write ( turnOutcomeJson ( outcome ) + "\n" ) ;
14641522 else process . stdout . write ( "\n" + theme . dim ( "✗ turn aborted" ) + "\n" ) ;
1523+ steerEnd = "cancelled" ;
14651524 return "aborted" ;
14661525 }
14671526 // ChatTurnError means the Renderer already painted "✗ <msg>" for the
@@ -1493,9 +1552,18 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
14931552 buf . clear ( ) ;
14941553 buf . insert ( recovered ) ;
14951554 }
1555+ // A hosted auth failure saved the prompt with its notes already in it.
1556+ if ( ! authFailure ) steerEnd = "retryable" ;
14961557 return "failed" ;
14971558 } finally {
14981559 turnAbort = null ;
1560+ // Notes that never reached the model are reported either way: kept for
1561+ // the next turn, or — when the operator cancelled — not retained.
1562+ if ( steerEnd === "cancelled" ) steerChannel . cancelTurn ( ) ;
1563+ else steerChannel . endTurn ( ) ;
1564+ if ( steerEnd === "retryable" ) {
1565+ steerChannel . restoreNextTurnNotes ( carriedSteers , `turn ${ steerTurn } failed; kept for the retry` ) ;
1566+ }
14991567 }
15001568 } ;
15011569
@@ -1591,10 +1659,11 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
15911659 // ── mid-turn Enter: bypass commands + type-ahead queueing ──
15921660 if ( busy ) {
15931661 if ( t . startsWith ( "/steer " ) ) {
1594- steering = t . slice ( 7 ) . trim ( ) || steering ;
15951662 remember ( buf . value ) ;
15961663 buf . commit ( buf . value ) ;
1597- if ( steering ) process . stdout . write ( `\n🎯 Steering set: "${ steering } "\n` ) ;
1664+ // Acknowledged through the channel: accepted/applied by the running
1665+ // turn, refused, or deferred to the next turn — never assumed.
1666+ steerChannel . steer ( t . slice ( 7 ) ) . catch ( ( ) => { /* acks are best-effort output */ } ) ;
15981667 return ;
15991668 }
16001669 if ( t . startsWith ( "/btw " ) ) {
@@ -1675,9 +1744,9 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
16751744 if ( t . startsWith ( "/steer " ) || t === "/steer" ) {
16761745 const guidance = t . slice ( 6 ) . trim ( ) ;
16771746 if ( ! guidance ) { process . stdout . write ( "usage: /steer <guidance>\n" ) ; repaint ( ) ; return ; }
1678- steering = guidance ;
1679- process . stdout . write ( `🎯 Steering set: " ${ guidance } "\n` ) ;
1680- repaint ( ) ; return ;
1747+ // No turn is running: kept, in order, for the next one (the ack repaints).
1748+ steerChannel . steer ( guidance ) . catch ( ( ) => { /* acks are best-effort output */ } ) ;
1749+ return ;
16811750 }
16821751 if ( t . startsWith ( "/btw " ) || t === "/btw" ) {
16831752 const note = t . slice ( 4 ) . trim ( ) ;
@@ -1716,7 +1785,7 @@ export async function repl(ctx: AppContext, skillOpts: TurnSkillOptions = {}): P
17161785 } else process . stdout . write ( "No safely rejected task is ready. Use /auth status for details.\n" ) ;
17171786 } else if ( sub === "new" ) {
17181787 process . stdout . write ( authRepair . startNewConversation ( ) ) ;
1719- queue . length = 0 ; steering = null ; btwNotes . length = 0 ;
1788+ queue . length = 0 ; steerChannel . clearNextTurn ( ) ; btwNotes . length = 0 ;
17201789 } else if ( sub === "draft" ) {
17211790 if ( buf . value ) process . stdout . write ( "Current draft is still in the input line; clear it before restoring the earlier draft.\n" ) ;
17221791 else if ( ! heldDraft ) process . stdout . write ( "No earlier type-ahead draft is saved.\n" ) ;
0 commit comments