refactor(schedule): simplify request zone authority

packages/context/time-context/src/index.ts

@@ -10,19 +10,13 @@ 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'
+import {
+  deriveClientTimeZoneContext,
+  renderTimeZoneContext,
+} from './request-zone.ts'
 
-export type {
-  ClientTimeZoneAuthority,
-  SessionTimeZoneAuthority,
-  TimeContextAuthority,
-  TimeContextMessageSource,
-} from './authority.ts'
-export { decodeTimeContextSource, renderTimeContextAuthority } from './authority.ts'
+export type { ClientTimeZoneContext } from './request-zone.ts'
+export { deriveClientTimeZoneContext, renderTimeZoneContext } from './request-zone.ts'
 
 /** Cordis plugin name used by loader diagnostics. */
 export const name = 'time-context'
@@ -44,7 +38,6 @@ 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. */
@@ -73,7 +66,7 @@ function formatDuration(elapsedMs: number): string {
   return parts.join(' ')
 }
 
-/** Find the latest model-visible event, excluding this plugin's pending append. */
+/** Find the latest model-visible event before the current proposal. */
 function precedingMessageTime(agent: Agent): number | undefined {
   for (const event of [...agent.session.events].reverse()) {
     switch (event.type) {
@@ -114,40 +107,18 @@ 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 }
+/** Collect already-entered and proposed messages belonging to one open turn. */
+function requestMessages(agent: Agent, turn: number, proposed: readonly UserMessage[]): UserMessage[] {
+  const start = agent.session.events.findLastIndex(
+    event => event.type === 'turn/start' && event.data.turn === turn,
+  )
+  const entered = start < 0
+    ? []
+    : agent.session.events.slice(start + 1).flatMap(event => event.type === 'user/message' ? [event.data] : [])
+  return [...entered, ...proposed]
 }
 
+/** Render one durable time reading. */
 function renderText(
   now: number,
   turn: number,
@@ -155,75 +126,17 @@ function renderText(
   previous: number | undefined,
   formatter: Intl.DateTimeFormat,
   displayTimeZone: string,
-  authority: TimeContextAuthority,
+  sessionTimeZone: string | undefined,
+  messages: readonly UserMessage[],
 ): string {
   const elapsed = previous === undefined ? 'unavailable' : formatDuration(now - previous)
   const baseline = step === 1 ? 'model-visible message' : 'step context'
+  const client = deriveClientTimeZoneContext(messages)
   return `Time sampled while preparing turn ${turn}, step ${step}: ${formatTimestamp(now, formatter, displayTimeZone)}\n`
-    + `${renderTimeContextAuthority(authority)}\n`
+    + `${renderTimeZoneContext(sessionTimeZone, client)}\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 && (
@@ -238,9 +151,10 @@ function validateRefreshInterval(refreshIntervalMs: number | undefined): void {
 
 /**
  * Register a prepended pre-step listener for the lifetime of `ctx`.
- * @param ctx - plugin context; the listener is disposed with it.
- * @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.
+ * @param ctx - Plugin context; the listener is disposed with it.
+ * @param config - Time zone and durable refresh scheduling configuration.
+ * @returns A disposer that prevents an in-flight listener from contributing.
+ * @throws When the refresh interval or configured/process time zone is invalid.
  */
 export function apply(ctx: Context, config: Config): () => void {
   const timeZone = config.timeZone
@@ -268,8 +182,6 @@ export function apply(ctx: Context, config: Config): () => void {
   }
   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. */
@@ -286,200 +198,52 @@ export function apply(ctx: Context, config: Config): () => void {
     return created
   }
 
-  /** Build one current reading without placing it in the inbox or decision. */
+  /** Build one current reading after downstream pre-step transforms settle. */
   const readingFor = (
     agent: Agent,
-    position: PreparationPosition,
+    turn: number,
+    step: number,
     messages: readonly UserMessage[],
-  ): { message: UserMessage; fingerprint: string } => {
+  ): UserMessage => {
     const now = Date.now()
-    const previous = position.step === 1
+    const previous = step === 1
       ? precedingMessageTime(agent)
-      : precedingStepContextTime(agent, position.turn)
+      : precedingStepContextTime(agent, 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),
-    }
+    return createUserMessage({
+      content: [{
+        type: 'text',
+        text: renderText(
+          now,
+          turn,
+          step,
+          previous,
+          formatter,
+          displayTimeZone,
+          sessionTimeZone,
+          requestMessages(agent, turn, messages),
+        ),
+      }],
+      source: { kind: 'plugin', plugin: name },
+    })
   }
 
-  /** 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
+    const wasAborted = (): boolean => signal.aborted
     if (wasDisposed()) return next()
     const decision = await next()
-    if (wasDisposed()) return decision
-    const staged = assemblyAuthorities.get(agent)
-    if (decision.kind === 'reject' || signal.aborted) {
-      if (samePosition(staged, { turn, step })) closeAssembly(staged)
+    if (wasDisposed() || wasAborted() || decision.kind === 'reject'
+      || (step === 1 && decision.messages.length === 0)) {
       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)
@@ -487,51 +251,16 @@ export function apply(ctx: Context, config: Config): () => void {
         && now >= lastInjection
         && now - lastInjection < refreshIntervalMs) return decision
     }
-    const reading = readingFor(agent, { turn, step }, decision.messages)
     return {
       kind: 'enter',
       messages: [
         ...decision.messages,
-        reading.message,
+        readingFor(agent, turn, step, decision.messages),
       ],
     }
   }, { 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()
   }
 }