import { existsSync, mkdirSync } from "node:fs"; import { access as fsAccess, readFile as fsReadFile } from "node:fs/promises"; import { extname, isAbsolute, join, relative, resolve, sep } from "node:path"; import type { AssistantMessage, Model, TextContent } from "@earendil-works/pi-ai"; import type { PathMetadata } from "@earendil-works/pi-coding-agent"; import { type ContextUsage, type CreateAgentSessionOptions, createAgentSession, createCodingTools, DefaultResourceLoader, getAgentDir, type LoadExtensionsResult, ModelRegistry, ModelRuntime, type PromptTemplate, type ResourceDiagnostic, type ResourceLoader, SessionManager, SettingsManager, type Skill, type Theme, type ToolDefinition, } from "@earendil-works/pi-coding-agent"; import type { Static, TSchema } from "typebox"; import { Check, Convert } from "typebox/value"; import { type AgentHistoryEntry, compactAgentHistory } from "./agent-history.js"; import { applyToolPolicy } from "./agent-registry.js"; import { resolveWorkflowCompactionPolicy, type WorkflowCompactionPolicyDecision, type WorkflowCompactionPolicyName, } from "./compaction-policy.js"; import { emitCompactionTelemetry } from "./compaction-telemetry.js"; import { DEFAULT_CONTEXT_MODE, filterSkillsByName, isShareableResourceLoaderConfig, needsResourceLoader, resolveContextMode, resourceLoaderFlags, type SystemPromptMode, } from "./context-mode.js"; import { classifyProviderLimit, WorkflowError, WorkflowErrorCode } from "./errors.js"; import { type GuardCtxReadOptions, guardCtxReadPath } from "./lean-ctx-guardrail.js"; import { isApiBilledModel, isSecurityModel, loadModelTierConfig, type ModelRoleConfig, type ModelTierConfig, resolveRoleModel, resolveTierModel, SECURITY_ROLE, } from "./model-tier-config.js"; import { createStructuredOutputTool, type StructuredOutputCapture } from "./structured-output.js"; /** * Find a JSON object/array in free-form text: a fenced ```json block if present, * else the first balanced {...} or [...]. Best-effort (the schema check is the * real gate). Returns the raw JSON string, or undefined when none is found. */ function findJsonBlock(text: string): string | undefined { const fence = text.match(/```(?:json)?\s*([\s\S]*?)```/i); if (fence?.[1]) return fence[1].trim(); const start = text.search(/[{[]/); if (start === -1) return undefined; const open = text[start]; const close = open === "{" ? "}" : "]"; let depth = 0; for (let i = start; i < text.length; i++) { if (text[i] === open) depth++; else if (text[i] === close && --depth === 0) return text.slice(start, i + 1); } return undefined; } /** * Last-resort structured-output recovery: extract a JSON block from prose, coerce * it toward the schema, and accept it only if it then validates. Never fabricates * — returns undefined unless the parsed value genuinely satisfies the schema. */ export function extractValidated(text: string, schema: TSchema): T | undefined { const json = findJsonBlock(text); if (json === undefined) return undefined; let parsed: unknown; try { parsed = JSON.parse(json); } catch { return undefined; } try { const converted = Convert(schema, parsed); if (Check(schema, converted)) return converted as T; } catch { // typebox can throw on exotic schemas; treat as no match. } return undefined; } /** * The last assistant message's terminal metadata (stopReason/errorMessage). The pi * SDK does NOT throw provider usage/quota limits — it records them as an assistant * message with stopReason "error" and an errorMessage. This is the only place that * metadata is observable to the workflow layer. */ export function lastAssistantError(messages: unknown[]): { stopReason?: string; errorMessage?: string } | undefined { for (let i = messages.length - 1; i >= 0; i--) { const message = messages[i] as Partial | undefined; if (message?.role !== "assistant") continue; return { stopReason: message.stopReason, errorMessage: message.errorMessage }; } return undefined; } /** * If the subagent's turn ended in a provider usage/quota/rate-limit error, throw a * PROVIDER_USAGE_LIMIT WorkflowError carrying the real provider message + reset hint. * Gated on stopReason === "error" so a successful turn whose text merely mentions * "rate limit" is never misclassified. recoverable:false so the run checkpoints * (paused) rather than being retried into the same wall or collapsed to a silent null. */ export function throwIfProviderLimit(messages: unknown[], label?: string): void { const err = lastAssistantError(messages); if (err?.stopReason !== "error") return; const { matched, resetHint } = classifyProviderLimit(err.errorMessage); if (!matched) return; throw new WorkflowError( err.errorMessage ?? "Provider usage/quota limit reached", WorkflowErrorCode.PROVIDER_USAGE_LIMIT, { recoverable: false, agentLabel: label, resetHint }, ); } /** Minimal session surface resolveStructuredOutput needs (real session or a test double). */ export interface StructuredSession { prompt(text: string): Promise; setActiveToolsByName?(names: string[]): void; messages: unknown[]; } /** * Resolve a schema agent's result. If the tool was called, return the captured * value. Otherwise re-prompt up to maxSchemaRetries (tools restricted to * structured_output), then try strict schema-validated prose extraction, else * throw SCHEMA_NONCOMPLIANCE (non-recoverable — surfaced, never a silent null). * Module-level with an injected `lastText` so it is unit-testable. */ export async function resolveStructuredOutput( session: StructuredSession, capture: StructuredOutputCapture, schema: TSchema, options: { maxSchemaRetries?: number; signal?: AbortSignal; label?: string; checkContextCap?: (pendingPrompt?: string) => void; }, lastText: (messages: unknown[]) => string, ): Promise { options.checkContextCap?.(); if (capture.called) return capture.value as T; const maxRetries = Math.max(0, options.maxSchemaRetries ?? 2); // Restrict to the schema tool so the only useful next action is calling it // (takes effect on the next prompt turn). Best-effort. try { session.setActiveToolsByName?.(["structured_output"]); } catch { // ignore — the re-prompt alone still drives most models to comply } const requestRepair = async (attempt: number): Promise => { if (attempt >= maxRetries || capture.called) return; const repairText = "You did not call the structured_output tool. Call structured_output now as your only action, with the required fields filled in. Do not write a prose answer."; options.checkContextCap?.(repairText); if (options.signal?.aborted) throw new Error("Subagent was aborted"); await session.prompt(repairText); options.checkContextCap?.(); await requestRepair(attempt + 1); }; await requestRepair(0); if (capture.called) return capture.value as T; const extracted = extractValidated(lastText(session.messages), schema); if (extracted !== undefined) { console.warn( "[workflow] structured_output recovered from prose extraction (the model never called the tool); prefer a tool-reliable model", ); return extracted; } // A repair re-prompt can itself hit the provider limit. Surface that as the real // (recoverable) cause instead of the misleading non-recoverable SCHEMA_NONCOMPLIANCE. throwIfProviderLimit(session.messages, options.label); throw new WorkflowError( "Subagent did not produce valid structured_output after repair attempts", WorkflowErrorCode.SCHEMA_NONCOMPLIANCE, { recoverable: false, agentLabel: options.label }, ); } /** * Resolve which concrete model spec a subagent should use. Precedence, most * specific first (see issue #142): * 1. options.model — an explicit per-agent model (also carries agentType / * phase model, which the workflow layer folds into options.model). * 2. options.modelRole + optional options.modelRoute — the semantic routing * role (worker/conductor/advisor/security) resolved from the roles block * of the model routing config. A named route is an exact specialist * selection, never a fallback ladder: an unknown/absent route does NOT * fall back to the role default; it surfaces an error upstream. * 3. options.tier — DEPRECATED size tier (small/medium/big), resolved via the * legacy tiers block, falling back to the session's main model when the * tier has no configured entry. Kept for migration of saved workflows and * journals. * 4. DEFAULT ROLE — when none of the above are set but a config exists, * untagged agents default to the `conductor` role (the old `medium` tier's * successor) so a configured profile affects the whole workflow, not just * agents the script tagged. Fresh-install conductor == the session model, * so this is a no-op until the user customizes via `/workflows-models`. * Returns undefined when nothing applies, so the session default is used. * * Policy guards (issue #142): the security-only model (Fable) is rejected * outside the `security` role — after every source above, including an * explicit `model`. API-billed (google-ai-studio) models are rejected from * role defaults/routes; an exact `model` use requires `googleBillingOptIn`. * These throw rather than silently drop to the session model, because a * silent downgrade of a required security gate is unsafe. Callers that want * the legacy non-throwing behavior (e.g. old tests) can pass `{ enforcePolicy: false }`. * * `loadConfig` is injectable for testing; it defaults to reading from disk. */ export interface ResolveAgentModelSpecOptions { model?: string; /** Semantic routing role (worker/conductor/advisor/security). */ modelRole?: string; /** Named specialist route within the role (e.g. escalation, long-context). */ modelRoute?: string; /** DEPRECATED size tier (small/medium/big). Migration input only. */ tier?: string; modelTierConfig?: ModelRoleConfig | null; /** Visible operator opt-in to use an API-billed (google-ai-studio) exact model. */ googleBillingOptIn?: boolean; /** The role the caller is running under (for security-model enforcement). Defaults to modelRole. */ securityRoleContext?: string; /** When false, skip the security/billing policy guards (legacy/test path). */ enforcePolicy?: boolean; } export function resolveAgentModelSpec( options: ResolveAgentModelSpecOptions | { model?: string; tier?: string; modelTierConfig?: ModelTierConfig | null }, mainModel: string | undefined, loadConfig: () => ModelRoleConfig | null = loadModelTierConfig, ): string | undefined { const opts = options as ResolveAgentModelSpecOptions; const config = Object.hasOwn(opts, "modelTierConfig") ? (opts.modelTierConfig ?? null) : loadConfig(); const enforcePolicy = opts.enforcePolicy !== false; // 1. Explicit model (also carries agentType/phase model folded by the workflow layer). if (opts.model) { if (enforcePolicy) enforceModelPolicy(opts.model, opts.securityRoleContext ?? opts.modelRole, opts.googleBillingOptIn); return opts.model; } // 2. Semantic role + optional named route. if (opts.modelRole) { const resolved = resolveRoleModel(opts.modelRole, opts.modelRoute, config); if (resolved) { if (enforcePolicy) enforceModelPolicy(resolved, opts.securityRoleContext ?? opts.modelRole, opts.googleBillingOptIn); return resolved; } // Unknown role/route or unavailable explicit route: do NOT silently fall // back to the session model. The workflow layer surfaces this as an error; // here we return undefined only when there is genuinely no config, so a // direct agent.run() caller gets the session default (matches legacy tier // behavior for a missing config). The workflow layer distinguishes the // "requested but unavailable" case and throws. if (config) { // A role was requested and a config exists but the role/route didn't // resolve — surface undefined so the caller can decide; the workflow // layer throws an actionable error for this. return undefined; } return undefined; } // 3. DEPRECATED tier (migration compatibility). if (opts.tier) { const tierModel = (config ? resolveTierModel(opts.tier, config) : undefined) ?? mainModel; if (tierModel && enforcePolicy) { enforceModelPolicy(tierModel, opts.securityRoleContext ?? opts.modelRole, opts.googleBillingOptIn); } return tierModel; } // 4. Untagged agent: default to the conductor role (old `medium` tier successor) // when a config exists, so a configured profile affects the whole workflow. if (config) { const conductor = resolveRoleModel("conductor", undefined, config); if (conductor) { if (enforcePolicy) enforceModelPolicy(conductor, "conductor", opts.googleBillingOptIn); return conductor; } // Fall back to legacy medium tier if roles block is absent (tier-only profile). const medium = resolveTierModel("medium", config); if (medium) return medium; } return undefined; } /** * Enforce the security/billing policy on a resolved model spec. Throws an * actionable error (never a silent downgrade) when: * - the model is the security-only Fable model but the role is not `security`; * - the model is API-billed (google-ai-studio) without an explicit opt-in. */ function enforceModelPolicy( modelSpec: string, role: string | undefined, googleBillingOptIn: boolean | undefined, ): void { if (isSecurityModel(modelSpec) && role !== SECURITY_ROLE) { throw new WorkflowError( `model "${modelSpec}" is reserved for the security role; it cannot be used with role "${role ?? "(none)"}"`, WorkflowErrorCode.SCHEMA_NONCOMPLIANCE, { recoverable: false }, ); } if (isApiBilledModel(modelSpec) && !googleBillingOptIn) { throw new WorkflowError( `model "${modelSpec}" is API-billed and must never be a role default/route/fallback; exact use requires a visible googleBillingOptIn opt-in`, WorkflowErrorCode.SCHEMA_NONCOMPLIANCE, { recoverable: false }, ); } } export interface WorkflowAgentOptions { cwd?: string; /** Extra tools available to the subagent in addition to the structured output tool. */ tools?: ToolDefinition[]; /** * Base createAgentSession options. When a tool-authority fence is active, * `tools`, `customTools`, and `resourceLoader` are re-resolved after this layer * so session defaults cannot widen the fence. */ session?: Partial; /** Extra system guidance prepended to every subagent task. */ instructions?: string; /** * The session's main model (`provider/modelId`). Used as a fallback when * resolving opts.tier and no model-tiers.json config exists. Without this, * a workflow using `{ tier: "small" }` would log a warning and fall through * to the session default when no config is saved yet. */ mainModel?: string; /** * Host session ModelRegistry shared with subagents (upstream #49 port). When * set, tier/phase/model routing resolves against the same registry as the * main Pi session — including providers registered dynamically by extensions * — instead of an isolated disk registry. Absent → disk fallback. */ modelRegistry?: ModelRegistry; } /** * List the user's currently available models (those with auth configured) as * `provider/modelId` specs. Used to tell the workflow author which models it may * route agents to. Best-effort: returns [] if the registry can't be built. */ interface DiskModelContext { runtime: ModelRuntime; registry: ModelRegistry; } interface DiskModelContextState { ready?: DiskModelContext; pending?: Promise; } const diskModelContexts = new Map(); /** * Build one disk-backed runtime per Pi agent directory. Promise memoization * prevents concurrent subagents from racing duplicate runtime initialization; * rejection clears the cache so a transient startup failure remains retryable. */ function ensureDiskModelContext(agentDir = getAgentDir()): Promise { const state = diskModelContexts.get(agentDir) ?? {}; if (state.ready) return Promise.resolve(state.ready); if (state.pending) return state.pending; state.pending = ModelRuntime.create({ authPath: join(agentDir, "auth.json"), modelsPath: join(agentDir, "models.json"), }) .then((runtime) => { const ready = { runtime, registry: new ModelRegistry(runtime) }; state.ready = ready; state.pending = undefined; return ready; }) .catch((error: unknown) => { state.pending = undefined; diskModelContexts.delete(agentDir); throw error; }); diskModelContexts.set(agentDir, state); return state.pending; } function readyDiskRegistryOrStart(): ModelRegistry | undefined { try { const agentDir = getAgentDir(); const ready = diskModelContexts.get(agentDir)?.ready?.registry; if (ready) return ready; void ensureDiskModelContext(agentDir).catch(() => { // Best-effort discovery only. A real agent run retries and surfaces errors. }); } catch { // Keep the synchronous discovery API non-throwing. } return undefined; } export function listAvailableModelSpecs(registry?: ModelRegistry): string[] { try { const resolved = registry ?? readyDiskRegistryOrStart(); return resolved?.getAvailable().map((model) => `${model.provider}/${model.id}`) ?? []; } catch { return []; } } // ───────────────────────────────────────────────────────────────────────────── // #135: shared immutable resource-loader discovery cache. // // Every workflow subagent currently constructs its OWN DefaultResourceLoader and // calls `reload()`, which runs the full package-manager resolve + extension/skill/ // prompt/theme/context-file discovery. The discovered set is IDENTICAL across // subagents sharing the same `(cwd, agentDir)`; only the per-agent filters // (`noContextFiles`, `noSkills`, `systemPrompt`, `appendSystemPrompt`, // `skillsOverride`) differ, and those are pure read-time filters over one // full-discovery base. `isShareableResourceLoaderConfig` documents + tests that // contract. // // Safety: a subagent session built via `createAgentSession` (the workflow path) // only READS the loader during construction — it never calls `bindExtensions`, so // `extendResourcesFromExtensions` (the only path that mutates the loader via // `resources_discover` hooks) never runs. `ImmutableResourceLoader` therefore // serves per-agent filtered views from a shared pre-reloaded base WITHOUT // mutating it. Its `extendResources` is copy-on-write (per-instance) and its // `reload` is a no-op, so even a future caller that triggers discovery hooks or a // session reload cannot cross-contaminate the shared base. Measured (see // tests/resource-loader-retention-bench.test.ts): one shared reload collapses a // 14-subagent fan-out from ~1.7s of redundant reload work to ~9ms of snapshot // reads, with zero per-agent discovery cost. // ───────────────────────────────────────────────────────────────────────────── interface SharedDiscoveryState { ready?: DefaultResourceLoader; pending?: Promise; } /** * Structural mirror of the SDK's `ResourceExtensionPaths` (not exported from the * package index). Used only by `ImmutableResourceLoader.extendResources`. */ interface ResourceExtensionPaths { skillPaths?: Array<{ path: string; metadata: PathMetadata }>; promptPaths?: Array<{ path: string; metadata: PathMetadata }>; themePaths?: Array<{ path: string; metadata: PathMetadata }>; } const sharedDiscoveryLoaders = new Map(); /** * Get-or-start the single shared discovery loader for `(cwd, agentDir)`. The * first caller pays the one-time `reload()` (extension/skill/prompt/theme * discovery); every subsequent subagent in the fan-out reuses the pre-reloaded * snapshot. Promise-memoization prevents concurrent subagents from racing * duplicate reloads; rejection clears the cache so a transient failure stays * retryable. The returned loader is the SHARED base — callers wrap it in an * `ImmutableResourceLoader` facade before handing it to a session so per-agent * filters never mutate the base. */ function ensureSharedDiscoveryLoader(cwd: string, agentDir: string): Promise { const key = `${cwd}\0${agentDir}`; const state = sharedDiscoveryLoaders.get(key) ?? {}; if (state.ready) return Promise.resolve(state.ready); if (state.pending) return state.pending; state.pending = (async () => { // The shared discovery loader uses its OWN SettingsManager (independent of // any per-agent compaction override settings) — discovery does not depend on // per-agent compaction settings, and the session gets its own SettingsManager. const settingsManager = SettingsManager.create(cwd, agentDir); const loader = new DefaultResourceLoader({ cwd, agentDir, settingsManager }); await loader.reload(); state.ready = loader; state.pending = undefined; sharedDiscoveryLoaders.set(key, state); return loader; })().catch((error: unknown) => { state.pending = undefined; sharedDiscoveryLoaders.delete(key); throw error; }); sharedDiscoveryLoaders.set(key, state); return state.pending; } /** * Clear the shared discovery cache. Exposed for tests and for an explicit * extension-reload handoff (issue #135 scope item 4): the host calls this when * the extension set changes (e.g. after `/n` reload) so the next fan-out * rediscovers against the new set instead of replaying a stale snapshot. The * handoff is a single function call; the next `ensureSharedDiscoveryLoader` * rebuilds one active set. */ export function invalidateSharedDiscoveryLoaders(): void { sharedDiscoveryLoaders.clear(); } /** * A read-only, per-agent facade over the shared discovery loader. Applies the * context-mode + skills-allowlist filters at READ time without mutating the * shared base, so N subagents share one discovery reload. `reload()` is a no-op * (the base is already loaded; `createAgentSession` never reloads an injected * loader). `extendResources()` is a no-op with an observable warning — the * workflow subagent path never calls it (no `bindExtensions`), so this is purely * defense-in-depth against a future `resources_discover` hook. */ export class ImmutableResourceLoader implements ResourceLoader { private readonly base: ResourceLoader; private readonly noContextFiles: boolean; private readonly noSkills: boolean; private readonly systemPromptOverride: string | undefined; private readonly appendSystemPromptOverride: string[] | undefined; private readonly skillsFilter: | ((base: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }) => { skills: Skill[]; diagnostics: ResourceDiagnostic[]; }) | undefined; constructor(options: { base: ResourceLoader; noContextFiles: boolean; noSkills: boolean; systemPrompt: string | undefined; appendSystemPrompt: string[] | undefined; skillsFilter?: (base: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }) => { skills: Skill[]; diagnostics: ResourceDiagnostic[]; }; }) { this.base = options.base; this.noContextFiles = options.noContextFiles; this.noSkills = options.noSkills; this.systemPromptOverride = options.systemPrompt; this.appendSystemPromptOverride = options.appendSystemPrompt; this.skillsFilter = options.skillsFilter; } getExtensions(): LoadExtensionsResult { // Extensions are never filtered by context-mode/skills policy — the tool // authority fence (applyToolPolicy) governs which extension TOOLS are active, // not the loader. The full extension set is shared as-is. return this.base.getExtensions(); } getSkills(): { skills: Skill[]; diagnostics: ResourceDiagnostic[] } { if (this.noSkills) return { skills: [], diagnostics: [] }; const baseResult = this.base.getSkills(); if (!this.skillsFilter) return baseResult; return this.skillsFilter(baseResult); } getPrompts(): { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] } { return this.base.getPrompts(); } getThemes(): { themes: Theme[]; diagnostics: ResourceDiagnostic[] } { return this.base.getThemes(); } getAgentsFiles(): { agentsFiles: Array<{ path: string; content: string }> } { // noContextFiles drops the project AGENTS.md/context files — a read-time // filter, not a different discovery. (isolated/scoped modes use this.) if (this.noContextFiles) return { agentsFiles: [] }; return this.base.getAgentsFiles(); } getSystemPrompt(): string | undefined { // systemPromptMode "replace" installs the role prompt AS the base system // prompt; "append" leaves the discovered prompt intact. return this.systemPromptOverride ?? this.base.getSystemPrompt(); } getSystemPromptSource(): { path: string } | undefined { // When the role prompt replaces the base, there is no source file. if (this.systemPromptOverride !== undefined) return undefined; // Floor-safe: the Pi SDK floor (0.80.8) ResourceLoader has no // getSystemPromptSource; guard so this builds against both the floor and // latest (which added the method). Returns undefined when the base can't // report a source, matching the override-absent branch above. const base = this.base as { getSystemPromptSource?: () => { path: string } | undefined }; return base.getSystemPromptSource?.(); } getAppendSystemPrompt(): string[] { // inheritMainRules:false → [] (block .pi/APPEND_SYSTEM.md); undefined → base. if (this.appendSystemPromptOverride !== undefined) return this.appendSystemPromptOverride; return this.base.getAppendSystemPrompt(); } getAppendSystemPromptSources(): Array<{ path: string }> { if (this.appendSystemPromptOverride !== undefined) return []; // Floor-safe: the Pi SDK floor (0.80.8) ResourceLoader has no // getAppendSystemPromptSources; guard so this builds against both the floor // and latest (which added the method). Returns [] when the base can't report // sources, matching the override-present branch above. const base = this.base as { getAppendSystemPromptSources?: () => Array<{ path: string }> }; return base.getAppendSystemPromptSources?.() ?? []; } /** * No-op (defense-in-depth). The workflow subagent path never calls this: a * session built via `createAgentSession` does not invoke `bindExtensions`, so * `extendResourcesFromExtensions` (the only caller of `extendResources`) never * runs, and no installed extension registers a `resources_discover` hook * today. If a future extension DOES register one and the host binds extensions * into a subagent session, per-session discovery cannot be served by the shared * immutable base — emit an observable warning rather than silently dropping * the paths or silently contaminating the shared base. */ extendResources(paths: ResourceExtensionPaths): void { const total = (paths.skillPaths?.length ?? 0) + (paths.promptPaths?.length ?? 0) + (paths.themePaths?.length ?? 0); if (total > 0) { console.warn( "[workflow] ImmutableResourceLoader.extendResources: per-session resources_discover is not supported under shared discovery; paths ignored. Use invalidateSharedDiscoveryLoaders() to rebuild the shared set after an extension-set change.", ); } } /** No-op: the shared base is already loaded. `createAgentSession` never reloads * an injected loader, and re-loading the shared base would defeat the cache * and risk cross-agent mutation. A no-op keeps the facade immutable. */ async reload(): Promise { // Intentionally empty. The shared discovery loader was reloaded once when // first cached; this facade only reads it. } } /** Real token/cost usage for a single subagent run, read from the SDK session. */ export interface AgentUsage { input: number; output: number; cacheRead: number; cacheWrite: number; total: number; cost: number; } export type AgentContextWindowLevel = "ok" | "warn" | "critical" | "over" | "unknown"; export interface AgentContextWindowStats { /** Tokens currently occupying the model context, usually provider input tokens for the completed turn. */ contextTokens: number; /** Model/runtime context window, when known. */ runtimeContextWindow?: number; /** Reserved response/scratch tokens subtracted from runtimeContextWindow, when known. */ reserve?: number; /** Runtime window minus reserve. */ effectiveWindow?: number; /** contextTokens / effectiveWindow, when effectiveWindow is known. */ occupancy?: number; /** Highest threshold crossed by this measurement. */ threshold?: number; /** Human-readable severity for UI/telemetry. */ level: AgentContextWindowLevel; /** Optional hard cap supplied by workflow policy. */ maxContextTokens?: number; /** True when contextTokens exceeded maxContextTokens. */ exceededMaxContextTokens?: boolean; /** Human-readable warning for UI/log/telemetry. */ warning?: string; } export type WorkflowCtxReadGuardrailOptions = Omit; export interface AgentRunOptions { label?: string; schema?: TSchemaDef; tools?: ToolDefinition[]; instructions?: string; signal?: AbortSignal; /** * Called once with this subagent's real usage, read from the session right * before disposal. Fires on both the success and error paths so partial * usage is never lost. `total === 0` means the provider reported no usage. */ onUsage?: (usage: AgentUsage) => void; /** * Model spec for this subagent: either `provider/modelId` (unambiguous) or a * bare `modelId`. When it can't be resolved, the session default is used and * a warning is logged. When omitted, the session default applies. */ model?: string; /** * Semantic routing role (worker | conductor | advisor | security). Resolved * from the `roles` block of the model routing config (see /workflows-models). * An optional `modelRoute` selects a named specialist route within the role. * Precedence: explicit `model` > agentType model > `modelRole`(+`modelRoute`) * > deprecated `tier` > untagged conductor default > session model. */ modelRole?: string; /** Named specialist route within `modelRole` (e.g. escalation, long-context). */ modelRoute?: string; /** * Visible operator opt-in to use an API-billed (google-ai-studio) exact * `model`. Required when `model` is a google-ai-studio spec; otherwise the * call is rejected (API-billed models are never role defaults/routes). */ googleBillingOptIn?: boolean; /** * DEPRECATED size tier ("small" | "medium" | "big"), resolved from the legacy * `tiers` block of the model routing config (see /workflows-models). Kept only * as a migration input for existing saved workflows and journals; new authoring * guidance uses `modelRole`. An explicit `model` always takes priority. * @deprecated Use `modelRole` (small→worker, medium→conductor, big→advisor). */ tier?: string; /** * Model routing config snapshotted by runWorkflow. When present (including * null), run() must not re-read model-tiers.json mid-run. */ modelTierConfig?: ModelRoleConfig | null; /** * Per-run ModelRegistry override (wins over the agent's shared registry and * the disk fallback). Lets a host thread its live registry into a single run. */ modelRegistry?: ModelRegistry; /** Called with the resolved model id once known (for display/telemetry). */ onModelResolved?: (modelId: string) => void; /** Called when `model`/`tier`/phase resolved to a spec that wasn't found (fell back to session default). */ onModelFallback?: (requestedSpec: string) => void; /** Called with a compact snapshot of this subagent's message/tool history. */ onHistory?: (history: AgentHistoryEntry[]) => void; /** Called with model-window occupancy stats for this subagent, when measurable. */ onContextWindow?: (stats: AgentContextWindowStats) => void; /** Per-subagent compaction policy. "auto" makes local/no-cache models compact earlier. */ compactionPolicy?: WorkflowCompactionPolicyName | null; /** Workflow run id/phase used to scope compaction telemetry emitted by this subagent. */ workflowRunId?: string; phase?: string; /** Hard cap for provider input/context tokens for this subagent. */ maxContextTokens?: number; /** Override the model output/reserve tokens used to compute effective context window. */ contextReserveTokens?: number; /** Run this agent in a different working directory (e.g. an isolated worktree). */ cwd?: string; /** * Directory to persist this subagent's NDJSON transcript into. When set, * a real (file-backed) SessionManager is used so the full subagent message * stream survives session disposal — matching Claude Code's per-subagent * `agent-.jsonl` transcript. When omitted, an in-memory session is used * (ad-hoc `agent()` with no run context) and nothing is written to disk. */ transcriptDir?: string; /** * Restrict the subagent's coding tools to these names (an agentType * definition's `tools` allowlist). Undefined = all coding tools. The * structured_output tool is always added after this filter, so a schema * still works under a restrictive allowlist. */ toolNames?: string[]; /** Remove these coding-tool names after the allowlist (an agentType `disallowedTools` denylist). */ disallowedToolNames?: string[]; /** Optional read-path guardrail options supplied by a harness_config expansion. */ ctxReadGuardrail?: WorkflowCtxReadGuardrailOptions; /** * With `schema`: how many extra repair turns to allow if the model finishes * without calling structured_output. Each retry re-prompts (tools restricted to * structured_output) before falling back to strict prose extraction. Default 2. */ maxSchemaRetries?: number; /** * Context-inheritance posture for this subagent (expands to the three * primitives below). When omitted, the explicit fields — else `inherit` — * apply. See context-mode.ts. Default `inherit` == today's behavior. */ contextMode?: string; /** Load project AGENTS.md / context files into the subagent session. Default true. */ inheritProjectContext?: boolean; /** "append": base prompt intact, role-as-task (default); "replace": role IS the base system prompt. */ systemPromptMode?: SystemPromptMode; /** Load skills into the subagent session. Default true. */ inheritSkills?: boolean; /** * Per-agent **skills allowlist**: when set, the subagent loads ONLY the named * skills (matched by skill `name`), regardless of `inheritSkills`/`contextMode` * skill posture. An empty array is a fence that yields ZERO skills (equivalent * to `inheritSkills:false`); `undefined` preserves today's behavior (the * resolved context posture decides whether skills load at all). * * Precedence: `skills` wins over `inheritSkills`/`contextMode` for the skills * channel — when set, a custom resource loader is always constructed (even * under `legacy`) and a `skillsOverride` filter keeps only the named skills. * Names that match no discovered skill warn (console) and are skipped; the run * never fails on an unknown name. */ skills?: string[]; /** * Inherit the main-agent append channel (`.pi/APPEND_SYSTEM.md`) into this * subagent. Default false: the main session's orchestration-only rules do not * leak into subagents (OpenCode-style). Set true (or use the `legacy` mode) to * restore the pre-feature behavior where subagents inherited them. */ inheritMainRules?: boolean; /** * The agentType role prompt to install AS the system prompt when the resolved * `systemPromptMode` is "replace". The workflow layer passes the agent `.md` * body here (and omits it from the task to avoid duplication). Ignored unless * the resolved mode is "replace". */ systemPromptText?: string; /** * Read-only fence: when true, the subagent never receives write tools * (edit, bash, write). The fence is the last filter step so it cannot be * bypassed by an allowlist from `harness_config` or `agentType`. */ readOnly?: boolean; /** * TEST-ONLY: override the live tool registry/active set seen by the * pre-dispatch tool assertion. This lets the test inject the exact * broken-harness divergence (missing or inactive authorized tool) that * the guard exists to catch, without depending on an SDK regression. * Production code never sets this. * * @internal Test-only seam — stripped from emitted declarations via * tsconfig `stripInternal`. It is NOT part of the public `AgentRunOptions` * surface and is not governed by semver. The compiled `.js` still reads it * (runtime behavior is unchanged); only the public type surface hides it. * `src/` ships in `package.json` `files[]`, so it remains readable in source * to a determined reader — this hides it from the API type surface, not from * source inspection. See issue #147. */ __testToolRegistryOverride?: { liveToolNames?: string[]; activeToolNames?: string[]; }; } export type AgentRunResult = TSchemaDef extends TSchema ? Static : string; export function createGuardedReadOperations(cwd: string, guardrail: WorkflowCtxReadGuardrailOptions) { const remappedPaths = new Map(); const remap = (absolutePath: string): string => remappedPaths.get(absolutePath) ?? resolveGuardedReadPath(cwd, absolutePath, guardrail); return { async access(absolutePath: string): Promise { const guardedPath = resolveGuardedReadPath(cwd, absolutePath, guardrail); await fsAccess(guardedPath); if (guardedPath === absolutePath) { remappedPaths.delete(absolutePath); } else { remappedPaths.set(absolutePath, guardedPath); } }, async readFile(absolutePath: string): Promise { return await fsReadFile(remap(absolutePath)); }, async detectImageMimeType(absolutePath: string): Promise { return imageMimeType(remap(absolutePath)); }, }; } export function applyCtxReadGuardrailToTools( baseTools: ToolDefinition[], cwd: string, guardrail: WorkflowCtxReadGuardrailOptions, ): ToolDefinition[] { const guardedReadTool = createCodingTools(cwd, { read: { operations: createGuardedReadOperations(cwd, guardrail) }, }).find((tool) => tool.name === "read"); if (!guardedReadTool) return baseTools; return baseTools.map((tool) => (tool.name === "read" ? guardedReadTool : tool)); } function resolveGuardedReadPath(cwd: string, absolutePath: string, guardrail: WorkflowCtxReadGuardrailOptions): string { const normalizedPath = relativeToCwd(cwd, absolutePath); if (!normalizedPath) throw new Error(`Path escapes the repository: ${absolutePath}`); const outcome = guardCtxReadPath(normalizedPath, { cwd, ...guardrail }); if (outcome.ok && outcome.normalizedPath) return resolve(cwd, outcome.normalizedPath); throw new Error([outcome.reason, outcome.fallbackHint].filter(Boolean).join(" ")); } function relativeToCwd(cwd: string, absolutePath: string): string | undefined { const normalized = relative(cwd, absolutePath).split(sep).join("/") || "."; if (normalized === ".." || normalized.startsWith("../") || isAbsolute(normalized)) return undefined; return normalized; } function imageMimeType(path: string): string | null { switch (extname(path).toLowerCase()) { case ".jpg": case ".jpeg": return "image/jpeg"; case ".png": return "image/png"; case ".gif": return "image/gif"; case ".webp": return "image/webp"; default: return null; } } function hasCtxReadGuardrailOptions( value: WorkflowCtxReadGuardrailOptions | undefined, ): value is WorkflowCtxReadGuardrailOptions { return value !== undefined && Object.values(value).some((entry) => entry !== undefined); } export function buildAgentContextWindowStats( usage: Pick, options: { runtimeContextWindow?: number; reserve?: number; maxContextTokens?: number } = {}, ): AgentContextWindowStats { const contextTokens = usage.input > 0 ? usage.input : usage.total; const runtimeContextWindow = positiveIntegerField(options.runtimeContextWindow); const reserve = positiveIntegerField(options.reserve); const effectiveWindow = runtimeContextWindow !== undefined ? Math.max(1, runtimeContextWindow - (reserve ?? 0)) : undefined; const occupancy = effectiveWindow !== undefined ? contextTokens / effectiveWindow : undefined; const threshold = occupancy === undefined ? undefined : occupancy >= 0.95 ? 0.95 : occupancy >= 0.85 ? 0.85 : occupancy >= 0.7 ? 0.7 : undefined; const exceededMaxContextTokens = options.maxContextTokens !== undefined && contextTokens > options.maxContextTokens; const level: AgentContextWindowLevel = exceededMaxContextTokens ? "over" : occupancy === undefined ? "unknown" : occupancy >= 1 ? "over" : occupancy >= 0.95 ? "critical" : occupancy >= 0.7 ? "warn" : "ok"; const warning = buildContextWindowWarning({ contextTokens, effectiveWindow, occupancy, threshold, maxContextTokens: options.maxContextTokens, exceededMaxContextTokens, }); return { contextTokens, runtimeContextWindow, reserve, effectiveWindow, occupancy, threshold, level, maxContextTokens: options.maxContextTokens, exceededMaxContextTokens, warning, }; } export function buildContextWindowStatsForSession( usage: Pick, contextUsage: ContextUsage | undefined, options: { runtimeContextWindow?: number; reserve?: number; maxContextTokens?: number } = {}, ): AgentContextWindowStats { const currentContextTokens = positiveIntegerField(contextUsage?.tokens); const contextTokens = currentContextTokens ?? (usage.input > 0 ? usage.input : usage.total); return buildAgentContextWindowStats( { input: contextTokens, total: contextTokens }, { runtimeContextWindow: positiveIntegerField(contextUsage?.contextWindow) ?? options.runtimeContextWindow, reserve: options.reserve, maxContextTokens: options.maxContextTokens, }, ); } function buildContextWindowWarning(input: { contextTokens: number; effectiveWindow?: number; occupancy?: number; threshold?: number; maxContextTokens?: number; exceededMaxContextTokens?: boolean; }): string | undefined { if (input.exceededMaxContextTokens && input.maxContextTokens !== undefined) { return `context tokens ${input.contextTokens.toLocaleString()} exceeded configured cap ${input.maxContextTokens.toLocaleString()}`; } if (input.occupancy === undefined || input.threshold === undefined) return undefined; const pct = Math.round(input.occupancy * 100); const effective = input.effectiveWindow ? `/${input.effectiveWindow.toLocaleString()}` : ""; return `context window ${pct}% used (${input.contextTokens.toLocaleString()}${effective} tokens)`; } function positiveIntegerField(value: unknown): number | undefined { return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.floor(value) : undefined; } function emitCompactionPolicyTelemetry( decision: WorkflowCompactionPolicyDecision, input: { label?: string; phase?: string; workflowRunId?: string; modelSpec?: string; model?: Partial>; baseSettings: { reserveTokens: number; keepRecentTokens: number }; }, ): void { const configuredWindow = positiveIntegerField(input.model?.contextWindow); const reserve = decision.settings?.reserveTokens ?? input.baseSettings.reserveTokens; emitCompactionTelemetry({ type: "workflow_compaction_policy", workflowRunId: input.workflowRunId, phase: input.phase, trigger: "agent_start", configuredWindow, runtimeContextWindow: configuredWindow, reserve, effectiveWindow: configuredWindow ? Math.max(1, configuredWindow - reserve) : undefined, compactionKeepRecentTokens: decision.settings?.keepRecentTokens ?? input.baseSettings.keepRecentTokens, compactionPolicy: decision.policy, compactionPolicyReason: decision.reason, compactionCacheValue: decision.cacheValue, compactor: "pi-sdk-auto", suppressedByCacheHot: false, source: [input.model?.provider, input.model?.id].filter(Boolean).join("/") || input.modelSpec || undefined, }); } export class WorkflowAgent { private readonly cwd: string; private readonly baseTools: ToolDefinition[]; /** Whether the default tool universe should include resource-loader extension tools. */ private readonly includeResourceLoaderTools: boolean; private readonly sessionOptions: Partial; private readonly instructions?: string; private readonly mainModel?: string; /** Host session registry shared across all runs of this agent (upstream #49 port). */ private readonly sharedRegistry?: ModelRegistry; /** Registry facade for an explicitly supplied session runtime. */ private sessionRegistry?: ModelRegistry; constructor(options: WorkflowAgentOptions = {}) { this.cwd = options.cwd ?? process.cwd(); this.baseTools = options.tools ?? createCodingTools(this.cwd); // An explicit run-level tool set is an authority fence (saved read-only // review workflows depend on this). The default coding-tool path also // inherits installed extension tools discovered by Pi's resource loader. this.includeResourceLoaderTools = options.tools === undefined; this.sessionOptions = options.session ?? {}; this.instructions = options.instructions; this.mainModel = options.mainModel; this.sharedRegistry = options.modelRegistry; } /** * Extract the ModelRuntime from a ModelRegistry. * The runtime field is private in the type declaration, but the JS runtime * exposes it as a plain property. Isolated in one narrow helper so the * structural access is contained and reviewed. */ private static extractRuntime(registry: ModelRegistry): ModelRuntime { const runtime = (registry as unknown as { runtime?: unknown }).runtime; if ( !runtime || typeof runtime !== "object" || typeof (runtime as ModelRuntime).getModel !== "function" || typeof (runtime as ModelRuntime).complete !== "function" ) { throw new Error( "The host Pi ModelRegistry does not expose its ModelRuntime; refusing to create a provider-incomplete subagent session", ); } return runtime as ModelRuntime; } /** * Resolve the registry used for model lookup. Precedence: per-run registry, * shared host registry, explicit session runtime, then disk fallback. */ private async getRegistry(perRun?: ModelRegistry): Promise { if (perRun) return perRun; if (this.sharedRegistry) return this.sharedRegistry; if (this.sessionOptions.modelRuntime) { this.sessionRegistry ??= new ModelRegistry(this.sessionOptions.modelRuntime); return this.sessionRegistry; } return (await ensureDiskModelContext()).registry; } /** * Resolve the runtime passed to createAgentSession with the same precedence as * model lookup. Sharing the host runtime preserves dynamically registered * providers; a missing future SDK bridge fails clearly in extractRuntime(). */ private async getRuntime(perRun?: ModelRegistry): Promise { if (perRun) return WorkflowAgent.extractRuntime(perRun); if (this.sharedRegistry) return WorkflowAgent.extractRuntime(this.sharedRegistry); if (this.sessionOptions.modelRuntime) return this.sessionOptions.modelRuntime; return (await ensureDiskModelContext()).runtime; } /** * Resolve a model spec to a Model. Accepts `provider/modelId` (unambiguous) * or a bare `modelId` (prefers auth-configured models, then any known model). * Returns undefined when nothing matches. */ private async resolveModel(spec: string, perRun?: ModelRegistry): Promise | undefined> { const registry = await this.getRegistry(perRun); const slash = spec.indexOf("/"); if (slash > 0) { return registry.find(spec.slice(0, slash), spec.slice(slash + 1)); } return registry.getAvailable().find((m) => m.id === spec) ?? registry.getAll().find((m) => m.id === spec); } private async resolveSettingsDefaultModel( settingsManager: SettingsManager, perRun?: ModelRegistry, ): Promise | undefined> { const provider = settingsManager.getDefaultProvider(); const modelId = settingsManager.getDefaultModel(); if (provider && modelId) return (await this.getRegistry(perRun)).find(provider, modelId); return modelId ? await this.resolveModel(modelId, perRun) : undefined; } async run( prompt: string, options: AgentRunOptions = {}, ): Promise> { const capture: StructuredOutputCapture = { called: false, value: undefined }; const toolAuthorityPolicyActive = !this.includeResourceLoaderTools || options.toolNames !== undefined || Boolean(options.disallowedToolNames?.length) || options.readOnly === true || options.schema !== undefined || hasCtxReadGuardrailOptions(options.ctxReadGuardrail); // Per-call cwd (e.g. a worktree) needs coding tools bound to that directory, // since tools capture their cwd at construction and can't be relocated. const runCwd = options.cwd ?? this.cwd; const baseToolsForCwd = runCwd === this.cwd ? this.baseTools : createCodingTools(runCwd); const baseTools = hasCtxReadGuardrailOptions(options.ctxReadGuardrail) ? applyCtxReadGuardrailToTools(baseToolsForCwd, runCwd, options.ctxReadGuardrail) : baseToolsForCwd; // Apply the agentType tool policy BEFORE adding structured_output, so a // restrictive allowlist never strips the schema tool. const customTools: ToolDefinition[] = applyToolPolicy( [...baseTools, ...(options.tools ?? [])], options.toolNames, options.disallowedToolNames, { readOnly: options.readOnly }, ); if (options.schema) { customTools.push(createStructuredOutputTool({ schema: options.schema, capture }) as unknown as ToolDefinition); } // Resolve the model spec (explicit model > tier > session default). This // composes with phase-based routing in workflow.ts, which only supplies // options.model when a phase pattern matches — so an explicit model wins. const modelSpec = resolveAgentModelSpec(options, this.mainModel); // Resolve a requested model spec to a Model object. A given-but-unresolved // spec falls back to the session default (with a warning) rather than failing. let resolvedModel: Model | undefined; if (modelSpec) { resolvedModel = await this.resolveModel(modelSpec, options.modelRegistry); if (resolvedModel) { options.onModelResolved?.(`${resolvedModel.provider}/${resolvedModel.id}`); } else { console.warn(`[workflow] model "${modelSpec}" not found; using session default`); options.onModelFallback?.(modelSpec); } } const agentDir = getAgentDir(); // Single SettingsManager shared by the session and (when built) the loader, so // the subagent inherits the user's default provider/model exactly as today. const settingsManager = SettingsManager.create(this.cwd, agentDir); const settingsDefaultModel = await this.resolveSettingsDefaultModel(settingsManager, options.modelRegistry); const mainModelFallback = modelSpec === undefined && this.mainModel ? await this.resolveModel(this.mainModel, options.modelRegistry) : undefined; const activeModel = resolvedModel ?? (this.sessionOptions.model as Partial> | undefined) ?? mainModelFallback ?? settingsDefaultModel; const baseCompactionSettings = settingsManager.getCompactionSettings(); const resolvedModelSpec = resolvedModel ? modelSpec : undefined; const compactionPolicy = resolveWorkflowCompactionPolicy({ requested: options.compactionPolicy, modelSpec: resolvedModelSpec, model: activeModel, contextWindow: positiveIntegerField(activeModel?.contextWindow), }); if (compactionPolicy.settings) { settingsManager.applyOverrides({ compaction: compactionPolicy.settings }); } emitCompactionPolicyTelemetry(compactionPolicy, { label: options.label, phase: options.phase, workflowRunId: options.workflowRunId, modelSpec: resolvedModelSpec, model: activeModel, baseSettings: baseCompactionSettings, }); // Resolve the context-inheritance posture (run options are the runtime layer; // any frontmatter layer was already folded in by the workflow layer, which // passes explicit primitives that win over a mode). `inherit` (the default) // resolves to needsResourceLoader === false, so the block below is skipped and // the session is constructed exactly as before — the backward-compat gate. const { primitives: ctx, unknownMode } = resolveContextMode(undefined, { contextMode: options.contextMode, inheritProjectContext: options.inheritProjectContext, systemPromptMode: options.systemPromptMode, inheritSkills: options.inheritSkills, inheritMainRules: options.inheritMainRules, }); if (unknownMode) { console.warn(`[workflow] unknown contextMode "${unknownMode}"; using "${DEFAULT_CONTEXT_MODE}"`); } let resourceLoader: ResourceLoader | undefined; // A per-agent skills allowlist forces a custom loader even under `legacy` // (where needsResourceLoader is false): the full skill set is discovered with // noSkills:false, then skillsOverride filters it down to the named skills. // An empty allowlist is a fence → zero skills (noSkills:true), mirroring // applyToolPolicy's empty-allowlist semantics. const skillsAllowlist = Array.isArray(options.skills) ? options.skills : undefined; const skillsAllowlistActive = skillsAllowlist !== undefined; if (skillsAllowlistActive || needsResourceLoader(ctx)) { // #135: serve this subagent from the SHARED immutable discovery loader. The // full extension/skill/prompt/theme/context-file set is discovered ONCE per // (cwd, agentDir) and reused across the whole fan-out; per-agent filters are // applied at read-time by ImmutableResourceLoader without mutating the // shared base. `isShareableResourceLoaderConfig` documents + tests that // every flag below is a pure read-time filter over one full-discovery base. // If a future flag breaks that property, it fails the check and we fall back // to a per-agent loader (no cross-agent mutation). const flags = resourceLoaderFlags(ctx, options.systemPromptText); const allowlist = skillsAllowlist ?? []; const noSkills = skillsAllowlistActive ? allowlist.length === 0 : flags.noSkills; const skillsFilter = skillsAllowlistActive ? (base: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }) => { const filtered = filterSkillsByName(base.skills, allowlist); for (const name of filtered.unknown) { console.warn(`[workflow] skills allowlist: no skill named "${name}" (skipped)`); } return { skills: filtered.skills, diagnostics: base.diagnostics }; } : undefined; if (isShareableResourceLoaderConfig(ctx, skillsAllowlistActive)) { const sharedBase = await ensureSharedDiscoveryLoader(runCwd, agentDir); resourceLoader = new ImmutableResourceLoader({ base: sharedBase, noContextFiles: flags.noContextFiles, noSkills, systemPrompt: flags.systemPrompt, appendSystemPrompt: flags.appendSystemPrompt, ...(skillsFilter ? { skillsFilter } : {}), }); } else { // Conservative fallback: a non-shareable configuration (none today, but // the check guards against a future flag that needs a different // discovery set). Build a per-agent loader as before — no sharing. const loader = new DefaultResourceLoader({ cwd: runCwd, agentDir, settingsManager, noContextFiles: flags.noContextFiles, noSkills, systemPrompt: flags.systemPrompt, appendSystemPrompt: flags.appendSystemPrompt, ...(skillsFilter ? { skillsOverride: skillsFilter as (base: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }) => { skills: Skill[]; diagnostics: ResourceDiagnostic[]; }, } : {}), }); await loader.reload(); resourceLoader = loader; } } // Pi's session-level `tools` allowlist governs built-ins, SDK custom tools, // AND tools registered by resource-loader extensions. When the caller did // not supply an explicit run-level tool universe, preserve those installed // extension tools while applying the same per-agent allow/deny/readOnly // policy. An explicit run-level tool set remains a closed authority fence. let sessionResourceLoader = this.sessionOptions.resourceLoader ?? resourceLoader; if (toolAuthorityPolicyActive && this.includeResourceLoaderTools && !sessionResourceLoader) { // #135: the tool-authority inspection loader enumerates extension tool names // from the SHARED discovery base (read-only). It is never mutated by the // session (the session uses `sessionResourceLoader` for its own reads), so // sharing is unconditionally safe here and avoids a redundant reload per // subagent. const sharedBase = await ensureSharedDiscoveryLoader(runCwd, agentDir); sessionResourceLoader = new ImmutableResourceLoader({ base: sharedBase, noContextFiles: false, noSkills: false, systemPrompt: undefined, appendSystemPrompt: undefined, }); } const resourceToolCandidates = toolAuthorityPolicyActive && this.includeResourceLoaderTools ? (sessionResourceLoader ?.getExtensions() .extensions.flatMap((extension) => [...extension.tools.keys()].map((name) => ({ name }))) ?? []) : []; const authorizedResourceToolNames = applyToolPolicy( resourceToolCandidates, options.toolNames, options.disallowedToolNames, { readOnly: options.readOnly }, ).map((tool) => tool.name); const resolvedToolNames = [...new Set([...customTools.map((tool) => tool.name), ...authorizedResourceToolNames])]; // Persist the subagent's full message stream to disk when a transcript dir is // provided (workflow runs), so a failed run is debuggable — matching Claude // Code's per-subagent `agent-.jsonl` transcript. Ad-hoc `agent()` with no // run context keeps the in-memory session so nothing is written to disk. let sessionManager: SessionManager; if (options.transcriptDir) { try { if (!existsSync(options.transcriptDir)) mkdirSync(options.transcriptDir, { recursive: true }); } catch { // Best-effort: SessionManager.create will also mkdirSync. Never let a // transient FS failure downgrade a run to in-memory silently. } sessionManager = SessionManager.create(runCwd, options.transcriptDir); } else { sessionManager = SessionManager.inMemory(); } const { session } = await createAgentSession({ cwd: runCwd, agentDir, sessionManager, // Use real SettingsManager to inherit user's default provider/model settings. // SettingsManager.inMemory() doesn't load ~/.pi/settings.json, so subagents // would fall back to the first available model (e.g. openai-codex) which may // not have valid auth, causing silent empty responses. settingsManager, customTools, // Preserve historical unrestricted-session precedence. A real authority // fence below reasserts the inspected definitions/loader/allowlist after it. ...(resourceLoader ? { resourceLoader } : {}), ...this.sessionOptions, ...(toolAuthorityPolicyActive ? { // Otherwise session.customTools could replace the filtered definitions. customTools, // Keep the inspected loader and session loader identical so a second, // uninspected loader cannot introduce tools outside the authority set. ...(sessionResourceLoader ? { resourceLoader: sessionResourceLoader } : {}), // An empty array is deny-all across built-in, custom, and extension tools. tools: resolvedToolNames, } : {}), // Per-call model wins over any sessionOptions.model. ...(resolvedModel ? { model: resolvedModel } : {}), // Share the host runtime with the subagent session so it resolves models // against the same providers as the main session (per-run > shared). modelRuntime: await this.getRuntime(options.modelRegistry), }); let removeAbortListener: (() => void) | undefined; let removeSessionListener: (() => void) | undefined; let lastHistoryEmit = 0; let usageEmitted = false; const emitHistory = () => options.onHistory?.(compactAgentHistory(session.messages)); const readUsageAndContext = (): { usage: AgentUsage; contextWindow: AgentContextWindowStats } | undefined => { try { const { tokens, cost } = session.getSessionStats(); const usage: AgentUsage = { input: tokens.input, output: tokens.output, cacheRead: tokens.cacheRead, cacheWrite: tokens.cacheWrite, total: tokens.total, cost, }; const contextUsage = session.getContextUsage(); return { usage, contextWindow: buildContextWindowStatsForSession(usage, contextUsage, { runtimeContextWindow: positiveIntegerField(activeModel?.contextWindow), reserve: positiveIntegerField(options.contextReserveTokens) ?? positiveIntegerField(activeModel?.maxTokens), maxContextTokens: positiveIntegerField(options.maxContextTokens), }), }; } catch { // Usage/context stats are best-effort; never let stats failure mask the real result/error. return undefined; } }; const throwIfContextCapExceeded = (contextWindow: AgentContextWindowStats | undefined): void => { if (!contextWindow?.exceededMaxContextTokens) return; throw new WorkflowError( contextWindow.warning ?? "Subagent exceeded maxContextTokens", WorkflowErrorCode.CONTEXT_WINDOW_EXCEEDED, { recoverable: false, agentLabel: options.label, details: contextWindow }, ); }; /** * Estimate the token cost of a pending prompt that has not yet been appended to * the session context. Uses the same 4-chars/token heuristic as the faux * provider (pi-ai/providers/faux `estimateTokens`) and workflow.ts estimateTokens, * so the cap check sees the projected post-prompt context consistently. */ const estimatePromptTokens = (text: string): number => Math.ceil((text ?? "").length / 4); /** * Pre-request hard cap (#143). `maxContextTokens` is documented as a hard * workflow-subagent context cap, but the SDK enforces it only after a turn * completes. A tool-using agent can therefore dispatch additional provider * requests after crossing the cap. The SDK emits `turn_start` (awaited by the * agent loop) before every provider request in the tool loop — turn 1's * `turn_start` is skipped by the loop, so the pre-`session.prompt()` check * below covers the first request and this listener covers continuations. * `getContextUsage()` returns a pre-request estimate; aborting the session * stops the in-flight run before `streamAssistantResponse` calls the provider. */ const preRequestContextCapError = (pendingPromptText?: string): WorkflowError | undefined => { if (options.maxContextTokens === undefined) return undefined; const current = readUsageAndContext(); if (!current) return undefined; // Project the context to include the pending prompt that session.prompt() // has not yet appended (the SDK loop skips `turn_start` on the first turn, // so the pre-request check must account for the prompt it is about to send). // When pendingPromptText is omitted, projected == current.contextTokens, // preserving the existing no-pending behavior (still blocks when the // current context already exceeds the cap). const projected = current.contextWindow.contextTokens + (pendingPromptText ? estimatePromptTokens(pendingPromptText) : 0); if (projected <= options.maxContextTokens) return undefined; return new WorkflowError( current.contextWindow.warning ?? `Subagent would exceed maxContextTokens after pending prompt (projected ${projected.toLocaleString()} > cap ${options.maxContextTokens.toLocaleString()})`, WorkflowErrorCode.CONTEXT_WINDOW_EXCEEDED, { recoverable: false, agentLabel: options.label, details: current.contextWindow }, ); }; const enforcePreRequestContextCap = (pendingPromptText?: string): void => { const capError = preRequestContextCapError(pendingPromptText); if (!capError) return; // Abort the in-flight run so it settles cleanly, then surface the cap failure. // Swallow the abort's own rejection (waitForIdle) — the cap error is the cause. void session.abort().catch(() => {}); throw capError; }; const emitUsageAndContext = (enforceCap: boolean): void => { if (usageEmitted) return; usageEmitted = true; if (!options.onUsage && !options.onContextWindow && options.maxContextTokens === undefined) return; const current = readUsageAndContext(); if (!current) return; try { options.onUsage?.(current.usage); } catch { // Usage hooks are diagnostic only; cap enforcement must still run. } try { options.onContextWindow?.(current.contextWindow); } catch { // Context hooks are diagnostic only; cap enforcement must still run. } if (enforceCap) throwIfContextCapExceeded(current.contextWindow); }; const maybeEmitHistory = () => { if (!options.onHistory) return; const now = Date.now(); if (now - lastHistoryEmit < 250) return; lastHistoryEmit = now; emitHistory(); }; try { if (options.signal?.aborted) throw new Error("Subagent was aborted"); // #140: defense-in-depth pre-dispatch tool snapshot. Verify every name in // resolvedToolNames exists in session.getAllTools() and appears in // session.getActiveToolNames() at session-construction time, before // session.prompt() is called. This runs INSIDE the try/finally below so a // rejection disposes the session — no SDK session/extension leak. // // This is DEFENSE-IN-DEPTH, not a fail-closed guarantee: the SDK's // `before_agent_start` resource-loader hooks run INSIDE session.prompt() // at the true pre-provider boundary, i.e. AFTER this snapshot is taken. // A hook calling pi.setActiveTools()/setActiveToolsByName() can still // remove an authorized tool after this check passes and this guard will // not catch it. The SDK exposes no caller-usable pre-provider hook to // validate at that boundary (PromptOptions.preflightResult is an // after-the-fact boolean observer for RPC mode, not a veto), so closing // that window requires an upstream Pi SDK change — see issue #146. The // check only applies when an authority policy is active // (resolvedToolNames is the authority set); unrestricted sessions keep // their default tool set. if (toolAuthorityPolicyActive && resolvedToolNames.length > 0) { // The live registry/active set is the oracle. A test-only override // (AgentRunOptions.__testToolRegistryOverride) can simulate a broken // harness where an authorized tool is missing/inactive so the guard's // rejection path is verifiable; production never sets it and the real // session-inspection branches below are byte-identical to the override. const override = options.__testToolRegistryOverride; const liveTools = override ? new Set(override.liveToolNames ?? []) : new Set(session.getAllTools().map((t) => t.name)); const activeTools = override ? new Set(override.activeToolNames ?? []) : new Set(session.getActiveToolNames()); const missing = resolvedToolNames.filter((name) => !liveTools.has(name)); const inactive = resolvedToolNames.filter((name) => liveTools.has(name) && !activeTools.has(name)); if (missing.length > 0 || inactive.length > 0) { const detail = [ missing.length ? `missing from registry: ${missing.join(", ")}` : "", inactive.length ? `present but inactive: ${inactive.join(", ")}` : "", ] .filter(Boolean) .join("; "); throw new WorkflowError( `Subagent "${options.label ?? "agent"}" authorized tools not available before dispatch (${detail}); refusing to invoke the model without the required tools`, WorkflowErrorCode.HARNESS_NOT_WIRED, { recoverable: false, agentLabel: options.label }, ); } } if (options.signal) { const onAbort = () => void session.abort(); options.signal.addEventListener("abort", onAbort, { once: true }); removeAbortListener = () => options.signal?.removeEventListener("abort", onAbort); } removeSessionListener = session.subscribe((event: { type: string; [key: string]: unknown }) => { maybeEmitHistory(); // Pre-request hard cap (#143): `turn_start` fires (and is awaited by the // agent loop) before each provider request in the tool loop. The first // turn's `turn_start` is skipped by the SDK loop, so the pre-prompt check // below covers turn 1; this listener covers tool-loop continuations. if (event.type === "turn_start") { enforcePreRequestContextCap(); } if (event.type === "compaction_start") { emitCompactionTelemetry({ type: "precompact", workflowRunId: options.workflowRunId, phase: options.phase, trigger: event.reason as string | undefined, configuredWindow: positiveIntegerField(activeModel?.contextWindow), reserve: compactionPolicy.settings?.reserveTokens ?? baseCompactionSettings.reserveTokens, compactionKeepRecentTokens: compactionPolicy.settings?.keepRecentTokens ?? baseCompactionSettings.keepRecentTokens, compactionPolicy: compactionPolicy.policy, compactionPolicyReason: compactionPolicy.reason, compactionCacheValue: compactionPolicy.cacheValue, recommended: compactionPolicy.policy === "aggressive-local", suppressedByCacheHot: false, compactor: "pi-sdk-auto", }); } else if (event.type === "compaction_end") { const result = event.result as { tokensBefore?: number } | undefined; emitCompactionTelemetry({ type: "compaction_result", workflowRunId: options.workflowRunId, phase: options.phase, trigger: event.reason as string | undefined, beforeTokens: result?.tokensBefore, compactionPolicy: compactionPolicy.policy, compactionPolicyReason: compactionPolicy.reason, compactionCacheValue: compactionPolicy.cacheValue, compactor: "pi-sdk-auto", error: typeof event.errorMessage === "string" ? event.errorMessage : undefined, }); } }); // Pre-request hard cap (#143): the SDK loop skips `turn_start` on the first // turn, so enforce the cap before the first provider request here — including // the pending prompt itself, which session.prompt() has not appended yet. const builtPrompt = this.buildPrompt(prompt, options as AgentRunOptions, Boolean(options.schema)); enforcePreRequestContextCap(builtPrompt); await session.prompt(builtPrompt); if (options.signal?.aborted) throw new Error("Subagent was aborted"); // The SDK buries a provider usage/quota limit in the assistant message rather // than throwing; detect it here (before the schema/empty-text branches) so it // is classified as a recoverable checkpoint, not a SCHEMA_NONCOMPLIANCE failure // (schema path) or a silent empty-output null (non-schema path). throwIfProviderLimit(session.messages, options.label); if (options.schema) { const structured = (await resolveStructuredOutput( session, capture, options.schema, { ...options, checkContextCap: (pendingPrompt?: string) => { const err = preRequestContextCapError(pendingPrompt); if (err) throw err; }, }, (m) => this.lastAssistantText(m), )) as AgentRunResult; emitUsageAndContext(true); return structured; } const text = this.lastAssistantText(session.messages); if (!text.trim()) { throw new WorkflowError("Subagent produced no assistant output", WorkflowErrorCode.AGENT_EMPTY_OUTPUT, { recoverable: true, agentLabel: options.label, }); } emitUsageAndContext(true); return text as AgentRunResult; } catch (error) { emitUsageAndContext(true); throw error; } finally { removeAbortListener?.(); removeSessionListener?.(); try { emitHistory(); } catch { // History is diagnostic only; never let it mask the real result/error. } session.dispose(); } } private buildPrompt(prompt: string, options: AgentRunOptions, structured: boolean): string { const parts = [ this.instructions, options.instructions, options.label ? `Task label: ${options.label}` : undefined, prompt, ].filter(Boolean); if (structured) { parts.push( [ "Final output contract:", "- Your final action MUST be a structured_output tool call.", "- The structured_output arguments are the return value of this subagent.", "- Do not emit a prose final answer instead of structured_output.", "- If you need to inspect files or run commands first, do so, then call structured_output exactly once.", ].join("\n"), ); } return parts.join("\n\n"); } private lastAssistantText(messages: unknown[]): string { for (let i = messages.length - 1; i >= 0; i--) { const message = messages[i] as Partial | undefined; if (message?.role !== "assistant" || !Array.isArray(message.content)) continue; const text = message.content .filter((part): part is TextContent => part.type === "text") .map((part) => part.text) .join(""); if (text.trim()) return text; } return ""; } }