feat(goal): drive same-session goal rounds
This commit is contained in:
Tianyi Cui
@@ -262,6 +262,10 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
||||
signature: 'get(agent: Agent): GoalView | undefined',
|
||||
jsDoc: '/**\n * Read the current goal for one exact live agent.\n * @param agent - owning live agent.\n * @returns a fresh view or `undefined` when no goal is current.\n * @throws {@link GoalError} when the agent is not the registry\'s live instance.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'disarm(agent: Agent): GoalView | undefined',
|
||||
jsDoc: '/**\n * Remove process-local continuation authority without changing durable goal\n * phase or revision. Lifecycle owners use this before unloading a driver;\n * a later human-authorized {@link resume} records the new activation edge.\n * @param agent - owning live agent.\n * @returns a fresh disarmed view, or `undefined` when no goal is current.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'create(agent: Agent, request: CreateGoalRequest): GoalView',
|
||||
jsDoc: '/**\n * Create and arm a goal. A completed goal may be replaced; every other\n * current phase must be cleared or resumed instead.\n * @param agent - owning live agent.\n * @param request - objective and optional round cap.\n * @returns the created live view.\n */',
|
||||
@@ -663,6 +667,13 @@ export const EVENT_API: readonly EventApiEntry[] = [
|
||||
jsDoc: '/**\n * A declarative agent entry failed before it could publish a live agent.\n * Consumers that buffer work for the configured identity use this\n * transient signal to reject that work instead of waiting forever. Normal\n * factory teardown suppresses failures from the cancelled startup attempt.\n * @param sessionId - exact shared agent/session identity that failed startup.\n * @param error - persistence, setup, or publication failure.\n * @mode emit\n */',
|
||||
summary: 'A declarative agent entry failed before it could publish a live agent.',
|
||||
},
|
||||
{
|
||||
name: 'agent/cancel-requested',
|
||||
mode: 'emit',
|
||||
signature: '\'agent/cancel-requested\'(this: Scoped<Agent>, agent: Agent, reason: string): void',
|
||||
jsDoc: '/**\n * Effective broad cancellation was requested, before queued/steering work\n * is cleared or the active step is aborted. This observe-only notification\n * cannot veto cancellation; listener failures are contained.\n * @param agent - the agent whose current work is being cancelled.\n * @param reason - resolved cancellation reason, including the default.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */',
|
||||
summary: 'Effective broad cancellation was requested, before queued/steering work is cleared or the active step is aborted.',
|
||||
},
|
||||
{
|
||||
name: 'agent/created',
|
||||
mode: 'emit',
|
||||
|
||||
@@ -54,7 +54,7 @@ The driver owns one agent for its lifetime and runs inside `ctx.agents.withIniti
|
||||
|
||||
Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
|
||||
|
||||
Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery observes a closed failed step, and a retry rebuilds the request from the durable log in a new numbered step. Cancellation clears pending work and aborts the current step without leaking to the next prompt; undispatched model tool calls receive synthetic `tool/call` and aborted result pairs. Terminal continuation stops remain authoritative through turn close and durability flush.
|
||||
Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery observes a closed failed step, and a retry rebuilds the request from the durable log in a new numbered step. Effective cancellation resolves its reason and emits `agent/cancel-requested` before clearing pending work or aborting the current step; notification failures are contained, queued work added by an observer is included in the same broad clear, and idle cancellation emits nothing. Undispatched model tool calls receive synthetic `tool/call` and aborted result pairs. Terminal continuation stops remain authoritative through turn close and durability flush.
|
||||
|
||||
Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, then drains accepted batch context before the turn closes through the normal abort path.
|
||||
|
||||
|
||||
@@ -333,13 +333,18 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
|
||||
cancel(reason?: string): void {
|
||||
const resolvedReason = reason ?? 'cancelled'
|
||||
// Arm only for current work; an idle marker would cancel the next prompt.
|
||||
if (this._status === 'running' || this.currentAbort !== undefined || this.#inbox.hasQueued || this.#inbox.hasSteering) {
|
||||
this.cancelRequested = true
|
||||
// Capture the resolved reason for the marker-only windows (pre-step /
|
||||
// continuation). The mid-step path reads it from abort.signal.reason
|
||||
// below; the marker path reads it via the LoopHandle's cancelReason().
|
||||
this.cancelReason = reason ?? 'cancelled'
|
||||
this.cancelReason = resolvedReason
|
||||
// Coordination consumers must update their own state before this call
|
||||
// clears the inbox or aborts the step. Notification failures are
|
||||
// contained by the fused dispatcher and cannot veto cancellation.
|
||||
agentEvents(this.loopCtx, this).emit('agent/cancel-requested', resolvedReason)
|
||||
}
|
||||
// Drop all pending queued + steering work (un-started prompts never run; the
|
||||
// cancelled turn's steering is not re-enqueued). Cleared directly even when
|
||||
@@ -349,7 +354,7 @@ export class ReactLoopAgent implements Agent {
|
||||
// Interrupt an in-flight step immediately (the running turn observes the
|
||||
// abort and ends `aborted`). The marker covers the windows where no step is
|
||||
// running (pre-step, continuation).
|
||||
this.currentAbort?.abort(reason ?? 'cancelled')
|
||||
this.currentAbort?.abort(resolvedReason)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
* @module dsh-agent-loop/tests/cancel
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { type Message } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
@@ -55,6 +55,33 @@ function userTexts(agent: Agent): string[] {
|
||||
}
|
||||
|
||||
describe('Agent.cancel()', () => {
|
||||
it('notifies every observer before clearing work and contains listener failures', async () => {
|
||||
const adapter = new MockAdapter([textResponse('must remain unused')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('cancel-event'), { provider: 'mock', model: 'mock' })
|
||||
const warned = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const seen: string[] = []
|
||||
ctx.on('agent/cancel-requested', (subject, reason) => {
|
||||
if (subject !== agent) return
|
||||
seen.push(`first:${reason}`)
|
||||
subject.send([{ type: 'text', text: 'queued by cancel observer' }])
|
||||
throw new Error('observer failed')
|
||||
})
|
||||
ctx.on('agent/cancel-requested', (subject, reason) => {
|
||||
if (subject === agent) seen.push(`second:${reason}`)
|
||||
})
|
||||
|
||||
send(agent, 'drop me')
|
||||
agent.cancel()
|
||||
await new Promise(resolve => setTimeout(resolve, 30))
|
||||
agent.cancel('idle no-op')
|
||||
|
||||
expect(seen).toEqual(['first:cancelled', 'second:cancelled'])
|
||||
expect(userTexts(agent)).toEqual([])
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
expect(warned).toHaveBeenCalledWith(expect.stringContaining('agent/cancel-requested'))
|
||||
})
|
||||
|
||||
it('cancel() on an idle agent with nothing queued is a no-op; the next prompt runs (F2 leak guard)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('reply')])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
@@ -44,7 +44,7 @@ Agent *creation* is provided by the plugin implementing `AgentFactory` (`dsh-age
|
||||
|
||||
The lifecycle edges have two important local caveats. `agent/created` runs after scoped setup and after both session and agent registry entries exist. Setup is trusted composition-only code; the immediately following non-vetoing `agent/session-start` notification is the first supported startup injection point. `agent/disposed` always means the exact agent has left the registry. AgentLoop emits it after its driver is quiescent, while ordered teardown may still be detaching the session and unwinding the scope; custom agents registered directly own any stronger driver-ordering contract themselves.
|
||||
|
||||
Most interception points are cooperative waterfalls returning seam-specific decisions. `agent/pre-step` and `agent/post-step` are serial checkpoints around a step's durable work, while `agent/request-error` is the failed-model-request recovery waterfall: a retry opens a new numbered step after the failed step closes. `agent/turn-stop` is the terminal serial fold: it runs after ordinary continuation and steering folding, and a returned stop remains in force through turn close and flush so later steering cannot create an extra step or turn. Ordinary queued prompts remain intact. The full rationale for scoped dispatch and terminal settlement is in the [agent-scope runtime-design RFC](../../../docs/rfc/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way).
|
||||
Most interception points are cooperative waterfalls returning seam-specific decisions. `agent/pre-step` and `agent/post-step` are serial checkpoints around a step's durable work, while `agent/request-error` is the failed-model-request recovery waterfall: a retry opens a new numbered step after the failed step closes. `agent/turn-stop` is the terminal serial fold: it runs after ordinary continuation and steering folding, and a returned stop remains in force through turn close and flush so later steering cannot create an extra step or turn. Ordinary queued prompts remain intact. Effective broad cancellation first emits the observe-only `agent/cancel-requested` with its resolved reason, then clears queues and aborts; notification failures are contained and cannot veto the stop. The full rationale for scoped dispatch and terminal settlement is in the [agent-scope runtime-design RFC](../../../docs/rfc/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way).
|
||||
|
||||
`PromptDecision.additionalContexts` is an array so every injected context keeps its own source, envelope, and metadata. A `ContinuationDecision` reason is narrower: it becomes a `steering/message`, not a `context/message`, and therefore carries only content and source.
|
||||
|
||||
@@ -57,7 +57,7 @@ The handle every plugin programs against:
|
||||
- `agent.send(content, options?)` — queue a message; starts a turn when idle. Content and resolved source become one detached, deeply frozen lossless-JSON record before `agent/queued` and enqueue; invalid data throws synchronously, and caller or notification-listener in-place mutation cannot change the log or model input (`agent/prompt-submit` still rewrites by returning replacement content).
|
||||
- `agent.steer(content, options?)` — steer a running turn (inject between steps); uses the same owned acceptance boundary and behaves like `send` when idle
|
||||
- `agent.inject(content, options?)` — accept detached in-session context without running the model; the next request sees its `context/message`. `options.envelope` defaults to the canonical `<context>` framing and may be `'raw'` when the caller owns a complete familiar frame; `options.meta` persists opaque JSON state without rendering it. While a turn is open it joins that turn, deferring FIFO while the current tool batch executes and draining before turn close if execution is interrupted; while idle it is wrapped in a one-shot `injection` turn and durability checkpoint ([the turn-enclosure invariant](../../../docs/rfc/implemented/architecture/2026-06-15-turn-enclosure-invariant.md)).
|
||||
- `agent.cancel(reason?)` — cancel ALL pending work: clears the queued + steering FIFOs, aborts the in-flight step, and drops a turn about to start (the pre-step window) so a queued-but-not-started prompt never runs. A UI/ACP `session/cancel` maps to this. The single public stop primitive. Idle with nothing pending → a safe no-op.
|
||||
- `agent.cancel(reason?)` — cancel ALL pending work: an effective call emits `agent/cancel-requested` before it clears the queued + steering FIFOs, aborts the in-flight step, and drops a turn about to start (the pre-step window). Observers may synchronize their own state but cannot veto cancellation. A UI/ACP `session/cancel` maps to this. The single public stop primitive. Idle with nothing pending → a safe no-op with no notification.
|
||||
- `agent.whenIdle()` — resolve once the agent reaches quiescence after settling out of `running` (idle → immediately; disposed → awaits the loop exit). A non-owner's quiescence-observation hook: it observes the work settling WITHOUT tearing the agent down. Teardown is separate — a lifecycle owner stops and unregisters via `AgentHandle.dispose()`, which awaits the loop exit directly.
|
||||
- `agent.session`, `agent.status`, `agent.options`, `agent.id`
|
||||
|
||||
|
||||
@@ -120,9 +120,10 @@ export interface Agent {
|
||||
|
||||
/**
|
||||
* Clear queued and steering work, including work waiting to start, and abort
|
||||
* the active step. The supplied reason is preserved across pre-step and active
|
||||
* cancellation windows, and `whenIdle()` resolves after cancellation reaches
|
||||
* quiescence. Idle cancellation is a no-op and does not arm a later cancel.
|
||||
* the active step. An effective call first emits `agent/cancel-requested` with
|
||||
* the resolved reason. The supplied reason is preserved across pre-step and
|
||||
* active cancellation windows, and `whenIdle()` resolves after cancellation
|
||||
* reaches quiescence. Idle cancellation is a no-op and does not arm a later cancel.
|
||||
*/
|
||||
cancel(reason?: string): void
|
||||
|
||||
@@ -173,6 +174,16 @@ declare module 'cordis' {
|
||||
* @mode emit
|
||||
*/
|
||||
'agent/queued'(this: Scoped<Agent>, agent: Agent, content: ContentBlock[], info: { source: MessageSource; steering: boolean }): void
|
||||
/**
|
||||
* Effective broad cancellation was requested, before queued/steering work
|
||||
* is cleared or the active step is aborted. This observe-only notification
|
||||
* cannot veto cancellation; listener failures are contained.
|
||||
* @param agent - the agent whose current work is being cancelled.
|
||||
* @param reason - resolved cancellation reason, including the default.
|
||||
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
|
||||
* @mode emit
|
||||
*/
|
||||
'agent/cancel-requested'(this: Scoped<Agent>, agent: Agent, reason: string): void
|
||||
|
||||
// ---- session lifecycle (emit) ----
|
||||
/**
|
||||
|
||||
@@ -5,6 +5,7 @@ The goal family owns durable objective state independently of the model-facing t
|
||||
| Package | Role | ctx key |
|
||||
|---|---|---|
|
||||
| `goal/` | Event-sourced goal lifecycle, replay fold, compare-and-set mutations, and process-local activation | `ctx.goals` |
|
||||
| `goal-session/` | Same-session goal-round admission, outcome mapping, and lifecycle race fencing | — |
|
||||
| `tool-goal/` | Model-facing read/create/update tools with execution-time authority checks | — |
|
||||
|
||||
Goal state is part of the owning session log. Consumers depend on `dsh-goal`, not on the concrete agent loop; continuation behavior belongs in a separate plugin on the public agent seams.
|
||||
|
||||
62
packages/goal/goal-session/README.md
Normal file
62
packages/goal/goal-session/README.md
Normal file
@@ -0,0 +1,62 @@
|
||||
# @deepseek-ai/dsh-goal-session
|
||||
|
||||
Same-session continuation driver for [`ctx.goals`](../goal/README.md). It turns an active, armed goal into sequential [goal rounds](../../../docs/glossary.md#goal-round) through the public `Agent` and session seams; the [same-session driver RFC](../../../docs/rfc/implemented/feature/2026-07-19-same-session-goal-round-driver.md) owns the race and lifecycle rationale.
|
||||
|
||||
## Composition
|
||||
|
||||
```yaml
|
||||
- id: goal
|
||||
name: '@deepseek-ai/dsh-goal'
|
||||
|
||||
- id: tool-goal
|
||||
name: '@deepseek-ai/dsh-tool-goal'
|
||||
|
||||
- id: goal-session
|
||||
name: '@deepseek-ai/dsh-goal-session'
|
||||
```
|
||||
|
||||
The plugin has no tunable configuration. `maxGoalRounds` belongs to the goal definition, while the model-facing blocked threshold belongs to [`dsh-tool-goal`](../tool-goal/README.md); duplicating either value in the driver could produce divergent policy.
|
||||
|
||||
## Round contract
|
||||
|
||||
When an exact live agent is idle with an active, armed goal and remaining capacity, the driver first checkpoints pending goal mutations, then reserves `roundsStarted + 1` for the current `{ goalId, revision }`. It queues one `<goal_round>` prompt with `GoalMessageSource`. Admission through `agent/prompt-submit` verifies the complete queued record and current goal both before and after downstream prompt hooks; only the accepted `user/message` increments `roundsStarted`. A reservation rejected as stale does not consume the round number.
|
||||
|
||||
One goal round owns one ordinary session turn, and that turn may contain several model/tool steps. Human messages remain ordinary turns and do not consume the goal cap. If human work enters the inbox before a reservation or joins its pending batch, automatic work yields until that work settles; a pending automatic prompt in a mixed batch is rejected and re-reserved only after the agent becomes idle.
|
||||
|
||||
The retained prompt names the JSON-quoted objective and `round/maxGoalRounds`, treats the current workspace, tool results, and durable session state as authoritative, requires evidence before completion, and tells the model to leave the goal active when work remains. Quoting preserves multiline or tag-like objective text as data. Goal lifecycle mutations still require the independent authority checks in `dsh-tool-goal`.
|
||||
|
||||
## Settlement policy
|
||||
|
||||
| Durable turn outcome | Goal action | Automatic retry |
|
||||
|---|---|---|
|
||||
| `completed` with goal still active and armed | admit the next round, or mark `budget-limited` at the cap | yes |
|
||||
| broad cancellation / `aborted` | `paused` | no |
|
||||
| `error` with `RATE_LIMIT` | `usage-limited` | no |
|
||||
| other `error`, `max-tokens`, or a non-stale prompt rejection | `blocked` | no |
|
||||
| durability failure, disposal, interruption, or unknown future outcome | disarm or block for inspection | no |
|
||||
|
||||
A goal mutation made during its round supersedes settlement of the older revision. Completion, pause, blocking, and edits therefore remain authoritative even if the physical turn closes afterward. No abnormal result is retried automatically.
|
||||
|
||||
## Lifecycle and durability
|
||||
|
||||
`goal/changed` creates a durability obligation. Before queuing work, the driver awaits `ctx.sessions.flush()` and rechecks both the goal revision and competing input after the await. A closing flush failure arrives through `agent/error`; the driver disarms before another round can start.
|
||||
|
||||
Activation is never inherited when this plugin loads over an existing agent. `GoalService.disarm()` removes process-local authority without changing durable phase, revision, or history; explicit human-authorized resume records the later reactivation. The same rule applies after session resume and fork through the goal domain's `agent/session-start` handling.
|
||||
|
||||
Cancellation is observe-before-act: the concrete loop emits `agent/cancel-requested` before clearing queues or aborting a step, allowing this plugin to pause and disarm the exact active goal. Plugin teardown closes admission, disarms every live goal, cancels an admitted round, and awaits the driver plus agent quiescence while its event fence remains installed.
|
||||
|
||||
## Model Experience
|
||||
|
||||
### Goal-round prompt
|
||||
|
||||
**What the model sees**: Each admitted round is one retained user-role `<goal_round>` block naming the full objective and positive round number. Earlier human messages, goal-state snapshots, assistant output, and tool records remain in the same session history.
|
||||
|
||||
**Token effect**: One fixed instruction block plus the objective is added per admitted round. Later requests resend retained rounds until compaction shadows them; no fresh agent or copied conversation prefix is created.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **No independent evaluator** — the model-facing goal policy decides when evidence is sufficient for completion and whether a blocker is semantically unchanged; evaluator-backed certification remains deferred.
|
||||
- **Same-session execution only** — this package deliberately does not spawn a fresh agent, fork a session prefix, or implement Ralph-style independent attempts; that workflow belongs to its own plugin layer.
|
||||
- **Accepted-queue unload race** — Cordis plugin unload is asynchronous. A goal prompt already accepted by the agent inbox can begin and consume its round before unload starts; teardown then cancels the request, disarms the goal, and awaits quiescence. No later round starts.
|
||||
- **Round cap, not resource budget** — token, currency, time, and provider quota policies remain independent; `RATE_LIMIT` only maps an observed provider stop into `usage-limited`.
|
||||
- **No abnormal auto-retry** — transient provider and persistence failures require a later human-authorized resume rather than an implicit retry policy.
|
||||
43
packages/goal/goal-session/package.json
Normal file
43
packages/goal/goal-session/package.json
Normal file
@@ -0,0 +1,43 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-goal-session",
|
||||
"description": "Race-fenced same-session goal-round driver",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "lib/index.js",
|
||||
"types": "lib/types/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./lib/types/index.d.ts",
|
||||
"default": "./lib/index.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
"lib/types/**/*.d.ts",
|
||||
"lib/types/**/*.d.ts.map",
|
||||
"src"
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-goal": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent-loop": "workspace:^",
|
||||
"@deepseek-ai/dsh-agent-loop-testkit": "workspace:^",
|
||||
"@deepseek-ai/dsh-goal": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-loader-smoke": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-system-prompt": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
}
|
||||
}
|
||||
443
packages/goal/goal-session/src/index.ts
Normal file
443
packages/goal/goal-session/src/index.ts
Normal file
@@ -0,0 +1,443 @@
|
||||
/**
|
||||
* Same-session goal-round driver over public agent, session, and goal seams.
|
||||
* @module @deepseek-ai/dsh-goal-session
|
||||
*/
|
||||
|
||||
import { isDeepStrictEqual } from 'node:util'
|
||||
import { FiberState } from 'cordis'
|
||||
import type { Context } from 'cordis'
|
||||
import type { Agent, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import type { GoalMessageSource, GoalRef, GoalView } from '@deepseek-ai/dsh-goal'
|
||||
import { assertNever } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session, SessionEvent, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import { classifyGoalRound } from './outcome.ts'
|
||||
import type { GoalRoundOutcome } from './outcome.ts'
|
||||
import { renderGoalRoundPrompt } from './prompt.ts'
|
||||
|
||||
export { classifyGoalRound } from './outcome.ts'
|
||||
export type { GoalRoundOutcome } from './outcome.ts'
|
||||
export { renderGoalRoundPrompt } from './prompt.ts'
|
||||
|
||||
export const name = 'goal-session'
|
||||
export const inject = ['agents', 'goals', 'sessions']
|
||||
|
||||
const STALE_ROUND_REASON = 'stale goal-round reservation'
|
||||
|
||||
/** Identity reserved before a goal continuation enters the agent inbox. */
|
||||
interface RoundIdentity {
|
||||
readonly goalId: GoalRef['id']
|
||||
readonly revision: number
|
||||
readonly round: number
|
||||
}
|
||||
|
||||
/** One queued or admitted attempt, retained until its physical turn settles. */
|
||||
interface RoundAttempt extends RoundIdentity {
|
||||
readonly content: ContentBlock[]
|
||||
phase: 'queued' | 'admitted'
|
||||
turn: number | undefined
|
||||
reason: TurnEndReason | undefined
|
||||
rejectedReason: string | undefined
|
||||
stale: boolean
|
||||
}
|
||||
|
||||
/** Serialized process-local scheduling state for one exact Agent lifecycle. */
|
||||
interface DriverState {
|
||||
readonly agent: Agent
|
||||
attempt: RoundAttempt | undefined
|
||||
openTurn: number | undefined
|
||||
competingQueued: boolean
|
||||
needsCheckpoint: boolean
|
||||
requested: boolean
|
||||
run: Promise<void> | undefined
|
||||
stopping: boolean
|
||||
readonly flushFailedTurns: Set<number>
|
||||
}
|
||||
|
||||
/** Whether a source identifies an automatic, positive-numbered goal round. */
|
||||
function isGoalRoundSource(source: MessageSource): source is GoalMessageSource {
|
||||
return source.kind === 'goal' && source.round > 0
|
||||
}
|
||||
|
||||
/** Compare a source to one reserved identity. */
|
||||
function sameRound(source: GoalMessageSource, round: RoundIdentity): boolean {
|
||||
return source.goalId === round.goalId
|
||||
&& source.revision === round.revision
|
||||
&& source.round === round.round
|
||||
}
|
||||
|
||||
/** Compare the complete queued record to the driver's reservation. */
|
||||
function sameQueued(content: ContentBlock[], source: MessageSource, attempt: RoundAttempt): boolean {
|
||||
return isGoalRoundSource(source) && sameRound(source, attempt) && isDeepStrictEqual(content, attempt.content)
|
||||
}
|
||||
|
||||
/** Exact current ref for a view. */
|
||||
function goalRef(goal: GoalView): GoalRef {
|
||||
return { id: goal.id, revision: goal.revision }
|
||||
}
|
||||
|
||||
/** Human-readable unexpected values for logs. */
|
||||
function renderThrown(value: unknown): string {
|
||||
return value instanceof Error ? value.message : String(value)
|
||||
}
|
||||
|
||||
/** Install automatic same-session continuation and its race fences. */
|
||||
export function apply(ctx: Context): void {
|
||||
const states = new Map<Agent, DriverState>()
|
||||
|
||||
/** Create state for an exact currently live agent. */
|
||||
function stateFor(agent: Agent): DriverState {
|
||||
const existing = states.get(agent)
|
||||
if (existing !== undefined) return existing
|
||||
const state: DriverState = {
|
||||
agent,
|
||||
attempt: undefined,
|
||||
openTurn: undefined,
|
||||
competingQueued: false,
|
||||
needsCheckpoint: false,
|
||||
requested: false,
|
||||
run: undefined,
|
||||
stopping: false,
|
||||
flushFailedTurns: new Set(),
|
||||
}
|
||||
states.set(agent, state)
|
||||
return state
|
||||
}
|
||||
|
||||
/** Read only when the exact Agent remains live. */
|
||||
function currentGoal(state: DriverState): GoalView | undefined {
|
||||
if (ctx.agents.get(state.agent.id) !== state.agent || state.agent.status === 'disposed') return undefined
|
||||
return ctx.goals.get(state.agent)
|
||||
}
|
||||
|
||||
/** Whether this exact lifecycle is quiescent with no competing prompt. */
|
||||
function readyToDrive(state: DriverState): boolean {
|
||||
return ctx.fiber.state === FiberState.ACTIVE
|
||||
&& !state.stopping
|
||||
&& ctx.agents.get(state.agent.id) === state.agent
|
||||
&& state.agent.status === 'idle'
|
||||
&& !state.competingQueued
|
||||
}
|
||||
|
||||
/** Recheck every condition that an awaited checkpoint may have changed. */
|
||||
function readyAfterCheckpoint(state: DriverState): boolean {
|
||||
return readyToDrive(state) && !state.needsCheckpoint
|
||||
}
|
||||
|
||||
/** Remove automatic authority while preserving the durable phase. */
|
||||
function disarm(state: DriverState): void {
|
||||
try {
|
||||
const goal = currentGoal(state)
|
||||
if (goal?.activation === 'armed') ctx.goals.disarm(state.agent)
|
||||
} catch (error: unknown) {
|
||||
ctx.logger.warn(`goal-session: could not disarm agent "${state.agent.id}": ${renderThrown(error)}`)
|
||||
}
|
||||
}
|
||||
|
||||
/** Apply one closed-round outcome only to the exact still-current revision. */
|
||||
function applyOutcome(state: DriverState, goal: GoalView, outcome: GoalRoundOutcome): void {
|
||||
const ref = goalRef(goal)
|
||||
switch (outcome.kind) {
|
||||
case 'continue':
|
||||
return
|
||||
case 'pause':
|
||||
ctx.goals.pause(state.agent, ref)
|
||||
return
|
||||
case 'usage-limited':
|
||||
ctx.goals.markUsageLimited(state.agent, ref)
|
||||
return
|
||||
case 'blocked':
|
||||
ctx.goals.block(state.agent, ref)
|
||||
return
|
||||
case 'disarm':
|
||||
ctx.goals.disarm(state.agent)
|
||||
return
|
||||
/* v8 ignore next 2 -- GoalRoundOutcome is closed and every member is handled above */
|
||||
default:
|
||||
assertNever(outcome, 'goal round outcome')
|
||||
}
|
||||
}
|
||||
|
||||
/** Process a settled attempt, then reserve at most one next round. */
|
||||
async function drive(state: DriverState): Promise<void> {
|
||||
const { agent } = state
|
||||
if (!readyToDrive(state)) return
|
||||
|
||||
if (state.needsCheckpoint) {
|
||||
state.needsCheckpoint = false
|
||||
try {
|
||||
await ctx.sessions.flush(agent.session)
|
||||
} catch (error: unknown) {
|
||||
ctx.logger.warn(`goal-session: durability checkpoint failed for agent "${agent.id}": ${renderThrown(error)}`)
|
||||
const goal = currentGoal(state)
|
||||
if (goal !== undefined) applyOutcome(state, goal, { kind: 'disarm', reason: 'durability-failed' })
|
||||
return
|
||||
}
|
||||
// A mutation or ordinary prompt may have arrived while the checkpoint
|
||||
// was settling. Give it its own checkpoint / turn before reserving.
|
||||
if (!readyAfterCheckpoint(state)) return
|
||||
}
|
||||
|
||||
const attempt = state.attempt
|
||||
if (attempt !== undefined) {
|
||||
if (attempt.reason === undefined) return
|
||||
state.attempt = undefined
|
||||
const turn = attempt.turn
|
||||
/* v8 ignore next -- a closed attempt acquired its turn at turn/start */
|
||||
if (turn === undefined) throw new Error('settled goal-round attempt lacks a turn')
|
||||
const durable = !state.flushFailedTurns.delete(turn)
|
||||
const goal = currentGoal(state)
|
||||
if (goal !== undefined && goal.id === attempt.goalId && goal.revision === attempt.revision
|
||||
&& goal.phase === 'active' && goal.activation === 'armed') {
|
||||
const outcome = attempt.phase === 'queued' && attempt.rejectedReason !== undefined && !attempt.stale
|
||||
? { kind: 'blocked', reason: 'rejected', detail: attempt.rejectedReason } as const
|
||||
: classifyGoalRound(attempt.reason, durable)
|
||||
if (!attempt.stale) applyOutcome(state, goal, outcome)
|
||||
}
|
||||
if (!readyToDrive(state)) return
|
||||
}
|
||||
|
||||
const goal = currentGoal(state)
|
||||
if (goal === undefined || goal.phase !== 'active' || goal.activation !== 'armed') return
|
||||
if (goal.roundsStarted >= goal.maxGoalRounds) {
|
||||
ctx.goals.markBudgetLimited(agent, goalRef(goal))
|
||||
return
|
||||
}
|
||||
|
||||
const round = goal.roundsStarted + 1
|
||||
const content = renderGoalRoundPrompt(goal, round)
|
||||
const reservation: RoundAttempt = {
|
||||
goalId: goal.id,
|
||||
revision: goal.revision,
|
||||
round,
|
||||
content,
|
||||
phase: 'queued',
|
||||
turn: undefined,
|
||||
reason: undefined,
|
||||
rejectedReason: undefined,
|
||||
stale: false,
|
||||
}
|
||||
state.attempt = reservation
|
||||
try {
|
||||
agent.send(content, {
|
||||
source: { kind: 'goal', goalId: goal.id, revision: goal.revision, round },
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
state.attempt = undefined
|
||||
ctx.logger.warn(`goal-session: could not queue round ${round} for agent "${agent.id}": ${renderThrown(error)}`)
|
||||
const latest = currentGoal(state)
|
||||
if (latest !== undefined && latest.id === goal.id && latest.revision === goal.revision
|
||||
&& latest.phase === 'active' && latest.activation === 'armed') {
|
||||
ctx.goals.block(agent, goalRef(latest))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Coalesce triggers onto one agent-local serialized driver. */
|
||||
function requestDrive(state: DriverState): void {
|
||||
/* v8 ignore next -- teardown may race a final trigger after synchronously closing admission */
|
||||
if (state.stopping) return
|
||||
state.requested = true
|
||||
if (state.run !== undefined) return
|
||||
let run: Promise<void>
|
||||
try {
|
||||
run = ctx.agents.withoutInitiator(async () => {
|
||||
while (state.requested && !state.stopping) {
|
||||
state.requested = false
|
||||
try {
|
||||
await drive(state)
|
||||
} catch (error: unknown) {
|
||||
ctx.logger.warn(`goal-session: driver failed for agent "${state.agent.id}": ${renderThrown(error)}`)
|
||||
disarm(state)
|
||||
}
|
||||
}
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
ctx.logger.warn(`goal-session: could not start driver for agent "${state.agent.id}": ${renderThrown(error)}`)
|
||||
disarm(state)
|
||||
return
|
||||
}
|
||||
state.run = run
|
||||
const retire = (): void => {
|
||||
state.run = undefined
|
||||
if (state.requested && !state.stopping) requestDrive(state)
|
||||
}
|
||||
void run.then(retire, (error: unknown) => {
|
||||
ctx.logger.warn(`goal-session: driver task rejected for agent "${state.agent.id}": ${renderThrown(error)}`)
|
||||
disarm(state)
|
||||
retire()
|
||||
})
|
||||
}
|
||||
|
||||
// One composite effect owns every listener and the quiescent close. Cordis
|
||||
// unloads sibling effects concurrently; nesting makes the close run first
|
||||
// and keeps the admission fence installed until its drain settles.
|
||||
ctx.effect(function* () {
|
||||
/** Mark a post-turn persistence failure before idle scheduling can run. */
|
||||
ctx.on('agent/error', (agent, turn) => {
|
||||
const state = stateFor(agent)
|
||||
const last = agent.session.events.at(-1)
|
||||
if (last?.type !== 'turn/end' || last.data.turn !== turn) return
|
||||
if (state.attempt?.turn === turn) state.flushFailedTurns.add(turn)
|
||||
disarm(state)
|
||||
})
|
||||
|
||||
ctx.on('agent/created', (agent) => { stateFor(agent) })
|
||||
ctx.on('agent/disposed', (agent) => { states.delete(agent) })
|
||||
ctx.on('agent/session-start', (agent) => {
|
||||
const state = stateFor(agent)
|
||||
state.attempt = undefined
|
||||
state.openTurn = undefined
|
||||
state.competingQueued = false
|
||||
state.needsCheckpoint = false
|
||||
state.flushFailedTurns.clear()
|
||||
})
|
||||
ctx.on('agent/status', (agent, status) => {
|
||||
const state = stateFor(agent)
|
||||
if (status === 'disposed') {
|
||||
state.stopping = true
|
||||
return
|
||||
}
|
||||
if (status === 'idle') {
|
||||
state.competingQueued = false
|
||||
requestDrive(state)
|
||||
}
|
||||
})
|
||||
ctx.on('agent/queued', (agent, content, info) => {
|
||||
const state = stateFor(agent)
|
||||
const attempt = state.attempt
|
||||
if (attempt !== undefined && sameQueued(content, info.source, attempt)) return
|
||||
state.competingQueued = true
|
||||
if (attempt?.phase === 'queued') attempt.stale = true
|
||||
})
|
||||
ctx.on('agent/cancel-requested', (agent, reason) => {
|
||||
const state = stateFor(agent)
|
||||
state.attempt = undefined
|
||||
state.competingQueued = false
|
||||
const goal = currentGoal(state)
|
||||
if (goal?.phase === 'active' && goal.activation === 'armed') {
|
||||
applyOutcome(state, goal, { kind: 'pause', reason })
|
||||
}
|
||||
})
|
||||
ctx.on('goal/changed', (agent) => {
|
||||
const state = stateFor(agent)
|
||||
state.needsCheckpoint = true
|
||||
requestDrive(state)
|
||||
})
|
||||
|
||||
ctx.on('session/event', (session: Session, event: SessionEvent) => {
|
||||
const agent = ctx.agents.get(session.id)
|
||||
if (agent === undefined || agent.session !== session) return
|
||||
const state = stateFor(agent)
|
||||
switch (event.type) {
|
||||
case 'turn/start':
|
||||
state.openTurn = event.data.turn
|
||||
if (state.attempt !== undefined && isGoalRoundSource(event.data.trigger.source)
|
||||
&& sameRound(event.data.trigger.source, state.attempt)) {
|
||||
state.attempt.turn = event.data.turn
|
||||
}
|
||||
return
|
||||
case 'user/message':
|
||||
if (state.attempt !== undefined && isGoalRoundSource(event.data.source)
|
||||
&& sameRound(event.data.source, state.attempt)) {
|
||||
state.attempt.phase = 'admitted'
|
||||
/* v8 ignore next -- this driver's admitted message always follows its observed turn/start */
|
||||
if (state.openTurn !== undefined) state.attempt.turn = state.openTurn
|
||||
}
|
||||
return
|
||||
case 'prompt/blocked':
|
||||
if (state.attempt !== undefined && state.attempt.phase === 'queued'
|
||||
&& isGoalRoundSource(event.data.source) && sameRound(event.data.source, state.attempt)) {
|
||||
/* v8 ignore next -- this driver's rejected message always follows its observed turn/start */
|
||||
if (state.openTurn !== undefined) state.attempt.turn = state.openTurn
|
||||
state.attempt.rejectedReason = event.data.reason
|
||||
if (event.data.reason === STALE_ROUND_REASON) state.attempt.stale = true
|
||||
}
|
||||
return
|
||||
case 'turn/end':
|
||||
if (state.attempt?.turn === event.data.turn) state.attempt.reason = event.data.reason
|
||||
/* v8 ignore next -- balanced live turns close the open turn just observed by this listener */
|
||||
if (state.openTurn === event.data.turn) state.openTurn = undefined
|
||||
return
|
||||
default:
|
||||
return
|
||||
}
|
||||
})
|
||||
|
||||
/** Fail closed unless the queued prompt still owns the exact live revision. */
|
||||
function validReservation(
|
||||
state: DriverState,
|
||||
content: ContentBlock[],
|
||||
source: GoalMessageSource,
|
||||
): boolean {
|
||||
const attempt = state.attempt
|
||||
const goal = currentGoal(state)
|
||||
return ctx.fiber.state === FiberState.ACTIVE
|
||||
&& !state.stopping && attempt !== undefined && attempt.phase === 'queued'
|
||||
&& !attempt.stale && sameQueued(content, source, attempt)
|
||||
&& goal !== undefined && goal.id === source.goalId && goal.revision === source.revision
|
||||
&& goal.phase === 'active' && goal.activation === 'armed'
|
||||
&& source.round === goal.roundsStarted + 1
|
||||
}
|
||||
|
||||
ctx.on('agent/prompt-submit', async (agent, content, source, next): Promise<PromptDecision> => {
|
||||
if (!isGoalRoundSource(source)) return next()
|
||||
const state = stateFor(agent)
|
||||
let valid = false
|
||||
try {
|
||||
valid = validReservation(state, content, source)
|
||||
} catch (error: unknown) {
|
||||
ctx.logger.warn(`goal-session: admission check failed for agent "${agent.id}": ${renderThrown(error)}`)
|
||||
disarm(state)
|
||||
}
|
||||
if (!valid) {
|
||||
const attempt = state.attempt
|
||||
if (attempt !== undefined && sameRound(source, attempt)) attempt.stale = true
|
||||
return { kind: 'block', reason: STALE_ROUND_REASON }
|
||||
}
|
||||
const decision = await next()
|
||||
if (decision.kind === 'block') return decision
|
||||
try {
|
||||
valid = validReservation(state, content, source)
|
||||
} catch (error: unknown) {
|
||||
ctx.logger.warn(`goal-session: post-admission check failed for agent "${agent.id}": ${renderThrown(error)}`)
|
||||
disarm(state)
|
||||
valid = false
|
||||
}
|
||||
if (!valid) {
|
||||
const attempt = state.attempt
|
||||
if (attempt !== undefined && sameRound(source, attempt)) attempt.stale = true
|
||||
return { kind: 'block', reason: STALE_ROUND_REASON }
|
||||
}
|
||||
return decision
|
||||
})
|
||||
|
||||
// Loading a lifecycle driver over existing agents never inherits hidden
|
||||
// automatic authority from an earlier producer instance.
|
||||
for (const agent of ctx.agents.list()) {
|
||||
const state = stateFor(agent)
|
||||
disarm(state)
|
||||
}
|
||||
|
||||
// Yielded after listener registration, so this close runs first and the
|
||||
// composite effect removes listeners only after its promise settles.
|
||||
yield async () => {
|
||||
const waits: Promise<void>[] = []
|
||||
for (const state of states.values()) {
|
||||
state.stopping = true
|
||||
disarm(state)
|
||||
const attempt = state.attempt
|
||||
if (attempt !== undefined) {
|
||||
attempt.stale = true
|
||||
if (attempt.phase === 'admitted' && state.agent.status === 'running') {
|
||||
state.agent.cancel('goal-session driver disposed')
|
||||
}
|
||||
waits.push(state.agent.whenIdle())
|
||||
}
|
||||
if (state.run !== undefined) waits.push(state.run)
|
||||
}
|
||||
await Promise.allSettled(waits)
|
||||
states.clear()
|
||||
}
|
||||
}, 'goal-session lifecycle')
|
||||
}
|
||||
48
packages/goal/goal-session/src/outcome.ts
Normal file
48
packages/goal/goal-session/src/outcome.ts
Normal file
@@ -0,0 +1,48 @@
|
||||
/** Typed settlement policy for one admitted same-session goal round. */
|
||||
|
||||
import type { TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
|
||||
/** Driver action derived from one closed goal-owned turn. */
|
||||
export type GoalRoundOutcome =
|
||||
| { readonly kind: 'continue' }
|
||||
| { readonly kind: 'pause'; readonly reason: string }
|
||||
| { readonly kind: 'usage-limited'; readonly message: string }
|
||||
| {
|
||||
readonly kind: 'blocked'
|
||||
readonly reason: 'error' | 'max-tokens' | 'rejected' | 'unknown'
|
||||
readonly detail: string
|
||||
}
|
||||
| { readonly kind: 'disarm'; readonly reason: 'durability-failed' | 'disposed' | 'interrupted' }
|
||||
|
||||
/**
|
||||
* Classify one closed goal round without mutating goal state.
|
||||
* @param reason - durable reason from the round's `turn/end`.
|
||||
* @param durable - whether the closing flush reached its durability checkpoint.
|
||||
* @returns the single driver action; no abnormal outcome requests an automatic retry.
|
||||
*/
|
||||
export function classifyGoalRound(reason: TurnEndReason, durable: boolean): GoalRoundOutcome {
|
||||
if (!durable) return { kind: 'disarm', reason: 'durability-failed' }
|
||||
const extensibleReason: { readonly kind: string } = reason
|
||||
switch (reason.kind) {
|
||||
case 'completed':
|
||||
return { kind: 'continue' }
|
||||
case 'aborted':
|
||||
return { kind: 'pause', reason: reason.reason ?? 'cancelled' }
|
||||
case 'error':
|
||||
return reason.code === 'RATE_LIMIT'
|
||||
? { kind: 'usage-limited', message: reason.message }
|
||||
: { kind: 'blocked', reason: 'error', detail: reason.message }
|
||||
case 'max-tokens':
|
||||
return { kind: 'blocked', reason: 'max-tokens', detail: 'model output reached max tokens' }
|
||||
case 'rejected':
|
||||
return { kind: 'blocked', reason: 'rejected', detail: reason.reason }
|
||||
case 'disposed':
|
||||
return { kind: 'disarm', reason: 'disposed' }
|
||||
case 'interrupted':
|
||||
return { kind: 'disarm', reason: 'interrupted' }
|
||||
// TurnEndReason is merge-extensible. An unknown producer cannot opt into
|
||||
// automatic retry merely by adding a tag; stop for inspection instead.
|
||||
default:
|
||||
return { kind: 'blocked', reason: 'unknown', detail: `unknown turn outcome: ${extensibleReason.kind}` }
|
||||
}
|
||||
}
|
||||
26
packages/goal/goal-session/src/prompt.ts
Normal file
26
packages/goal/goal-session/src/prompt.ts
Normal file
@@ -0,0 +1,26 @@
|
||||
/** Model-visible continuation prompt for one same-session goal round. */
|
||||
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { GoalView } from '@deepseek-ai/dsh-goal'
|
||||
|
||||
/**
|
||||
* Render the complete goal-round instruction retained in session history.
|
||||
* @param goal - exact active goal revision being admitted.
|
||||
* @param round - next positive round number.
|
||||
* @returns a fresh one-block prompt for `Agent.send()`.
|
||||
*/
|
||||
export function renderGoalRoundPrompt(goal: GoalView, round: number): ContentBlock[] {
|
||||
return [{
|
||||
type: 'text',
|
||||
text: '<goal_round>\n'
|
||||
+ `Objective: ${JSON.stringify(goal.objective)}\n`
|
||||
+ `Round: ${round}/${goal.maxGoalRounds}\n\n`
|
||||
+ 'Continue working toward the objective in this same session. Treat the current workspace, '
|
||||
+ 'tool results, and durable session state as authoritative; inspect them instead of assuming '
|
||||
+ 'earlier narration is still current. Make concrete progress and verify the result. Before '
|
||||
+ 'claiming completion, gather evidence that the whole objective is achieved, read the current '
|
||||
+ 'goal, and mark it complete. If work remains, leave the goal active for the next round. Follow '
|
||||
+ 'the configured goal-tool policy before reporting a blocker.\n'
|
||||
+ '</goal_round>',
|
||||
}]
|
||||
}
|
||||
138
packages/goal/goal-session/tests/goal-session.e2e.ts
Normal file
138
packages/goal/goal-session/tests/goal-session.e2e.ts
Normal file
@@ -0,0 +1,138 @@
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
|
||||
import { mkdtemp, readFile, readdir, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { foldGoal } from '@deepseek-ai/dsh-goal'
|
||||
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
|
||||
|
||||
const binScript = fileURLToPath(new URL('../../../examples/stdio-demo/src/bin.ts', import.meta.url))
|
||||
const configPath = fileURLToPath(new URL(
|
||||
'../../../../examples/echo-agent/tests/fixtures/goal/goal-session/cordis.yml',
|
||||
import.meta.url,
|
||||
))
|
||||
const repoTsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url))
|
||||
const PROCESS_TIMEOUT_MS = 30_000
|
||||
const TEST_TIMEOUT_MS = PROCESS_TIMEOUT_MS + 15_000
|
||||
|
||||
let child: ChildProcessWithoutNullStreams | undefined
|
||||
let workdir: string | undefined
|
||||
|
||||
afterEach(async () => {
|
||||
if (child !== undefined && child.exitCode === null) child.kill('SIGKILL')
|
||||
child = undefined
|
||||
if (workdir !== undefined) await rm(workdir, { recursive: true, force: true })
|
||||
workdir = undefined
|
||||
})
|
||||
|
||||
/** Recursively locate persistence JSONL files in one temporary root. */
|
||||
async function jsonlFiles(dir: string): Promise<string[]> {
|
||||
const entries = await readdir(dir, { withFileTypes: true })
|
||||
const paths = await Promise.all(entries.map(async (entry) => {
|
||||
const path = join(dir, entry.name)
|
||||
if (entry.isDirectory()) return jsonlFiles(path)
|
||||
return entry.isFile() && entry.name.endsWith('.jsonl') ? [path] : []
|
||||
}))
|
||||
return paths.flat()
|
||||
}
|
||||
|
||||
/** Run the complete deterministic human-turn plus two-round composition. */
|
||||
async function runComposition(): Promise<{ stdout: string; stderr: string }> {
|
||||
workdir = await mkdtemp(join(tmpdir(), 'goal-session-e2e-'))
|
||||
const cwd = workdir
|
||||
return new Promise((resolve, reject) => {
|
||||
const launch = resolveExampleLaunch({
|
||||
srcBin: binScript,
|
||||
configArgs: [configPath],
|
||||
tsconfigPath: repoTsconfig,
|
||||
exposeInternals: true,
|
||||
env: {
|
||||
DSH_HOME: join(cwd, '.dsh'),
|
||||
DSH_AGENTS_HOME: join(cwd, '.agents'),
|
||||
},
|
||||
})
|
||||
const proc = spawn(launch.command, launch.args, {
|
||||
cwd,
|
||||
env: { ...process.env, ...launch.env },
|
||||
stdio: ['pipe', 'pipe', 'pipe'],
|
||||
})
|
||||
child = proc
|
||||
let stdout = ''
|
||||
let stderr = ''
|
||||
let inputClosed = false
|
||||
proc.stdout.setEncoding('utf8')
|
||||
proc.stdout.on('data', (chunk: string) => {
|
||||
stdout += chunk
|
||||
if (!inputClosed && stdout.includes('ROUND TWO COMPLETE') && stdout.includes('\n> ')) {
|
||||
inputClosed = true
|
||||
proc.stdin.end()
|
||||
}
|
||||
})
|
||||
proc.stderr.setEncoding('utf8')
|
||||
proc.stderr.on('data', (chunk: string) => { stderr += chunk })
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
proc.kill('SIGKILL')
|
||||
reject(new Error(
|
||||
`goal-session e2e did not exit within ${PROCESS_TIMEOUT_MS / 1_000}s. stdout:\n${stdout}\nstderr:\n${stderr}`,
|
||||
))
|
||||
}, PROCESS_TIMEOUT_MS)
|
||||
proc.on('exit', (code) => {
|
||||
clearTimeout(timer)
|
||||
if (code === 0) resolve({ stdout, stderr })
|
||||
else reject(new Error(`goal-session e2e exited ${String(code)}. stdout:\n${stdout}\nstderr:\n${stderr}`))
|
||||
})
|
||||
proc.on('error', (error) => { clearTimeout(timer); reject(error) })
|
||||
proc.stdin.write('start\n')
|
||||
})
|
||||
}
|
||||
|
||||
describe('same-session goal rounds through a real Loader, app, and stdio process', () => {
|
||||
it('persists two exact rounds and stops the completion turn without another request', async () => {
|
||||
const { stdout, stderr } = await runComposition()
|
||||
expect(stderr).not.toContain('UNHANDLED')
|
||||
expect(stdout).toContain('goal-session e2e ready.')
|
||||
expect(stdout).toContain('GOAL CREATED')
|
||||
expect(stdout).toContain('ROUND ONE')
|
||||
expect(stdout).toContain('ROUND TWO COMPLETE')
|
||||
|
||||
const logs = await jsonlFiles(join(workdir as string, '.sessions'))
|
||||
expect(logs).toHaveLength(1)
|
||||
const lines = (await readFile(logs[0] as string, 'utf8')).trimEnd().split('\n')
|
||||
const events = lines.slice(1).map(line => JSON.parse(line) as SessionEvent)
|
||||
|
||||
const calls = events.filter(event => event.type === 'tool/call')
|
||||
expect(calls.map(event => event.data.name)).toEqual(['create_goal', 'get_goal', 'update_goal'])
|
||||
expect(events.filter(event => event.type === 'step/start')).toHaveLength(5)
|
||||
expect(events.filter(event => event.type === 'tool/result').every(event => !event.data.isError)).toBe(true)
|
||||
|
||||
const rounds = events.filter(event => event.type === 'user/message'
|
||||
&& event.data.source.kind === 'goal')
|
||||
expect(rounds).toHaveLength(2)
|
||||
const roundNumbers: number[] = []
|
||||
const revisions: number[] = []
|
||||
const prompts: string[] = []
|
||||
for (const event of events) {
|
||||
if (event.type !== 'user/message' || event.data.source.kind !== 'goal') continue
|
||||
roundNumbers.push(event.data.source.round)
|
||||
revisions.push(event.data.source.revision)
|
||||
prompts.push(event.data.content.find(block => block.type === 'text')?.text ?? '')
|
||||
}
|
||||
expect(roundNumbers).toEqual([1, 2])
|
||||
expect(revisions).toEqual([1, 1])
|
||||
expect(prompts[0]).toContain('Round: 1/2')
|
||||
expect(prompts[1]).toContain('Round: 2/2')
|
||||
|
||||
expect(foldGoal(events)).toMatchObject({
|
||||
goal: {
|
||||
objective: 'Complete two deterministic same-session rounds',
|
||||
phase: 'complete',
|
||||
revision: 2,
|
||||
maxGoalRounds: 2,
|
||||
},
|
||||
roundsStarted: 2,
|
||||
})
|
||||
}, TEST_TIMEOUT_MS)
|
||||
})
|
||||
648
packages/goal/goal-session/tests/goal-session.spec.ts
Normal file
648
packages/goal/goal-session/tests/goal-session.spec.ts
Normal file
@@ -0,0 +1,648 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { agentEvents } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
||||
import GoalService, { GoalId } from '@deepseek-ai/dsh-goal'
|
||||
import type { GoalView } from '@deepseek-ai/dsh-goal'
|
||||
import { LlmAdapter } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import * as goalSession from '../src/index.ts'
|
||||
|
||||
type ScriptEntry = StreamChunk[] | Error | 'hang' | ((options: GenerateOptions) => StreamChunk[])
|
||||
|
||||
/** Small request-recording adapter with controllable failure and cancellation. */
|
||||
class ScriptedAdapter extends LlmAdapter {
|
||||
readonly requests: GenerateOptions[] = []
|
||||
|
||||
constructor(private readonly script: ScriptEntry[]) {
|
||||
super()
|
||||
}
|
||||
|
||||
override async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
this.requests.push(options)
|
||||
const entry = this.script.shift()
|
||||
if (entry === undefined) throw new Error('ScriptedAdapter: script exhausted')
|
||||
if (entry instanceof Error) throw entry
|
||||
if (entry === 'hang') {
|
||||
yield { type: 'block-start', index: 0, blockType: 'text' }
|
||||
yield { type: 'text-delta', index: 0, text: 'partial' }
|
||||
await new Promise<void>((_resolve, reject) => {
|
||||
if (options.signal?.aborted) {
|
||||
reject(new Error('aborted'))
|
||||
return
|
||||
}
|
||||
options.signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
|
||||
})
|
||||
return
|
||||
}
|
||||
const chunks = typeof entry === 'function' ? entry(options) : entry
|
||||
for (const chunk of chunks) yield chunk
|
||||
}
|
||||
}
|
||||
|
||||
/** One successful text response. */
|
||||
function textResponse(text: string): StreamChunk[] {
|
||||
return [
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text } },
|
||||
{ type: 'finish', reason: { kind: 'stop' } },
|
||||
]
|
||||
}
|
||||
|
||||
/** One successful response cut off at the model output limit. */
|
||||
function maxTokensResponse(text: string): StreamChunk[] {
|
||||
return [
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text } },
|
||||
{ type: 'finish', reason: { kind: 'max-tokens' } },
|
||||
]
|
||||
}
|
||||
|
||||
/** Complete request history as a single string for ordering assertions. */
|
||||
function requestText(request: GenerateOptions): string {
|
||||
return request.messages
|
||||
.flatMap(message => message.content)
|
||||
.filter(block => block.type === 'text')
|
||||
.map(block => block.text)
|
||||
.join('\n')
|
||||
}
|
||||
|
||||
interface Harness {
|
||||
readonly ctx: Context
|
||||
readonly adapter: ScriptedAdapter
|
||||
readonly agent: Agent
|
||||
readonly driver: Awaited<ReturnType<Context['plugin']>>
|
||||
}
|
||||
|
||||
const contexts: Context[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.allSettled(contexts.splice(0).map(context => context.fiber.dispose()))
|
||||
})
|
||||
|
||||
/** Mount a real loop with only its model scripted. */
|
||||
async function harness(script: ScriptEntry[]): Promise<Harness> {
|
||||
const ctx = new Context()
|
||||
contexts.push(ctx)
|
||||
await mountAgentLoopTestDependencies(ctx)
|
||||
await ctx.plugin(GoalService)
|
||||
const driver = await ctx.plugin(goalSession)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
const adapter = new ScriptedAdapter(script)
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId(`goal-session-${Math.random()}`), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
return { ctx, adapter, agent, driver }
|
||||
}
|
||||
|
||||
/** Await a stable goal projection selected by the caller. */
|
||||
async function waitForGoal(
|
||||
ctx: Context,
|
||||
agent: Agent,
|
||||
predicate: (goal: GoalView | undefined) => boolean,
|
||||
): Promise<GoalView | undefined> {
|
||||
await vi.waitFor(() => {
|
||||
expect(predicate(ctx.goals.get(agent))).toBe(true)
|
||||
})
|
||||
return ctx.goals.get(agent)
|
||||
}
|
||||
|
||||
/** Await a specific number of dispatched model requests. */
|
||||
async function waitForRequests(adapter: ScriptedAdapter, count: number): Promise<void> {
|
||||
await vi.waitFor(() => {
|
||||
expect(adapter.requests).toHaveLength(count)
|
||||
})
|
||||
}
|
||||
|
||||
describe('goal-round outcome policy', () => {
|
||||
it.each([
|
||||
[{ kind: 'completed' }, true, { kind: 'continue' }],
|
||||
[{ kind: 'aborted', reason: 'operator stopped' }, true, { kind: 'pause', reason: 'operator stopped' }],
|
||||
[{ kind: 'aborted' }, true, { kind: 'pause', reason: 'cancelled' }],
|
||||
[{ kind: 'error', step: 1, message: 'slow down', code: 'RATE_LIMIT' }, true,
|
||||
{ kind: 'usage-limited', message: 'slow down' }],
|
||||
[{ kind: 'error', step: 1, message: 'broken' }, true,
|
||||
{ kind: 'blocked', reason: 'error', detail: 'broken' }],
|
||||
[{ kind: 'max-tokens' }, true,
|
||||
{ kind: 'blocked', reason: 'max-tokens', detail: 'model output reached max tokens' }],
|
||||
[{ kind: 'rejected', reason: 'policy' }, true,
|
||||
{ kind: 'blocked', reason: 'rejected', detail: 'policy' }],
|
||||
[{ kind: 'disposed' }, true, { kind: 'disarm', reason: 'disposed' }],
|
||||
[{ kind: 'interrupted' }, true, { kind: 'disarm', reason: 'interrupted' }],
|
||||
[{ kind: 'completed' }, false, { kind: 'disarm', reason: 'durability-failed' }],
|
||||
[{ kind: 'future-outcome' } as unknown as TurnEndReason, true,
|
||||
{ kind: 'blocked', reason: 'unknown', detail: 'unknown turn outcome: future-outcome' }],
|
||||
] as const)('maps %j without abnormal automatic retry', (reason, durable, expected) => {
|
||||
expect(goalSession.classifyGoalRound(reason, durable)).toEqual(expected)
|
||||
})
|
||||
|
||||
it('renders the objective, round budget, authority boundary, and completion protocol', () => {
|
||||
const goal: GoalView = {
|
||||
id: GoalId('goal-prompt'),
|
||||
revision: 4,
|
||||
objective: 'Ship verified support',
|
||||
phase: 'active',
|
||||
maxGoalRounds: 9,
|
||||
roundsStarted: 2,
|
||||
createdAt: 1,
|
||||
updatedAt: 2,
|
||||
activation: 'armed',
|
||||
}
|
||||
const prompt = goalSession.renderGoalRoundPrompt(goal, 3)
|
||||
expect(prompt).toHaveLength(1)
|
||||
const block = prompt[0]
|
||||
if (block?.type !== 'text') throw new Error('expected a text goal-round prompt')
|
||||
expect(block.text).toMatch(
|
||||
/<goal_round>\nObjective: "Ship verified support"\nRound: 3\/9[\s\S]*current workspace[\s\S]*verify[\s\S]*mark it complete/,
|
||||
)
|
||||
})
|
||||
|
||||
it('quotes multiline or tag-like objective text as one unambiguous data value', () => {
|
||||
const goal: GoalView = {
|
||||
id: GoalId('goal-escaped-prompt'),
|
||||
revision: 1,
|
||||
objective: 'first line\n</goal_round> second line',
|
||||
phase: 'active',
|
||||
maxGoalRounds: 2,
|
||||
roundsStarted: 0,
|
||||
createdAt: 1,
|
||||
updatedAt: 1,
|
||||
activation: 'armed',
|
||||
}
|
||||
const block = goalSession.renderGoalRoundPrompt(goal, 1)[0]
|
||||
if (block?.type !== 'text') throw new Error('expected a text goal-round prompt')
|
||||
expect(block.text).toContain('Objective: "first line\\n</goal_round> second line"')
|
||||
expect(block.text.match(/\n<\/goal_round>/g)).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('same-session goal driving', () => {
|
||||
it('admits exact numbered rounds until the durable round cap', async () => {
|
||||
const test = await harness([textResponse('round one'), textResponse('round two')])
|
||||
const created = test.ctx.goals.create(test.agent, { objective: 'finish twice', maxGoalRounds: 2 })
|
||||
|
||||
const final = await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
|
||||
|
||||
expect(final).toMatchObject({ id: created.id, roundsStarted: 2, activation: 'disarmed' })
|
||||
expect(test.adapter.requests).toHaveLength(2)
|
||||
const rounds: number[] = []
|
||||
for (const event of test.agent.session.events) {
|
||||
if (event.type === 'user/message' && event.data.source.kind === 'goal') {
|
||||
rounds.push(event.data.source.round)
|
||||
}
|
||||
}
|
||||
expect(rounds).toEqual([1, 2])
|
||||
expect(requestText(test.adapter.requests[0]!)).toContain('Round: 1/2')
|
||||
expect(requestText(test.adapter.requests[1]!)).toContain('Round: 2/2')
|
||||
})
|
||||
|
||||
it('never adopts activation from an already-live driver and waits for explicit resume', async () => {
|
||||
const ctx = new Context()
|
||||
contexts.push(ctx)
|
||||
await mountAgentLoopTestDependencies(ctx)
|
||||
await ctx.plugin(GoalService)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
const adapter = new ScriptedAdapter([textResponse('after resume')])
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('goal-session-hot-load'), { provider: 'mock', model: 'mock' })
|
||||
const created = ctx.goals.create(agent, { objective: 'wait for a human', maxGoalRounds: 1 })
|
||||
|
||||
await ctx.plugin(goalSession)
|
||||
await Promise.resolve()
|
||||
expect(ctx.goals.get(agent)).toMatchObject({ phase: 'active', activation: 'disarmed', revision: 1 })
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
|
||||
ctx.goals.resume(agent, created)
|
||||
await waitForGoal(ctx, agent, goal => goal?.phase === 'budget-limited')
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['rate limit', Object.assign(new Error('slow down'), { code: 'RATE_LIMIT' }), 'usage-limited'],
|
||||
['request error', new Error('provider broke'), 'blocked'],
|
||||
['max tokens', maxTokensResponse('unfinished'), 'blocked'],
|
||||
] as const)('stops after a %s without an automatic retry', async (_label, response, phase) => {
|
||||
const test = await harness([response])
|
||||
test.ctx.goals.create(test.agent, { objective: 'stop safely', maxGoalRounds: 8 })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === phase)
|
||||
|
||||
expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' })
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('maps a downstream prompt veto to blocked without admitting the round', async () => {
|
||||
const test = await harness([])
|
||||
test.ctx.on('agent/prompt-submit', (_agent, _content, source, next) => source.kind === 'goal'
|
||||
? Promise.resolve({ kind: 'block', reason: 'deployment policy' })
|
||||
: next())
|
||||
test.ctx.goals.create(test.agent, { objective: 'respect policy' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
|
||||
|
||||
expect(goal?.roundsStarted).toBe(0)
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
expect(test.agent.session.events.some(event => event.type === 'prompt/blocked'
|
||||
&& event.data.reason === 'deployment policy')).toBe(true)
|
||||
})
|
||||
|
||||
it('does not reserve again when a stopped-goal observer queues ordinary work', async () => {
|
||||
const test = await harness([textResponse('human follow-up')])
|
||||
test.ctx.on('agent/prompt-submit', (_agent, _content, source, next) => source.kind === 'goal'
|
||||
? Promise.resolve({ kind: 'block', reason: 'stop this round' })
|
||||
: next())
|
||||
test.ctx.on('goal/changed', (agent, change) => {
|
||||
if (change.operation === 'block') agent.send([{ type: 'text', text: 'inspect the blocker' }])
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'stop and inspect' })
|
||||
|
||||
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
|
||||
await waitForRequests(test.adapter, 1)
|
||||
await test.agent.whenIdle()
|
||||
|
||||
expect(requestText(test.adapter.requests[0]!)).toContain('inspect the blocker')
|
||||
})
|
||||
|
||||
it('pauses and drops a reserved round when cancellation lands before admission', async () => {
|
||||
const test = await harness([])
|
||||
const cancel = test.ctx.on('agent/queued', (agent, _content, info) => {
|
||||
if (agent === test.agent && info.source.kind === 'goal') {
|
||||
cancel()
|
||||
agent.cancel('operator cancelled pending goal')
|
||||
}
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'do not start yet' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused')
|
||||
|
||||
expect(goal).toMatchObject({ roundsStarted: 0, activation: 'disarmed' })
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
expect(test.agent.session.events.some(event => event.type === 'user/message'
|
||||
&& event.data.source.kind === 'goal')).toBe(false)
|
||||
})
|
||||
|
||||
it('pauses an admitted round when cancellation aborts an active step', async () => {
|
||||
const test = await harness(['hang'])
|
||||
test.ctx.goals.create(test.agent, { objective: 'stop in flight' })
|
||||
await waitForRequests(test.adapter, 1)
|
||||
|
||||
test.agent.cancel('operator stopped active goal')
|
||||
await test.agent.whenIdle()
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused')
|
||||
|
||||
expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' })
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('lets already-queued human work finish before reserving the next round', async () => {
|
||||
const test = await harness([textResponse('human answer'), textResponse('goal answer')])
|
||||
test.ctx.goals.create(test.agent, { objective: 'continue after the human', maxGoalRounds: 1 })
|
||||
test.agent.send([{ type: 'text', text: 'human goes first' }])
|
||||
|
||||
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
|
||||
|
||||
expect(test.adapter.requests).toHaveLength(2)
|
||||
expect(requestText(test.adapter.requests[0]!)).toContain('human goes first')
|
||||
expect(requestText(test.adapter.requests[0]!)).not.toContain('<goal_round>')
|
||||
expect(requestText(test.adapter.requests[1]!)).toContain('<goal_round>')
|
||||
})
|
||||
|
||||
it('makes a reserved round stale when a listener queues human work behind it', async () => {
|
||||
const test = await harness([textResponse('human batch'), textResponse('later goal')])
|
||||
let inserted = false
|
||||
test.ctx.on('agent/queued', (agent, _content, info) => {
|
||||
if (agent !== test.agent || info.source.kind !== 'goal' || inserted) return
|
||||
inserted = true
|
||||
agent.send([{ type: 'text', text: 'human joined the pending batch' }])
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'yield to nested human input', maxGoalRounds: 1 })
|
||||
|
||||
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
|
||||
|
||||
expect(test.adapter.requests).toHaveLength(2)
|
||||
expect(requestText(test.adapter.requests[0]!)).toContain('human joined the pending batch')
|
||||
expect(requestText(test.adapter.requests[0]!)).not.toContain('<goal_round>')
|
||||
expect(requestText(test.adapter.requests[1]!)).toContain('<goal_round>')
|
||||
})
|
||||
|
||||
it('blocks a queued reservation made stale by a goal edit and continues the new revision', async () => {
|
||||
const test = await harness([textResponse('new revision')])
|
||||
let edited = false
|
||||
test.ctx.on('agent/queued', (agent, _content, info) => {
|
||||
if (agent !== test.agent || info.source.kind !== 'goal' || edited) return
|
||||
edited = true
|
||||
const current = test.ctx.goals.get(agent)
|
||||
if (current === undefined) throw new Error('missing goal during queued edit')
|
||||
test.ctx.goals.edit(agent, current, { objective: 'new objective' })
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'old objective', maxGoalRounds: 1 })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'budget-limited')
|
||||
|
||||
expect(goal).toMatchObject({ revision: 3, objective: 'new objective', roundsStarted: 1 })
|
||||
const blocked = test.agent.session.events.find(event => event.type === 'prompt/blocked')
|
||||
expect(blocked?.type === 'prompt/blocked' ? blocked.data.reason : undefined)
|
||||
.toBe('stale goal-round reservation')
|
||||
const admitted = test.agent.session.events.find(event => event.type === 'user/message'
|
||||
&& event.data.source.kind === 'goal')
|
||||
expect(admitted?.type === 'user/message' && admitted.data.source.kind === 'goal'
|
||||
? admitted.data.source.revision
|
||||
: undefined).toBe(2)
|
||||
})
|
||||
|
||||
it('rechecks revision after downstream prompt hooks before admitting', async () => {
|
||||
const test = await harness([textResponse('new revision')])
|
||||
let edited = false
|
||||
test.ctx.on('agent/prompt-submit', (agent, _content, source, next) => {
|
||||
if (source.kind === 'goal' && !edited) {
|
||||
edited = true
|
||||
const current = test.ctx.goals.get(agent)
|
||||
if (current === undefined) throw new Error('missing goal during prompt edit')
|
||||
test.ctx.goals.edit(agent, current, { objective: 'edited downstream' })
|
||||
}
|
||||
return next()
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'edit during admission', maxGoalRounds: 1 })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'budget-limited')
|
||||
|
||||
expect(goal).toMatchObject({ objective: 'edited downstream', roundsStarted: 1 })
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
expect(test.agent.session.events.some(event => event.type === 'prompt/blocked'
|
||||
&& event.data.reason === 'stale goal-round reservation')).toBe(true)
|
||||
})
|
||||
|
||||
it('disarms without dispatch when a durability checkpoint fails', async () => {
|
||||
const test = await harness([])
|
||||
test.ctx.on('session/flush', () => Promise.reject(new Error('disk unavailable')))
|
||||
test.ctx.goals.create(test.agent, { objective: 'do not outrun storage' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
|
||||
|
||||
expect(goal).toMatchObject({ phase: 'active', roundsStarted: 0 })
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('contains a checkpoint failure after a clear notification leaves no current goal', async () => {
|
||||
const test = await harness([])
|
||||
test.ctx.on('session/flush', () => Promise.reject(new Error('clear checkpoint failed')))
|
||||
agentEvents(test.ctx, test.agent).emit('goal/changed', {
|
||||
operation: 'clear',
|
||||
ref: { id: GoalId('cleared-goal'), revision: 2 },
|
||||
})
|
||||
await new Promise<void>((resolve) => { setImmediate(resolve) })
|
||||
|
||||
expect(test.ctx.goals.get(test.agent)).toBeUndefined()
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('disarms an admitted round whose closing durability checkpoint fails', async () => {
|
||||
const test = await harness([textResponse('not durable')])
|
||||
test.ctx.on('session/flush', (session) => {
|
||||
const lastStart = session.events.findLast(event => event.type === 'turn/start')
|
||||
if (lastStart?.type === 'turn/start' && lastStart.data.trigger.kind === 'message'
|
||||
&& lastStart.data.trigger.source.kind === 'goal') {
|
||||
return Promise.reject(new Error('round flush failed'))
|
||||
}
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'checkpoint the result' })
|
||||
|
||||
const goal = await waitForGoal(
|
||||
test.ctx,
|
||||
test.agent,
|
||||
current => current?.roundsStarted === 1 && current.activation === 'disarmed',
|
||||
)
|
||||
|
||||
expect(goal?.phase).toBe('active')
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('blocks the goal when a custom agent rejects the otherwise valid send', async () => {
|
||||
const test = await harness([])
|
||||
vi.spyOn(test.agent, 'send').mockImplementationOnce(() => {
|
||||
throw new Error('queue rejected')
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'handle queue failure' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
|
||||
|
||||
expect(goal).toMatchObject({ roundsStarted: 0, activation: 'disarmed' })
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('preserves a custom agent side effect when send disarms before throwing', async () => {
|
||||
const test = await harness([])
|
||||
vi.spyOn(test.agent, 'send').mockImplementationOnce(() => {
|
||||
test.ctx.goals.disarm(test.agent)
|
||||
throw new Error('queue rejected after disarm')
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'preserve the newer activation state' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
|
||||
|
||||
expect(goal).toMatchObject({ phase: 'active', roundsStarted: 0 })
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('contains a driver read failure and removes continuation authority', async () => {
|
||||
const test = await harness([])
|
||||
let flushes = 0
|
||||
test.ctx.on('session/flush', () => {
|
||||
flushes += 1
|
||||
if (flushes !== 2) return
|
||||
vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => {
|
||||
throw new Error('corrupt projection')
|
||||
})
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'fail the driver closed' })
|
||||
await new Promise<void>((resolve) => { setImmediate(resolve) })
|
||||
|
||||
const goal = test.ctx.goals.get(test.agent)
|
||||
|
||||
expect(goal?.phase).toBe('active')
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('contains synchronous scheduler startup failure', async () => {
|
||||
const test = await harness([])
|
||||
vi.spyOn(test.ctx.agents, 'withoutInitiator').mockImplementationOnce(() => {
|
||||
throw 'scheduler closed'
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'fail startup closed' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
|
||||
|
||||
expect(goal?.phase).toBe('active')
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('contains an asynchronously rejected scheduler task', async () => {
|
||||
const test = await harness([])
|
||||
vi.spyOn(test.ctx.agents, 'withoutInitiator').mockImplementationOnce(
|
||||
() => Promise.reject(new Error('scheduler task rejected')),
|
||||
)
|
||||
test.ctx.goals.create(test.agent, { objective: 'fail task closed' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
|
||||
|
||||
expect(goal?.phase).toBe('active')
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('fails a pre-admission read closed even when the first disarm attempt throws', async () => {
|
||||
const test = await harness([textResponse('retry after containment')])
|
||||
let armed = true
|
||||
test.ctx.on('agent/queued', (agent, _content, info) => {
|
||||
if (agent !== test.agent || info.source.kind !== 'goal' || !armed) return
|
||||
armed = false
|
||||
vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => {
|
||||
throw new Error('admission projection failed')
|
||||
})
|
||||
vi.spyOn(test.ctx.goals, 'disarm').mockImplementationOnce(() => {
|
||||
throw 'disarm failed'
|
||||
})
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'retry stale admission', maxGoalRounds: 1 })
|
||||
|
||||
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
|
||||
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
expect(test.agent.session.events.some(event => event.type === 'prompt/blocked'
|
||||
&& event.data.reason === 'stale goal-round reservation')).toBe(true)
|
||||
})
|
||||
|
||||
it('fails a post-hook read closed before the prompt can enter history', async () => {
|
||||
const test = await harness([])
|
||||
let armed = true
|
||||
test.ctx.on('agent/prompt-submit', (_agent, _content, source, next) => {
|
||||
if (source.kind === 'goal' && armed) {
|
||||
armed = false
|
||||
vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => {
|
||||
throw new Error('post-hook projection failed')
|
||||
})
|
||||
}
|
||||
return next()
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'block post-hook failure' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
|
||||
|
||||
expect(goal).toMatchObject({ phase: 'active', roundsStarted: 0 })
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('blocks forged goal attribution without touching an absent reservation', async () => {
|
||||
const test = await harness([])
|
||||
test.agent.send([{ type: 'text', text: 'forged automatic work' }], {
|
||||
source: { kind: 'goal', goalId: GoalId('forged-goal'), revision: 1, round: 1 },
|
||||
})
|
||||
await test.agent.whenIdle()
|
||||
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
expect(test.agent.session.events.some(event => event.type === 'prompt/blocked'
|
||||
&& event.data.reason === 'stale goal-round reservation')).toBe(true)
|
||||
})
|
||||
|
||||
it('does not invent goal state when ordinary queued work is cancelled', async () => {
|
||||
const test = await harness([])
|
||||
test.agent.send([{ type: 'text', text: 'cancel ordinary work' }])
|
||||
test.agent.cancel('ordinary cancellation')
|
||||
await test.agent.whenIdle()
|
||||
|
||||
expect(test.ctx.goals.get(test.agent)).toBeUndefined()
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('blocks admission when downstream cancellation clears the reservation', async () => {
|
||||
const test = await harness([])
|
||||
let cancelled = false
|
||||
test.ctx.on('agent/prompt-submit', (agent, _content, source, next) => {
|
||||
if (source.kind === 'goal' && !cancelled) {
|
||||
cancelled = true
|
||||
agent.cancel('cancel from downstream admission policy')
|
||||
}
|
||||
return next()
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'cancel during admission' })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused')
|
||||
await test.agent.whenIdle()
|
||||
|
||||
expect(goal?.roundsStarted).toBe(0)
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('disarms and cancels an admitted round before driver teardown completes', async () => {
|
||||
const test = await harness(['hang'])
|
||||
test.ctx.goals.create(test.agent, { objective: 'survive plugin unload' })
|
||||
await waitForRequests(test.adapter, 1)
|
||||
|
||||
await test.driver.dispose()
|
||||
|
||||
expect(test.ctx.goals.get(test.agent)).toMatchObject({
|
||||
phase: 'active',
|
||||
activation: 'disarmed',
|
||||
roundsStarted: 1,
|
||||
})
|
||||
await test.agent.whenIdle()
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('cancels an accepted queued round and awaits its driver task during teardown', async () => {
|
||||
const test = await harness([])
|
||||
let unloading: Promise<void> | undefined
|
||||
test.ctx.on('agent/queued', (agent, _content, info) => {
|
||||
if (agent === test.agent && info.source.kind === 'goal' && unloading === undefined) {
|
||||
unloading = Promise.resolve(test.driver.dispose())
|
||||
}
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'unload while queued' })
|
||||
await vi.waitFor(() => { expect(unloading).toBeDefined() })
|
||||
await unloading
|
||||
|
||||
expect(test.ctx.goals.get(test.agent)).toMatchObject({
|
||||
phase: 'active',
|
||||
activation: 'disarmed',
|
||||
roundsStarted: 1,
|
||||
})
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('resets process-local scheduling state at a session-start edge', async () => {
|
||||
const test = await harness([textResponse('after explicit resume')])
|
||||
const created = test.ctx.goals.create(test.agent, { objective: 'restart safely', maxGoalRounds: 1 })
|
||||
agentEvents(test.ctx, test.agent).emit('agent/session-start', 'resume')
|
||||
await Promise.resolve()
|
||||
|
||||
expect(test.ctx.goals.get(test.agent)).toMatchObject({ activation: 'disarmed', roundsStarted: 0 })
|
||||
expect(test.adapter.requests).toHaveLength(0)
|
||||
|
||||
test.ctx.goals.resume(test.agent, created)
|
||||
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('ignores session events without an exact owning agent and retires disposed agent state', async () => {
|
||||
const test = await harness([])
|
||||
const orphan = test.ctx.sessions.create(SessionId('goal-session-orphan'))
|
||||
orphan.append('turn/start', {
|
||||
turn: 1,
|
||||
trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'test' } },
|
||||
})
|
||||
orphan.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
|
||||
const handle = await test.ctx.agents.create({
|
||||
sessionId: SessionId('goal-session-disposed'),
|
||||
agentOptions: { provider: 'mock', model: 'mock' },
|
||||
})
|
||||
await handle.dispose()
|
||||
|
||||
expect(test.ctx.agents.get(handle.agent.id)).toBeUndefined()
|
||||
})
|
||||
})
|
||||
30
packages/goal/goal-session/tsconfig.json
Normal file
30
packages/goal/goal-session/tsconfig.json
Normal file
@@ -0,0 +1,30 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib/types"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../core/agent"
|
||||
},
|
||||
{
|
||||
"path": "../goal"
|
||||
},
|
||||
{
|
||||
"path": "../../llm/llm"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -15,7 +15,7 @@ Event-sourced same-session goal state. The service retains one current completio
|
||||
|
||||
## Service contract
|
||||
|
||||
`ctx.goals` accepts only the exact live `Agent` instance registered under its id. `get()` returns a detached `GoalView`; mutations use a `GoalRef { id, revision }` compare-and-set fence and reject stale refs. The service exposes create, edit, pause, resume, complete, block, usage-limit, budget-limit, and clear verbs through the generated [service catalog](../../../docs/cordis-catalog/services.md).
|
||||
`ctx.goals` accepts only the exact live `Agent` instance registered under its id. `get()` returns a detached `GoalView`; mutations use a `GoalRef { id, revision }` compare-and-set fence and reject stale refs. The service exposes create, edit, pause, resume, complete, block, usage-limit, budget-limit, and clear verbs through the generated [service catalog](../../../docs/cordis-catalog/services.md). `disarm()` is the lifecycle-only exception: it removes process-local continuation authority without writing a revision or emitting a mutation.
|
||||
|
||||
At most one goal is current. Creation produces an active revision-one goal and arms it. A non-complete goal must be edited, transitioned, or cleared; a completed goal may be replaced by a globally fresh id. Edits retain phase and activation. Pause, completion, blocking, limit transitions, and clear disarm activation. Resume accepts a stopped phase or a disarmed active goal only while the configured round cap has remaining capacity; an active armed goal rejects the redundant operation.
|
||||
|
||||
@@ -23,7 +23,7 @@ Every non-clear mutation appends a complete versioned snapshot through `agent.in
|
||||
|
||||
Injection may append immediately or wait in an active tool-batch FIFO. The service overlays accepted pending changes in memory and reconciles each exact payload when it enters the log, so consecutive model-tool mutations see their own latest revisions without treating an unlogged cache as durable state. `goal/changed` fires after the append or enqueue succeeds; listener failures are contained.
|
||||
|
||||
Activation is never persisted. A fresh cache and every `agent/session-start` edge disarm it even when replay finds an active durable phase. Session resume and fork therefore retain the objective, phase, revisions, and admitted-round count without initiating work; a later explicit resume mutation must arm continuation.
|
||||
Activation is never persisted. A fresh cache and every `agent/session-start` edge disarm it even when replay finds an active durable phase. A continuation driver also calls `disarm()` before unload or after durability uncertainty. Session resume, fork, and driver replacement therefore retain the objective, phase, revisions, and admitted-round count without initiating work; a later explicit resume mutation must arm continuation.
|
||||
|
||||
## Extension points
|
||||
|
||||
|
||||
@@ -139,6 +139,21 @@ export class GoalService extends Service {
|
||||
return this.view(cache)
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove process-local continuation authority without changing durable goal
|
||||
* phase or revision. Lifecycle owners use this before unloading a driver;
|
||||
* a later human-authorized {@link resume} records the new activation edge.
|
||||
* @param agent - owning live agent.
|
||||
* @returns a fresh disarmed view, or `undefined` when no goal is current.
|
||||
*/
|
||||
disarm(agent: Agent): GoalView | undefined {
|
||||
this.assertLive(agent)
|
||||
const cache = this.cache(agent.session)
|
||||
this.sync(agent.session, cache)
|
||||
cache.activation = 'disarmed'
|
||||
return this.view(cache)
|
||||
}
|
||||
|
||||
/**
|
||||
* Create and arm a goal. A completed goal may be replaced; every other
|
||||
* current phase must be cleared or resumed instead.
|
||||
|
||||
@@ -235,6 +235,20 @@ describe('GoalService creation and replay', () => {
|
||||
expect(() => foldGoal(session.events)).not.toThrow()
|
||||
})
|
||||
|
||||
it('lets a lifecycle owner disarm without writing a durable revision', async () => {
|
||||
const { ctx, agent, session } = await harness()
|
||||
const goal = ctx.goals.create(agent, { objective: 'survive driver reload' })
|
||||
const before = session.events.length
|
||||
expect(ctx.goals.disarm(agent)).toMatchObject({
|
||||
id: goal.id,
|
||||
revision: goal.revision,
|
||||
phase: 'active',
|
||||
activation: 'disarmed',
|
||||
})
|
||||
expect(session.events).toHaveLength(before)
|
||||
expect(ctx.goals.resume(agent, goal)).toMatchObject({ revision: 2, activation: 'armed' })
|
||||
})
|
||||
|
||||
it('requires the exact live registry instance for reads and mutations', async () => {
|
||||
const { ctx, agent } = await harness()
|
||||
const impostor = { ...agent, session: new Session(agent.id) }
|
||||
|
||||
@@ -28,6 +28,7 @@ function adapt<K extends ScopedEventName>(
|
||||
}
|
||||
|
||||
const scopedSubjectResolvers = Object.freeze({
|
||||
'agent/cancel-requested': adapt<'agent/cancel-requested'>(args => args[0]),
|
||||
'agent/created': adapt<'agent/created'>(args => args[0]),
|
||||
'agent/disposed': adapt<'agent/disposed'>(args => args[0]),
|
||||
'agent/error': adapt<'agent/error'>(args => args[0]),
|
||||
|
||||
Reference in New Issue
Block a user