Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(schedule): add absolute-time reminders

Browse Source
This commit is contained in:
pku-xht
2026-08-06 15:03:48 +08:00
committed by Tianyi Cui
parent 9e61b7d1b1
commit d61059364e
54 changed files with 3169 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
packages/context/time-context/README.md
View File

@@ -2,7 +2,7 @@
English | [中文](README.zh.md)
Opt-in durable context with the current zoned time and elapsed time sampled during model-request preparation. `dsh-agent-spine-demo` and shipped examples do not mount it. Decision record: [the durable time-context Agent Note](../../../.agents/notes/implemented/feature/2026-07-16-durable-per-step-time-context.md).
Opt-in durable context with the current zoned time, Session and request-zone authority, and elapsed time sampled during model-request preparation. Default compositions do not mount it; the opt-in Schedule Web overlay does. Decision record: [the durable time-context Agent Note](../../../.agents/notes/implemented/feature/2026-07-16-durable-per-step-time-context.md).
## Config
@@ -10,25 +10,29 @@ Opt-in durable context with the current zoned time and elapsed time sampled duri
- id: time-context
name: '@deepseek-ai/dsh-time-context'
config:
timeZone: Asia/Shanghai # optional IANA override; omit for the process zone
timeZone: Asia/Shanghai # optional fallback for headerless Sessions; omit for the process zone
refreshIntervalMs: 60000 # optional; omit or set to 0 for every eligible attempt
```
When `timeZone` is omitted, the plugin resolves the Node process's system zone once at plugin load. Node honors `TZ`; without that override, the host or container supplies the zone. An explicit `timeZone` must be an IANA identifier and is validated at plugin load.
When a Session has `SessionHeader.timeZone`, that immutable IANA zone formats its readings. A headerless Session instead uses the configured fallback; when `timeZone` is omitted, the plugin resolves the Node process's system zone once at plugin load. Node honors `TZ`; without that override, the host or container supplies the fallback. An explicit `timeZone` is validated at plugin load but does not override a Session-owned zone.
`refreshIntervalMs` must be a non-negative safe integer. Omission or `0` adds context to every eligible entering pre-step whose signal is not already aborted. A positive value adds it only when the session has no earlier time-context injection, wall time moved backward, or at least that many milliseconds have elapsed since the latest injection.
`refreshIntervalMs` must be a non-negative safe integer. Omission or `0` adds context to every eligible request preparation whose final pre-step decision contains input and whose signal is not already aborted. A positive value adds it only when the session has no earlier time-context injection, wall time moved backward, or at least that many milliseconds have elapsed since the latest injection.
## Timing semantics
The plugin prepends an `agent/pre-step` listener. When an injection is due and the downstream decision enters the proposed step, it adds one sourced `UserMessage` to the returned batch. AgentLoop records that context after `step/start` and before ordinary automatic compaction with source `{ kind: 'plugin', plugin: 'time-context' }`. A suppressed, rejected, or failed pre-step records nothing.
The plugin opens a narrow authority envelope in `system-prompt/assemble` and closes it around `agent/pre-step`. It captures already-claimed input, and each user steering message admitted during asynchronous assembly is followed synchronously by a superseding same-step authority. AgentLoop includes the envelope's non-authority messages in the downstream pre-step proposal; after downstream edits, discards, or filtering settle, time-context derives the final authority from that decision.
An entering step records its downstream messages followed by exactly one final time-context `UserMessage` after `step/start`. Its source is `{ kind: 'plugin', plugin: 'time-context', authority }`, where `authority` identifies the proposed turn and step, the Session zone as `resolved` or `unavailable`, and the current request's client zones as `resolved`, `mixed`, or `missing`. An empty downstream decision consumes the envelope without opening a step or request.
If preparation exits before `step/start`, AgentLoop removes the envelope before closing the turn. It may settle an appendable final authority inside that failed turn, but an append failure drops the authority instead of leaving it pending. Cancellation cannot generate another authority after it wins; plugin disposal removes pending authorities and an in-flight listener contributes nothing after disposal. Steering and unrelated inbox work retain their ordinary cancellation policy, and no authority for an old turn or step can leak into a later request.
Positive-interval scheduling scans the raw durable session events for the latest `user/message` with that source, including a reading shadowed by compaction. The schedule therefore applies across turns and resumed processes without process-local cache state. It reduces append frequency and history growth but never removes an existing reading, and sessions schedule independently.
Step 1 measures from the latest preceding model-visible message, including the prompt that opened the turn. Later steps measure from the preceding time-context event in the same turn. Both baselines use durable session-event timestamps; backward wall-clock movement clamps elapsed time to zero. A missing first-step baseline, or a later step with no earlier same-turn reading because interval suppression skipped it, reports `unavailable`.
Step 1 measures from the latest durable model-visible message before the current proposal; the prompt entering that same step has not been appended yet. Later steps measure from the preceding time-context event in the same turn. Both baselines use durable session-event timestamps; backward wall-clock movement clamps elapsed time to zero. A missing first-step baseline, or a later step with no earlier same-turn reading because interval suppression skipped it, reports `unavailable`.
A time reading records an entered pre-step batch, not a completed step or transmitted request. A later request-preparation failure can therefore leave the reading in history, but a downstream pre-step listener that rejects or fails prevents it from being recorded.
A time reading records request preparation, not a completed step or transmitted request. A later request-preparation failure can therefore leave the reading in history, and a no-step failure can settle an already-sampled authority inside its failed turn.
The separately published `./invariant` companion checks each plugin-attributed reading against the open turn, next pre-step position, elapsed baseline, and durable event time. Its rendered timestamp must parse and cannot postdate the event; process suspension between sampling and append does not invalidate the reading.
The separately published `./invariant` companion strictly decodes each plugin-attributed authority and checks it against the open turn, next pre-step position, elapsed baseline, and durable event time. Its rendered timestamp must parse and cannot postdate the event; process suspension between sampling and append does not invalidate the reading.
The time reading stays in derived conversation history until a later compaction shadows it. Request headers contain no time-context state. Request reconstruction uses the complete durable surface prefix after each `step/start`, so transmitted requests need not map one-to-one to readings: request preparation can fail after step entry, while interval suppression can let a request reuse existing history without adding one.
@@ -38,12 +42,14 @@ The time reading stays in derived conversation history until a later compaction
#### What the model sees
On each preparation attempt that injects, one source-tagged context message containing the two lines below. `<timestamp>` is an ISO-shaped local timestamp with numeric offset and IANA zone; durations use compact whole-second units. Positive intervals can leave an attempted step without a new reading.
On each preparation attempt that injects, one source-tagged context message contains the four lines below. `<timestamp>` is an ISO-shaped local timestamp with numeric offset and IANA zone; durations use compact whole-second units. The Session line reports the immutable Session zone or `unavailable`, and the client line reports one resolved zone, a sorted mixed set, or `missing`. Positive intervals can leave an attempted step without a new reading.
##### First step
```markdown
Time sampled while preparing turn <turn>, step 1: <timestamp>
Session time zone: <iana-zone-or-unavailable>.
Client time zone for this request: <iana-zone-or-mixed-set-or-missing>.
Elapsed since the preceding model-visible message: <duration-or-unavailable>.
```
@@ -51,12 +57,14 @@ Elapsed since the preceding model-visible message: <duration-or-unavailable>.
```markdown
Time sampled while preparing turn <turn>, step <step>: <timestamp>
Session time zone: <iana-zone-or-unavailable>.
Client time zone for this request: <iana-zone-or-mixed-set-or-missing>.
Elapsed since the preceding step context: <duration-or-unavailable>.
```
#### Token effect
Each injected two-line message accumulates until compaction shadows it. A positive interval reduces additions; omission or `0` adds one for every eligible preparation attempt.
Each injected four-line message accumulates until compaction shadows it. A positive interval reduces additions; omission or `0` adds one for every eligible preparation attempt.
#### KV Cache effect
@@ -66,5 +74,6 @@ Append-only; newly visible content follows the reusable request prefix and does
- **Whole-second display** — timestamps and durations omit sub-second precision even though durable event times retain milliseconds.
- **Session-event baseline** — elapsed time starts from durable append timestamps, not a client transport's original send timestamp.
- **Process-local default zone** — omission uses the Node process's `TZ`, host, or container zone captured at plugin load, not a remote user's zone; configure an explicit IANA zone when those differ.
- **Headerless fallback zone** — a Session without `SessionHeader.timeZone` renders through the configured or process fallback but reports Session authority as `unavailable`; consumers that require unambiguous local-time interpretation must request an explicit zone.
- **Immutable Session zone** — a Session zone does not change when another browser resumes it. The per-request client authority reports disagreement instead of silently changing the displayed default.
- **History cost between compactions** — omission or `0` retains one reading for every eligible preparation attempt, including attempts later cancelled or failed; a positive interval reduces but does not eliminate this cost.

135
packages/context/time-context/src/authority.ts Normal file
View File

@@ -0,0 +1,135 @@
/** Machine-readable Session and request-zone authority carried by time-context messages. */
/** Session-owned zone authority included in each time-context reading. */
export type SessionTimeZoneAuthority =
| { readonly kind: 'resolved'; readonly timeZone: string }
| { readonly kind: 'unavailable' }
/** Client-zone provenance of the messages entering one proposed step. */
export type ClientTimeZoneAuthority =
| { readonly kind: 'resolved'; readonly timeZone: string }
| { readonly kind: 'mixed'; readonly timeZones: string[] }
| { readonly kind: 'missing' }
/** Machine-readable time authority shared by model context and Schedule tools. */
export interface TimeContextAuthority {
readonly turn: number
readonly step: number
readonly session: SessionTimeZoneAuthority
readonly client: ClientTimeZoneAuthority
}
/** Source shape owned by the time-context plugin. */
export interface TimeContextMessageSource {
kind: 'plugin'
plugin: 'time-context'
authority: TimeContextAuthority
}
declare module '@deepseek-ai/dsh-llm' {
interface MessageSourceMap {
'time-context': TimeContextMessageSource
}
}
/** Whether an unknown value is one ordinary JSON object. */
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
}
/** Require one object to carry exactly the named keys. */
function hasExactKeys(value: Record<string, unknown>, expected: readonly string[]): boolean {
const keys = Object.keys(value).sort()
const wanted = [...expected].sort()
return keys.length === wanted.length && keys.every((key, index) => key === wanted[index])
}
/** Decode one non-empty zone name without re-owning Host canonicalization. */
function zone(value: unknown): string {
if (typeof value !== 'string' || value.length === 0) {
throw new TypeError('time-context authority time zone must be a non-empty string')
}
return value
}
/** Decode the Session branch of one authority value. */
function sessionAuthority(value: unknown): SessionTimeZoneAuthority {
if (!isRecord(value)) throw new TypeError('time-context Session authority must be an object')
if (value['kind'] === 'unavailable' && hasExactKeys(value, ['kind'])) return { kind: 'unavailable' }
if (value['kind'] === 'resolved' && hasExactKeys(value, ['kind', 'timeZone'])) {
return { kind: 'resolved', timeZone: zone(value['timeZone']) }
}
throw new TypeError('time-context Session authority has an invalid shape')
}
/** Decode the request-client branch of one authority value. */
function clientAuthority(value: unknown): ClientTimeZoneAuthority {
if (!isRecord(value)) throw new TypeError('time-context client authority must be an object')
if (value['kind'] === 'missing' && hasExactKeys(value, ['kind'])) return { kind: 'missing' }
if (value['kind'] === 'resolved' && hasExactKeys(value, ['kind', 'timeZone'])) {
return { kind: 'resolved', timeZone: zone(value['timeZone']) }
}
if (value['kind'] === 'mixed' && hasExactKeys(value, ['kind', 'timeZones'])) {
const values = value['timeZones']
if (!Array.isArray(values)
|| !values.every((item): item is string => typeof item === 'string' && item.length > 0)
|| values.length < 2) {
throw new TypeError('time-context mixed client authority must contain at least two zones')
}
const timeZones = [...new Set(values)].sort()
if (timeZones.length !== values.length || timeZones.some((item, index) => item !== values[index])) {
throw new TypeError('time-context mixed client zones must be unique and sorted')
}
return { kind: 'mixed', timeZones }
}
throw new TypeError('time-context client authority has an invalid shape')
}
/**
* Decode the strict durable source attached to a time-context message.
* @param value - Untrusted message source.
* @returns Detached machine authority and its fixed plugin discriminator.
*/
export function decodeTimeContextSource(value: unknown): TimeContextMessageSource {
if (!isRecord(value) || !hasExactKeys(value, ['kind', 'plugin', 'authority'])
|| value['kind'] !== 'plugin' || value['plugin'] !== 'time-context') {
throw new TypeError('time-context message source has an invalid shape')
}
const authority = value['authority']
if (!isRecord(authority) || !hasExactKeys(authority, ['turn', 'step', 'session', 'client'])) {
throw new TypeError('time-context authority has an invalid shape')
}
const turn = authority['turn']
const step = authority['step']
if (!Number.isSafeInteger(turn) || (turn as number) < 1
|| !Number.isSafeInteger(step) || (step as number) < 1) {
throw new TypeError('time-context authority turn and step must be positive safe integers')
}
return {
kind: 'plugin',
plugin: 'time-context',
authority: {
turn: turn as number,
step: step as number,
session: sessionAuthority(authority['session']),
client: clientAuthority(authority['client']),
},
}
}
/**
* Render the machine authority as concise model-visible policy.
* @param authority - Session and request-zone authority for one proposed step.
* @returns The two policy lines appended to the time-context reading.
*/
export function renderTimeContextAuthority(authority: TimeContextAuthority): string {
const session = authority.session.kind === 'resolved'
? authority.session.timeZone
: 'unavailable'
const client = authority.client.kind === 'resolved'
? authority.client.timeZone
: authority.client.kind === 'mixed'
? `mixed ${JSON.stringify(authority.client.timeZones)}`
: 'missing'
return `Session time zone: ${session}.\nClient time zone for this request: ${client}.`
}

402
packages/context/time-context/src/index.ts
View File

@@ -9,6 +9,20 @@ import type { Context } from 'cordis'
import z from 'schemastery'
import type { Agent, PreStepDecision } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { UserMessage } from '@deepseek-ai/dsh-llm'
import { renderTimeContextAuthority } from './authority.ts'
import type {
ClientTimeZoneAuthority,
TimeContextAuthority,
} from './authority.ts'
export type {
ClientTimeZoneAuthority,
SessionTimeZoneAuthority,
TimeContextAuthority,
TimeContextMessageSource,
} from './authority.ts'
export { decodeTimeContextSource, renderTimeContextAuthority } from './authority.ts'
/** Cordis plugin name used by loader diagnostics. */
export const name = 'time-context'
@@ -18,7 +32,7 @@ export const inject = ['agents']
/** Request-preparation clock formatting and append scheduling. Invalid values fail plugin load. */
export interface Config {
/** IANA time zone used for the rendered timestamp. Omit to resolve the Node process's system zone at plugin load. */
/** Fallback display zone for headerless Sessions. Omit to use the process zone. */
timeZone?: string
/** Minimum milliseconds between durable injections in one session. Omit or set to 0 to inject at every eligible step. */
refreshIntervalMs?: number
@@ -30,6 +44,7 @@ export const Config: z<Config> = z.object({
refreshIntervalMs: z.number(),
})
type TimestampPart = 'day' | 'hour' | 'minute' | 'month' | 'second' | 'timeZoneName' | 'year'
/** Format an epoch millisecond value as an ISO-shaped timestamp with offset and IANA zone. */
@@ -99,20 +114,116 @@ function latestInjectionTime(agent: Agent): number | undefined {
return undefined
}
/** Read the Host-validated client zone from one ordinary user-rpc message. */
function clientTimeZone(message: UserMessage): string | undefined {
const source = message.source
return source.kind === 'user'
&& 'clientTimeZone' in source
&& typeof source.clientTimeZone === 'string'
? source.clientTimeZone
: undefined
}
/** Derive all distinct client zones in the current request chain. */
function requestClientTimeZones(agent: Agent, turn: number, messages: readonly UserMessage[]): string[] {
const zones = new Set<string>()
for (const event of [...agent.session.events].reverse()) {
if (event.type === 'turn/start' && event.data.turn === turn) break
if (event.type !== 'user/message') continue
const zone = clientTimeZone(event.data)
if (zone !== undefined) zones.add(zone)
}
for (const message of messages) {
const zone = clientTimeZone(message)
if (zone !== undefined) zones.add(zone)
}
return [...zones].sort()
}
/** Close the request-zone set into the machine authority union. */
function clientAuthority(timeZones: string[]): ClientTimeZoneAuthority {
const [timeZone, ...remaining] = timeZones
if (timeZone === undefined) return { kind: 'missing' }
if (remaining.length === 0) return { kind: 'resolved', timeZone }
return { kind: 'mixed', timeZones }
}
function renderText(
now: number,
turn: number,
step: number,
previous: number | undefined,
formatter: Intl.DateTimeFormat,
timeZone: string,
displayTimeZone: string,
authority: TimeContextAuthority,
): string {
const elapsed = previous === undefined ? 'unavailable' : formatDuration(now - previous)
const baseline = step === 1 ? 'model-visible message' : 'step context'
return `Time sampled while preparing turn ${turn}, step ${step}: ${formatTimestamp(now, formatter, timeZone)}\n`
return `Time sampled while preparing turn ${turn}, step ${step}: ${formatTimestamp(now, formatter, displayTimeZone)}\n`
+ `${renderTimeContextAuthority(authority)}\n`
+ `Elapsed since the preceding ${baseline}: ${elapsed}.`
}
interface PreparationPosition {
turn: number
step: number
}
interface ClaimedPreparation extends PreparationPosition {
messages: UserMessage[]
}
interface AssemblyAuthorityState extends PreparationPosition {
agent: Agent
claimed: readonly UserMessage[]
deferredIds: Set<string>
handledIds: Set<string>
accepting: boolean
lastFingerprint?: string
lastMessageId?: UserMessage['id']
readonly signal: AbortSignal
readonly onAbort: () => void
}
/** Derive the next unopened step while one turn is in pre-step preparation. */
function preparationPosition(agent: Agent): PreparationPosition | undefined {
for (const event of [...agent.session.events].reverse()) {
switch (event.type) {
case 'step/start':
case 'turn/end':
return undefined
case 'step/end':
return { turn: event.data.turn, step: event.data.step + 1 }
case 'turn/start':
return { turn: event.data.turn, step: 1 }
default:
break
}
}
return undefined
}
/** Whether two preparation coordinates identify the same unopened step. */
function samePosition<T extends PreparationPosition>(
left: T | undefined,
right: PreparationPosition,
): left is T {
return left?.turn === right.turn && left.step === right.step
}
/** Whether one message is a time-context reading for an exact preparation. */
function isAuthorityMessage(
message: UserMessage,
position: PreparationPosition,
): boolean {
const source = message.source
return source.kind === 'plugin'
&& source.plugin === name
&& 'authority' in source
&& source.authority.turn === position.turn
&& source.authority.step === position.step
}
/** Reject refresh intervals that cannot represent an exact elapsed-millisecond threshold. */
function validateRefreshInterval(refreshIntervalMs: number | undefined): void {
if (refreshIntervalMs !== undefined && (
@@ -131,37 +242,244 @@ function validateRefreshInterval(refreshIntervalMs: number | undefined): void {
* @param config - time zone and durable refresh scheduling configuration.
* @throws when the refresh interval is invalid or the configured or process time zone cannot be resolved.
*/
export function apply(ctx: Context, config: Config): void {
export function apply(ctx: Context, config: Config): () => void {
const timeZone = config.timeZone
const refreshIntervalMs = config.refreshIntervalMs
validateRefreshInterval(refreshIntervalMs)
let formatter: Intl.DateTimeFormat
const createFormatter = (selectedTimeZone?: string): Intl.DateTimeFormat => new Intl.DateTimeFormat('en-US', {
...(selectedTimeZone === undefined ? {} : { timeZone: selectedTimeZone }),
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hourCycle: 'h23',
timeZoneName: 'longOffset',
})
let fallbackFormatter: Intl.DateTimeFormat
try {
formatter = new Intl.DateTimeFormat('en-US', {
...(timeZone === undefined ? {} : { timeZone }),
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hourCycle: 'h23',
timeZoneName: 'longOffset',
})
fallbackFormatter = createFormatter(timeZone)
} catch (error: unknown) {
const message = timeZone === undefined
? 'time-context: failed to resolve the system time zone'
: `time-context: invalid IANA timeZone ${JSON.stringify(timeZone)}`
throw new Error(message, { cause: error })
}
const resolvedTimeZone = formatter.resolvedOptions().timeZone
const fallbackTimeZone = fallbackFormatter.resolvedOptions().timeZone
const formatters = new Map<string, Intl.DateTimeFormat>([[fallbackTimeZone, fallbackFormatter]])
const claimedPreparations = new Map<Agent, ClaimedPreparation>()
const assemblyAuthorities = new Map<Agent, AssemblyAuthorityState>()
let disposed = false
/** Resolve one Session-owned formatter without making the process zone authoritative. */
const formatterFor = (selectedTimeZone: string): Intl.DateTimeFormat => {
const existing = formatters.get(selectedTimeZone)
if (existing !== undefined) return existing
let created: Intl.DateTimeFormat
try {
created = createFormatter(selectedTimeZone)
} catch (error: unknown) {
throw new Error(`time-context: invalid Session time zone ${JSON.stringify(selectedTimeZone)}`, { cause: error })
}
formatters.set(selectedTimeZone, created)
return created
}
/** Build one current reading without placing it in the inbox or decision. */
const readingFor = (
agent: Agent,
position: PreparationPosition,
messages: readonly UserMessage[],
): { message: UserMessage; fingerprint: string } => {
const now = Date.now()
const previous = position.step === 1
? precedingMessageTime(agent)
: precedingStepContextTime(agent, position.turn)
const sessionTimeZone = agent.session.header.timeZone
const authority: TimeContextAuthority = {
turn: position.turn,
step: position.step,
session: sessionTimeZone === undefined
? { kind: 'unavailable' }
: { kind: 'resolved', timeZone: sessionTimeZone },
client: clientAuthority(requestClientTimeZones(agent, position.turn, messages)),
}
const displayTimeZone = sessionTimeZone ?? fallbackTimeZone
const formatter = sessionTimeZone === undefined
? fallbackFormatter
: formatterFor(sessionTimeZone)
return {
message: createUserMessage({
content: [{
type: 'text',
text: renderText(
now,
position.turn,
position.step,
previous,
formatter,
displayTimeZone,
authority,
),
}],
source: { kind: 'plugin', plugin: name, authority },
}),
fingerprint: JSON.stringify(authority),
}
}
/** Messages added after assembly opened, excluding deferred pre-existing work. */
const assemblyMessages = (state: AssemblyAuthorityState): UserMessage[] =>
state.agent.inbox.nextStep.filter(message => !state.deferredIds.has(message.id))
/** Stop accepting late steering while retaining the state for boundary cleanup. */
const closeAssembly = (state: AssemblyAuthorityState): void => {
state.accepting = false
}
/** Forget one preparation and detach its cancellation observer. */
const clearAssembly = (agent: Agent, state = assemblyAuthorities.get(agent)): void => {
if (state === undefined) return
state.accepting = false
state.signal.removeEventListener('abort', state.onAbort)
if (assemblyAuthorities.get(agent) === state) assemblyAuthorities.delete(agent)
}
/** Append one same-step authority after the messages that caused it. */
const stageAuthority = (state: AssemblyAuthorityState, force: boolean): void => {
if (disposed || !state.accepting) return
const reading = readingFor(
state.agent,
state,
[...state.claimed, ...assemblyMessages(state)],
)
if (!force && reading.fingerprint === state.lastFingerprint) return
state.agent.inject(reading.message)
state.lastFingerprint = reading.fingerprint
state.lastMessageId = reading.message.id
}
/**
* Capture messages claimed for the unopened step. The system-prompt
* assembly itself does not receive this batch, so the preparation listener
* preserves its request-zone provenance explicitly.
*/
ctx.on('agent/inbox/claimed', ({ agent, message, turn }) => {
if (disposed) return
const position = preparationPosition(agent)
if (position === undefined || position.turn !== turn) return
const existing = claimedPreparations.get(agent)
if (!samePosition(existing, position)) {
claimedPreparations.set(agent, { ...position, messages: [message] })
return
}
existing.messages.push(message)
})
/**
* Open the narrow assembly window before downstream prompt providers run.
* The initial authority enters the ordinary next-step outbox; AgentLoop
* drains its closed envelope only after pre-step accepts the step.
*/
ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
if (disposed) return next()
const agent = context.agent
const signal = context.signal
const position = agent === undefined ? undefined : preparationPosition(agent)
if (agent === undefined || signal === undefined || position === undefined || signal.aborted) {
return next()
}
if (samePosition(assemblyAuthorities.get(agent), position)) return next()
clearAssembly(agent)
const now = Date.now()
if (refreshIntervalMs !== undefined && refreshIntervalMs > 0) {
const lastInjection = latestInjectionTime(agent)
if (lastInjection !== undefined
&& now >= lastInjection
&& now - lastInjection < refreshIntervalMs) return next()
}
const claimed = claimedPreparations.get(agent)
const state = {
...position,
agent,
claimed: samePosition(claimed, position) ? [...claimed.messages] : [],
deferredIds: new Set(agent.inbox.nextStep.map(message => message.id)),
handledIds: new Set<string>(),
accepting: true,
signal,
onAbort: () => {},
} satisfies AssemblyAuthorityState
state.onAbort = () => { closeAssembly(state) }
assemblyAuthorities.set(agent, state)
signal.addEventListener('abort', state.onAbort, { once: true })
try {
stageAuthority(state, true)
return await next()
} finally {
closeAssembly(state)
}
}, { prepend: true })
/** A late steering message supersedes the authority synchronously behind it. */
ctx.on('agent/inbox/inserted', ({ agent, message }) => {
if (disposed) return
const state = assemblyAuthorities.get(agent)
if (state === undefined || !state.accepting
|| state.deferredIds.has(message.id)
|| !agent.inbox.nextStep.some(candidate => candidate.id === message.id)
|| message.source.kind !== 'user') return
const handledByReplacement = state.handledIds.has(message.id)
stageAuthority(state, !handledByReplacement)
state.handledIds.add(message.id)
})
/** Recompute after an edit/discard, but do not resurrect a cleared inbox. */
ctx.on('agent/inbox/discarded', ({ agent, message }) => {
if (disposed) return
const state = assemblyAuthorities.get(agent)
if (state === undefined || !state.accepting
|| state.deferredIds.has(message.id)
|| message.source.kind !== 'user') return
if (!agent.inbox.nextStep.some(candidate => isAuthorityMessage(candidate, state))) {
closeAssembly(state)
return
}
stageAuthority(state, false)
state.handledIds = new Set(
assemblyMessages(state)
.filter(candidate => candidate.source.kind === 'user')
.map(candidate => candidate.id),
)
})
ctx.on('agent/pre-step', async (
{ agent, turn, step, signal },
next,
): Promise<PreStepDecision> => {
const wasDisposed = (): boolean => disposed
if (wasDisposed()) return next()
const decision = await next()
if (decision.kind === 'reject' || signal.aborted) return decision
if (wasDisposed()) return decision
const staged = assemblyAuthorities.get(agent)
if (decision.kind === 'reject' || signal.aborted) {
if (samePosition(staged, { turn, step })) closeAssembly(staged)
return decision
}
if (samePosition(staged, { turn, step })) {
closeAssembly(staged)
const reading = readingFor(agent, { turn, step }, decision.messages)
if (reading.fingerprint !== staged.lastFingerprint) {
const replaced = staged.lastMessageId === undefined
? false
: agent.inbox.replace(staged.lastMessageId, reading.message)
if (!replaced) agent.inject(reading.message)
staged.lastFingerprint = reading.fingerprint
staged.lastMessageId = reading.message.id
}
return decision
}
if (decision.messages.length === 0) return decision
const now = Date.now()
if (refreshIntervalMs !== undefined && refreshIntervalMs > 0) {
const lastInjection = latestInjectionTime(agent)
@@ -169,19 +487,51 @@ export function apply(ctx: Context, config: Config): void {
&& now >= lastInjection
&& now - lastInjection < refreshIntervalMs) return decision
}
const previous = step === 1
? precedingMessageTime(agent)
: precedingStepContextTime(agent, turn)
const text = renderText(now, turn, step, previous, formatter, resolvedTimeZone)
const reading = readingFor(agent, { turn, step }, decision.messages)
return {
kind: 'enter',
messages: [
...decision.messages,
createUserMessage({
content: [{ type: 'text', text }],
source: { kind: 'plugin', plugin: name, form: 'snapshot', sections: [{ name, text }] },
}),
reading.message,
],
}
}, { prepend: true })
/** Step/turn/lifecycle boundaries release request-only bookkeeping. */
ctx.on('session/event', (session, event) => {
if (disposed) return
if (event.type !== 'step/start' && event.type !== 'turn/end') return
const agent = ctx.agents.get(session.id)
if (agent === undefined || agent.session !== session) return
clearAssembly(agent)
if (event.type === 'turn/end') claimedPreparations.delete(agent)
})
ctx.on('agent/status', (agent, status) => {
if (disposed) return
if (status !== 'idle') return
clearAssembly(agent)
claimedPreparations.delete(agent)
})
ctx.on('agent/disposed', (agent) => {
if (disposed) return
clearAssembly(agent)
claimedPreparations.delete(agent)
})
return () => {
disposed = true
for (const [agent, state] of assemblyAuthorities) {
closeAssembly(state)
for (const message of [...agent.inbox.nextStep]) {
if (!isAuthorityMessage(message, state)) continue
try {
agent.inbox.remove(message.id)
} catch (error: unknown) {
ctx.logger.warn(`time-context: failed to discard authority during dispose: ${String(error)}`)
}
}
clearAssembly(agent, state)
}
claimedPreparations.clear()
}
}

62
packages/context/time-context/src/invariant.ts
View File

@@ -3,12 +3,15 @@
import type { Context } from 'cordis'
import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import { decodeTimeContextSource, renderTimeContextAuthority } from './authority.ts'
const PACKAGE_NAME = '@deepseek-ai/dsh-time-context'
const SOURCE_NAME = 'time-context'
const READING = new RegExp(
'^Time sampled while preparing turn (\\d+), step (\\d+): '
+ '(\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(?:Z|[+-]\\d{2}:\\d{2})\\[[^\\]]+\\])\\n'
+ 'Session time zone: ([^.]+)\\.\\n'
+ 'Client time zone for this request: (.+)\\.\\n'
+ 'Elapsed since the preceding (model-visible message|step context): '
+ '(?:unavailable|(?:(?:\\d+d )?(?:\\d+h )?(?:\\d+m )?\\d+s))\\.$',
)
@@ -18,27 +21,43 @@ export const name = 'time-context-invariant'
/** Service required before the companion can reserve package ownership. */
export const inject = ['invariants']
/** Derive the entered step boundary at which a time-context reading may append. */
/**
* Derive the step preparation owned by a time-context reading. A normal
* reading follows `step/start`; a pre-step failure may settle context-only
* output in the still-open turn before that boundary.
*/
function preparationPosition(history: readonly SessionEvent[], fail: InvariantFailure): { turn: number; step: number } {
for (const event of history.slice().reverse()) {
let openTurn: number | undefined
let openStep: number | undefined
let nextStep = 1
for (const event of history) {
switch (event.type) {
case 'step/start':
return { turn: event.data.turn, step: event.data.step }
case 'turn/start':
case 'step/end':
case 'turn/end':
case 'request/header':
case 'assistant/chunk':
case 'assistant/message':
case 'tool/call':
case 'tool/result':
fail('time-context reading must be appended at a prompt boundary')
case 'turn/start': {
openTurn = event.data.turn
openStep = undefined
nextStep = 1
break
}
case 'step/start': {
openStep = event.data.step
break
}
case 'step/end': {
openStep = undefined
nextStep = event.data.step + 1
break
}
case 'turn/end': {
openTurn = undefined
openStep = undefined
break
}
default:
break
}
}
fail('time-context reading must be appended at a prompt boundary')
if (openTurn === undefined) fail('time-context reading must be appended inside an open turn')
return { turn: openTurn, step: openStep ?? nextStep }
}
/** Validate one plugin-attributed time reading against its session position and timestamp. */
@@ -62,7 +81,20 @@ function validateReading(
if (turn !== expected.turn || step !== expected.step) {
fail(`time-context reading names turn ${turn}/step ${step}, expected turn ${expected.turn}/step ${expected.step}`)
}
const baseline = match[4]
let source: ReturnType<typeof decodeTimeContextSource>
try {
source = decodeTimeContextSource(event.data.source)
} catch (error: unknown) {
fail(error instanceof Error ? error.message : String(error))
}
if (source.authority.turn !== turn || source.authority.step !== step) {
fail('time-context text and source authority name different positions')
}
const renderedAuthority = `Session time zone: ${match[4]}.\nClient time zone for this request: ${match[5]}.`
if (renderedAuthority !== renderTimeContextAuthority(source.authority)) {
fail('time-context text and source authority describe different zones')
}
const baseline = match[6]
if ((step === 1) !== (baseline === 'model-visible message')) {
fail(`time-context step ${step} uses the wrong elapsed-time baseline ${JSON.stringify(baseline)}`)
}

51
packages/context/time-context/tests/invariant.spec.ts
View File

@@ -22,13 +22,27 @@ function event(
content?: unknown[],
plugin = 'time-context',
): SessionEvent<'user/message'> {
const position = /turn (\d+), step (\d+):/.exec(text)
const turn = Number(position?.[1] ?? '1')
const step = Number(position?.[2] ?? '1')
return {
type: 'user/message',
seq: 0,
time,
data: createUserMessage({
content: (content ?? [{ type: 'text', text }]) as ContentBlock[],
source: { kind: 'plugin', plugin },
source: plugin === 'time-context'
? {
kind: 'plugin',
plugin,
authority: {
turn,
step,
session: { kind: 'unavailable' },
client: { kind: 'missing' },
},
}
: { kind: 'plugin', plugin },
}),
}
}
@@ -40,6 +54,8 @@ function reading(
timestamp = '2026-07-14T00:00:00+00:00[UTC]',
): string {
return `Time sampled while preparing turn ${turn}, step ${step}: ${timestamp}\n`
+ 'Session time zone: unavailable.\n'
+ 'Client time zone for this request: missing.\n'
+ `Elapsed since the preceding ${baseline}: unavailable.`
}
@@ -63,9 +79,19 @@ function preparing(turn: number, step: number): Session {
}
function appendReading(session: Session, text: string): void {
const position = /turn (\d+), step (\d+):/.exec(text)
session.append('user/message', createUserMessage({
content: [{ type: 'text', text }],
source: { kind: 'plugin', plugin: 'time-context' },
source: {
kind: 'plugin',
plugin: 'time-context',
authority: {
turn: Number(position?.[1] ?? '1'),
step: Number(position?.[2] ?? '1'),
session: { kind: 'unavailable' },
client: { kind: 'missing' },
},
},
}), { surfaceOp: 'append' })
}
@@ -73,6 +99,8 @@ describe('time-context invariants', () => {
it('accepts a reading whose turn, step, baseline, and timestamp agree', async () => {
const ctx = await setup()
const text = 'Time sampled while preparing turn 2, step 3: 2026-07-14T00:00:00+00:00[UTC]\n'
+ 'Session time zone: unavailable.\n'
+ 'Client time zone for this request: missing.\n'
+ 'Elapsed since the preceding step context: 4m 2s.'
expect(() => { ctx.emit('session/event', preparing(2, 3), event(text)) }).not.toThrow()
})
@@ -129,20 +157,25 @@ describe('time-context invariants', () => {
const session = preparing(1, 2)
session.append('turn/end', { turn: 1, reason: { kind: 'aborted', reason: { kind: 'user' } } })
expect(() => { ctx.emit('session/event', session, event(reading('1', '2', 'step context'))) })
.toThrow(/at a prompt boundary/)
.toThrow(/inside an open turn/)
})
it('rejects a reading outside a prompt boundary', async () => {
it('accepts context-only settlement before step/start', async () => {
const ctx = await setup()
const session = Session.create(SessionId('time-invariant-turn-only'))
session.append('turn/start', { turn: 1 })
expect(() => { ctx.emit('session/event', session, event(reading())) }).not.toThrow()
})
it('rejects a reading outside its open preparation', async () => {
const ctx = await setup()
const ended = preparing(1, 1)
ended.append('step/end', { turn: 1, step: 1 })
expect(() => { ctx.emit('session/event', ended, event(reading())) }).toThrow(/at a prompt boundary/)
const notEntered = Session.create(SessionId('time-invariant-turn-only'))
notEntered.append('turn/start', { turn: 1 })
expect(() => { ctx.emit('session/event', notEntered, event(reading())) }).toThrow(/at a prompt boundary/)
expect(() => { ctx.emit('session/event', ended, event(reading())) })
.toThrow(/expected turn 1\/step 2/)
expect(() => {
ctx.emit('session/event', Session.create(SessionId('time-invariant-empty')), event(reading()))
}).toThrow(/at a prompt boundary/)
}).toThrow(/inside an open turn/)
})
it.each([

382
packages/context/time-context/tests/time-context.spec.ts
View File

@@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { createUserMessage, CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk, UserMessage } from '@deepseek-ai/dsh-llm'
import { Session, SessionId, type SessionEvent } from '@deepseek-ai/dsh-session'
import AgentRegistry, { agentEvents, Inbox, type Agent } from '@deepseek-ai/dsh-agent'
import { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
@@ -79,19 +79,35 @@ async function fire(
turn: number,
step: number,
signal: AbortSignal = SIGNAL,
messages: UserMessage[] = [],
): Promise<void> {
const fallback = messages.length === 0
? createUserMessage({
content: [],
source: { kind: 'plugin', plugin: 'time-context-test-proposal' },
})
: undefined
const proposal = fallback === undefined ? messages : [fallback]
const decision = await agentEvents(ctx, agent).waterfall(
'agent/pre-step',
{ messages: [], turn, step, signal },
() => Promise.resolve({ kind: 'enter' as const, messages: [] }),
{ messages: proposal, turn, step, signal },
() => Promise.resolve({ kind: 'enter' as const, messages: proposal }),
)
if (decision.kind === 'enter') {
for (const message of decision.messages) {
if (message.id === fallback?.id) continue
agent.session.append('user/message', message, { surfaceOp: 'append' })
}
}
}
function rpcMessage(text: string, clientTimeZone: string): UserMessage {
return createUserMessage({
content: [{ type: 'text', text }],
source: { kind: 'user', clientTimeZone } as never,
})
}
function textResponse(text: string): StreamChunk[] {
return [
{ type: 'block-start', index: 0, blockType: 'text' },
@@ -145,6 +161,77 @@ function requestText(request: GenerateOptions): string {
}
describe('durable step context', () => {
it('uses the immutable Session zone and the current request message zone', async () => {
const { ctx } = await mount()
const id = SessionId('session-zone')
const session = Session.create(id, [], {
version: 0,
id,
createdAt: BASE,
timeZone: 'Asia/Shanghai',
})
session.append('turn/start', { turn: 1 })
await fire(ctx, sessionAgent(session), 1, 1, SIGNAL, [
rpcMessage('local request', 'Asia/Shanghai'),
])
expect(contextTexts(session)[0]).toContain(
'2026-07-14T08:00:00+08:00[Asia/Shanghai]',
)
const reading = session.events.at(-1)
expect(reading).toMatchObject({
type: 'user/message',
data: {
source: {
kind: 'plugin',
plugin: 'time-context',
authority: {
turn: 1,
step: 1,
session: { kind: 'resolved', timeZone: 'Asia/Shanghai' },
client: { kind: 'resolved', timeZone: 'Asia/Shanghai' },
},
},
},
})
})
it('reports sorted mixed zones from the current request chain without changing the Session zone', async () => {
const { ctx } = await mount()
const id = SessionId('mixed-zone')
const session = Session.create(id, [], {
version: 0,
id,
createdAt: BASE,
timeZone: 'Asia/Shanghai',
})
session.append('turn/start', { turn: 1 })
session.append('user/message', rpcMessage('first tab', 'Asia/Shanghai'), {
surfaceOp: 'append',
})
await fire(ctx, sessionAgent(session), 1, 1, SIGNAL, [
rpcMessage('second tab', 'America/New_York'),
])
const reading = session.events.at(-1)
expect(reading).toMatchObject({
type: 'user/message',
data: {
source: {
authority: {
session: { kind: 'resolved', timeZone: 'Asia/Shanghai' },
client: {
kind: 'mixed',
timeZones: ['America/New_York', 'Asia/Shanghai'],
},
},
},
},
})
})
it('records turn, step, zoned time, and the preceding model-visible message baseline', async () => {
const { ctx } = await mount({ timeZone: 'Asia/Shanghai' })
const session = Session.create(SessionId('first'))
@@ -155,23 +242,22 @@ describe('durable step context', () => {
expect(contextTexts(session)).toEqual([
'Time sampled while preparing turn 1, step 1: 2026-07-15T09:01:01+08:00[Asia/Shanghai]\n'
+ 'Session time zone: unavailable.\n'
+ 'Client time zone for this request: missing.\n'
+ 'Elapsed since the preceding model-visible message: 1d 1h 1m 1s.',
])
const event = session.events.at(-1)
expect(event?.type).toBe('user/message')
if (event?.type !== 'user/message') throw new Error('missing time context')
// The reading is a `snapshot`-form context: one named contribution whose
// text is exactly what the model read, so a consumer attributes it without
// re-splitting prose.
expect(event.data.source).toEqual({
kind: 'plugin',
plugin: 'time-context',
form: 'snapshot',
sections: [{
name: 'time-context',
text: 'Time sampled while preparing turn 1, step 1: 2026-07-15T09:01:01+08:00[Asia/Shanghai]\n'
+ 'Elapsed since the preceding model-visible message: 1d 1h 1m 1s.',
}],
authority: {
turn: 1,
step: 1,
session: { kind: 'unavailable' },
client: { kind: 'missing' },
},
})
expect(event.surfaceOp).toBe('append')
})
@@ -203,6 +289,8 @@ describe('durable step context', () => {
expect(contextTexts(session)[1]).toBe(
'Time sampled while preparing turn 3, step 2: 2026-07-14T00:01:01+00:00[UTC]\n'
+ 'Session time zone: unavailable.\n'
+ 'Client time zone for this request: missing.\n'
+ 'Elapsed since the preceding step context: 1m 1s.',
)
})
@@ -372,9 +460,9 @@ describe('configuration and lifecycle', () => {
describe('real agent-loop request history', () => {
it.each([
['throws'],
['cancels'],
] as const)('does not commit a preparation reading when a downstream pre-step listener %s', async (mode) => {
['throws', 1],
['cancels', 0],
] as const)('settles preparation context when a downstream pre-step listener %s', async (mode, expectedContexts) => {
const adapter = new ScriptedAdapter([textResponse('unused')])
const ctx = await loopHarness(adapter)
ctx.on('agent/pre-step', ({ agent: subject }, next) => {
@@ -387,12 +475,274 @@ describe('real agent-loop request history', () => {
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'start' }], source: { kind: 'user' } }))
await agent.whenIdle()
expect(contextTexts(agent.session)).toHaveLength(0)
expect(contextTexts(agent.session)).toHaveLength(expectedContexts)
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.some(event => event.type === 'step/start')).toBe(false)
await ctx.fiber.dispose()
})
it('drains late assembly steering between initial and superseding same-step authorities', async () => {
const adapter = new ScriptedAdapter([textResponse('done')])
const ctx = await loopHarness(adapter)
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
let proposedTexts: string[] = []
ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
if (context.agent !== undefined) {
entered.resolve(undefined)
await release.promise
}
return next()
})
ctx.on('agent/pre-step', async ({ messages }, next) => {
proposedTexts = messages.flatMap(message => message.content)
.filter(block => block.type === 'text')
.map(block => block.text)
return next()
})
const agent = ctx.agentLoop.create(SessionId('late-steering'), { provider: 'mock', model: 'mock' })
agent.followup(rpcMessage('start in Shanghai', 'Asia/Shanghai'))
await entered.promise
agent.steer(rpcMessage('switch to New York', 'America/New_York'))
release.resolve(undefined)
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
expect(agent.inbox.hasPending).toBe(false)
const enteredMessages = agent.session.events.filter(
(event): event is SessionEvent<'user/message'> => event.type === 'user/message',
)
const texts = enteredMessages.map(message =>
message.data.content.find(block => block.type === 'text')?.text)
expect(texts).toEqual([
'start in Shanghai',
'switch to New York',
expect.stringContaining('Client time zone for this request: mixed ["America/New_York","Asia/Shanghai"].'),
])
expect(proposedTexts).toEqual([
'start in Shanghai',
'switch to New York',
])
const authorities = enteredMessages
.filter(message => message.data.source.kind === 'plugin')
.map(message => message.data.source.kind === 'plugin' && 'authority' in message.data.source
? message.data.source.authority
: undefined)
expect(authorities).toEqual([
expect.objectContaining({
turn: 1,
step: 1,
client: {
kind: 'mixed',
timeZones: ['America/New_York', 'Asia/Shanghai'],
},
}),
])
await ctx.fiber.dispose()
})
it('collapses edited and discarded late steering to one truthful final authority', async () => {
const adapter = new ScriptedAdapter([textResponse('done')])
const ctx = await loopHarness(adapter)
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
if (context.agent !== undefined) {
entered.resolve(undefined)
await release.promise
}
return next()
})
const agent = ctx.agentLoop.create(SessionId('edited-late-steering'), {
provider: 'mock',
model: 'mock',
})
agent.followup(rpcMessage('start in Shanghai', 'Asia/Shanghai'))
await entered.promise
const edited = rpcMessage('switch to New York', 'America/New_York')
agent.steer(edited)
const replacement = rpcMessage('stay in Shanghai', 'Asia/Shanghai')
expect(agent.inbox.replace(edited.id, replacement)).toBe(true)
const discarded = rpcMessage('temporary New York tab', 'America/New_York')
agent.steer(discarded)
expect(agent.inbox.remove(discarded.id)).toBe(true)
release.resolve(undefined)
await agent.whenIdle()
expect(agent.inbox.hasPending).toBe(false)
const request = requestText(adapter.requests[0]!)
expect(request).toContain('start in Shanghai')
expect(request).toContain('stay in Shanghai')
expect(request).not.toContain('switch to New York')
expect(request).not.toContain('temporary New York tab')
expect(request).not.toContain('Client time zone for this request: mixed')
const authorities = agent.session.events.filter(event =>
event.type === 'user/message'
&& event.data.source.kind === 'plugin'
&& event.data.source.plugin === 'time-context')
expect(authorities).toHaveLength(1)
expect(authorities[0]).toMatchObject({
data: {
source: {
authority: {
turn: 1,
step: 1,
client: { kind: 'resolved', timeZone: 'Asia/Shanghai' },
},
},
},
})
await ctx.fiber.dispose()
})
it('does not let preparation authority create a step after downstream suppression', async () => {
const adapter = new ScriptedAdapter([textResponse('unused')])
const ctx = await loopHarness(adapter)
ctx.on('agent/pre-step', async (_payload, next) => {
const decision = await next()
return decision.kind === 'reject' ? decision : { kind: 'enter', messages: [] }
})
const agent = ctx.agentLoop.create(SessionId('suppressed-preparation'), {
provider: 'mock',
model: 'mock',
})
agent.followup(rpcMessage('suppress this prompt', 'Asia/Shanghai'))
await agent.whenIdle()
expect(adapter.requests).toEqual([])
expect(agent.session.events.some(event => event.type === 'step/start')).toBe(false)
expect(contextTexts(agent.session)).toEqual([])
expect(agent.inbox.hasPending).toBe(false)
await ctx.fiber.dispose()
})
it('settles authorities but preserves steering when keep-inbox cancellation wins assembly', async () => {
const adapter = new ScriptedAdapter([textResponse('resumed')])
const ctx = await loopHarness(adapter)
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
let blocked = true
ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
if (blocked && context.agent !== undefined) {
entered.resolve(undefined)
await release.promise
}
return next()
})
const agent = ctx.agentLoop.create(SessionId('cancelled-assembly'), { provider: 'mock', model: 'mock' })
const steering = rpcMessage('preserve this steering', 'America/New_York')
agent.followup(rpcMessage('start', 'Asia/Shanghai'))
await entered.promise
agent.steer(steering)
agent.cancel({ kind: 'user' }, { keepInbox: true })
blocked = false
release.resolve(undefined)
await agent.whenIdle()
expect(agent.session.events.some(event => event.type === 'step/start')).toBe(false)
expect(contextTexts(agent.session)).toHaveLength(1)
expect(agent.inbox.nextStep).toEqual([steering])
expect(agent.inbox.nextStep.some(message =>
message.source.kind === 'plugin' && message.source.plugin === 'time-context')).toBe(false)
const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
const lastAuthority = agent.session.events.findLast(event =>
event.type === 'user/message'
&& event.data.source.kind === 'plugin'
&& event.data.source.plugin === 'time-context')
expect(lastAuthority?.seq).toBeLessThan(turnEnd?.seq ?? -1)
agent.followup(rpcMessage('wake', 'America/New_York'))
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
expect(requestText(adapter.requests[0]!)).toContain('preserve this steering')
await ctx.fiber.dispose()
})
it('does not contribute after its disposer wins an in-flight pre-step', async () => {
const adapter = new ScriptedAdapter([textResponse('done')])
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx)
await ctx.plugin(AgentLoop, { agents: [] })
const stopTimeContext = timeContext.apply(ctx, {})
ctx.llm.registerAdapter(['mock'], adapter)
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
ctx.on('agent/pre-step', async (_payload, next) => {
entered.resolve(undefined)
await release.promise
return next()
})
const agent = ctx.agentLoop.create(SessionId('dispose-inflight-pre-step'), {
provider: 'mock',
model: 'mock',
})
agent.followup(rpcMessage('continue without disposed context', 'Asia/Shanghai'))
await entered.promise
stopTimeContext()
release.resolve(undefined)
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
expect(requestText(adapter.requests[0]!)).not.toContain('Time sampled while preparing')
expect(contextTexts(agent.session)).toEqual([])
expect(agent.inbox.nextStep).toEqual([])
await ctx.fiber.dispose()
})
it('drops a rejected context append instead of leaking its authority to the next turn', async () => {
const adapter = new ScriptedAdapter([textResponse('resumed')])
const ctx = await loopHarness(adapter)
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
let blocked = true
ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
if (blocked && context.agent !== undefined) {
entered.resolve(undefined)
await release.promise
}
return next()
})
const agent = ctx.agentLoop.create(SessionId('context-append-rejection'), {
provider: 'mock',
model: 'mock',
})
const originalAppend = agent.session.append.bind(agent.session)
let rejectContext = true
vi.spyOn(agent.session, 'append').mockImplementation(((type, data, options) => {
if (rejectContext && type === 'user/message'
&& (data as UserMessage).source.kind === 'plugin'
&& (data as UserMessage).source.plugin === 'time-context') {
rejectContext = false
throw new Error('context append unavailable')
}
return originalAppend(type, data, options)
}) as typeof agent.session.append)
agent.followup(rpcMessage('start', 'Asia/Shanghai'))
await entered.promise
agent.cancel({ kind: 'user' }, { keepInbox: true })
blocked = false
release.resolve(undefined)
await agent.whenIdle()
expect(contextTexts(agent.session)).toHaveLength(0)
expect(agent.inbox.nextStep.some(message =>
message.source.kind === 'plugin' && message.source.plugin === 'time-context')).toBe(false)
agent.followup(rpcMessage('wake', 'Asia/Shanghai'))
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
const request = requestText(adapter.requests[0]!)
expect(request).toContain('Time sampled while preparing turn 2, step 1:')
expect(request).not.toContain('Time sampled while preparing turn 1, step 1:')
await ctx.fiber.dispose()
})
it('persists one ordered context per request, accumulates readings, and leaves system headers unchanged', async () => {
const adapter = new ScriptedAdapter([toolCallResponse(), textResponse('done')])
const ctx = await loopHarness(adapter)