refactor(agent): run maintenance between turns

This commit is contained in:
_Kerman
2026-08-03 10:53:08 +08:00
parent 0efbb7d87a
commit b4258a2c4a
6 changed files with 145 additions and 60 deletions

View File

@@ -27,6 +27,8 @@ function expectedFailure(error: ManualCompactionError): CommandResult {
kind: 'error', kind: 'error',
text: 'Compaction is unavailable because this process has an active compaction, or the agent is not idle.', text: 'Compaction is unavailable because this process has an active compaction, or the agent is not idle.',
} }
case 'cancelled':
return { kind: 'error', text: 'Compaction cancelled.' }
case 'changed': case 'changed':
return { return {
kind: 'error', kind: 'error',

View File

@@ -365,44 +365,54 @@ export class BasicCompactService extends CompactService {
* Force one useful idle-session compaction below the pressure threshold, and * Force one useful idle-session compaction below the pressure threshold, and
* resolve only after its standalone marker pair is durably checkpointed. * resolve only after its standalone marker pair is durably checkpointed.
* @param agent - idle agent whose next-turn admission this call reserves. * @param agent - idle agent whose next-turn admission this call reserves.
* @param signal - command-owned cancellation forwarded to summarization. * @param signal - cancellation scoped to this compaction request.
* @returns the committed result, or `null` when no safe useful range exists. * @returns the committed result, or `null` when no safe useful range exists.
*/ */
override async compactNow( override compactNow(agent: Agent, signal: AbortSignal): Promise<CompactionResult | null> {
agent: Agent,
signal: AbortSignal,
): Promise<CompactionResult | null> {
signal.throwIfAborted() signal.throwIfAborted()
const releaseTurnAdmission = agent.reserveTurnAdmission() try {
if (releaseTurnAdmission === undefined) { return agent.runMaintenance(async (agentSignal) => {
const operationSignal = AbortSignal.any([agentSignal, signal])
try {
operationSignal.throwIfAborted()
const range = selectCompactableRange(
agent.session,
this.ctx.tokenMeter.measure(agent.session),
0,
)
if (range === null) return null
return await compactSurfaceRegion(
this.regionDependencies(),
agent.session,
range.start,
range.end,
agent,
{
owner: null,
stability: 'selected-span',
flush: () => this.ctx.sessions.flush(agent.session),
},
operationSignal,
)
} catch (error: unknown) {
if (agentSignal.aborted && operationSignal.reason === agentSignal.reason) {
throw new ManualCompactionError(
'cancelled',
'manual compaction was cancelled',
{ cause: error },
)
}
operationSignal.throwIfAborted()
throw error
}
})
} catch (error: unknown) {
throw new ManualCompactionError( throw new ManualCompactionError(
'busy', 'busy',
'manual compaction requires an idle agent with no waking queued work', 'manual compaction requires an idle agent with no waking queued work',
{ cause: error },
) )
} }
try {
const range = selectCompactableRange(
agent.session,
this.ctx.tokenMeter.measure(agent.session),
0,
)
if (range === null) return null
return await compactSurfaceRegion(
this.regionDependencies(),
agent.session,
range.start,
range.end,
agent,
{
owner: null,
stability: 'selected-span',
flush: () => this.ctx.sessions.flush(agent.session),
},
signal,
)
} finally {
releaseTurnAdmission()
}
} }
/** Bind the effective token meter and dynamically dispatched summarizer hook. */ /** Bind the effective token meter and dynamically dispatched summarizer hook. */

View File

@@ -22,7 +22,13 @@ export { COMPACT_CHECKPOINT_SOURCE, isCompactCheckpointSource } from './checkpoi
export type CompactionTrigger = 'pressure' | 'context-overflow' export type CompactionTrigger = 'pressure' | 'context-overflow'
/** Expected failure classes for an explicit idle-session compaction request. */ /** Expected failure classes for an explicit idle-session compaction request. */
export type ManualCompactionErrorCode = 'busy' | 'changed' | 'summary' | 'commit' | 'persistence' export type ManualCompactionErrorCode =
| 'busy'
| 'cancelled'
| 'changed'
| 'summary'
| 'commit'
| 'persistence'
/** /**
* Expected manual-compaction failure suitable for a direct human-command result. * Expected manual-compaction failure suitable for a direct human-command result.
@@ -59,7 +65,14 @@ export interface CompactAgentContext {
* other compaction transactions. * other compaction transactions.
*/ */
export interface ManualCompactAgentContext extends CompactAgentContext { export interface ManualCompactAgentContext extends CompactAgentContext {
reserveTurnAdmission(): (() => void) | undefined /**
* Run a non-turn maintenance operation only while the agent is idle, withholding later
* waking input until it settles.
* @param task - operation whose fulfillment or rejection is preserved, with an agent-owned cancellation signal.
* @throws synchronously when the agent is already active.
* @returns the task promise.
*/
runMaintenance<T>(task: (signal: AbortSignal) => Promise<T>): Promise<T>
} }
declare module 'cordis' { declare module 'cordis' {
@@ -102,21 +115,22 @@ export abstract class CompactService extends Service {
/** /**
* Explicitly compact useful history even below automatic pressure thresholds. * Explicitly compact useful history even below automatic pressure thresholds.
* Implementations reserve idle turn admission synchronously before any * Implementations synchronously start an idle task before any asynchronous
* asynchronous work, select a useful range without writing on a no-op, then * work, select a useful range without writing on a no-op, then
* append a standalone `compact/start` before summarization. That durable * append a standalone `compact/start` before summarization. That durable
* marker is the compaction lock until one `compact/end` attempt. Later waking * marker is the compaction lock until one `compact/end` attempt. Later waking
* prompts remain accepted in FIFO order and start only after the optional * prompts remain accepted in FIFO order and start only after the optional
* durability checkpoint and admission release. Context injected while the * durability checkpoint and idle-task settlement. Context injected while the
* summary runs may sit between the marker pair; only the selected span must * summary runs may sit between the marker pair; only the selected span must
* remain stable. * remain stable.
* *
* @param agent - idle agent whose durable history should be compacted. * @param agent - idle agent whose durable history should be compacted.
* @param signal - command-owned cancellation forwarded to summarization. * @param signal - cancellation scoped to this compaction request.
* @returns the compaction result, or `null` when no safe useful range exists. * @returns the compaction result, or `null` when no safe useful range exists.
* @throws {@link ManualCompactionError} for expected busy, changed-span, * @throws {@link ManualCompactionError} for expected busy, agent-cancellation,
* summarization/shrink, commit-stage, or persistence failures, and the exact * changed-span, summarization/shrink, commit-stage, or persistence failures;
* abort reason when cancelled. Failed attempts remain visible in the log. * an aborted request preserves its exact abort reason. Failed attempts remain
* visible in the log.
*/ */
abstract compactNow( abstract compactNow(
agent: ManualCompactAgentContext, agent: ManualCompactAgentContext,

View File

@@ -35,6 +35,12 @@ import { executeToolCalls } from './tool-calls.ts'
type Phase = type Phase =
| { kind: 'idle'; lastTurn: number } | { kind: 'idle'; lastTurn: number }
| {
kind: 'maintenance'
abort: AbortController
lastTurn: number
wakeRequested: boolean
}
| { kind: 'collecting'; abort: AbortController; lastTurn: number } | { kind: 'collecting'; abort: AbortController; lastTurn: number }
| { kind: 'running'; abort: AbortController; turn: number; step: number } | { kind: 'running'; abort: AbortController; turn: number; step: number }
@@ -57,7 +63,7 @@ function requestProposal(header: EpochHeader): LlmCallConfig {
export class ReactLoopAgent implements Agent { export class ReactLoopAgent implements Agent {
readonly inbox: Inbox readonly inbox: Inbox
private phase: Phase private phase: Phase
private driverDone: Promise<void> = Promise.resolve() private activityDone: Promise<void> = Promise.resolve()
/** The agent-scoped registration boundary; the lifecycle owner unwinds it after the driver exits. */ /** The agent-scoped registration boundary; the lifecycle owner unwinds it after the driver exits. */
readonly scope: Scope readonly scope: Scope
@@ -85,7 +91,7 @@ export class ReactLoopAgent implements Agent {
} }
get status(): AgentStatus { get status(): AgentStatus {
return this.phase.kind === 'idle' ? 'idle' : 'running' return this.phase.kind === 'idle' || this.phase.kind === 'maintenance' ? 'idle' : 'running'
} }
/** Commit a phase and publish its externally visible status transition. */ /** Commit a phase and publish its externally visible status transition. */
@@ -99,7 +105,7 @@ export class ReactLoopAgent implements Agent {
} }
send(message: UserMessage, target: InboxTarget, wakeup: boolean): void { send(message: UserMessage, target: InboxTarget, wakeup: boolean): void {
// Waking input cannot join an aborted pre-step or turn, so it starts the next turn. // Waking input cannot join an aborted activity, so it starts the next turn.
const wakingAfterAbort = wakeup && this.phase.kind !== 'idle' && this.phase.abort.signal.aborted const wakingAfterAbort = wakeup && this.phase.kind !== 'idle' && this.phase.abort.signal.aborted
const resolvedTarget = wakingAfterAbort ? 'next-turn' : target const resolvedTarget = wakingAfterAbort ? 'next-turn' : target
this.inbox.splice(resolvedTarget, Infinity, 0, [message]) this.inbox.splice(resolvedTarget, Infinity, 0, [message])
@@ -119,15 +125,44 @@ export class ReactLoopAgent implements Agent {
} }
cancel(cause: AgentCancelCause, options: CancelOptions = {}): void { cancel(cause: AgentCancelCause, options: CancelOptions = {}): void {
if (!options.keepInbox) this.inbox.clear() if (!options.keepInbox) {
this.inbox.clear()
if (this.phase.kind === 'maintenance') this.phase.wakeRequested = false
}
if (this.phase.kind !== 'idle') this.phase.abort.abort(cause) if (this.phase.kind !== 'idle') this.phase.abort.abort(cause)
} }
/** Reserve a driver before deferring idle pre-step processing. */ runMaintenance<T>(task: (signal: AbortSignal) => Promise<T>): Promise<T> {
if (this.phase.kind !== 'idle') throw new Error(`agent "${this.id}" already has active work`)
const done = Promise.withResolvers<void>()
const maintenance: Phase = {
kind: 'maintenance',
abort: new AbortController(),
lastTurn: this.phase.lastTurn,
wakeRequested: false,
}
this.setPhase(maintenance)
this.activityDone = done.promise
return (async () => {
try {
return await task(maintenance.abort.signal)
} finally {
this.setPhase({ kind: 'idle', lastTurn: maintenance.lastTurn })
if (maintenance.wakeRequested) this.scheduleKick()
done.resolve()
}
})()
}
/** Schedule one driver, or remember its wake behind maintenance. */
private scheduleKick(): void { private scheduleKick(): void {
if (this.phase.kind === 'maintenance') {
if (!this.phase.abort.signal.aborted) this.phase.wakeRequested = true
return
}
if (this.phase.kind !== 'idle') return if (this.phase.kind !== 'idle') return
const driver = Promise.withResolvers<void>() const driver = Promise.withResolvers<void>()
this.driverDone = driver.promise this.activityDone = driver.promise
this.setPhase({ kind: 'collecting', abort: new AbortController(), lastTurn: this.phase.lastTurn }) this.setPhase({ kind: 'collecting', abort: new AbortController(), lastTurn: this.phase.lastTurn })
queueMicrotask(() => { queueMicrotask(() => {
this.loopCtx.agents.withInitiator(this, () => this.kick()).then(driver.resolve, driver.reject) this.loopCtx.agents.withInitiator(this, () => this.kick()).then(driver.resolve, driver.reject)
@@ -135,10 +170,10 @@ export class ReactLoopAgent implements Agent {
} }
async whenIdle(): Promise<void> { async whenIdle(): Promise<void> {
let driver: Promise<void> let activity: Promise<void>
do { do {
await (driver = this.driverDone) await (activity = this.activityDone)
} while (driver !== this.driverDone) } while (activity !== this.activityDone)
} }
/** Report one failure at its live boundary, then preserve it for driver containment. */ /** Report one failure at its live boundary, then preserve it for driver containment. */
@@ -184,7 +219,7 @@ export class ReactLoopAgent implements Agent {
/** Claimed input stays unowned until `turn/start` commits. */ /** Claimed input stays unowned until `turn/start` commits. */
private async turn(): Promise<boolean> { private async turn(): Promise<boolean> {
if (this.phase.kind === 'idle') { if (this.phase.kind === 'idle' || this.phase.kind === 'maintenance') {
this.throwError(new Error(`agent "${this.id}": turn without driver reservation`)) this.throwError(new Error(`agent "${this.id}": turn without driver reservation`))
} }
const abort = this.phase.kind === 'collecting' ? this.phase.abort : new AbortController() const abort = this.phase.kind === 'collecting' ? this.phase.abort : new AbortController()

View File

@@ -100,9 +100,9 @@ export interface Agent {
/** /**
* Clear queued and steering work — unless `keepInbox` — and abort the active * Clear queued and steering work — unless `keepInbox` — and abort the active
* turn. The first cause wins for the active turn. Idle cancellation is a * turn or between-turn task. The first cause wins for that activity. With no
* no-op and does not arm later work. * active activity, cancellation is a no-op and does not arm later work.
* @param cause - the stable caller intent carried by the current turn signal. * @param cause - the stable caller intent carried by the active operation signal.
* @param options - cancellation options; `keepInbox` preserves pending work. * @param options - cancellation options; `keepInbox` preserves pending work.
*/ */
cancel(cause: AgentCancelCause, options?: CancelOptions): void cancel(cause: AgentCancelCause, options?: CancelOptions): void
@@ -115,6 +115,17 @@ export interface Agent {
*/ */
whenIdle(): Promise<void> whenIdle(): Promise<void>
/**
* Run one non-turn maintenance task from the true idle phase. The task starts
* synchronously after claiming that phase; later waking input remains in the
* inbox until the task settles, while public status stays `idle`.
* `whenIdle()` follows both the task and any waking work released behind it.
* @param task - operation whose fulfillment or rejection is preserved, with a signal aborted by {@link cancel}.
* @throws synchronously when turn-driving or another maintenance task already owns the agent.
* @returns the task promise.
*/
runMaintenance<T>(task: (signal: AbortSignal) => Promise<T>): Promise<T>
/** /**
* Route identified input to an inbox boundary and optionally wake the driver. * Route identified input to an inbox boundary and optionally wake the driver.
* Waking input submitted after active cancellation is queued for the next turn. * Waking input submitted after active cancellation is queued for the next turn.

View File

@@ -973,6 +973,19 @@ export function createTuiChat(
void shutdown(true) void shutdown(true)
} }
/** Cancel the active turn or standalone compaction, preserving work queued behind compaction. */
const cancelActive = (): boolean => {
if (agent.status === 'running') {
agent.cancel({ kind: 'user' })
return true
}
if (compacting !== undefined) {
agent.cancel({ kind: 'user' }, { keepInbox: true })
return true
}
return false
}
/** Swap the palette and all derived themes for the given terminal color scheme. */ /** Swap the palette and all derived themes for the given terminal color scheme. */
const applyColorScheme = (scheme: TerminalColorScheme): void => { const applyColorScheme = (scheme: TerminalColorScheme): void => {
if (scheme === currentScheme) return if (scheme === currentScheme) return
@@ -1033,8 +1046,8 @@ export function createTuiChat(
chat.addChild(new Text(palette.bold(palette.accent('Keyboard shortcuts')), 0, 0)) chat.addChild(new Text(palette.bold(palette.accent('Keyboard shortcuts')), 0, 0))
chat.addChild(new Text([ chat.addChild(new Text([
'Enter send • Shift/Alt+Enter newline • Up/Down prompt history', 'Enter send • Shift/Alt+Enter newline • Up/Down prompt history',
'Esc cancel turn • Ctrl+O cycle cards (collapse/expand/hide) • Ctrl+R toggle reasoning • Ctrl+L redraw', 'Esc cancel active work • Ctrl+O cycle cards (collapse/expand/hide) • Ctrl+R toggle reasoning • Ctrl+L redraw',
'Ctrl+C cancel while running; clear input or exit while idle • Ctrl+D exit', 'Ctrl+C cancel active work; clear input or exit while idle • Ctrl+D exit',
'', '',
...commandLines, ...commandLines,
'/skill:<name> [instructions] — load a skill into the conversation', '/skill:<name> [instructions] — load a skill into the conversation',
@@ -1518,14 +1531,12 @@ export function createTuiChat(
ui.requestRender(true) ui.requestRender(true)
return { consume: true } return { consume: true }
} }
if (matchesKey(data, Key.escape) && agent.status === 'running') { if (matchesKey(data, Key.escape) && cancelActive()) {
agent.cancel({ kind: 'user' })
return { consume: true } return { consume: true }
} }
if (matchesKey(data, Key.ctrl('c'))) { if (matchesKey(data, Key.ctrl('c'))) {
if (agent.status === 'running') { if (cancelActive()) return { consume: true }
agent.cancel({ kind: 'user' }) if (editor.getText() !== '') {
} else if (editor.getText() !== '') {
editor.setText('') editor.setText('')
} else { } else {
requestExit() requestExit()
@@ -1533,7 +1544,9 @@ export function createTuiChat(
return { consume: true } return { consume: true }
} }
if (matchesKey(data, Key.ctrl('d'))) { if (matchesKey(data, Key.ctrl('d'))) {
if (agent.status === 'running') appendNotice('Cancel the active turn before exiting.', 'warning') if (agent.status === 'running' || compacting !== undefined) {
appendNotice('Cancel active work before exiting.', 'warning')
}
else requestExit() else requestExit()
return { consume: true } return { consume: true }
} }