Merge remote-tracking branch 'origin/master' into codex/project-instruction-files

# Conflicts: # AGENTS.md # docs/architecture.md # docs/event-producer-consumer.md # docs/module-graph.md # docs/rfc/INDEX.md # packages/bash/bash-local/tests/executor.spec.ts # packages/bash/bash-local/tests/run.spec.ts # packages/core/agent-core/src/index.ts # packages/core/agent-core/tests/agent-core.spec.ts # packages/fs/fs-local/src/fsio.ts # packages/ui/acp-agent/package.json # packages/ui/acp-agent/src/index.ts # packages/ui/acp-agent/tsconfig.json # packages/ui/stdio-agent/package.json # packages/ui/stdio-agent/tsconfig.json # packages/util/README.md # pnpm-lock.yaml # tsconfig.build.json # tsconfig.json
2026-07-09 23:04:41 +08:00
parent 15c7a27f39 d58a55ad59
commit 7b54768ee7
437 changed files with 23888 additions and 1656 deletions
@@ -157,13 +157,17 @@ export interface LoopHandle {
 *       drain steering → session('steering/message')  ⟵ catches late steering
 *       assembly = ctx.systemPrompt.assemble({agent}) ⟵ waterfall system-prompt/assemble; renderPrompt
 *                                                        (persona section + {{variables}}) IS the full prompt
- *       await ctx.serial('agent/pre-step')            ⟵ surface mutation (compaction) OUTSIDE the step
+ *       prefix ??= waterfall agent/session-prefix     ⟵ once per loop instance (first step): frozen
+ *                                                        session prefix; logged on the header, never
+ *                                                        session history
+ *       await ctx.serial('agent/pre-step', …, prefix) ⟵ surface mutation (compaction) OUTSIDE the step;
+ *                                                        pressure gates see the prefix the request carries
 *       boundary = session.deriveMessages()           ⟵ the reconstruction boundary: snapshot in the
 *       session('step/start')                            same sync frame, strictly before step/start
 *       config = waterfall agent/request(config)      ⟵ frozen seed; a returned replacement switches
 *       session('request/header'|'request/header-delta')  ⟵ the header event this request owes the
 *                                                        log (initial/resume anchor, delta, fallback)
- *       req = freeze({header..., messages: boundary, sessionId, signal})
+ *       req = freeze({header..., messages: prefix+boundary, sessionId, signal})
 *       stream ctx.llm.stream(req)                    ⟵ waterfall llm/stream (raw chunks, frozen req)
 *         session('assistant/chunk')
 *       msg = waterfall agent/step-result             ⟵ BEFORE the log append, so the
@@ -185,6 +189,9 @@ export interface LoopHandle {
 *     re-enqueue leftover steering as queued          ⟵ steering is never stranded
 *   idle (emit agent/status) unless more queued
 * ```
+ * @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
+ * @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
+ * @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
 */
 export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
  // Per-instance transmission bookkeeping: whether THIS loop instance has
@@ -469,6 +476,50 @@ async function runTurn(
        break
      }

+      // Compose the session prefix ONCE per loop instance, lazily before the
+      // instance's first pre-step: request-only messages placed in front of
+      // the ENTIRE derived history on every request this instance sends. It
+      // MUST precede the pre-step seam so compaction gates on THIS instance's
+      // prefix — reading a previous instance's logged prefix would let a
+      // resumed/forked instance whose contributor grew skip compaction and
+      // ship an over-window first request. The result is deep-cloned
+      // (decoupled from listener-held references), deep-frozen, and cached on
+      // the transmission bookkeeping, so reuse is structural — the prefix
+      // cannot change mid-session and the provider prefix cache holds by
+      // construction (resume = a new instance = a recompose, anchored by its
+      // 'resume' snapshot). The prefix is not session history — the header
+      // event in runStep is its only durable record
+      // (EpochHeader.messagePrefix). The frozen empty seed serves both the
+      // listener chain and the no-listener fallback: a contribution is a
+      // RETURNED extension of `await next()`, never an in-place push. This
+      // runs OUTSIDE the step, before the boundary snapshot: a composing
+      // listener's session append lands before the boundary and joins the
+      // CURRENT request.
+      if (transmission.sessionPrefix === undefined) {
+        const emptyPrefix: Message[] = deepFreeze([])
+        const composed = await ctx.waterfall(
+          'agent/session-prefix', agent, emptyPrefix, abort.signal,
+          () => Promise.resolve(emptyPrefix),
+        )
+
+        // Interruption landing during prefix composition: mirror the assembly
+        // window above — drop the about-to-start step without running the
+        // seam, and DISCARD the composition instead of caching it. An
+        // abort-aware listener may have returned a degraded fallback under
+        // the firing signal; committing it would ship a prefix no request
+        // ever used (and no header ever logged) on this instance's next real
+        // request. The next turn recomposes under a live signal — the cache
+        // only ever holds a fully composed prefix. The cache-hit path needs
+        // no such check: nothing awaits between the assembly check above and
+        // the pre-step seam.
+        if (handle.isCancelled() || handle.isDisposed()) {
+          handle.setAbort(undefined)
+          reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
+          break
+        }
+        transmission.sessionPrefix = deepFreeze(structuredClone(composed))
+      }
+
      // Pre-step surface-mutation checkpoint (compaction), fired OUTSIDE the
      // step: after `turn/start` (and the prior step's close) but before
      // `step/start`, so a compaction's log-only `compact/*` records and its
@@ -479,8 +530,10 @@ async function runTurn(
      // concurrent listeners cannot interleave their `session.append`s. A
      // throwing listener escapes to the outer catch, which closes the (not-yet-
      // open) step as a no-op and ends the turn via failTurn — a broken
-      // pre-step plugin ends the turn, not the loop.
-      await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, abort.signal)
+      // pre-step plugin ends the turn, not the loop. The composed session
+      // prefix rides along so token-pressure listeners count everything the
+      // request will actually carry.
+      await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, transmission.sessionPrefix, abort.signal)

      // Interruption landing during the pre-step seam: do not open an empty step.
      if (handle.isCancelled() || handle.isDisposed()) {
@@ -671,11 +724,12 @@ function drainSteering(agent: ReactLoopAgent, turn: number): boolean {
 }

 /** One step: build the request from the boundary snapshot + the step's
- * header → log the header event the request owes → stream model → record →
- * execute tools. The caller assembles the system prompt, fires the
- * `agent/pre-step` seam, snapshots the derivation, and opens the step BEFORE
- * calling this, so `boundaryMessages` is exactly the surface prefix at
- * step/start and already reflects any compaction. */
+ * header → compose the session prefix if this instance has none yet → log
+ * the header event the request owes → stream model → record → execute
+ * tools. The caller assembles the
+ * system prompt, fires the `agent/pre-step` seam, snapshots the derivation,
+ * and opens the step BEFORE calling this, so `boundaryMessages` is exactly
+ * the surface prefix at step/start and already reflects any compaction. */
 async function runStep(
  ctx: Context,
  agent: ReactLoopAgent,
@@ -715,22 +769,30 @@ async function runStep(
    throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
  }

+  // The session prefix was composed (once per instance) before this step's
+  // pre-step seam — the caller guarantees it, so the cache is always set here.
+  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
+  const sessionPrefix = transmission.sessionPrefix!
+
  // The request header (the log's request/header* vocabulary): canonical form,
-  // recorded before dispatch so the log always explains the request.
+  // recorded before dispatch so the log always explains the request —
+  // including the session prefix, which no other event carries.
  const header = canonicalHeader({
    config,
    ...system ? { system } : {},
    ...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
+    ...sessionPrefix.length > 0 ? { messagePrefix: sessionPrefix } : {},
  })
  recordRequestHeader(session, transmission, header)

  // Build and freeze: the request is a pure function of (boundary snapshot,
  // logged header) — llm/stream listeners and adapters read it, mutation
  // throws. sessionId + frozen is the loop-built marker the dev invariant
-  // keys on.
+  // keys on. Message order: header.messagePrefix, then the boundary
+  // snapshot — the reconstruction equation the invariant recomputes.
  const request: GenerateOptions = deepFreeze({
    model: header.config.model,
-    messages: boundaryMessages,
+    messages: [...header.messagePrefix ?? [], ...boundaryMessages],
    ...header.system !== undefined ? { system: header.system } : {},
    ...header.tools !== undefined ? { tools: header.tools } : {},
    ...header.config.temperature !== undefined ? { temperature: header.config.temperature } : {},
@@ -877,7 +939,11 @@ function withoutToolCalls(message: Message): Message {
  return { ...message, content: message.content.filter(block => block.type !== 'tool-call') }
 }

-/** The last turn number in a (possibly seeded) session log, or 0. */
+/**
+ * The last turn number in a (possibly seeded) session log, or 0.
+ * @param session - the session whose log is scanned for the latest `turn/start`.
+ * @returns the latest `turn/start`'s turn number, or 0 when the log has none (the next turn is this plus one).
+ */
 export function lastTurnNumber(session: Session): number {
  const lastStart = session.events.findLast(event => event.type === 'turn/start')
  return lastStart?.data.turn ?? 0
@@ -891,6 +957,8 @@ export function lastTurnNumber(session: Session): number {
 * returns to idle), so status is not a reliable open-turn signal. Used by
 * `inject()` to choose between appending into an open turn vs. wrapping the
 * injection in its own one-shot turn (the turn-enclosure RFC).
+ * @param session - the session whose log is inspected.
+ * @returns true when the log's last turn boundary is a `turn/start` with no matching `turn/end` yet.
 */
 export function isTurnOpen(session: Session): boolean {
  const last = session.events.findLast(e => e.type === 'turn/start' || e.type === 'turn/end')