diff --git a/.pylon/features.yaml b/.pylon/features.yaml index a8cc132686..33797f3c2d 100644 --- a/.pylon/features.yaml +++ b/.pylon/features.yaml @@ -265,3 +265,19 @@ decisions: revisit_when: - Prime upstream exposes an equivalent generation-scoped, post-attach public proof that is false before attach and after invalidation. - Pylon and Comet can remove the fork SDK token/accessor without enabling optional behavior from a server offer, version, or method presence. + + child-scoped-provider-identity: + area: runtime-reliability + state: shipped + owner: pylon-prime-integration + decision: retain + pylon_refs: + - https://github.com/pylon-code/prime-agent/issues/22 + - https://github.com/pylon-code/prime-agent/issues/23 + upstream_refs: + - https://github.com/PrimeIntellect-ai/prime-agent/tree/a903d4b6768f484bd6d459b7b0aa7dee38e461e2 + fork_change: child-scoped-provider-identity-v1 + upstream_support: Prime through a903d4b6768f constructs inline RLM children and side questions with the root session's onPayload/onResponse closures, and calls compaction, branch summarization, and refinement completeSimple with no payload hook. Every derived request therefore either claims the parent's provider session identity while sending a divergent history or arrives with no identity at all. + revisit_when: + - Prime routes subagent and derived-request provider hooks through the owning session's extension runner. + - An upstream extension contract exposes per-agent provider identity that supersedes the scoped session-id view. diff --git a/.pylon/upstream-review.md b/.pylon/upstream-review.md index 633f485a63..2decd48991 100644 --- a/.pylon/upstream-review.md +++ b/.pylon/upstream-review.md @@ -118,3 +118,11 @@ This ledger records Prime upstream evidence and the decision taken for each over - Large-transcript handling prepares one immutable payload, bounds framing and drain waits, avoids quadratic private-buffer shifting, and preserves spill ownership and cleanup across cancellation, crash, and stale generations. Stock/current `v0.8.1` supervisor and worker directions retain their mixed-version fallback. - The pre-ledger source candidate `8b504e3774875c241c5d0d3b4b588a09f4aa3f8e` passed `npm run check`, package build, 246 conflict-affected exact-head tests after rebase, 16 real supervisor-process tests with 8 fixture-gated skips, stock/current compatibility in both directions, a 36 MiB exact-package transfer, a 10,000-message preparation probe, a 131,000-fragment framing probe, and two independent adversarial reviews. The ledger correction changes the exact head and therefore requires renewed targeted checks and hosted CI before merge. - Revisit when Prime upstream supplies the same capability-gated fresh-generation identity, attachment-local retry containment, mixed-version behavior, and bounded preparation/framing guarantees without weakening Pylon's correlated lifecycle or cleanup contracts. + +## 2026-08-30 — child-scoped provider identity for subagents and derived requests + +- Upstream baseline: `PrimeIntellect-ai/prime-agent@a903d4b6768f484bd6d459b7b0aa7dee38e461e2`; this fix is client-local and does not advance `reviewed_upstream_commit`. +- Reviewed current upstream `agent-session.ts`, `side-question.ts`, `sdk.ts`, `compaction/`, and `refinement/`, plus upstream issues and pull requests for provider identity, `metadata.user_id`, and `before_provider_request`. Upstream has the same defect and no equivalent work; only issue #832 touches provider hooks, and it adds unrelated header/settled hooks. +- `child-scoped-provider-identity`: **retain**. Inline RLM children now run their payload, response, and context hooks through their own extension runner, converging with the daemon path that already gets a per-child runner from `createAgentSession`. Side questions and the compaction, branch-summary, refine, and auto-refine-review passes run through a scoped view of the owning session whose `getSessionId()` returns `/`; every other accessor still reports the owning session. The extension contract stays additive: existing `before_provider_request` handlers keep working and simply observe one identity per agent instead of the parent's for all of them. +- Out of scope by design: `SessionManager.sessionId` still changes on fork and branch. A genuinely divergent history deserves a new provider key, so persisting identity across forks is a separate decision. +- Validation: `npm run check` clean. `test/suite/regressions/23-child-provider-identity.test.ts` passes 3/3 and fails on the pre-fix inline-child wiring. Adjacent suites pass: side questions, fast-mode children, compaction (suite, extensions, summary reasoning), refinement, subagent runtime host, subagent model selection, subagent terminal messages, agent-session runtime, recursion, context tree, concurrent sessions, daemon agent connection, and the `packages/ai` faux provider — 620 passes with 8 skips across 18 files. diff --git a/packages/ai/.changes/23-faux-provider-payload-hook.md b/packages/ai/.changes/23-faux-provider-payload-hook.md new file mode 100644 index 0000000000..69ed29d9eb --- /dev/null +++ b/packages/ai/.changes/23-faux-provider-payload-hook.md @@ -0,0 +1 @@ +- Added `onPayload` support and recorded request payloads (`getSentPayloads`, `clearSentPayloads`) to the faux provider so payload-rewriting hosts and extensions can be tested ([#23](https://github.com/pylon-code/prime-agent/issues/23)). diff --git a/packages/ai/src/providers/faux.ts b/packages/ai/src/providers/faux.ts index 7bba1b72ec..9bed351325 100644 --- a/packages/ai/src/providers/faux.ts +++ b/packages/ai/src/providers/faux.ts @@ -113,6 +113,20 @@ export interface RegisterFauxProviderOptions { }; } +/** + * Stand-in for a provider's serialized request body. Real providers build an + * API-shaped payload and hand it to `onPayload` before sending; the faux + * provider builds this instead so hosts and extensions that stamp identity + * onto the payload are exercised and observable. + */ +export interface FauxRequestPayload { + model: string; + systemPrompt?: string; + messages: Message[]; + sessionId?: string; + metadata?: Record; +} + export interface FauxProviderRegistration { api: string; models: [Model, ...Model[]]; @@ -122,6 +136,9 @@ export interface FauxProviderRegistration { setResponses: (responses: FauxResponseStep[]) => void; appendResponses: (responses: FauxResponseStep[]) => void; getPendingResponseCount: () => number; + /** Payloads as they stood after `onPayload`, in request order. */ + getSentPayloads: () => unknown[]; + clearSentPayloads: () => void; unregister: () => void; } @@ -401,6 +418,7 @@ export function registerFauxProvider(options: RegisterFauxProviderOptions = {}): const tokensPerSecond = options.tokensPerSecond; const state = { callCount: 0 }; const promptCache = new Map(); + const sentPayloads: unknown[] = []; const modelDefinitions = options.models?.length ? options.models @@ -435,6 +453,15 @@ export function registerFauxProvider(options: RegisterFauxProviderOptions = {}): queueMicrotask(async () => { try { + const payload: FauxRequestPayload = { + model: requestModel.id, + systemPrompt: context.systemPrompt, + messages: context.messages, + sessionId: streamOptions?.sessionId, + metadata: streamOptions?.metadata, + }; + const replacedPayload = await streamOptions?.onPayload?.(payload, requestModel); + sentPayloads.push(replacedPayload ?? payload); await streamOptions?.onResponse?.({ status: 200, headers: {} }, requestModel); if (!step) { let message = createErrorMessage( @@ -492,6 +519,12 @@ export function registerFauxProvider(options: RegisterFauxProviderOptions = {}): getPendingResponseCount() { return pendingResponses.length; }, + getSentPayloads() { + return [...sentPayloads]; + }, + clearSentPayloads() { + sentPayloads.length = 0; + }, unregister() { unregisterApiProviders(sourceId); }, diff --git a/packages/coding-agent/.changes/23-child-provider-identity.md b/packages/coding-agent/.changes/23-child-provider-identity.md new file mode 100644 index 0000000000..0e5c737050 --- /dev/null +++ b/packages/coding-agent/.changes/23-child-provider-identity.md @@ -0,0 +1 @@ +- Fixed inline subagents, side questions, compaction, branch summarization, and refinement sending provider requests under the parent session's identity or none at all ([#23](https://github.com/pylon-code/prime-agent/issues/23)). diff --git a/packages/coding-agent/docs/extensions.md b/packages/coding-agent/docs/extensions.md index 4fd2912808..cb5a71d71d 100644 --- a/packages/coding-agent/docs/extensions.md +++ b/packages/coding-agent/docs/extensions.md @@ -649,6 +649,13 @@ pi.on("before_provider_request", (event, ctx) => { This is mainly useful for debugging provider serialization and cache behavior. +The hook fires for every provider request the session is responsible for, and `ctx.sessionManager.getSessionId()` identifies the conversation the request belongs to rather than the session you started from: + +- A subagent's requests report the subagent's own session id, whether it runs inline or in the daemon. Handlers that key provider state on the session id therefore see one identity per agent instead of the parent's for all of them. +- Requests that carry a conversation the session never sent report a derived id of the form `/`: `side:` for a side question, and `compaction`, `branch-summary`, `refine`, or `auto-refine-review` for the summarization passes. Everything else on `ctx.sessionManager` still describes the owning session. + +Treat the id as opaque: match on the `` prefix if you need to group derived work with its session. + #### after_provider_response Fired after an HTTP response is received and before its stream body is consumed. Handlers run in extension load order. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index a2de9a3d54..e6013046d7 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -117,6 +117,7 @@ import { normalizeHeartbeatDeliveryMode } from "./cron-jobs.js"; import { DEFAULT_THINKING_LEVEL } from "./defaults.js"; import { exportSessionToHtml, type ToolHtmlRenderer } from "./export-html/index.js"; import { createToolHtmlRenderer } from "./export-html/tool-renderer.js"; +import { createExtensionProviderHooks, type ExtensionProviderHooks } from "./extension-provider-hooks.js"; import { type ContextUsage, type ExtensionCommandContextActions, @@ -1354,6 +1355,19 @@ export class AgentSession { this._mcpManager?.refresh(); } + /** + * Provider hooks for requests made on this session's behalf that send a + * conversation this session never sent — side questions and the + * summarization/refinement passes. `scope` derives a child identity from the + * session id so those requests reach the provider identified, without + * claiming the session's own provider session key for a divergent history. + * + * The runner is resolved per request, so `/reload` is picked up. + */ + createScopedProviderHooks(scope: string): ExtensionProviderHooks { + return createExtensionProviderHooks(() => this._extensionRunner, { sessionIdScope: scope }); + } + /** * Set the RLM heartbeat controller after construction. Used by * print/headless mode to attach an in-process heartbeat scheduler @@ -8015,7 +8029,16 @@ export class AgentSession { const { summary, firstKeptEntryId, tokensBefore, details } = extensionCompaction ?? - (await compact(preparation, model, apiKey, headers, customInstructions, signal, this.thinkingLevel)); + (await compact( + preparation, + model, + apiKey, + headers, + customInstructions, + signal, + this.thinkingLevel, + this.createScopedProviderHooks("compaction").onPayload, + )); if (signal.aborted) { throw new Error("Compaction cancelled"); @@ -8512,6 +8535,7 @@ export class AgentSession { headers, signal, this.thinkingLevel, + this.createScopedProviderHooks("auto-refine-review").onPayload, ); } @@ -8751,6 +8775,7 @@ export class AgentSession { headers, signal, this.thinkingLevel, + this.createScopedProviderHooks("refine").onPayload, ); if (this._disposed || signal.aborted) { throw new Error("Refinement cancelled because the session was disposed."); @@ -10002,6 +10027,14 @@ export class AgentSession { childSessionManager.appendThinkingLevelChange(options.thinkingLevel); childSessionManager.appendServiceTierChange(options.serviceTier); + // The child's extension hooks must run against the child's own runner, or + // every request it makes claims this session's provider identity and the + // parent and its children interleave on one provider session key. The + // runner only exists once the child AgentSession builds it below, which is + // why the hooks resolve it late — the same shape the SDK root uses. + const childExtensionRunnerRef: { current?: ExtensionRunner } = {}; + const childProviderHooks = createExtensionProviderHooks(() => childExtensionRunnerRef.current); + const childAgent = new Agent({ initialState: { systemPrompt: "", @@ -10011,11 +10044,11 @@ export class AgentSession { tools: [], }, convertToLlm: this.agent.convertToLlm, - transformContext: this.agent.transformContext, + transformContext: childProviderHooks.transformContext, streamFn: this.agent.streamFn, getApiKey: this.agent.getApiKey, - onPayload: this.agent.onPayload, - onResponse: this.agent.onResponse, + onPayload: childProviderHooks.onPayload, + onResponse: childProviderHooks.onResponse, steeringMode: this.settingsManager.getSteeringMode(), followUpMode: this.settingsManager.getFollowUpMode(), sessionId: childSessionManager.getSessionId(), @@ -10035,6 +10068,7 @@ export class AgentSession { resourceLoader: this._resourceLoader, customTools: options.customTools, modelRegistry: this._modelRegistry, + extensionRunnerRef: childExtensionRunnerRef, initialActiveToolNames: options.activeToolNames, allowedToolNames: options.allowedToolNames, includeGoals: options.includeGoals, @@ -12049,6 +12083,7 @@ export class AgentSession { customInstructions, replaceInstructions, reserveTokens: branchSummarySettings.reserveTokens, + onPayload: this.createScopedProviderHooks("branch-summary").onPayload, }); if (result.aborted) { return { cancelled: true, aborted: true }; diff --git a/packages/coding-agent/src/core/compaction/branch-summarization.ts b/packages/coding-agent/src/core/compaction/branch-summarization.ts index a23f825f05..e384089742 100644 --- a/packages/coding-agent/src/core/compaction/branch-summarization.ts +++ b/packages/coding-agent/src/core/compaction/branch-summarization.ts @@ -8,6 +8,7 @@ import type { AgentMessage } from "@earendil-works/pi-agent-core"; import type { Model } from "@earendil-works/pi-ai"; import { completeSimple } from "@earendil-works/pi-ai"; +import type { ProviderPayloadHook } from "../extension-provider-hooks.js"; import { convertToLlm, createBranchSummaryMessage, @@ -72,6 +73,11 @@ export interface GenerateBranchSummaryOptions { replaceInstructions?: boolean; /** Tokens reserved for prompt + LLM response (default 16384) */ reserveTokens?: number; + /** + * Provider payload hook carrying the owning session's identity. Expected to + * be scoped: the summarization prompt is not the session's conversation. + */ + onPayload?: ProviderPayloadHook; } /** * Collect entries that should be summarized when navigating from one position to another. @@ -249,7 +255,16 @@ export async function generateBranchSummary( entries: SessionEntry[], options: GenerateBranchSummaryOptions, ): Promise { - const { model, apiKey, headers, signal, customInstructions, replaceInstructions, reserveTokens = 16384 } = options; + const { + model, + apiKey, + headers, + signal, + customInstructions, + replaceInstructions, + reserveTokens = 16384, + onPayload, + } = options; const contextWindow = model.contextWindow || 128000; const tokenBudget = contextWindow - reserveTokens; @@ -282,7 +297,7 @@ export async function generateBranchSummary( const response = await completeSimple( model, { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, - { apiKey, headers, signal, maxTokens: 2048 }, + { apiKey, headers, signal, onPayload, maxTokens: 2048 }, ); if (response.stopReason === "aborted") { return { aborted: true }; diff --git a/packages/coding-agent/src/core/compaction/compaction.ts b/packages/coding-agent/src/core/compaction/compaction.ts index cdda5ed080..db5aa5c99a 100644 --- a/packages/coding-agent/src/core/compaction/compaction.ts +++ b/packages/coding-agent/src/core/compaction/compaction.ts @@ -8,6 +8,7 @@ import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { AssistantMessage, Model, Usage } from "@earendil-works/pi-ai"; import { completeSimple } from "@earendil-works/pi-ai"; +import type { ProviderPayloadHook } from "../extension-provider-hooks.js"; import { convertToLlm, createBranchSummaryMessage, @@ -504,6 +505,10 @@ export function buildSummarizationPrompt(customInstructions?: string, previousSu /** * Generate a summary of the conversation using the LLM. * If previousSummary is provided, uses the update prompt to merge. + * + * `onPayload` carries the owning session's provider identity so the call is + * not anonymous at the provider; it is expected to be scoped, because the + * summarization prompt is not the session's own conversation. */ export async function generateSummary( currentMessages: AgentMessage[], @@ -515,6 +520,7 @@ export async function generateSummary( customInstructions?: string, previousSummary?: string, thinkingLevel?: ThinkingLevel, + onPayload?: ProviderPayloadHook, ): Promise { const maxTokens = Math.floor(0.8 * reserveTokens); @@ -538,8 +544,8 @@ export async function generateSummary( const completionOptions = model.reasoning && thinkingLevel && thinkingLevel !== "off" - ? { maxTokens, signal, apiKey, headers, reasoning: thinkingLevel } - : { maxTokens, signal, apiKey, headers }; + ? { maxTokens, signal, apiKey, headers, onPayload, reasoning: thinkingLevel } + : { maxTokens, signal, apiKey, headers, onPayload }; const response = await completeSimple( model, @@ -678,6 +684,7 @@ export async function compact( customInstructions?: string, signal?: AbortSignal, thinkingLevel?: ThinkingLevel, + onPayload?: ProviderPayloadHook, ): Promise { const { firstKeptEntryId, @@ -704,6 +711,7 @@ export async function compact( customInstructions, previousSummary, thinkingLevel, + onPayload, ) : Promise.resolve("No prior history."), generateTurnPrefixSummary( @@ -714,6 +722,7 @@ export async function compact( headers, signal, thinkingLevel, + onPayload, ), ]); summary = `${historyResult}\n\n---\n\n**Turn Context (split turn):**\n\n${turnPrefixResult}`; @@ -728,6 +737,7 @@ export async function compact( customInstructions, previousSummary, thinkingLevel, + onPayload, ); } const { readFiles, modifiedFiles } = computeFileLists(fileOps); @@ -756,6 +766,7 @@ async function generateTurnPrefixSummary( headers?: Record, signal?: AbortSignal, thinkingLevel?: ThinkingLevel, + onPayload?: ProviderPayloadHook, ): Promise { const maxTokens = Math.floor(0.5 * reserveTokens); // Smaller budget for turn prefix const llmMessages = convertToLlm(messages); @@ -773,8 +784,8 @@ async function generateTurnPrefixSummary( model, { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, model.reasoning && thinkingLevel && thinkingLevel !== "off" - ? { maxTokens, signal, apiKey, headers, reasoning: thinkingLevel } - : { maxTokens, signal, apiKey, headers }, + ? { maxTokens, signal, apiKey, headers, onPayload, reasoning: thinkingLevel } + : { maxTokens, signal, apiKey, headers, onPayload }, ); if (response.stopReason === "error") { diff --git a/packages/coding-agent/src/core/extension-provider-hooks.ts b/packages/coding-agent/src/core/extension-provider-hooks.ts new file mode 100644 index 0000000000..39124e3368 --- /dev/null +++ b/packages/coding-agent/src/core/extension-provider-hooks.ts @@ -0,0 +1,71 @@ +/** + * Provider-request hooks bound to one session's extension runner. + * + * Anthropic-protocol providers carry session identity only in the payload + * (`metadata.user_id`), which extensions stamp from `before_provider_request`. + * The hook therefore has to resolve the runner of the session that owns the + * request; sharing a parent's hook makes every child request claim the + * parent's provider session. + */ + +import type { AgentMessage } from "@earendil-works/pi-agent-core"; +import type { SimpleStreamOptions } from "@earendil-works/pi-ai"; +import type { ExtensionRunner } from "./extensions/index.js"; + +export type ProviderPayloadHook = NonNullable; +export type ProviderResponseHook = NonNullable; + +export interface ExtensionProviderHooks { + onPayload: ProviderPayloadHook; + onResponse: ProviderResponseHook; + transformContext: (messages: AgentMessage[]) => Promise; +} + +export interface ExtensionProviderHookOptions { + /** + * Derives a child identity from the owning session's id for requests that + * carry a history the session itself never sent (side questions, + * summarization). Without it those requests would key the provider's + * session cache to the owner while replaying a divergent conversation. + */ + sessionIdScope?: string; +} + +/** + * Build provider hooks that resolve their runner at call time. + * + * Late resolution is required in both directions: an `Agent` is constructed + * before the `AgentSession` that owns its runner, and `/reload` replaces the + * runner on a live session. + */ +export function createExtensionProviderHooks( + getRunner: () => ExtensionRunner | undefined, + options: ExtensionProviderHookOptions = {}, +): ExtensionProviderHooks { + const { sessionIdScope } = options; + return { + onPayload: async (payload) => { + const runner = getRunner(); + if (!runner?.hasHandlers("before_provider_request")) { + return payload; + } + return runner.emitBeforeProviderRequest(payload, { sessionIdScope }); + }, + onResponse: async (response) => { + const runner = getRunner(); + if (!runner?.hasHandlers("after_provider_response")) { + return; + } + await runner.emit({ + type: "after_provider_response", + status: response.status, + headers: response.headers, + }); + }, + transformContext: async (messages) => { + const runner = getRunner(); + if (!runner) return messages; + return runner.emitContext(messages, { sessionIdScope }); + }, + }; +} diff --git a/packages/coding-agent/src/core/extensions/index.ts b/packages/coding-agent/src/core/extensions/index.ts index ba96318205..bf48f0c075 100644 --- a/packages/coding-agent/src/core/extensions/index.ts +++ b/packages/coding-agent/src/core/extensions/index.ts @@ -20,6 +20,7 @@ export type { ForkHandler, NavigateTreeHandler, NewSessionHandler, + ScopedEmitOptions, ShutdownHandler, SwitchSessionHandler, } from "./runner.js"; diff --git a/packages/coding-agent/src/core/extensions/runner.ts b/packages/coding-agent/src/core/extensions/runner.ts index 172f725003..fe9595cac4 100644 --- a/packages/coding-agent/src/core/extensions/runner.ts +++ b/packages/coding-agent/src/core/extensions/runner.ts @@ -9,7 +9,7 @@ import { type Theme, theme } from "../../modes/interactive/theme/theme.js"; import type { ResourceDiagnostic } from "../diagnostics.js"; import type { KeybindingsConfig } from "../keybindings.js"; import type { ModelRegistry } from "../model-registry.js"; -import type { SessionManager } from "../session-manager.js"; +import type { ReadonlySessionManager, SessionManager } from "../session-manager.js"; import type { BuildSystemPromptOptions } from "../system-prompt.js"; import type { BeforeAgentStartEvent, @@ -232,6 +232,35 @@ const noOpUIContext: ExtensionUIContext = { setToolsExpanded: () => {}, }; +/** + * Read-only session view whose id identifies work derived from a session + * rather than the session itself. Everything else still reports the owning + * session, so an extension keyed on `getSessionId()` gets a distinct provider + * identity without losing access to the real transcript. + */ +function scopedSessionManagerView(base: SessionManager, scope: string): ReadonlySessionManager { + return { + getCwd: () => base.getCwd(), + getSessionDir: () => base.getSessionDir(), + getSessionId: () => `${base.getSessionId()}/${scope}`, + getSessionFile: () => base.getSessionFile(), + getLeafId: () => base.getLeafId(), + getLeafEntry: () => base.getLeafEntry(), + getEntry: (id) => base.getEntry(id), + getLabel: (id) => base.getLabel(id), + getBranch: (fromId) => base.getBranch(fromId), + getHeader: () => base.getHeader(), + getEntries: () => base.getEntries(), + getTree: () => base.getTree(), + getSessionName: () => base.getSessionName(), + }; +} + +export interface ScopedEmitOptions { + /** Derives a child provider identity from the owning session's id. */ + sessionIdScope?: string; +} + export class ExtensionRunner { private extensions: Extension[]; private runtime: ExtensionRuntime; @@ -575,7 +604,7 @@ export class ExtensionRunner { * Create an ExtensionContext for use in event handlers and tool execution. * Context values are resolved at call time, so changes via bindCore/bindUI are reflected. */ - createContext(): ExtensionContext { + createContext(sessionIdScope?: string): ExtensionContext { const runner = this; const getModel = this.getModel; return { @@ -593,7 +622,9 @@ export class ExtensionRunner { }, get sessionManager() { runner.assertActive(); - return runner.sessionManager; + return sessionIdScope + ? scopedSessionManagerView(runner.sessionManager, sessionIdScope) + : runner.sessionManager; }, get modelRegistry() { runner.assertActive(); @@ -861,8 +892,8 @@ export class ExtensionRunner { return undefined; } - async emitContext(messages: AgentMessage[]): Promise { - const ctx = this.createContext(); + async emitContext(messages: AgentMessage[], options: ScopedEmitOptions = {}): Promise { + const ctx = this.createContext(options.sessionIdScope); let currentMessages = structuredClone(messages); for (const ext of this.extensions) { @@ -893,8 +924,8 @@ export class ExtensionRunner { return currentMessages; } - async emitBeforeProviderRequest(payload: unknown): Promise { - const ctx = this.createContext(); + async emitBeforeProviderRequest(payload: unknown, options: ScopedEmitOptions = {}): Promise { + const ctx = this.createContext(options.sessionIdScope); let currentPayload = payload; for (const ext of this.extensions) { diff --git a/packages/coding-agent/src/core/refinement/refinement.ts b/packages/coding-agent/src/core/refinement/refinement.ts index 76cc701556..75e4aa8bec 100644 --- a/packages/coding-agent/src/core/refinement/refinement.ts +++ b/packages/coding-agent/src/core/refinement/refinement.ts @@ -15,6 +15,7 @@ import type { Model } from "@earendil-works/pi-ai"; import { completeSimple } from "@earendil-works/pi-ai"; import { getAgentDir } from "../../config.js"; import { serializeConversation } from "../compaction/utils.js"; +import type { ProviderPayloadHook } from "../extension-provider-hooks.js"; import { convertToLlm } from "../messages.js"; import type { CustomEntry } from "../session-manager.js"; @@ -887,6 +888,7 @@ export async function planRefinement( headers?: Record, signal?: AbortSignal, thinkingLevel?: ThinkingLevel, + onPayload?: ProviderPayloadHook, ): Promise { const id = generateRefinementId(); if (options.rollbackId) { @@ -930,7 +932,7 @@ export async function planRefinement( systemPrompt: REFINEMENT_SYSTEM_PROMPT, messages: [{ role: "user", content: [{ type: "text", text: userPrompt }], timestamp: Date.now() }], }, - { maxTokens: refinementMaxOutputTokens(model), signal, apiKey, headers }, + { maxTokens: refinementMaxOutputTokens(model), signal, apiKey, headers, onPayload }, ); if (response.stopReason === "error") { @@ -970,6 +972,7 @@ export async function reviewAutoRefine( headers?: Record, signal?: AbortSignal, thinkingLevel?: ThinkingLevel, + onPayload?: ProviderPayloadHook, ): Promise { const conversationText = serializeConversation(convertToLlm(messages)).slice(-40_000); const userPrompt = [ @@ -996,7 +999,7 @@ ${conversationText} systemPrompt: AUTO_REFINE_REVIEW_SYSTEM_PROMPT, messages: [{ role: "user", content: [{ type: "text", text: userPrompt }], timestamp: Date.now() }], }, - { maxTokens: autoRefineReviewMaxOutputTokens(model), signal, apiKey, headers }, + { maxTokens: autoRefineReviewMaxOutputTokens(model), signal, apiKey, headers, onPayload }, ); if (response.stopReason === "error") { throw new Error(`Auto-refine review failed: ${response.errorMessage || "Unknown error"}`); @@ -1021,8 +1024,20 @@ export async function refineHarness( headers?: Record, signal?: AbortSignal, thinkingLevel?: ThinkingLevel, + onPayload?: ProviderPayloadHook, ): Promise { - const plan = await planRefinement(messages, state, history, model, apiKey, options, headers, signal, thinkingLevel); + const plan = await planRefinement( + messages, + state, + history, + model, + apiKey, + options, + headers, + signal, + thinkingLevel, + onPayload, + ); return applyRefinementProposal(state, plan.proposal, { id: plan.id, rollbackOf: plan.rollbackOf, diff --git a/packages/coding-agent/src/core/sdk.ts b/packages/coding-agent/src/core/sdk.ts index 72b740e980..acd3fff984 100644 --- a/packages/coding-agent/src/core/sdk.ts +++ b/packages/coding-agent/src/core/sdk.ts @@ -8,6 +8,7 @@ import { formatNoModelsAvailableMessage } from "./auth-guidance.js"; import { AuthStorage } from "./auth-storage.js"; import type { AgentAutonomousConfig } from "./autonomous.js"; import { DEFAULT_THINKING_LEVEL } from "./defaults.js"; +import { createExtensionProviderHooks } from "./extension-provider-hooks.js"; import type { ExtensionRunner, LoadExtensionsResult, SessionStartEvent, ToolDefinition } from "./extensions/index.js"; import { McpManager } from "./mcp/mcp-manager.js"; import { convertToLlm } from "./messages.js"; @@ -273,6 +274,7 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} }; const extensionRunnerRef: { current?: ExtensionRunner } = {}; + const providerHooks = createExtensionProviderHooks(() => extensionRunnerRef.current); agent = new Agent({ initialState: { @@ -298,30 +300,10 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} headers: auth.headers || options?.headers ? { ...auth.headers, ...options?.headers } : undefined, }); }, - onPayload: async (payload, _model) => { - const runner = extensionRunnerRef.current; - if (!runner?.hasHandlers("before_provider_request")) { - return payload; - } - return runner.emitBeforeProviderRequest(payload); - }, - onResponse: async (response, _model) => { - const runner = extensionRunnerRef.current; - if (!runner?.hasHandlers("after_provider_response")) { - return; - } - await runner.emit({ - type: "after_provider_response", - status: response.status, - headers: response.headers, - }); - }, + onPayload: providerHooks.onPayload, + onResponse: providerHooks.onResponse, sessionId: sessionManager.getSessionId(), - transformContext: async (messages) => { - const runner = extensionRunnerRef.current; - if (!runner) return messages; - return runner.emitContext(messages); - }, + transformContext: providerHooks.transformContext, steeringMode: settingsManager.getSteeringMode(), followUpMode: settingsManager.getFollowUpMode(), transport: settingsManager.getTransport(), diff --git a/packages/coding-agent/src/core/side-question.ts b/packages/coding-agent/src/core/side-question.ts index 02d970cd8c..8f1c4cc812 100644 --- a/packages/coding-agent/src/core/side-question.ts +++ b/packages/coding-agent/src/core/side-question.ts @@ -1,8 +1,18 @@ import { Agent, type AgentMessage } from "@earendil-works/pi-agent-core"; import type { AssistantMessage, UserMessage } from "@earendil-works/pi-ai"; +import type { ExtensionProviderHooks } from "./extension-provider-hooks.js"; export type SideQuestionStatus = "running" | "complete" | "cancelled" | "error"; +/** + * The session a side question forks from. Structural so callers can pass an + * `AgentSession` without this module depending on it. + */ +export interface SideQuestionHost { + agent: Agent; + createScopedProviderHooks(scope: string): ExtensionProviderHooks; +} + export interface SideQuestionEvent { id: string; question: string; @@ -40,16 +50,21 @@ function readAssistantText(message: AgentMessage): string { } export function startSideQuestion( - parent: Agent, + host: SideQuestionHost, id: string, question: string, onEvent: (event: SideQuestionEvent) => void | Promise, previousTurns: SideQuestionTurn[] = [], ): SideQuestionRun { + const parent = host.agent; const model = parent.state.model; if (!model) { throw new Error("Select a model before asking a side question"); } + // A side question replays the main conversation plus turns the main session + // never sent, so it must not be stamped with the main session's provider + // identity: same key, divergent history discards the provider-side session. + const providerHooks = host.createScopedProviderHooks(`side:${id}`); // Each turn re-clones the live main conversation, so follow-ups always see // the newest main-thread context; earlier side turns are replayed after it. @@ -88,13 +103,13 @@ export function startSideQuestion( tools: [], }, convertToLlm: parent.convertToLlm, - transformContext: parent.transformContext, + transformContext: providerHooks.transformContext, streamFn: parent.streamFn, getApiKey: parent.getApiKey, - onPayload: parent.onPayload, - onResponse: parent.onResponse, + onPayload: providerHooks.onPayload, + onResponse: providerHooks.onResponse, shouldStopAfterTurn: () => true, - sessionId: parent.sessionId, + sessionId: parent.sessionId === undefined ? undefined : `${parent.sessionId}/side:${id}`, thinkingBudgets: parent.thinkingBudgets, transport: "sse", maxRetryDelayMs: parent.maxRetryDelayMs, diff --git a/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts index 0303c76edb..dc28624b28 100644 --- a/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts @@ -442,7 +442,7 @@ export class InProcessAgentConnection implements AgentConnection { throw new Error(`Side question already exists: ${id}`); } const run = startSideQuestion( - this.session.agent, + this.session, id, question, (event) => this.emit({ type: "side_question_event", event }), diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index cce909f09d..2c01240580 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -4395,7 +4395,7 @@ export class AgentDaemon { throw new Error("A side question is already running for this client and session"); } const run = startSideQuestion( - state.runtime.session.agent, + state.runtime.session, command.sideQuestionId, command.question, (event) => { diff --git a/packages/coding-agent/test/suite/harness.ts b/packages/coding-agent/test/suite/harness.ts index 97bda0d282..fcd6725477 100644 --- a/packages/coding-agent/test/suite/harness.ts +++ b/packages/coding-agent/test/suite/harness.ts @@ -5,7 +5,7 @@ import { existsSync, mkdirSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { AgentMessage, AgentTool } from "@earendil-works/pi-agent-core"; +import type { AgentTool } from "@earendil-works/pi-agent-core"; import { Agent } from "@earendil-works/pi-agent-core"; import type { FauxModelDefinition, FauxProviderRegistration, FauxResponseStep, Model } from "@earendil-works/pi-ai"; import { registerFauxProvider } from "@earendil-works/pi-ai"; @@ -14,6 +14,7 @@ import type { AgentObserveController } from "../../src/core/agent-observe.js"; import { AgentSession, type AgentSessionEvent, type AutoRefineReviewer } from "../../src/core/agent-session.js"; import { AuthStorage } from "../../src/core/auth-storage.js"; import type { AgentAutonomousConfig } from "../../src/core/autonomous.js"; +import { createExtensionProviderHooks } from "../../src/core/extension-provider-hooks.js"; import type { ExtensionRunner } from "../../src/core/extensions/index.js"; import { convertToLlm } from "../../src/core/messages.js"; import { ModelRegistry } from "../../src/core/model-registry.js"; @@ -147,6 +148,7 @@ export async function createHarness(options: HarnessOptions = {}): Promise extensionRunnerRef.current); const agent = new Agent({ getApiKey: () => (withConfiguredAuth ? "faux-key" : undefined), initialState: { @@ -155,29 +157,9 @@ export async function createHarness(options: HarnessOptions = {}): Promise { - const runner = extensionRunnerRef.current; - if (!runner?.hasHandlers("before_provider_request")) { - return payload; - } - return runner.emitBeforeProviderRequest(payload); - }, - onResponse: async (response) => { - const runner = extensionRunnerRef.current; - if (!runner?.hasHandlers("after_provider_response")) { - return; - } - await runner.emit({ - type: "after_provider_response", - status: response.status, - headers: response.headers, - }); - }, - transformContext: async (messages: AgentMessage[]) => { - const runner = extensionRunnerRef.current; - if (!runner) return messages; - return runner.emitContext(messages); - }, + onPayload: providerHooks.onPayload, + onResponse: providerHooks.onResponse, + transformContext: providerHooks.transformContext, }); const extensionsResult = options.extensionFactories ? await createTestExtensionsResult(options.extensionFactories, tempDir) diff --git a/packages/coding-agent/test/suite/regressions/23-child-provider-identity.test.ts b/packages/coding-agent/test/suite/regressions/23-child-provider-identity.test.ts new file mode 100644 index 0000000000..f3ed72baa0 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/23-child-provider-identity.test.ts @@ -0,0 +1,130 @@ +import { type FauxRequestPayload, fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { startSideQuestion } from "../../../src/core/side-question.js"; +import type { ExtensionAPI } from "../../../src/index.js"; +import { createHarness, type Harness } from "../harness.js"; + +/** + * Stands in for the Meridian extension: `metadata.user_id` is the only session + * identity channel Anthropic-protocol providers have, and extensions fill it + * from `ctx.sessionManager.getSessionId()`. + */ +function stampIdentity(observed: string[]) { + return (pi: ExtensionAPI) => { + pi.on("before_provider_request", (event, ctx) => { + const sessionId = ctx.sessionManager.getSessionId(); + observed.push(sessionId); + return { + ...(event.payload as FauxRequestPayload), + metadata: { user_id: JSON.stringify({ session_id: sessionId }) }, + }; + }); + }; +} + +function sentUserIds(harness: Harness): string[] { + return harness.faux.getSentPayloads().map((payload) => { + const userId = (payload as FauxRequestPayload).metadata?.user_id; + return typeof userId === "string" ? (JSON.parse(userId) as { session_id: string }).session_id : ""; + }); +} + +describe("#23 child-scoped provider identity", () => { + const harnesses: Harness[] = []; + + afterEach(() => { + while (harnesses.length > 0) { + harnesses.pop()?.cleanup(); + } + }); + + it("stamps each inline RLM child with its own session id, never the parent's", async () => { + const observed: string[] = []; + const harness = await createHarness({ + persistSession: true, + extensionFactories: [stampIdentity(observed)], + }); + harnesses.push(harness); + + harness.setResponses([fauxAssistantMessage("parent answer")]); + await harness.session.prompt("parent turn"); + const parentSessionId = harness.session.sessionId; + expect(observed).toEqual([parentSessionId]); + + // Two child turns plus the parent turns its own terminal notices trigger. + harness.setResponses([ + fauxAssistantMessage("child answer"), + fauxAssistantMessage("child answer"), + fauxAssistantMessage("parent noted"), + fauxAssistantMessage("parent noted"), + ]); + const first = await harness.session.runRlmChild("first child task"); + const second = await harness.session.runRlmChild("second child task"); + + await vi.waitFor(() => { + expect(harness.session.getRlmChildSession(first.rlm_child_id)?.getLastAssistantText()).toBe("child answer"); + expect(harness.session.getRlmChildSession(second.rlm_child_id)?.getLastAssistantText()).toBe("child answer"); + }); + + const childSessionIds = [first, second].map( + (child) => harness.session.getRlmChildSession(child.rlm_child_id)?.sessionId, + ); + expect(childSessionIds.every((id) => typeof id === "string" && id.length > 0)).toBe(true); + expect(new Set(childSessionIds).size).toBe(2); + expect(childSessionIds).not.toContain(parentSessionId); + + // Exactly the two child requests carried a child identity; before the fix + // every request in this run interleaved on the parent's key. + const nonParentIdentities = observed.filter((identity) => identity !== parentSessionId); + expect(nonParentIdentities).toHaveLength(2); + expect(new Set(nonParentIdentities)).toEqual(new Set(childSessionIds)); + expect(sentUserIds(harness)).toEqual(observed); + }); + + it("gives a side question an identity derived from its parent instead of the parent's own", async () => { + const observed: string[] = []; + const harness = await createHarness({ extensionFactories: [stampIdentity(observed)] }); + harnesses.push(harness); + + harness.setResponses([fauxAssistantMessage("main answer")]); + await harness.session.prompt("main turn"); + const parentSessionId = harness.session.sessionId; + + harness.setResponses([fauxAssistantMessage("side answer")]); + const run = startSideQuestion(harness.session, "question-1", "A side question?", () => {}); + await run.done; + + expect(observed).toEqual([parentSessionId, `${parentSessionId}/side:question-1`]); + expect(sentUserIds(harness)).toEqual(observed); + }); + + it("keys compaction summarization instead of reaching the provider anonymously", async () => { + const observed: string[] = []; + const harness = await createHarness({ + settings: { compaction: { keepRecentTokens: 1 } }, + persistSession: true, + extensionFactories: [stampIdentity(observed)], + }); + harnesses.push(harness); + + harness.setResponses([ + fauxAssistantMessage("one response"), + fauxAssistantMessage("two response"), + fauxAssistantMessage("model-generated summary"), + fauxAssistantMessage("model-generated turn summary"), + ]); + await harness.session.prompt("one"); + await harness.session.prompt("two"); + const parentSessionId = harness.session.sessionId; + const beforeCompaction = observed.length; + + await harness.session.compact(); + + const summarizationIdentities = observed.slice(beforeCompaction); + expect(summarizationIdentities.length).toBeGreaterThan(0); + for (const identity of summarizationIdentities) { + expect(identity).toBe(`${parentSessionId}/compaction`); + } + expect(sentUserIds(harness)).toEqual(observed); + }); +}); diff --git a/packages/coding-agent/test/suite/regressions/4509-side-questions.test.ts b/packages/coding-agent/test/suite/regressions/4509-side-questions.test.ts index e10da95722..07e664a22c 100644 --- a/packages/coding-agent/test/suite/regressions/4509-side-questions.test.ts +++ b/packages/coding-agent/test/suite/regressions/4509-side-questions.test.ts @@ -51,19 +51,16 @@ describe("ENG-4509 side questions", () => { }, ]); - const run = startSideQuestion( - harness.session.agent, - "question-1", - "What is the project codename?", - (event) => { - events.push(event); - }, - ); + const run = startSideQuestion(harness.session, "question-1", "What is the project codename?", (event) => { + events.push(event); + }); await run.done; expect(events.at(-1)).toMatchObject({ status: "complete", answer: "kestrel" }); expect(observedTransport).toBe("sse"); - expect(observedSessionId).toBe("cache-session"); + // Scoped off the main session (see #23): the side conversation is a + // divergent history and must not reuse the main provider session key. + expect(observedSessionId).toBe("cache-session/side:question-1"); expect(harness.session.messages).toEqual(messagesBefore); expect(harness.sessionManager.getEntries()).toEqual(entriesBefore); } finally { @@ -97,7 +94,7 @@ describe("ENG-4509 side questions", () => { const events: SideQuestionEvent[] = []; const run = startSideQuestion( - harness.session.agent, + harness.session, "turn-2", "Second side question?", (event) => { @@ -137,14 +134,9 @@ describe("ENG-4509 side questions", () => { const mainRun = harness.session.prompt("Run the main task."); await mainStarted.promise; const events: SideQuestionEvent[] = []; - const sideRun = startSideQuestion( - harness.session.agent, - "question-2", - "Can I ask this concurrently?", - (event) => { - events.push(event); - }, - ); + const sideRun = startSideQuestion(harness.session, "question-2", "Can I ask this concurrently?", (event) => { + events.push(event); + }); await sideRun.done; expect(harness.session.isStreaming).toBe(true); @@ -172,7 +164,7 @@ describe("ENG-4509 side questions", () => { }, ]); const events: SideQuestionEvent[] = []; - const run = startSideQuestion(harness.session.agent, "question-3", "Wait here", (event) => { + const run = startSideQuestion(harness.session, "question-3", "Wait here", (event) => { events.push(event); }); await sideStarted.promise; @@ -191,7 +183,7 @@ describe("ENG-4509 side questions", () => { try { const events: SideQuestionEvent[] = []; let shouldFail = true; - const run = startSideQuestion(harness.session.agent, "question-4", "Can this recover?", (event) => { + const run = startSideQuestion(harness.session, "question-4", "Can this recover?", (event) => { if (shouldFail) { shouldFail = false; throw new Error("event delivery failed"); diff --git a/packages/coding-agent/test/suite/regressions/4620-fast-mode-child-agents.test.ts b/packages/coding-agent/test/suite/regressions/4620-fast-mode-child-agents.test.ts index 2486129a5a..60c619e3c2 100644 --- a/packages/coding-agent/test/suite/regressions/4620-fast-mode-child-agents.test.ts +++ b/packages/coding-agent/test/suite/regressions/4620-fast-mode-child-agents.test.ts @@ -78,7 +78,7 @@ describe("ENG-4620 fast mode child agents", () => { }, ]); - const run = startSideQuestion(harness.session.agent, "question-1", "Check fast mode", () => {}); + const run = startSideQuestion(harness.session, "question-1", "Check fast mode", () => {}); await run.done; } finally { harness.cleanup();