feat(compact): add optional cancellation signal to the compact seam methods
This commit is contained in:
@@ -351,8 +351,8 @@ Implementations MUST honor:
|
|||||||
- **Blocking**: no compaction begins while another is in progress for the same session. The recommended mechanism is the log-recorded lock — append `compact/start` before the slow work and `compact/end` after (even on failure) — so the lock is visible to replay and crash recovery.
|
- **Blocking**: no compaction begins while another is in progress for the same session. The recommended mechanism is the log-recorded lock — append `compact/start` before the slow work and `compact/end` after (even on failure) — so the lock is visible to replay and crash recovery.
|
||||||
|
|
||||||
```ts cordis-catalog
|
```ts cordis-catalog
|
||||||
abstract compactIfNeeded( session: Session, systemPrompt?: string, model?: string, ): Promise<CompactionResult | null>
|
abstract compactIfNeeded( session: Session, systemPrompt?: string, model?: string, signal?: AbortSignal, ): Promise<CompactionResult | null>
|
||||||
abstract compactRegion( session: Session, start: number, end: number, model: string, ): Promise<CompactionResult>
|
abstract compactRegion( session: Session, start: number, end: number, model: string, signal?: AbortSignal, ): Promise<CompactionResult>
|
||||||
```
|
```
|
||||||
|
|
||||||
Source: [`packages/compact/compact/src/index.ts:57`](../../packages/compact/compact/src/index.ts)
|
Source: [`packages/compact/compact/src/index.ts:57`](../../packages/compact/compact/src/index.ts)
|
||||||
|
|||||||
@@ -18,8 +18,10 @@ Both methods are **abstract** — the backend owns the entire strategy (token es
|
|||||||
|
|
||||||
| Member | Semantics |
|
| Member | Semantics |
|
||||||
|---|---|
|
|---|---|
|
||||||
| `compactIfNeeded(session, systemPrompt?, model?)` | Estimate the history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. |
|
| `compactIfNeeded(session, systemPrompt?, model?, signal?)` | Estimate the history size; if over the backend's threshold, compact an older range via `compactRegion`, keeping recent context intact. Returns the `CompactionResult`, or `null` if nothing needed compacting. |
|
||||||
| `compactRegion(session, start, end, model)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start > end`. |
|
| `compactRegion(session, start, end, model, signal?)` | Forcibly summarize surface nodes `[start, end]` (inclusive seqs) into a single replacement node. **Throws** if a compaction is already in progress, if `start`/`end` aren't surface nodes, or if `start > end`. |
|
||||||
|
|
||||||
|
Both methods take an optional `signal: AbortSignal`. A backend that summarizes via `ctx.llm.stream()` **must** forward it into the call's `GenerateOptions.signal`, so an abort or fiber dispose tears down the in-flight summarization instead of leaving an orphaned model call running past the cancellation. The turn that the `compact/*` events belong to is not a parameter — it is recoverable from the log (the currently-open turn), so the backend stamps it without the caller supplying it.
|
||||||
|
|
||||||
## Surface contract
|
## Surface contract
|
||||||
|
|
||||||
|
|||||||
@@ -69,12 +69,17 @@ export abstract class CompactService extends Service {
|
|||||||
* @param session - the session whose surface may be compacted.
|
* @param session - the session whose surface may be compacted.
|
||||||
* @param systemPrompt - optional system prompt, counted toward the estimate.
|
* @param systemPrompt - optional system prompt, counted toward the estimate.
|
||||||
* @param model - optional summarization model (falls back to backend config).
|
* @param model - optional summarization model (falls back to backend config).
|
||||||
|
* @param signal - optional cancellation signal. A backend that summarizes via
|
||||||
|
* `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
|
||||||
|
* so an abort/dispose tears down the in-flight summarization rather than
|
||||||
|
* leaving an orphaned model call running past the cancellation.
|
||||||
* @returns the compaction result, or `null` if no compaction was needed.
|
* @returns the compaction result, or `null` if no compaction was needed.
|
||||||
*/
|
*/
|
||||||
abstract compactIfNeeded(
|
abstract compactIfNeeded(
|
||||||
session: Session,
|
session: Session,
|
||||||
systemPrompt?: string,
|
systemPrompt?: string,
|
||||||
model?: string,
|
model?: string,
|
||||||
|
signal?: AbortSignal,
|
||||||
): Promise<CompactionResult | null>
|
): Promise<CompactionResult | null>
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -88,6 +93,10 @@ export abstract class CompactService extends Service {
|
|||||||
* @param start - inclusive seq of the first surface node to compact.
|
* @param start - inclusive seq of the first surface node to compact.
|
||||||
* @param end - inclusive seq of the last surface node to compact.
|
* @param end - inclusive seq of the last surface node to compact.
|
||||||
* @param model - summarization model.
|
* @param model - summarization model.
|
||||||
|
* @param signal - optional cancellation signal. A backend that summarizes via
|
||||||
|
* `ctx.llm.stream()` MUST forward this into the call's `GenerateOptions.signal`
|
||||||
|
* so an abort/dispose tears down the in-flight summarization rather than
|
||||||
|
* leaving an orphaned model call running past the cancellation.
|
||||||
* @throws if compaction is already in progress, or if `start`/`end` are not
|
* @throws if compaction is already in progress, or if `start`/`end` are not
|
||||||
* valid surface nodes, or if `start > end`.
|
* valid surface nodes, or if `start > end`.
|
||||||
*/
|
*/
|
||||||
@@ -96,6 +105,7 @@ export abstract class CompactService extends Service {
|
|||||||
start: number,
|
start: number,
|
||||||
end: number,
|
end: number,
|
||||||
model: string,
|
model: string,
|
||||||
|
signal?: AbortSignal,
|
||||||
): Promise<CompactionResult>
|
): Promise<CompactionResult>
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -11,11 +11,27 @@ import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
|||||||
* declaration merge.
|
* declaration merge.
|
||||||
*/
|
*/
|
||||||
class StubCompactService extends CompactService {
|
class StubCompactService extends CompactService {
|
||||||
override async compactIfNeeded(_session: Session, _systemPrompt?: string, _model?: string): Promise<CompactionResult | null> {
|
/** Records the signal handed to the most recent call, to prove it threads through. */
|
||||||
|
lastSignal: AbortSignal | undefined
|
||||||
|
|
||||||
|
override async compactIfNeeded(
|
||||||
|
_session: Session,
|
||||||
|
_systemPrompt?: string,
|
||||||
|
_model?: string,
|
||||||
|
signal?: AbortSignal,
|
||||||
|
): Promise<CompactionResult | null> {
|
||||||
|
this.lastSignal = signal
|
||||||
return null
|
return null
|
||||||
}
|
}
|
||||||
|
|
||||||
override async compactRegion(session: Session, start: number, end: number, _model: string): Promise<CompactionResult> {
|
override async compactRegion(
|
||||||
|
session: Session,
|
||||||
|
start: number,
|
||||||
|
end: number,
|
||||||
|
_model: string,
|
||||||
|
signal?: AbortSignal,
|
||||||
|
): Promise<CompactionResult> {
|
||||||
|
this.lastSignal = signal
|
||||||
// Minimal stub honoring the lock + log-only event contract.
|
// Minimal stub honoring the lock + log-only event contract.
|
||||||
const startEvent = session.append('compact/start', { turn: 0 })
|
const startEvent = session.append('compact/start', { turn: 0 })
|
||||||
const summaryEvent = session.append('compact/summary', {
|
const summaryEvent = session.append('compact/summary', {
|
||||||
@@ -75,4 +91,17 @@ describe('CompactService seam', () => {
|
|||||||
expect(result.summarySeq).toBeGreaterThan(result.startSeq)
|
expect(result.summarySeq).toBeGreaterThan(result.startSeq)
|
||||||
expect(result.endSeq).toBeGreaterThan(result.summarySeq)
|
expect(result.endSeq).toBeGreaterThan(result.summarySeq)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('threads the cancellation signal through to the backend', async () => {
|
||||||
|
const ctx = new Context()
|
||||||
|
const svc = new StubCompactService(ctx)
|
||||||
|
const session = new Session(SessionId('s'))
|
||||||
|
const controller = new AbortController()
|
||||||
|
|
||||||
|
await svc.compactRegion(session, 0, 0, 'm', controller.signal)
|
||||||
|
expect(svc.lastSignal).toBe(controller.signal)
|
||||||
|
|
||||||
|
await svc.compactIfNeeded(session, undefined, undefined, controller.signal)
|
||||||
|
expect(svc.lastSignal).toBe(controller.signal)
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user