import { randomUUID } from "node:crypto"; import { chmod, readdir, realpath, unlink } from "node:fs/promises"; import { createServer, type Server } from "node:net"; import { tmpdir } from "node:os"; import { join, relative, resolve } from "node:path"; import type { ExtensionContext } from "@earendil-works/pi-coding-agent"; import type { RunEvent } from "pi-agents/src/run/events.js"; import { loadConfig, loadConfigVersion, loadPriceCatalog, pathsFor, rollbackConfig as rollbackConfigVersion, tryClaimAutoAdmission, updateConfig, type UltraPaths } from "../config/loader.js"; import { POLICY_CAPS } from "../config/defaults.js"; import { runBlocker } from "./blockers.js"; import type { AcceptedWarRoomSignal, ActiveRun, DelegatedExecution, DelegatedRunEvent, QueuedPiAgentsEvent, TraceHandoff, UsageAccountingState, WarRoomWriterFact } from "./run-state.js"; import { acceptWarRoomSignal, activateWarRoomParticipant, assertWarRoomTarget, closeWarRoomSignalBridge, completeWarRoomParticipantRun, issueWarRoomCredential, issueWarRoomRoundCredentials, liveWarRoomParticipantCount, parseWarRoomSignal, registerWarRoomRoundParticipants, registerWarRoomSpecialist, retireWarRoomCredential, setWarRoomParticipantState, unsettledWarRoomSpecialistRuns, warRoomWriterAttributionFacts, warRoomWriterFacts } from "./war-room-state.js"; import { runDoctor } from "./doctor.js"; import { affectedDirectory, checkoutPreflight } from "./preflight.js"; import { configFingerprint, expectedModelSavings, providerFailureStatus, redactLedgerValue, requestProvenance, sameJsonValue } from "./provenance.js"; import { classifyMicroAgentFailure, MicroAgentBackend } from "../backends/micro-agent-backend.js"; import { PiAgentsBackend, flowFor, plannedAgents, warRoomRoundFlow, warRoomSpecialistFlow, type PlannedAgent, type WarRoomBindings, type WarRoomFeeds } from "../backends/pi-agents-backend.js"; import { envelopePrompt, fitEnvelope } from "../context/envelope.js"; import { assertDeclaredArtifacts, isEmptyResult, mergeFacts, parseAgentResult, type AgentResultRole } from "../context/result-contracts.js"; import { discoverInterruptedRuns, materializeRecovery, recoveryContinuation, type RecoveryDecision } from "../context/recovery.js"; import { checklistProgress, plannedChecklist, saveLedger } from "../context/ledger.js"; import { loadProjectOverlay } from "../config/project-overlay.js"; import { appendRoleMemory, readRoleMemory, roleMemorySlice } from "../context/role-memory.js"; import { loadRoleDefinitions, roleShard } from "../workflows/role-definition.js"; import { assignExperiment, readExperiment, type ExperimentAssignmentResult } from "../experiments/runtime.js"; import { effectiveExcludedPaths } from "../security/excluded-defaults.js"; import { assertModelAllowed, scoutModelForLens, writerModelForLens } from "../models/model-policy.js"; import { rosterFor } from "../models/roster.js"; import { budgetAllocationFor, budgetFor, budgetPhase, derivedRate, estimateCredits, maySpawn, resolveCatalog, usableUsdCost, usdBudgetForCredits, type UsdModelCost } from "../models/price-catalog.js"; import { initialHealth, recordProviderResponse } from "../provider/health.js"; import { isRotatableFailure, nextRotationModel } from "../provider/rotation.js"; import { appendRawTrace, hmacId, loadOrCreateKey, pruneRawVault, readRawTrace, writeRawVault, type RawTraceRecord } from "../security/privacy.js"; import { assertAutoModeProfile, profileForAgentDir } from "../security/profiles.js"; import { assertWorkerPath, governanceViolationOf } from "../security/worker-scope.js"; import { TelemetryCollector } from "../telemetry/collector.js"; import { attributionUtility, validateAttributionManifest } from "../telemetry/attribution.js"; import { event, type EventIdentity, type UltraEventType } from "../telemetry/events.js"; import type { AgentFact, AgentResult, AttributionManifest, BaseEvent, ChecklistItem, DispatchRequest, DispatchResult, Mode, Policy, PriceCatalog, PrivacyClass, ProposedShard, ProviderHealth, TaskEnvelope, TaskShape, TaskLedger, ThinkingLevel, Topology, UltraConfig } from "../types.js"; import { reviewEnvelope, reviewPrompt, reviewQuorum, type ReviewRole } from "../workflows/review.js"; import { repairModel } from "../workflows/repair.js"; import { scoutShards } from "../workflows/scout.js"; import { swarmShards } from "../workflows/swarm.js"; import { BOARD_TYPES, parseBoardBatch, WAR_ROOM_MEMBERS, WarRoom, type BoardEvent, type BoardRecord, type CoordinationAction, type RoundMetrics, type WarRoomSignal, type WarRoomWorkerBinding } from "../workflows/war-room.js"; import { analyzeTask } from "./task-shape.js"; import { routeTask, warRoomRoundLimit } from "./routing.js"; import { contextAction } from "../context/compaction.js"; import { isTerminal, transition, type ControllerState } from "./state-machine.js"; import { isVerified, verify } from "../verification/verifier.js"; import { explicitAcceptanceCommand, inferAcceptanceCommand } from "../verification/infer-command.js"; import { EventStore } from "../telemetry/event-store.js"; import { buildWeeklyExportBundle, previewExport, type WeeklyExportBundle, writeWeeklyExport } from "../telemetry/weekly-export.js"; import { validateDecomposition } from "./decomposition-validator.js"; import { effectiveInitial, evaluateCoverage, shouldWiden } from "./progressive-widening.js"; import { allowedConcurrency, staggerDelayMs } from "../provider/backpressure.js"; import { formatTrace } from "../telemetry/live-trace.js"; import { integrateReviewedDiff, worktreeChangedPaths, worktreeDiff } from "../worktrees/integration.js"; import { createWorktree, removeWorktree, type Worktree } from "../worktrees/manager.js"; import { assertOwnedPaths } from "../worktrees/ownership.js"; export interface PendingRootTelemetry { samples: Array<{ sample: { model: string; thinkingLevel: string; turnIndex: number; callIndex: number; startedAt: number }; usage?: { input?: number; output?: number; cacheRead?: number; totalTokens?: number; cost?: { total?: number } }; errorClass?: string; started?: boolean; observedModel?: string; observedProvider?: string; }>; providers: Array<{ status: number; latencyMs: number; health: ProviderHealth }>; } function classifyPiAgentsError(error: string): string { const text = error.toLowerCase(); if (/(no api key|no credentials|unauthori[sz]ed|forbidden|\b401\b|\b403\b)/.test(text)) return "authentication"; if (/(rate.?limit|too many requests|\b429\b|overloaded)/.test(text)) return "rate-limited"; if (/(timed out|timeout)/.test(text)) return "timeout"; if (/(budget.*exceed|turn budget|agent duration)/.test(text)) return "budget-exhausted"; if (/(no models match|unknown model|model.*(?:not found|unavailable))/.test(text)) return "model-unavailable"; if (/(failed to spawn|failed to run delegated|exited before responding|rpc initialization|invalid json from delegated)/.test(text)) return "subprocess-start-failed"; return "pi-agents-node-failed"; } function presentationDetails(active: ActiveRun) { const changedFiles = active.ledger?.changedFiles.length ? active.ledger.changedFiles : active.attributionManifest?.changes.map((change) => change.path) ?? []; const acceptedFacts = active.ledger?.acceptedFacts.slice(0, 12).map(({ id, claim, confidence }) => ({ id, claim, confidence })) ?? []; return { changedFiles, acceptedFacts, ...(runBlocker(active.state, active.blockerReason) ?? {}) }; } /** * What one spawn is worth, as a share of the policy's hard cap. * * These were absolutes — 10 for a writer, 4 for a scout — calibrated on the credit scale of the * shipped Codex roster, where a scout call costs about 0.3 credits. Pricing a declared model at 125 * credits per USD makes that calibration bite: a scout is admitted only under 1.8 credits, which is * roughly $1.20 per million input tokens, so any scout tier at Sonnet's price would be refused at * every policy forever. As a fraction of `hardCap` the numbers scale with the budget the user chose, * and 0.5/0.2 reproduce 10 and 4 exactly at `balanced`, leaving the reference roster untouched. */ function spawnImpact(writer: boolean, policy: Policy): number { return (writer ? 0.5 : 0.2) * POLICY_CAPS[policy].hardCap; } /** One refusal, wherever a price cannot be established, naming the model and the way out. */ function unpricedModel(model: string, catalogPath: string): string { return `No credit price for ${model}. It is absent from price-catalog.json and the Pi model registry reports no usable USD cost for it. Add an entry for ${model} under "models" in ${catalogPath} (credits per 1,000,000 tokens), or drop it from models.tiers.`; } const SAFE_ROOT_TOOLS = new Set(["ultra_dispatch", "read", "write", "edit", "bash", "grep", "find", "ls"]); class DelegatedStartAbortedError extends Error {} function executionNodeId(execution: DelegatedExecution | undefined, nodeId: string): string { return execution ? `${execution.executionId}:${nodeId}` : nodeId; } function warRoomMetrics(room: WarRoom): RoundMetrics { return room.metrics(); } export class UltraController { readonly paths: UltraPaths; private config: UltraConfig | undefined; private telemetry: TelemetryCollector | undefined; private readonly active = new Map(); private readonly pendingPiAgentsEvents = new Map(); private piAgentsEventSequence = 0; private readonly settlingPiAgentsRuns = new Set(); private readonly settledPiAgentsRuns = new Set(); private health: ProviderHealth = initialHealth(); private readonly oneTaskBudgetClaimedVersions = new Set(); private autoLaunchReservationVersion: string | undefined; private readonly agents: PiAgentsBackend; constructor(readonly agentDir: string, pi: import("@earendil-works/pi-coding-agent").ExtensionAPI, private readonly onTrace?: (entry: BaseEvent) => void, private readonly onRawTrace?: (record: RawTraceRecord) => void) { this.paths = pathsFor(agentDir); this.agents = new PiAgentsBackend(pi); } async ready(): Promise { if (this.config) { await assertAutoModeProfile(this.agentDir, this.config.mode); return this.config; } const config = await loadConfig(this.paths); await assertAutoModeProfile(this.agentDir, config.mode); const profile = await profileForAgentDir(this.agentDir); if (profile && config.profile !== profile) throw new Error(`Profile config mismatch: expected ${profile}, found ${config.profile}`); this.config = config; this.telemetry = new TelemetryCollector(this.paths.events, this.paths.database, this.paths.hmacKey, config, config.profile); await pruneRawVault(this.paths.rawVault, config.telemetry.rawRetentionDays); await this.telemetry.pruneAnalytics(config.telemetry.analyticsRetentionDays); return config; } async rebuildProjection(): Promise { await this.ready(); await this.telemetry!.rebuildProjection(); } getConfig(): UltraConfig { if (!this.config) throw new Error("UltraPi is not ready"); return this.config; } getHealth(): ProviderHealth { return this.health; } async reserveAutoLaunch(): Promise { const config = await this.ready(); await this.assertConfigActivation(config); if (config.mode !== "auto") return; if (!config.budgets.acknowledged) throw new Error("Auto mode requires a saved budget or explicit force acknowledgement. Run /ultra-config budget or /ultra-config budget acknowledge."); if (config.budgets.weeklyCreditBudget !== undefined || config.budgets.dailyCreditBudget !== undefined) return; if (this.autoLaunchReservationVersion === config.configVersion) throw new Error("Auto mode requires a saved budget or explicit force acknowledgement. Run /ultra-config budget or /ultra-config budget acknowledge."); this.autoLaunchReservationVersion = config.configVersion; } releaseAutoLaunchReservation(): void { this.autoLaunchReservationVersion = undefined; } async claimAutoLaunchBudget(config = this.getConfig()): Promise { const current = config; await this.assertConfigActivation(current); if (!current.budgets.acknowledged || this.oneTaskBudgetClaimedVersions.has(current.configVersion)) throw new Error("Auto mode requires a saved budget or explicit force acknowledgement. Run /ultra-config budget or /ultra-config budget acknowledge."); if (current.budgets.weeklyCreditBudget !== undefined || current.budgets.dailyCreditBudget !== undefined) return false; if (!await tryClaimAutoAdmission(this.paths, current.configVersion)) throw new Error("Auto mode requires a saved budget or explicit force acknowledgement. Run /ultra-config budget or /ultra-config budget acknowledge."); this.oneTaskBudgetClaimedVersions.add(current.configVersion); if (this.config?.configVersion === current.configVersion) { const consumed = await updateConfig(this.paths, current, (next) => ({ ...next, budgets: { ...next.budgets, acknowledged: false } }), "budget:acknowledgement-consumed"); await this.activateConfig(current, consumed, "budget:acknowledgement-consumed", true); } return true; } private async admitAutoExecution(config: UltraConfig, alreadyAdmitted = false): Promise { await this.assertConfigActivation(config); if (config.mode !== "auto") { this.releaseAutoLaunchReservation(); return; } if (alreadyAdmitted) return; try { if (!config.budgets.acknowledged) throw new Error("Auto mode requires a saved budget or explicit force acknowledgement. Run /ultra-config budget or /ultra-config budget acknowledge."); if (config.budgets.weeklyCreditBudget !== undefined || config.budgets.dailyCreditBudget !== undefined) return; await this.claimAutoLaunchBudget(config); } finally { this.releaseAutoLaunchReservation(); } } private async record(identity: EventIdentity, eventType: UltraEventType, fields: Record = {}) { const active = this.active.get(identity.runId); if (eventType === "result.completed" && active?.terminalResultRecorded) return; if (eventType === "result.completed" && active) { active.terminalResultRecorded = true; // Every non-success ending passes through here, so capturing the reason once means a // path nobody enumerated still reaches the user with an explanation. if (fields.success === false && typeof fields.reason === "string") active.blockerReason ??= fields.reason; } const config = active?.config; let entry: BaseEvent; try { await this.ready(); entry = await this.telemetry!.record(identity, eventType, { ...fields, ...(config ? { configVersion: config.configVersion, policyVersion: config.policyVersion } : {}) }); } catch (error) { if (eventType === "result.completed" && active) active.terminalResultRecorded = false; throw error; } try { this.onTrace?.(entry); } catch {} return entry; } private async transition(runId: string, next: ControllerState, delegatedStopped = false): Promise { const active = this.active.get(runId); if (!active) throw new Error(`Unknown UltraPi run ${runId}`); if (!isTerminal(next)) { if (active.manualStopPending) await this.lifecycleSettlementCommitted(active); if (isTerminal(active.state)) return; if (active.terminalIntent) throw new Error(`UltraPi run ${runId} is settling ${active.terminalIntent}`); active.state = transition(active.state, next); await this.advanceChecklist(active, next); await this.record(active.identity, "route.changed", { state: next }); return; } if (active.terminalSettlement) { await active.terminalSettlement.catch(() => undefined); if (isTerminal(active.state)) return; } active.terminalIntent ??= next; const settlement = (async () => { if (!delegatedStopped) await this.stopDelegatedRuns(active); active.state = transition(active.state, next); active.reservedCredits = 0; await closeWarRoomSignalBridge(active); await this.recordEnvelopeMeasurement(active, Boolean(active.attributionManifest)).catch(() => undefined); await this.advanceChecklist(active, next); await this.record(active.identity, "route.changed", { state: next }); })(); active.terminalSettlement = settlement; try { await settlement; } finally { if (active.terminalSettlement === settlement) active.terminalSettlement = undefined; } } private async rememberScoutFacts(facts: readonly AgentFact[], shape: TaskShape, privacyClass: PrivacyClass): Promise { if (!facts.length) return; const secret = await loadOrCreateKey(this.paths.hmacKey); const entries = facts.filter((fact) => fact.confidence >= 0.7).slice(0, 8).map((fact) => ({ claim: fact.claim, intent: shape.intent, breadth: shape.breadth, paths: [...new Set(fact.evidence.flatMap((evidence) => evidence.path ? [evidence.path] : []))], })); await appendRoleMemory(this.paths.root, "scout", entries, secret, privacyClass); } private async scoutMemorySlice(shape: TaskShape): Promise { try { return roleMemorySlice(await readRoleMemory(this.paths.root, "scout"), shape); } catch { return []; } } private async advanceChecklist(active: ActiveRun, state: ControllerState): Promise { if (!active.ledger) return; const checklist = active.ledger.checklist ?? []; const index = checklist.findIndex((item) => item.id === state); if (index < 0) return; const advanced = checklist.map((item, position) => position < index ? { ...item, state: "done" as const } : position === index ? { ...item, state: isTerminal(state) ? "done" as const : "active" as const } : item); await this.persistLedger(active, { checklist: advanced }).catch(() => undefined); } private async persistLedger(active: ActiveRun, update: Partial = {}): Promise { if (!active.ledger) return; const config = active.config ?? this.getConfig(); const ledger = redactLedgerValue({ ...active.ledger, ...update, remainingBudget: budgetFor(active.policy ?? config.policy, active.spentCredits, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget), }, active.ledgerRedactions ?? []); active.ledger = ledger; await saveLedger(this.paths.ledgers, active.identity.runId, ledger); } private envelope(runId: string, request: DispatchRequest, topology: Topology, config: UltraConfig): TaskEnvelope { return { envelopeVersion: 1, trace: { runId, nodeId: topology }, goal: request.objective, taskSynopsis: request.objective.slice(0, 1000), scope: { paths: request.paths?.length ? request.paths : ["."], excludedPaths: effectiveExcludedPaths(request.excludePaths) }, lens: topology, evidenceTarget: request.acceptanceCommand ? "acceptance command" : "evidence-backed findings", knownFacts: [], openQuestions: [], constraints: ["No recursive orchestration", "Respect declared scope", "Return bounded structured result"], ...(request.acceptanceCommand ? { acceptance: { command: request.acceptanceCommand, expected: "exit code 0" } } : {}), privacyProfile: config.profile, outputContract: { schemaName: "AgentResult", maxOutputTokens: topology === "scout" ? config.scout.maxOutputTokens : config.boundedWriter.maxOutputTokens, maxFacts: 12, maxHypotheses: 8, maxUnknowns: 8 }, }; } private async recordGovernanceViolation(active: ActiveRun, error: unknown, nodeId: string, role: string): Promise { const violation = governanceViolationOf(error); // A node that fails its result contract delivered nothing while looking like it ran. That is // the silence the reviewer already reported; every role can produce it. const kind = violation?.kind ?? (classifyMicroAgentFailure(error) === "result-contract" ? "empty-result" : undefined); if (!kind) return; const scope = active.envelope?.scope; await this.record(active.identity, "governance.violation", { nodeId, role, kind, ...(scope ? { declaredScope: await this.traceHash(JSON.stringify(scope), "SCOPE") } : {}), ...(violation?.attemptedPath ? { attemptedPath: await this.traceHash(violation.attemptedPath, "PATH") } : {}), }).catch(() => undefined); } /** Reports a result that parsed but carries nothing. Observation only: the run continues. */ private async recordEmptyResult(active: ActiveRun, nodeId: string, role: string, result: AgentResult): Promise { if (!isEmptyResult(result)) return; await this.record(active.identity, "governance.violation", { nodeId, role, kind: "empty-result", schemaValid: true }).catch(() => undefined); } private async traceHash(value: string, kind: string): Promise { try { return hmacId(await loadOrCreateKey(this.paths.hmacKey), value, kind); } catch { return `<${kind}_UNAVAILABLE>`; } } private async storeRawTrace(active: ActiveRun, kind: "task" | "result" | "diagnostic" | "verification" | "workspace" | "feedback", fromNodeId: string, toNodeId: string, content: string): Promise { if (!(active.config ?? this.getConfig()).telemetry.rawVaultEnabled) return false; try { const record: RawTraceRecord = { timestamp: new Date().toISOString(), kind, runId: active.identity.runId, fromNodeId, toNodeId, content }; await appendRawTrace(this.paths.rawVault, this.paths.vaultKey, active.identity.runId, record); try { this.onRawTrace?.(record); } catch {} return true; } catch { return false; } } private async recordTaskAssignment(active: ActiveRun, assignment: PlannedAgent, roleOverride?: string, execution?: DelegatedExecution): Promise { const nodeId = executionNodeId(execution, assignment.nodeId); const content = assignment.task; const payloadChars = content.length; const payloadEstimatedTokens = Math.ceil(payloadChars / 4); const contentHash = await this.traceHash(content, "TASK"); const traceStored = await this.storeRawTrace(active, "task", "controller", nodeId, content); const writer = assignment.tools?.some((tool) => tool === "edit" || tool === "ultra_edit") ?? false; const role = roleOverride ?? (writer ? "writer" : active.topology === "deep" ? "deep" : "scout"); if (assignment.model) (execution?.plannedModels ?? (active.plannedModels ??= new Map())).set(nodeId, assignment.model); (execution?.resultRoles ?? (active.resultRoles ??= new Map())).set(nodeId, writer ? "writer" : "scout"); const handoff: TraceHandoff = { fromNodeId: "controller", toNodeId: nodeId, kind: "task-envelope", payloadChars, payloadEstimatedTokens, factIds: [], hypothesisIds: [], contentHash, traceStored, received: false }; (active.handoffs ??= []).push(handoff); const normalizedContext = content .replaceAll(active.identity.runId, "") .replaceAll(`\"nodeId\":\"${assignment.nodeId}\"`, "\"nodeId\":\"\"") .replaceAll(`:${assignment.nodeId}:`, "::"); const executionFields = execution ? { executionId: execution.executionId, piAgentsNodeInstance: assignment.nodeId } : {}; await this.record(active.identity, "task.context.measured", { nodeId, ...executionFields, contextHash: await this.traceHash(normalizedContext, "CONTEXT"), contextEstimatedTokens: Math.ceil(normalizedContext.length / 4) }); await this.record(active.identity, "agent.spawn.proposed", { nodeId, ...executionFields, role, ...(assignment.model ? { model: assignment.model } : {}), ...(assignment.thinking ? { thinkingLevel: assignment.thinking } : {}), taskChars: payloadChars, taskEstimatedTokens: payloadEstimatedTokens, contentHash, traceStored }); await this.record(active.identity, "handoff.sent", { fromNodeId: handoff.fromNodeId, toNodeId: handoff.toNodeId, kind: handoff.kind, payloadChars, payloadEstimatedTokens, factIds: [], hypothesisIds: [], contentHash, traceStored }); } private async recordPlannedAgents(active: ActiveRun, assignments: readonly PlannedAgent[], stage: "execution" | "escalation" = "execution", phase: DelegatedExecution["phase"] = "standard"): Promise { await this.assertSpawnsAllowed(active, assignments, stage); const execution: DelegatedExecution = { executionId: `exec-${(active.delegatedExecutionSequence ?? 0) + 1}`, phase, plannedModels: new Map(), resultRoles: new Map(), nodeKinds: new Map(), boardAuthors: new Map(), usageIncomplete: false, }; active.delegatedExecutionSequence = (active.delegatedExecutionSequence ?? 0) + 1; active.plannedModels = execution.plannedModels; active.resultRoles = execution.resultRoles; active.nodeKinds = execution.nodeKinds; active.delegatedUsageIncomplete = false; for (const assignment of assignments) await this.recordTaskAssignment(active, assignment, undefined, execution); return execution; } private bindDelegatedExecution(active: ActiveRun, piAgentsRunId: string, execution: DelegatedExecution): void { (active.delegatedExecutions ??= new Map()).set(piAgentsRunId, execution); (active.delegatedRunStates ??= new Map()).set(piAgentsRunId, "live"); active.piAgentsRunId = piAgentsRunId; const pending = this.pendingPiAgentsEvents.get(piAgentsRunId); if (pending?.length) { (active.piAgentsEventQueue ??= []).push(...pending); active.piAgentsEventQueue.sort((left, right) => left.sequence - right.sequence); this.pendingPiAgentsEvents.delete(piAgentsRunId); } } private liveDelegatedRunIds(active: ActiveRun): string[] { const runIds = new Set(); for (const piAgentsRunId of active.delegatedExecutions?.keys() ?? []) runIds.add(piAgentsRunId); if (active.piAgentsRunId) runIds.add(active.piAgentsRunId); return [...runIds].filter((piAgentsRunId) => { const state = active.delegatedRunStates?.get(piAgentsRunId); return state !== "stop-accepted" && state !== "settled" && !this.settlingPiAgentsRuns.has(piAgentsRunId) && !this.settledPiAgentsRuns.has(piAgentsRunId); }); } private async startDelegatedRun(active: ActiveRun, start: () => Promise, bind: (piAgentsRunId: string) => void): Promise { if (active.manualStopPending) { const settlement = active.terminalSettlement; if (settlement) await settlement.catch(() => undefined); } if (active.terminalIntent || isTerminal(active.state)) throw new DelegatedStartAbortedError("Delegated start blocked while its parent is terminal"); active.unboundDelegatedStarts = (active.unboundDelegatedStarts ?? 0) + 1; let released = false; const releaseUnbound = (resumeDrain: boolean) => { if (released) return; released = true; active.unboundDelegatedStarts = Math.max(0, (active.unboundDelegatedStarts ?? 1) - 1); if (resumeDrain && active.unboundDelegatedStarts === 0) void this.drainPiAgentsEvents(active).catch(() => undefined); }; let started: Promise; try { started = start(); } catch (error) { releaseUnbound(true); throw error; } let stale = false; const tracked = started.then(async (piAgentsRunId) => { bind(piAgentsRunId); releaseUnbound(false); if (active.terminalIntent || isTerminal(active.state)) { stale = true; await this.agents.stop(piAgentsRunId); (active.delegatedRunStates ??= new Map()).set(piAgentsRunId, "stop-accepted"); return; } }, () => { releaseUnbound(true); }); (active.inFlightStarts ??= new Set()).add(tracked); try { const piAgentsRunId = await started; try { await tracked; } catch (error) { if (stale) throw new DelegatedStartAbortedError(`Delegated run ${piAgentsRunId} could not stop after its parent started settling`, { cause: error }); throw error; } if (stale || active.terminalIntent || isTerminal(active.state)) throw new DelegatedStartAbortedError(`Delegated run ${piAgentsRunId} resolved after its parent started settling`); return piAgentsRunId; } finally { releaseUnbound(true); active.inFlightStarts?.delete(tracked); } } private async stopDelegatedRuns(active: ActiveRun): Promise { const failures: Array<{ runId?: string; error: unknown }> = []; const runIds = this.liveDelegatedRunIds(active); const starts = await Promise.allSettled([...(active.inFlightStarts ?? [])]); for (const result of starts) if (result.status === "rejected") failures.push({ error: result.reason }); const stops = await Promise.allSettled(runIds.map((piAgentsRunId) => this.agents.stop(piAgentsRunId))); for (let index = 0; index < stops.length; index += 1) { const result = stops[index]!; if (result.status === "fulfilled") { (active.delegatedRunStates ??= new Map()).set(runIds[index]!, "stop-accepted"); completeWarRoomParticipantRun(active, runIds[index]!); } else failures.push({ runId: runIds[index], error: result.reason }); } for (const failure of failures) { await this.recordSafeFailure(active, "workflow.failed", failure.runId ? `stop:${failure.runId}` : "delegated-start", failure.error instanceof Error ? failure.error.message : String(failure.error)).catch(() => undefined); } return failures.map((failure) => failure.error); } private async lifecycleSettlementCommitted(active: ActiveRun): Promise { if (active.manualStopPending) { const settlement = active.terminalSettlement; if (settlement) await settlement.catch(() => undefined); } return isTerminal(active.state) || Boolean(active.terminalIntent); } private delegatedRun(piAgentsRunId: string): { runId: string; active: ActiveRun; execution?: DelegatedExecution } | undefined { for (const [runId, active] of this.active) { const execution = active.delegatedExecutions?.get(piAgentsRunId); if (execution) return { runId, active, execution }; if (active.delegatedExecutionSequence === undefined && active.piAgentsRunId === piAgentsRunId) return { runId, active }; } return undefined; } private async assertSpawnsAllowed(active: ActiveRun, assignments: readonly PlannedAgent[], stage: "execution" | "escalation" = "execution"): Promise { const config = active.config ?? this.getConfig(); const policy = active.policy ?? config.policy; const catalog = await this.pricing(active, ...assignments.map((assignment) => assignment.model ?? active.model)); const allocation = budgetAllocationFor(policy); const stageCap = stage === "execution" ? allocation.explorationImplementation : allocation.explorationImplementation + allocation.verification + allocation.escalation; let reserved = 0; for (const assignment of assignments) { const model = assignment.model ?? active.model; const writer = assignment.tools?.some((tool) => tool === "edit" || tool === "ultra_edit") ?? false; const outputTokens = writer ? config.boundedWriter.maxOutputTokens : config.scout.maxOutputTokens; const estimated = estimateCredits(catalog, model, 8_000, 0, outputTokens); const state = budgetFor(policy, active.spentCredits + reserved, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget); const allocated = Math.min(state.internalStopTarget, stageCap); if (!maySpawn({ ...state, internalStopTarget: allocated }, estimated, writer ? 0.9 : 0.5, spawnImpact(writer, policy), 0.1, 0.1, !writer)) { await this.record(active.identity, "agent.spawn.rejected", { nodeId: assignment.nodeId, model, reason: "budget-governor", phase: budgetPhase(state), estimatedCredits: estimated, speculative: !writer }); throw new Error(`Budget governor blocks ${writer ? "spawn" : "speculative spawn"}`); } reserved += estimated; } } private async runScoutWaves(context: ExtensionContext, active: ActiveRun, envelope: TaskEnvelope, proposed: ProposedShard[], config: UltraConfig, policy: Policy) { const caps = POLICY_CAPS[policy]; const maxTotal = Math.min(config.scout.maxTotal, config.piAgentsBudgets.maxAgents, caps.maxTotal); let maxParallel = allowedConcurrency(this.health, Math.min(config.scout.maxParallel, config.piAgentsBudgets.maxParallelism, caps.maxParallel)); if (maxParallel === 0) throw new Error("Provider cooldown blocks scout dispatch"); const initial = validateDecomposition(proposed, maxTotal); let pending = initial.accepted.slice(0, Math.min(effectiveInitial(config, policy), maxParallel, maxTotal)); if (pending.length === 0) throw new Error(`Scout decomposition rejected: ${initial.rejected.map((entry) => entry.reason).join(", ")}`); const launched: ProposedShard[] = []; const results: Array<{ result: AgentResult; evidenceTarget: string }> = []; let wave = 1; let gapSequence = 0; while (pending.length > 0 && wave <= caps.maxWaves && launched.length < maxTotal) { await this.assertSpawnsAllowed(active, pending.map((shard) => ({ nodeId: shard.id, task: shard.objective, model: config.scout.model, tools: ["ultra_read"] }))); if (wave > 1) await this.assertBudgetAdmission(config.scout.model, "scout", config, pending.length, active); await this.record(active.identity, "decomposition.validated", { accepted: pending.length, rejected: wave === 1 ? initial.rejected.length : 0, wave }); const completed = (await Promise.all(pending.map(async (shard, index) => { await new Promise((resolvePromise) => setTimeout(resolvePromise, staggerDelayMs(index))); const workerEnvelope = fitEnvelope({ ...envelope, trace: { runId: active.identity.runId, nodeId: shard.id }, scope: { paths: shard.scope, excludedPaths: shard.excludedScope }, lens: shard.lens, evidenceTarget: shard.evidenceTarget }, config.context.envelopeHardCapTokens); const runShard = async (shardModel: string) => { const worker = new MicroAgentBackend(context, this.agentDir, shardModel, config.scout.thinking, async (event) => { if (event.type === "model.call.started") { await this.record(active.identity, event.type, { nodeId: shard.id, model: event.model, provider: event.provider, thinkingLevel: event.thinkingLevel, totalTokens: 0, latencyMs: 0, turnIndex: event.turnIndex, callIndex: event.callIndex, role: "scout" }); return; } if ("toolCallId" in event) { await this.recordRootTool(active.identity.runId, event.type, event.toolCallId, event.toolName, event.type === "tool.started" ? 0 : event.toolElapsedMs, shard.id, "scout"); return; } const errorClass = event.type === "model.call.failed" ? event.errorClass : undefined; const credits = await this.recordUsage(active, event, shardModel, shard.id, undefined, true, { thinkingLevel: event.thinkingLevel, totalTokens: event.totalTokens, latencyMs: event.turnElapsedMs, turnIndex: event.turnIndex, callIndex: event.callIndex, observedModel: event.model, observedProvider: event.provider, ...(errorClass ? { errorClass } : {}) }); const agentCredits = active.agentCredits ??= new Map(); agentCredits.set(shard.id, (agentCredits.get(shard.id) ?? 0) + credits); if (isTerminal(active.state)) throw new Error("MicroAgent stopped at the task budget limit"); }, async (failure) => { const health = this.providerResponse(failure.status, 0); await this.recordProviderHealth(active.identity.runId, failure.status, 0, "delegated", health).catch(() => undefined); }); await this.recordTaskAssignment(active, { nodeId: shard.id, task: envelopePrompt(workerEnvelope), model: shardModel, thinking: config.scout.thinking, tools: ["ultra_read", "ultra_grep", "ultra_find", "ultra_ls"] }); await this.record(active.identity, "agent.spawned", { nodeId: shard.id, role: "scout", lens: shard.lens, model: shardModel, thinkingLevel: config.scout.thinking, topology: active.topology, wave }); await this.recordReceivedHandoffs(active, shard.id); return await worker.run(workerEnvelope); }; try { const result = await this.withModelRotation(active, { model: scoutModelForLens(config, shard.lens), nodeId: shard.id, role: "scout", topology: "scout" }, runShard); if (isTerminal(active.state)) return undefined; await this.recordEmptyResult(active, shard.id, "scout", result); (active.agentFacts ??= new Map()).set(shard.id, result.facts); await this.record(active.identity, "agent.completed", { nodeId: shard.id, factsTotal: result.facts.length, outputTokens: 0, wave }); await this.recordResultHandoff(active, shard.id, result).catch(() => undefined); return { result, evidenceTarget: shard.evidenceTarget }; } catch (error) { await this.recordGovernanceViolation(active, error, shard.id, "scout"); await this.record(active.identity, "agent.failed", { nodeId: shard.id, errorClass: classifyMicroAgentFailure(error), wave }); return undefined; } }))).filter((entry): entry is { result: AgentResult; evidenceTarget: string } => entry !== undefined); if (isTerminal(active.state)) throw new Error("UltraPi budget exhausted while scouts were running"); launched.push(...pending); results.push(...completed); if (results.length === 0) throw new Error("All scouts failed before producing a structured result"); const merged = mergeFacts(results); const gaps = [...new Set(results.flatMap(({ result }) => [ ...result.unknowns, ...(result.resourceRequest?.reason ? [result.resourceRequest.reason] : []), ]).map((gap) => gap.trim()).filter((gap) => gap.length >= 8 && !/^(?:research|investigate|explore|look)(?: deeper| more)?[.!]?$/i.test(gap)))]; const facts = results.flatMap(({ result }) => result.facts); const confidence = merged.facts.length ? merged.facts.reduce((sum, fact) => sum + fact.confidence, 0) / merged.facts.length : 0; const catalog = await this.pricing(active, config.scout.model); const marginalCost = estimateCredits(catalog, config.scout.model, 8_000 * completed.length, 0, config.scout.maxOutputTokens * completed.length); const coverage = evaluateCoverage(facts, gaps.length, merged.contradictions.length, confidence, marginalCost, caps.internalStopTarget - active.spentCredits); await this.record(active.identity, "coverage.evaluated", { ...coverage, wave, totalAgents: launched.length }); if (!shouldWiden(coverage, policy, wave, gaps.length + merged.contradictions.length)) break; const candidates = gaps.map((gap) => ({ id: `gap-${++gapSequence}`, objective: `Resolve the concrete gap: ${gap}`, scope: envelope.scope.paths, excludedScope: envelope.scope.excludedPaths, lens: "gap", evidenceTarget: `Evidence resolving: ${gap}`, expectedOutput: "facts", canRunIndependently: true, canChangeFinalDecision: true, writeIntent: false } satisfies ProposedShard)); const candidateIds = new Set(candidates.map((shard) => shard.id)); const validated = validateDecomposition([...launched, ...candidates], maxTotal); maxParallel = allowedConcurrency(this.health, Math.min(config.scout.maxParallel, config.piAgentsBudgets.maxParallelism, caps.maxParallel)); if (maxParallel === 0) break; pending = validated.accepted.filter((shard) => candidateIds.has(shard.id)).slice(0, Math.min(maxParallel, maxTotal - launched.length)); wave += 1; } return mergeFacts(results); } /** * Runs `attempt` on `model`; if the provider rather than the work fails it, retries on the * next declared model at or above the same rank. A provider that stops answering is what * kills whole branches of a fleet silently, and the roster already says which model is the * next honest substitute. The fallback re-enters budget admission, so a rescue can be * refused on cost rather than taken for free. */ private async withModelRotation(active: ActiveRun, options: { model: string; nodeId: string; role: string; topology: Topology; critical?: boolean; maxRotations?: number }, attempt: (model: string) => Promise): Promise { const config = active.config ?? this.getConfig(); const roster = rosterFor(config); const exhausted = new Set(); const limit = options.maxRotations ?? 1; let model = options.model; for (let rotations = 0; ; rotations += 1) { exhausted.add(model); try { return await attempt(model); } catch (error) { const errorClass = classifyMicroAgentFailure(error); if (rotations >= limit || isTerminal(active.state) || !isRotatableFailure(errorClass)) throw error; const next = nextRotationModel(roster, options.model, options.critical ?? false, exhausted); if (!next) throw error; try { await this.assertBudgetAdmission(next, options.topology, config, 1, active, 0, false); this.captureModelCosts(active, [next]); } catch (blocked) { const reason = blocked instanceof Error && /registry entry is unavailable/.test(blocked.message) ? "model-unavailable" : "budget"; await this.record(active.identity, "model.rotation.blocked", { nodeId: options.nodeId, role: options.role, from: model, to: next, errorClass, reason }).catch(() => undefined); throw error; } await this.record(active.identity, "model.rotated", { nodeId: options.nodeId, role: options.role, from: model, to: next, errorClass }); model = next; } } } private captureModelCosts(active: ActiveRun, models: readonly string[]): void { const config = active.config ?? this.getConfig(); const costs = new Map(active.modelCosts); for (const reference of new Set(models)) { assertModelAllowed(config, reference); const slash = reference.indexOf("/"); const model = slash > 0 ? active.modelRegistry?.find(reference.slice(0, slash), reference.slice(slash + 1)) : undefined; if (!model) throw new Error(`Model registry entry is unavailable for ${reference}`); if (!model.cost) throw new Error(`Model registry entry for ${reference} declares no cost`); const rates = [model.cost, ...(model.cost.tiers ?? [])]; costs.set(reference, { input: Math.min(...rates.map((rate) => rate.input)), cacheRead: Math.min(...rates.map((rate) => rate.cacheRead)), output: Math.min(...rates.map((rate) => rate.output)), }); } active.modelCosts = costs; active.resolvedCatalog = undefined; } /** * The catalogue this run prices against: what `price-catalog.json` declares, plus an entry derived * from the Pi registry for every declared model the document does not name. Without this a model * outside the shipped Codex list costs nothing, and nothing that gates on credits — the phase, the * soft and hard caps, the weekly and daily windows — can ever fire. * * The catalogue file itself is never rewritten. `requestProvenance` stamps its version into every * request, so editing it mid-week would silently reinterpret events already on disk. */ private async pricing(active: ActiveRun, ...models: readonly string[]): Promise { const catalog = await loadPriceCatalog(this.paths); const wanted = models.filter((model) => model && !catalog.models[model] && !active.derivedRates?.has(model)); if (wanted.length) { this.captureModelCosts(active, wanted); const derived = new Map(active.derivedRates); for (const model of wanted) { const cost = active.modelCosts?.get(model); if (!usableUsdCost(cost)) throw new Error(unpricedModel(model, this.paths.priceCatalog)); derived.set(model, derivedRate(cost, catalog.creditsPerUsd)); } active.derivedRates = derived; active.resolvedCatalog = undefined; } active.resolvedCatalog ??= resolveCatalog(catalog, active.derivedRates ?? new Map()); return active.resolvedCatalog; } private async withDelegatedBudget(active: ActiveRun, config: UltraConfig, models: readonly string[], stage: "execution" | "escalation" = "execution"): Promise { const maxParallelism = allowedConcurrency(this.health, config.piAgentsBudgets.maxParallelism); if (maxParallelism === 0) throw new Error("Provider cooldown blocks delegated dispatch"); const throttled = { ...config, piAgentsBudgets: { ...config.piAgentsBudgets, maxParallelism } }; const [windows, catalog] = await Promise.all([this.windowSpend(), this.pricing(active, ...models)]); // A run whose every model is declared free has no USD ceiling to enforce. Keyed on the declared // price rather than on the profile name, so a free profile running a paid model still gets one. if (models.every((model) => { const rate = catalog.models[model]; return rate && rate.input === 0 && rate.cached === 0 && rate.output === 0; })) return throttled; const policy = active.policy ?? config.policy; const allocation = budgetAllocationFor(policy); const stageCap = stage === "execution" ? allocation.explorationImplementation : allocation.explorationImplementation + allocation.verification + allocation.escalation; const cap = Math.min(POLICY_CAPS[policy].internalStopTarget, stageCap); const pricedModels = models.map((model) => { const cost = active.modelCosts?.get(model); if (!cost) throw new Error(`Private model cost is unavailable for ${model}`); return { model, cost }; }); const otherReserved = [...this.active.values()].reduce((total, run) => total + (run === active || isTerminal(run.state) ? 0 : run.reservedCredits ?? 0), 0); const remaining = Math.min( cap - active.spentCredits, config.budgets.weeklyCreditBudget === undefined ? Infinity : config.budgets.weeklyCreditBudget - windows.week - otherReserved, config.budgets.dailyCreditBudget === undefined ? Infinity : config.budgets.dailyCreditBudget - windows.day - otherReserved, ); const derivedMaxCost = usdBudgetForCredits(catalog, pricedModels, remaining); if (derivedMaxCost === undefined) throw new Error(`Cannot enforce the delegated credit budget: ${pricedModels.map(({ model }) => model).join(", ")} has no usable USD cost in the Pi model registry`); const maxCost = Math.min(config.piAgentsBudgets.maxCost ?? derivedMaxCost, derivedMaxCost); if (!Number.isFinite(maxCost) || maxCost <= 0) throw new Error("Delegated budget requires a positive finite maxCost"); active.reservedCredits = remaining * maxCost / derivedMaxCost; return { ...throttled, piAgentsBudgets: { ...throttled.piAgentsBudgets, maxCost } }; } private resultIds(value: unknown, key: "facts" | "hypotheses"): string[] { if (!value || typeof value !== "object" || !Array.isArray((value as Record)[key])) return []; return ((value as Record)[key] as unknown[]).flatMap((entry) => { const id = entry && typeof entry === "object" ? (entry as Record).id : undefined; return typeof id === "string" ? [id] : []; }); } private async recordAttribution(active: ActiveRun, nodeId: string, result: AgentResult): Promise { const manifest = result.attributionManifest; if (!manifest) return; const factsByAgent = new Map(active.agentFacts); if (active.warRoomPhase === "writer") factsByAgent.set("warroom-board", warRoomWriterAttributionFacts(active)); factsByAgent.set(nodeId, result.facts); const facts = [...factsByAgent.values()].flat(); validateAttributionManifest(manifest, facts); active.agentFacts = factsByAgent; active.attributionManifest = manifest; await this.storeRawTrace(active, "result", nodeId, "attribution", JSON.stringify(manifest)); for (const [sourceNodeId, sourceFacts] of factsByAgent) { const ids = new Set(sourceFacts.map((fact) => fact.id)); const decisions = manifest.decisions.filter((decision) => decision.supportingFactIds.some((id) => ids.has(id))); const sourceManifest = { ...manifest, decisions }; const credits = active.agentCredits?.get(sourceNodeId) ?? 0; const utility = attributionUtility(sourceFacts, sourceManifest, credits); const used = sourceFacts.filter((fact) => manifest.usedFactIds.includes(fact.id)); const duplicates = new Set(manifest.rejectedFactIds.filter((entry) => entry.reason === "duplicate").map((entry) => entry.id)); await this.record(active.identity, "attribution.created", { nodeId: sourceNodeId, factsTotal: sourceFacts.length, factsUsed: used.length, uniqueFactsUsed: used.filter((fact) => !duplicates.has(fact.id)).length, duplicateFacts: sourceFacts.filter((fact) => duplicates.has(fact.id)).length, decisionCount: decisions.length, affectedDirectories: new Set(manifest.changes.map((change) => affectedDirectory(change.path))).size, ...utility, }); } } private async recordEnvelopeMeasurement(active: ActiveRun, complete: boolean): Promise { if (active.envelopeMeasurementRecorded || !active.handoffs?.some((handoff) => handoff.kind === "task-envelope")) return; const transmitted = new Set(active.envelope?.knownFacts.map((fact) => fact.id) ?? []); const usedEnvelopeFacts = complete && active.attributionManifest ? new Set(active.attributionManifest.usedFactIds.filter((id) => transmitted.has(id))).size : 0; active.envelopeMeasurementRecorded = true; await this.record(active.identity, "envelope.measured", { complete, transmittedEnvelopeFacts: transmitted.size, usedEnvelopeFacts }); } private async recordResultHandoff(active: ActiveRun, fromNodeId: string, value: unknown): Promise { let content: string; try { content = JSON.stringify(value); } catch { return; } const toNodeId = active.topology === "warroom" && active.warRoomPhase === "rounds" ? "blackboard" : "controller"; const payloadChars = content.length; const payloadEstimatedTokens = Math.ceil(payloadChars / 4); const facts = this.resultIds(value, "facts"); const hypotheses = this.resultIds(value, "hypotheses"); const contentHash = await this.traceHash(content, "HANDOFF"); const factIds = await Promise.all(facts.map((id) => this.traceHash(id, "FACT"))); const hypothesisIds = await Promise.all(hypotheses.map((id) => this.traceHash(id, "HYPOTHESIS"))); const handoff: TraceHandoff = { fromNodeId, toNodeId, kind: "result", payloadChars, payloadEstimatedTokens, factIds, hypothesisIds, contentHash, traceStored: await this.storeRawTrace(active, "result", fromNodeId, toNodeId, content), received: false, }; (active.handoffs ??= []).push(handoff); await this.record(active.identity, "handoff.sent", { fromNodeId, toNodeId, kind: handoff.kind, payloadChars, payloadEstimatedTokens, factIds: handoff.factIds, hypothesisIds: handoff.hypothesisIds, contentHash: handoff.contentHash, traceStored: handoff.traceStored }); if (active.topology === "warroom") await this.record(active.identity, "blackboard.event", { boardType: "result", fromNodeId, toNodeId, payloadChars, contentHash: handoff.contentHash }); if (toNodeId === "controller") await this.recordReceivedHandoffs(active, "controller"); } private async recordReceivedHandoffs(active: ActiveRun, toNodeId: string): Promise { for (const handoff of active.handoffs ?? []) { if (handoff.toNodeId !== toNodeId || handoff.received) continue; handoff.received = true; await this.record(active.identity, "handoff.received", { fromNodeId: handoff.fromNodeId, toNodeId, kind: handoff.kind, payloadChars: handoff.payloadChars, payloadEstimatedTokens: handoff.payloadEstimatedTokens, factIds: handoff.factIds, hypothesisIds: handoff.hypothesisIds, contentHash: handoff.contentHash, traceStored: handoff.traceStored }); } } async dispatch(context: ExtensionContext, request: DispatchRequest, pendingRootTelemetry?: PendingRootTelemetry): Promise { const championConfig = await this.ready(); if (!request.objective.trim()) throw new Error("ultra_dispatch requires an objective"); const autoLaunch = championConfig.mode === "auto" || request.mode === "auto"; const accountBeforeBudgetRejection = autoLaunch && !request.force && Boolean(pendingRootTelemetry?.samples.length); const ratio = (context.getContextUsage()?.percent ?? 0) / 100; let initialShape = analyzeTask({ ...request, contextRatio: ratio }); if (!request.acceptanceCommand && (initialShape.intent === "fix" || initialShape.intent === "change")) { const inferred = explicitAcceptanceCommand(request.objective) ?? await inferAcceptanceCommand(context.cwd, context.isProjectTrusted?.() === true); if (inferred) { request = { ...request, acceptanceCommand: inferred }; initialShape = analyzeTask({ ...request, contextRatio: ratio }); } } const requestHash = await this.telemetry!.requestHash(request.objective); const records = await Promise.all((await readdir(this.paths.experiments, { withFileTypes: true })) .filter((entry) => entry.isFile() && entry.name.endsWith(".json")) .map((entry) => readExperiment(this.paths, entry.name.slice(0, -5)))); const running = records.filter((record) => record.status === "running"); if (running.length > 1) throw new Error(`Multiple running UltraPi experiments: ${running.map((record) => record.experimentId).sort().join(", ")}`); const experiment = running[0]; const assignment: ExperimentAssignmentResult | undefined = experiment ? await assignExperiment(this.paths, experiment.experimentId, requestHash, { privacyClass: request.privacyClass ?? "restricted", risk: initialShape.risk, policy: request.policy ?? championConfig.policy, securitySensitive: /\b(auth|security|credential|secret|token|cve)\b/i.test(request.objective), migration: /\b(migration|migrate)\b/i.test(request.objective), destructive: /\b(delete|drop|truncate|destructive)\b/i.test(request.objective), productionIncident: /\b(production|prod(?:uction)? incident|outage)\b/i.test(request.objective), }) : undefined; const versioned = assignment ? await loadConfigVersion(this.paths, assignment.configVersion) : championConfig; const overlay = await loadProjectOverlay(context.cwd, versioned); const config = overlay.config; if (overlay.applied && config.forbiddenTopologies?.includes((request.mode ?? config.mode) as Mode)) throw new Error(`UltraPi project overlay forbids topology ${request.mode ?? config.mode}`); if (overlay.applied && config.pinnedAcceptanceCommand) request = { ...request, acceptanceCommand: config.pinnedAcceptanceCommand }; if (overlay.applied && config.projectScopePaths?.length && !request.paths?.length) request = { ...request, paths: config.projectScopePaths }; if (overlay.applied) request = { ...request, excludePaths: [...(request.excludePaths ?? []), ...(config.projectExcludedPaths ?? [])] }; await this.admitAutoExecution(config, Boolean(request.force) || accountBeforeBudgetRejection); const runId = randomUUID(); const identity: EventIdentity = { sessionId: await this.traceHash(context.sessionManager.getSessionId(), "SESSION"), taskId: runId, runId, ...(experiment ? { experimentId: experiment.experimentId } : {}) }; this.active.set(runId, { state: "RECEIVED", identity, topology: "direct", cwd: context.cwd, acceptanceCommand: request.acceptanceCommand, informational: ["answer", "investigate", "review"].includes(initialShape.intent), repairAttempts: 0, fingerprints: new Map(), model: config.root.model, spentCredits: 0, spentUsd: 0, policy: request.policy ?? config.policy, config, modelRegistry: context.modelRegistry, context, ledgerRedactions: [request.objective, request.acceptanceCommand ?? ""] }); try { const active = this.active.get(runId)!; if (assignment) await this.record(identity, assignment.event.eventType, assignment.event.fields); const availableModels = context.modelRegistry?.getAvailable?.().map((model) => `${model.provider}/${model.id}`).sort() ?? []; const priceCatalog = await loadPriceCatalog(this.paths); await this.record(identity, "request.received", { requestHash, privacyClass: request.privacyClass ?? "restricted", ...requestProvenance(config, availableModels, priceCatalog), }); await this.storeRawTrace(active, "task", "user", "root", request.objective); if (request.acceptanceCommand) await this.storeRawTrace(active, "verification", "controller", "verifier", request.acceptanceCommand); await this.storeRawTrace(active, "workspace", "controller", "root", await realpath(context.cwd)); if (config.telemetry.rawVaultEnabled) await writeRawVault(this.paths.rawVault, this.paths.vaultKey, request.objective).catch(() => undefined); await this.transition(runId, "PREFLIGHT"); for (const provider of pendingRootTelemetry?.providers ?? []) await this.recordProviderHealth(runId, provider.status, provider.latencyMs, "root", provider.health); for (const pending of pendingRootTelemetry?.samples ?? []) { const metadata = { thinkingLevel: pending.sample.thinkingLevel, totalTokens: pending.usage?.totalTokens ?? (pending.usage?.input ?? 0) + (pending.usage?.output ?? 0), latencyMs: Math.max(0, Date.now() - pending.sample.startedAt), turnIndex: pending.sample.turnIndex, callIndex: pending.sample.callIndex, observedModel: pending.observedModel, observedProvider: pending.observedProvider, ...(pending.errorClass ? { errorClass: pending.errorClass } : {}) }; if (!pending.started) await this.recordRootModelStarted(runId, pending.sample.model, { ...metadata, latencyMs: 0 }); await this.recordUsage(active, { input: pending.usage?.input ?? 0, output: pending.usage?.output ?? 0, cacheRead: pending.usage?.cacheRead ?? 0, cost: pending.usage?.cost?.total }, pending.sample.model, "root", undefined, true, metadata); } if (isTerminal(active.state)) throw new Error("UltraPi budget exhausted during the root decision"); if (accountBeforeBudgetRejection) { await this.admitAutoExecution(config); } const preflight = await checkoutPreflight(context.cwd, request, context, budgetFor(request.policy ?? config.policy, active.spentCredits, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget), this.health); await this.record(identity, "preflight.completed", preflight); await this.record(identity, "budget.allocated", { hardCap: POLICY_CAPS[request.policy ?? config.policy].hardCap, ...budgetAllocationFor(request.policy ?? config.policy) }); const shape = analyzeTask({ ...request, contextRatio: ratio, preflight }); const route = routeTask(shape, { ...config, policy: request.policy ?? config.policy }, request.mode ?? config.mode, ratio); active.topology = route.topology; active.model = route.model; const proposalSource = request.mode && request.mode !== "auto" ? "root-proposal" : "deterministic-preflight"; await this.record(identity, "route.proposed", { topology: route.proposedTopology, fanout: route.proposedFanout, model: route.proposedModel, source: proposalSource, proposalRequired: route.needsProposal }); assertModelAllowed(config, route.model, shape.risk === "high"); await this.assertBudgetAdmission(route.model, route.topology, config, shape.reasonCodes.includes("execution:first-proven") ? 3 : undefined, active); await this.record(identity, "request.sanitized", { requestExport: `intent=${shape.intent};breadth=${shape.breadth};coupling=${shape.coupling};uncertainty=${shape.uncertainty};risk=${shape.risk};verifiability=${shape.verifiability}` }); await this.record(identity, "request.classified", { ...shape, topology: route.topology, policy: request.policy ?? config.policy }); active.ledger = { goal: requestHash, checklist: plannedChecklist(route.topology), acceptedFacts: [], decisions: [{ id: "route", description: `topology:${route.topology}` }], changedFiles: [], verificationState: "pending", activeRisks: [`risk:${shape.risk}`], rejectedHypotheses: [], remainingBudget: budgetFor(request.policy ?? config.policy, active.spentCredits, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget), configVersion: config.configVersion, }; await this.persistLedger(active); const action = contextAction(ratio); if (action === "compact") { await new Promise((resolve, reject) => context.compact({ customInstructions: "Preserve the persisted UltraPi Task Ledger and current task intent; remove redundant tool output before the next phase.", onComplete: () => resolve(), onError: reject, })); await this.record(identity, "session.compacted", { action }); } else if (action !== "normal") await this.record(identity, "context.threshold", { action }); await this.transition(runId, "ROUTED"); await this.record(identity, "route.validated", { proposedTopology: route.proposedTopology, finalTopology: route.topology, reasonCodes: route.reasonCodes, fanoutProposed: route.proposedFanout, fanoutFinal: route.fanout, modelProposed: route.proposedModel, modelFinal: route.model, overridden: route.overridden, ...(route.overrideReason ? { overrideReason: route.overrideReason } : {}), proposalRequired: route.needsProposal }); const recalled = await this.scoutMemorySlice(shape); let envelope = fitEnvelope({ ...this.envelope(runId, request, route.topology, config), ...(recalled.length ? { knownFacts: recalled.map((claim, index) => ({ id: `M:${runId}:memory:${index + 1}`, claim })) } : {}), ...(shape.reasonCodes.includes("execution:first-proven") ? { lens: "first-proven-result" } : {}) }, config.context.envelopeHardCapTokens); let delegatedWriterModel: string | undefined; active.envelope = envelope; const rootSelection = { stableModel: config.root.model, cheapModel: config.scout.model, expectedBenefit: expectedModelSavings(priceCatalog, config.root.model, config.scout.model) }; if (route.topology === "direct") { return { runId, rootSelection: { ...rootSelection, model: route.model, thinking: route.thinking }, topology: "direct", status: "direct", summary: request.acceptanceCommand ? "DIRECT selected. Apply only the scoped root change; the controller will run the declared acceptance check after this turn." : "DIRECT selected; root synthesis is pending and no executable acceptance check was supplied.", shape, reasonCodes: route.reasonCodes }; } if (route.topology === "scout" || route.topology === "swarm") { await this.transition(runId, route.topology === "scout" ? "SCOUTING" : "SWARMING"); const builtIn = route.topology === "scout" ? scoutShards(shape) : swarmShards(envelope.scope.paths); const declaredRoles = (await loadRoleDefinitions(context.cwd).catch(() => [])).filter((role) => role.kind === "scout"); const shards = [...builtIn, ...declaredRoles.filter((role) => !builtIn.some((shard) => shard.id === role.name)).map((role) => roleShard(role, envelope.scope.paths))]; const merged = await this.runScoutWaves(context, active, envelope, shards, config, request.policy ?? config.policy); await this.rememberScoutFacts(merged.facts, shape, request.privacyClass ?? "restricted").catch(() => undefined); envelope = fitEnvelope({ ...envelope, knownFacts: merged.facts.map((fact) => ({ id: fact.id, claim: fact.claim })) }, config.context.envelopeHardCapTokens); active.envelope = envelope; await this.persistLedger(active, { acceptedFacts: merged.facts }); await this.transition(runId, "IMPLEMENTING"); if (route.topology === "scout") { const arbitrationRequired = merged.contradictions.length > 0; const model = arbitrationRequired ? config.arbitration.model : config.root.model; const facts = merged.facts.filter((fact) => envelope.knownFacts.some((fitted) => fitted.id === fact.id)); active.model = model; if (arbitrationRequired) await this.record(identity, "escalation.triggered", { reason: "contradictory-scout-evidence", model, contradictions: merged.contradictions.length }); return { runId, rootSelection: { ...rootSelection, ...(arbitrationRequired ? { stableModel: model } : {}), model, thinking: arbitrationRequired ? config.arbitration.thinking : "high" }, topology: "scout", status: "scheduled", summary: `SCOUT completed with ${facts.length} fitted facts and ${merged.duplicates.length} duplicates; root synthesis remains pending.`, shape, reasonCodes: [...route.reasonCodes, ...(arbitrationRequired ? ["routing:material-disagreement"] : [])], facts }; } delegatedWriterModel = merged.contradictions.length > 0 ? config.arbitration.model : writerModelForLens(config, envelope.lens); active.model = delegatedWriterModel; if (merged.contradictions.length > 0) await this.record(identity, "escalation.triggered", { reason: "contradictory-scout-evidence", model: delegatedWriterModel, contradictions: merged.contradictions.length }); } if (route.topology === "warroom") { active.warRoomMaxRounds = warRoomRoundLimit(request.mode ?? config.mode, config.warRoom.maxRounds); active.warRoom = new WarRoom(active.warRoomMaxRounds, config.warRoom.maxMessagesPerMember, ["quality", "max"].includes(request.policy ?? config.policy) ? "quality" : "balanced"); active.warRoomPhase = "rounds"; active.warRoomRound = 1; active.warRoomBatches = new Map(); active.warRoomSpecialists = new Set(); active.warRoom.post({ type: "request", author: "controller", claim: "Bounded typed blackboard requested" }); for (const member of WAR_ROOM_MEMBERS) active.warRoom.subscribe(member, BOARD_TYPES, active.warRoom.events.length); await this.record(identity, "blackboard.event", { boardType: "request", members: WAR_ROOM_MEMBERS.length, rounds: active.warRoomMaxRounds }); await this.openWarRoomSignalBridge(active); } if (route.topology === "deep") await this.transition(runId, "DEEP_RUNNING"); else if (route.topology === "warroom") await this.transition(runId, "WAR_ROOM"); active.worktree = await createWorktree(context.cwd, join(this.paths.root, "worktrees"), runId, "writer"); active.writerScope = [...envelope.scope.paths]; await this.record(identity, "worktree.created", { nodeId: "writer", baseCommit: active.worktree.baseCommit }); const flow = route.topology === "warroom" ? warRoomRoundFlow(envelope, config, active.worktree.path, 1, {}, issueWarRoomRoundCredentials(active)) : flowFor(route.topology, envelope, config, active.worktree.path, delegatedWriterModel); const assignments = plannedAgents(flow); const execution = await this.recordPlannedAgents(active, assignments, "execution", route.topology === "warroom" ? "warroom-round" : "standard"); const flowModels = assignments.flatMap((assignment) => assignment.model ? [assignment.model] : []); this.captureModelCosts(active, flowModels); const runtimeConfig = await this.withDelegatedBudget(active, config, flowModels); const delegatedTopology = route.topology as "swarm" | "deep" | "warroom"; const piAgentsRunId = await this.startDelegatedRun(active, () => this.agents.start(delegatedTopology, envelope, runtimeConfig, active.worktree!.path, delegatedWriterModel, flow), (startedRunId) => { this.bindDelegatedExecution(active, startedRunId, execution); if (route.topology === "warroom") registerWarRoomRoundParticipants(active, startedRunId); }); await this.record(identity, "workflow.started", { executionId: execution.executionId, piAgentsRunId, topology: route.topology, budgets: runtimeConfig.piAgentsBudgets }); await this.replayPendingPiAgentsEvents(piAgentsRunId); return { runId, topology: route.topology, status: "scheduled", summary: `${route.topology.toUpperCase()} scheduled as ${piAgentsRunId}. Completion is pending runtime verification.`, shape, reasonCodes: route.reasonCodes, piAgentsRunId }; } catch (error) { const active = this.active.get(runId); const committed = active ? await this.lifecycleSettlementCommitted(active) : false; if (active && !committed && !(error instanceof DelegatedStartAbortedError)) { await this.recordSafeFailure(active, "workflow.failed", "root", error instanceof Error ? error.message : String(error)).catch(() => undefined); await this.transition(runId, "FAILED").catch(() => undefined); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "FAILED", reason: "dispatch-failed" }).catch(() => undefined); } throw error; } } async stop(runId: string): Promise { const active = this.active.get(runId); if (!active) throw new Error(`Unknown UltraPi run ${runId}`); if (isTerminal(active.state)) { const failures = await this.stopDelegatedRuns(active); if (failures.length) throw new AggregateError(failures, `Failed to stop every delegated run: ${failures.map((error) => error instanceof Error ? error.message : String(error)).join("; ")}`); return; } if (active.terminalSettlement) { await active.terminalSettlement; if (isTerminal(active.state)) return; } active.terminalIntent = "CANCELLED"; active.manualStopPending = true; active.piAgentsEventGate = true; let settlement!: Promise; settlement = (async () => { const failures = await this.stopDelegatedRuns(active); if (failures.length) { active.manualStopPending = false; active.terminalIntent = undefined; if (active.terminalSettlement === settlement) active.terminalSettlement = undefined; try { await this.drainPiAgentsEvents(active, true, false); } catch (error) { failures.push(error); } finally { active.piAgentsEventGate = false; } throw new AggregateError(failures, `Failed to stop every delegated run: ${failures.map((error) => error instanceof Error ? error.message : String(error)).join("; ")}`); } active.manualStopPending = false; // Clearing the queue outright discarded queued node_completed events, and those carry // per-node usage that had not been priced yet -- the failure path immediately above // already drains for exactly this reason. Draining here too means a stopped run is // charged for the work it actually did instead of going free. try { await this.drainPiAgentsEvents(active, true, false); } catch { /* stopping must not fail on accounting */ } active.piAgentsEventQueue = []; active.piAgentsEventGate = false; if (active.terminalSettlement === settlement) active.terminalSettlement = undefined; if (isTerminal(active.state)) return; await this.transition(runId, "CANCELLED", true); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "CANCELLED", reason: "stopped-by-user" }); })(); active.terminalSettlement = settlement; try { await settlement; } finally { if (active.terminalSettlement === settlement) active.terminalSettlement = undefined; } } async blockRootModelSelection(runId: string): Promise { const active = this.active.get(runId); if (!active || isTerminal(active.state)) return; await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "root-model-unavailable" }); } async steer(runId: string, message: string): Promise { const active = this.active.get(runId); if (!active?.piAgentsRunId) throw new Error("Run has no active pi-agents workflow"); await this.agents.steer(active.piAgentsRunId, message); await this.record(active.identity, "agent.steered", { piAgentsRunId: active.piAgentsRunId, steeringHash: await this.traceHash(message, "STEERING") }); } private async openWarRoomSignalBridge(active: ActiveRun): Promise { if (active.warRoomSignalServer || !active.envelope) return; const signalPath = join(tmpdir(), `u${randomUUID().replaceAll("-", "").slice(0, 16)}.sock`); const server = createServer((socket) => { let buffer = ""; socket.setEncoding("utf8"); socket.on("data", (chunk: string) => { buffer += chunk; if (buffer.length > 32_000) { socket.destroy(); return; } const newline = buffer.indexOf("\n"); if (newline < 0) return; const line = buffer.slice(0, newline); buffer = buffer.slice(newline + 1); void (async () => { try { const raw = JSON.parse(line); const parsed = parseWarRoomSignal(raw); await active.warRoomSignalBindings?.get(parsed.credential)?.ready; const accepted = acceptWarRoomSignal(active, parsed); socket.end(`${JSON.stringify({ ok: true, cursor: accepted.record.cursor })}\n`, () => { this.scheduleWarRoomSignalDelivery(active, accepted); }); } catch (error) { socket.end(`${JSON.stringify({ ok: false, error: error instanceof Error ? error.message : String(error) })}\n`); } })(); }); }); await new Promise((resolvePromise, reject) => { const onError = (error: Error) => { server.off("listening", onListening); reject(error); }; const onListening = () => { server.off("error", onError); resolvePromise(); }; server.once("error", onError); server.once("listening", onListening); server.listen(signalPath); }); await chmod(signalPath, 0o600); active.warRoomSignalPath = signalPath; active.warRoomSignalServer = server; } private reserveWarRoomSpecialists(active: ActiveRun, actions: readonly CoordinationAction[]): void { for (const action of actions) { if (action.type !== "specialist-request" || active.warRoomSpecialistReservations?.has(action)) continue; const previous = active.warRoomSpecialistAdmission ?? Promise.resolve(); const closed = active.warRoomFinalizing || active.warRoomPhase !== "rounds"; const admission = previous.catch(() => undefined).then(async () => { if (closed) { await this.record(active.identity, "agent.spawn.rejected", { nodeId: "warroom-specialist", reason: "warroom-finalizing", gap: action.gap }); return; } await this.spawnWarRoomSpecialist(active, action, true); }); let tracked!: Promise; tracked = admission.finally(() => { if (active.warRoomSpecialistAdmission === tracked) active.warRoomSpecialistAdmission = undefined; }); active.warRoomSpecialistAdmission = tracked; (active.warRoomSpecialistReservations ??= new WeakMap()).set(action, tracked); } } private async spawnWarRoomSpecialist(active: ActiveRun, action: Extract, reserved = false): Promise { if (!reserved) { this.reserveWarRoomSpecialists(active, [action]); await active.warRoomSpecialistReservations?.get(action); return; } const config = active.config ?? this.getConfig(); if (!active.envelope || active.warRoomPhase !== "rounds" || await this.lifecycleSettlementCommitted(active)) { await this.record(active.identity, "agent.spawn.rejected", { nodeId: "warroom-specialist", reason: "warroom-closed", gap: action.gap }); return; } try { // Specialists have their own ceiling rather than borrowing the delegated parallelism budget, // which left exactly one slot beside the three members at its default. if ((active.warRoomSpecialists?.size ?? 0) >= config.warRoom.maxSpecialists || liveWarRoomParticipantCount(active) >= config.piAgentsBudgets.maxAgents) { await this.record(active.identity, "agent.spawn.rejected", { nodeId: "warroom-specialist", reason: "max-specialists", gap: action.gap }); return; } const specialistNumber = (active.warRoomSpecialists?.size ?? 0) + 1; const author = `specialist-${specialistNumber}`; const binding = issueWarRoomCredential(active, author, "$"); const flow = warRoomSpecialistFlow(active.envelope, config, active.worktree?.path ?? active.cwd, author, action.gap, action.events, binding); const assignments = plannedAgents(flow); try { await this.assertSpawnsAllowed(active, assignments, "escalation"); const execution = await this.recordPlannedAgents(active, assignments, "escalation", "warroom-specialist"); for (const assignment of assignments) execution.boardAuthors.set(executionNodeId(execution, assignment.nodeId), author); const models = assignments.flatMap((assignment) => assignment.model ? [assignment.model] : []); this.captureModelCosts(active, models); const runtimeConfig = await this.withDelegatedBudget(active, config, models, "escalation"); const piAgentsRunId = await this.startDelegatedRun(active, () => this.agents.start("warroom", active.envelope!, runtimeConfig, active.worktree?.path ?? active.cwd, undefined, flow), (startedRunId) => { this.bindDelegatedExecution(active, startedRunId, execution); (active.warRoomSpecialists ??= new Set()).add(author); registerWarRoomSpecialist(active, author, startedRunId, assignments[0]!.nodeId); }); await this.record(active.identity, "workflow.started", { executionId: execution.executionId, piAgentsRunId, topology: "warroom", phase: "warroom-specialist", specialist: author, gap: action.gap, relevantCursors: action.events.map((event) => event.cursor), budgets: runtimeConfig.piAgentsBudgets }); void this.replayPendingPiAgentsEvents(piAgentsRunId, false).catch(() => undefined); } catch (error) { if (!active.warRoomParticipants?.has(author)) retireWarRoomCredential(active, author); if (await this.lifecycleSettlementCommitted(active) || error instanceof DelegatedStartAbortedError) return; await this.record(active.identity, "agent.spawn.rejected", { nodeId: author, reason: "controller-governor", gap: action.gap, errorClass: error instanceof Error ? error.message : String(error) }); } } catch (error) { await this.record(active.identity, "agent.spawn.rejected", { nodeId: "warroom-specialist", reason: "controller-governor", gap: action.gap, errorClass: error instanceof Error ? error.message : String(error) }); } } private async executeWarRoomActions(active: ActiveRun, actions: readonly CoordinationAction[]): Promise { this.reserveWarRoomSpecialists(active, actions); if (await this.lifecycleSettlementCommitted(active)) return; for (const action of actions) { if (action.type === "steer") { const participant = active.warRoomParticipants?.get(action.target); if (!participant || participant.state !== "live") { await this.record(active.identity, "agent.steer.rejected", { nodeId: action.target, cursor: action.cursor, reason: participant ? `target-${participant.state}` : "target-missing" }); continue; } try { await this.agents.steer(participant.runId, action.message, participant.instance); await this.record(active.identity, "agent.steered", { piAgentsRunId: participant.runId, instance: participant.instance, cursor: action.cursor, steeringHash: await this.traceHash(action.message, "STEERING") }); } catch { await this.record(active.identity, "agent.steer.rejected", { nodeId: action.target, piAgentsRunId: participant.runId, instance: participant.instance, cursor: action.cursor, reason: "target-not-live" }); } continue; } await this.record(active.identity, "specialist.requested", { requester: action.requester, gap: action.gap, relevantCursors: action.events.map((event) => event.cursor), decision: "controller" }); await active.warRoomSpecialistReservations?.get(action); } } private scheduleWarRoomSignalDelivery(active: ActiveRun, accepted: AcceptedWarRoomSignal): void { if (accepted.duplicate) return; let tracked!: Promise; tracked = new Promise((resolvePromise) => setImmediate(resolvePromise)).then(async () => { const content = JSON.stringify(accepted.event); await this.storeRawTrace(active, "result", accepted.event.author, "blackboard", content); await this.record(active.identity, "blackboard.event", { boardType: accepted.event.type, author: accepted.event.author, cursor: accepted.record.cursor, round: accepted.record.round, payloadChars: content.length, payloadEstimatedTokens: accepted.record.estimatedTokens, signal: true }); await this.executeWarRoomActions(active, accepted.actions); }).catch(async () => { await this.record(active.identity, "agent.signal.failed", { nodeId: accepted.event.author, cursor: accepted.record.cursor, reason: "delivery-failed" }).catch(() => undefined); }).finally(() => { active.warRoomSignalDeliveries?.delete(tracked); }); (active.warRoomSignalDeliveries ??= new Set()).add(tracked); } async runs() { return this.status().activeRuns; } async trace(runId: string, raw = false): Promise { await this.ready(); if (!raw) return formatTrace((await new EventStore(this.paths.events).all()).filter((entry) => entry.runId === runId)); const records = await readRawTrace(this.paths.rawVault, this.paths.vaultKey, runId); return records.length ? records.map((record) => `[${record.timestamp}] ${record.kind} ${record.fromNodeId} → ${record.toNodeId}\n${record.content}`).join("\n\n") : "No encrypted raw trace for this run."; } async recoveryDecisions(): Promise { await this.ready(); return await discoverInterruptedRuns(this.paths.events, this.paths.ledgers); } async recover(context: ExtensionContext, oldRunId: string): Promise { await this.ready(); const decision = (await this.recoveryDecisions()).find((entry) => entry.runId === oldRunId); if (!decision) throw new Error(`No unfinished UltraPi run ${oldRunId}`); if (decision.kind === "blocked") throw new Error(`Recovery refused for ${oldRunId}: ${decision.reason}`); if (decision.kind !== "recoverable") throw new Error(`UltraPi run ${oldRunId} is already terminal`); let config: UltraConfig; try { config = await loadConfigVersion(this.paths, decision.ledger.configVersion); } catch { throw new Error(`Recovery refused for ${oldRunId}: config-provenance-mismatch`); } await this.admitAutoExecution(config); const configuredProfile = await profileForAgentDir(this.agentDir); if (config.policyVersion !== decision.sourcePolicyVersion || (decision.sourceConfigHash && configFingerprint(config) !== decision.sourceConfigHash) || (configuredProfile && configuredProfile !== config.profile)) { throw new Error(`Recovery refused for ${oldRunId}: config-provenance-mismatch`); } const records = await readRawTrace(this.paths.rawVault, this.paths.vaultKey, oldRunId); const task = [...records].reverse().find((record) => record.kind === "task" && record.fromNodeId === "user" && record.toNodeId === "root")?.content; const acceptanceCommand = [...records].reverse().find((record) => record.kind === "verification" && record.fromNodeId === "controller" && record.toNodeId === "verifier")?.content; const sourceWorkspace = [...records].reverse().find((record) => record.kind === "workspace" && record.fromNodeId === "controller" && record.toNodeId === "root")?.content; const currentWorkspace = await realpath(context.cwd); if (sourceWorkspace && sourceWorkspace !== currentWorkspace) throw new Error(`Recovery refused for ${oldRunId}: workspace-mismatch`); const continuation = recoveryContinuation(decision, sourceWorkspace ? task : undefined, sourceWorkspace ? acceptanceCommand : undefined); if (continuation.kind === "blocked") throw new Error(`Recovery refused for ${oldRunId}: ${continuation.reason}`); const metadata = materializeRecovery(decision, { runId: randomUUID(), sessionId: await this.traceHash(context.sessionManager.getSessionId(), "SESSION") }); const objective = continuation.kind === "scout" ? continuation.task : `Read-only recovery checkpoint ${oldRunId}`; const shape = analyzeTask({ objective, mode: "scout", contextRatio: (context.getContextUsage()?.percent ?? 0) / 100 }); const active: ActiveRun = { state: metadata.state, identity: metadata.identity, topology: "scout", cwd: currentWorkspace, ...(continuation.kind === "scout" ? { acceptanceCommand: continuation.acceptanceCommand } : {}), informational: ["answer", "investigate", "review"].includes(shape.intent), repairAttempts: 0, fingerprints: new Map(), model: config.scout.model, spentCredits: metadata.ledger.remainingBudget.spentCredits, spentUsd: 0, policy: decision.sourcePolicy, config, modelRegistry: context.modelRegistry, context, ledger: { ...metadata.ledger, changedFiles: [], verificationState: "pending", decisions: [...metadata.ledger.decisions, { id: "recovery", description: `recovered-from:${oldRunId}` }] }, ledgerRedactions: continuation.kind === "scout" ? [continuation.task, continuation.acceptanceCommand, currentWorkspace] : [], }; if (continuation.kind === "scout") { const admission = { ...active, state: "PREFLIGHT" as const }; this.active.set(metadata.identity.runId, admission); try { assertModelAllowed(config, config.scout.model); await this.assertBudgetAdmission(config.scout.model, "scout", { ...config, policy: decision.sourcePolicy }, undefined, admission); active.reservedCredits = admission.reservedCredits; } catch (error) { this.active.delete(metadata.identity.runId); throw error; } } this.active.set(metadata.identity.runId, active); try { const availableModels = context.modelRegistry?.getAvailable?.().map((model) => `${model.provider}/${model.id}`).sort() ?? []; const priceCatalog = await loadPriceCatalog(this.paths); await this.record({ sessionId: decision.sourceSessionId, taskId: decision.sourceTaskId, runId: oldRunId, ...(decision.sourceExperimentId ? { experimentId: decision.sourceExperimentId } : {}) }, "recovery.stopped", { continuedAsRunId: metadata.identity.runId, configVersion: config.configVersion, policyVersion: config.policyVersion, profile: config.profile }); await this.record(active.identity, "recovery.created", { recoveredFromRunId: oldRunId, previousState: metadata.previousState, sourceTopology: metadata.topology, mode: continuation.kind }); await this.record(active.identity, "request.received", { requestHash: active.ledger!.goal, privacyClass: "restricted", recoveredFromRunId: oldRunId, ...requestProvenance(config, availableModels, priceCatalog) }); if (continuation.kind === "scout") { await this.storeRawTrace(active, "task", "user", "root", continuation.task); await this.storeRawTrace(active, "verification", "controller", "verifier", continuation.acceptanceCommand); await this.storeRawTrace(active, "workspace", "controller", "root", currentWorkspace); } await this.persistLedger(active); await this.transition(active.identity.runId, "PREFLIGHT"); await this.record(active.identity, "request.classified", { topology: "scout", policy: decision.sourcePolicy, recoveredFromRunId: oldRunId }); await this.transition(active.identity.runId, "ROUTED"); let envelope = fitEnvelope({ ...this.envelope(active.identity.runId, { objective, mode: "scout", paths: shape.mentionedPaths }, "scout", config), knownFacts: metadata.ledger.acceptedFacts.map((fact) => ({ id: fact.id, claim: fact.claim })) }, config.context.envelopeHardCapTokens); active.envelope = envelope; await this.transition(active.identity.runId, "SCOUTING"); let acceptedFacts: AgentFact[]; if (continuation.kind === "ledger-handoff") { acceptedFacts = continuation.facts; await this.recordResultHandoff(active, "ledger", { schemaVersion: 1, status: "partial", summary: "Persisted read-only checkpoint", facts: acceptedFacts, hypotheses: [], unknowns: [], dependencies: [] }); } else { const merged = await this.runScoutWaves(context, active, envelope, scoutShards(shape), config, decision.sourcePolicy); acceptedFacts = mergeFacts([{ schemaVersion: 1, status: "complete", summary: "recovery", facts: [...metadata.ledger.acceptedFacts, ...merged.facts], hypotheses: [], unknowns: [], dependencies: [] }]).facts; } envelope = fitEnvelope({ ...envelope, knownFacts: acceptedFacts.map((fact) => ({ id: fact.id, claim: fact.claim })) }, config.context.envelopeHardCapTokens); active.envelope = envelope; await this.persistLedger(active, { acceptedFacts }); await this.transition(active.identity.runId, "IMPLEMENTING"); if (continuation.kind === "scout") { const facts = envelope.knownFacts.map((fact) => `- [${fact.id}] ${fact.claim}`).join("\n") || "- none"; return { runId: active.identity.runId, recoveredFromRunId: oldRunId, rootHandoff: `Continue recovered UltraPi run ${active.identity.runId}. Do not call ultra_dispatch again. Perform the task directly with the normal root tools. The controller restored the encrypted acceptance check and will run it when this turn ends.\n\nTask:\n${continuation.task}\n\nAccepted scout facts:\n${facts}`, topology: "scout", status: "scheduled", summary: `Recovered ${oldRunId}; root implementation and verification are pending.`, shape, reasonCodes: ["recovery:continued", "recovery:source-config"], facts: acceptedFacts, }; } await this.transition(active.identity.runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "recovery-read-only-handoff", recoveredFromRunId: oldRunId }); return { runId: active.identity.runId, recoveredFromRunId: oldRunId, topology: "scout", status: "blocked", summary: `Recovered ${oldRunId} as a read-only SCOUT handoff; mutation and verification were not replayed.`, shape, reasonCodes: ["recovery:read-only", "recovery:no-verification-replay"], facts: acceptedFacts }; } catch (error) { if (!isTerminal(active.state)) { await this.recordSafeFailure(active, "recovery.failed", "controller", error instanceof Error ? error.message : String(error)).catch(() => undefined); await this.transition(active.identity.runId, "FAILED").catch(() => undefined); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "FAILED", recoveredFromRunId: oldRunId }).catch(() => undefined); } throw error; } } private async assertConfigActivation(config: UltraConfig): Promise { await assertAutoModeProfile(this.agentDir, config.mode); const profile = await profileForAgentDir(this.agentDir); if (profile && profile !== config.profile) throw new Error(`Profile config mismatch: expected ${profile}, found ${config.profile}`); } private async activateConfig(previous: UltraConfig, next: UltraConfig, changeReason: string, budgetUpdated = false): Promise { await this.assertConfigActivation(next); this.config = next; this.telemetry?.close(); this.telemetry = new TelemetryCollector(this.paths.events, this.paths.database, this.paths.hmacKey, next, next.profile); const id = randomUUID(); const identity = { sessionId: "system", taskId: `config:${id}`, runId: `config:${id}` }; await this.record(identity, "config.changed", { version: next.configVersion, createdAt: new Date().toISOString(), configJson: next, configHash: configFingerprint(next), parentVersion: previous.configVersion, changeReason }); if (budgetUpdated) await this.record(identity, "budget.updated", { ...next.budgets }); return next; } async setMode(mode: Mode): Promise { const current = await this.ready(); await this.assertConfigActivation({ ...current, mode }); return this.activateConfig(current, await updateConfig(this.paths, current, (config) => ({ ...config, mode }), `mode:${mode}`), `mode:${mode}`); } async setPolicy(policy: Policy): Promise { const current = await this.ready(); await this.assertConfigActivation(current); return this.activateConfig(current, await updateConfig(this.paths, current, (config) => ({ ...config, policy, scout: { ...config.scout, initial: POLICY_CAPS[policy].initialScouts } }), `policy:${policy}`), `policy:${policy}`); } async setBudget(weeklyCreditBudget: number | undefined, dailyCreditBudget: number | undefined, acknowledged: boolean): Promise { if ((!weeklyCreditBudget || weeklyCreditBudget <= 0) && !acknowledged) throw new Error("Set a positive weekly budget or explicitly acknowledge one-task force mode"); const current = await this.ready(); await this.assertConfigActivation(current); return this.activateConfig(current, await updateConfig(this.paths, current, (config) => ({ ...config, budgets: { weeklyCreditBudget, dailyCreditBudget, acknowledged } }), "budget:updated"), "budget:updated", true); } async rollbackConfig(version: string): Promise { const current = await this.ready(); const target = await loadConfigVersion(this.paths, version); await this.assertConfigActivation(target); return this.activateConfig(current, await rollbackConfigVersion(this.paths, version), `rollback:${version}`); } providerResponse(status: number, latencyMs: number): ProviderHealth { const active = [...this.active.values()].find((run) => !isTerminal(run.state)); const config = active?.config ?? this.getConfig(); const ceiling = Math.min(config.scout.maxParallel, config.piAgentsBudgets.maxParallelism, POLICY_CAPS[active?.policy ?? config.policy].maxParallel); this.health = recordProviderResponse(this.health, status, latencyMs, ceiling); return structuredClone(this.health); } async recordSessionLifecycle(sessionId: string, eventType: "session.started" | "session.ended", reason: string): Promise { const safeId = await this.traceHash(sessionId, "SESSION"); await this.record({ sessionId: safeId, taskId: safeId, runId: safeId }, eventType, { reason }); } async recordProviderHealth(runId: string, status: number, latencyMs: number, source: "root" | "delegated", health: ProviderHealth = this.health): Promise { const active = this.active.get(runId); if (!active) return; await this.record(active.identity, "provider.health.changed", { source, status, latencyMs, state: health.state, concurrency: health.concurrency, recent429: health.recent429, recent5xx: health.recent5xx, ...(health.cooldownUntil ? { cooldownUntil: health.cooldownUntil } : {}) }); } async recordRootModelStarted(runId: string, model: string, metadata: { thinkingLevel: string; latencyMs: number; turnIndex: number; callIndex: number }): Promise { const active = this.active.get(runId); if (!active) return; await this.record(active.identity, "model.call.started", { nodeId: "root", model, provider: model.split("/", 1)[0] ?? "unknown", thinkingLevel: metadata.thinkingLevel, totalTokens: 0, latencyMs: metadata.latencyMs, turnIndex: metadata.turnIndex, callIndex: metadata.callIndex, role: "root" }); } async recordRootModelFailed(runId: string, model: string, metadata: { thinkingLevel: string; latencyMs: number; turnIndex: number; callIndex: number; errorClass: string; observedModel?: string; observedProvider?: string }, usage?: { input?: number; output?: number; cacheRead?: number; totalTokens?: number; cost?: { total?: number } }): Promise { const active = this.active.get(runId); if (!active) return; const terminalPresentation = isTerminal(active.state) && active.presentationQueued && !active.presentationUsageRecorded; if (isTerminal(active.state) && !terminalPresentation) return; if (terminalPresentation) active.presentationUsageRecorded = true; await this.recordUsage(active, { input: usage?.input ?? 0, output: usage?.output ?? 0, cacheRead: usage?.cacheRead ?? 0, cost: usage?.cost?.total }, model, "root", undefined, terminalPresentation, { ...metadata, totalTokens: usage?.totalTokens ?? (usage?.input ?? 0) + (usage?.output ?? 0) }); } async recordRootModelChanged(runId: string, previousModel: string, model: string, source: "set" | "cycle" | "restore"): Promise { const active = this.active.get(runId); if (!active || isTerminal(active.state) || previousModel === model) return; await this.record(active.identity, "model.changed", { previousModel, model, source }); } async recordRootSelectionCompleted(runId: string, selection: NonNullable, actualModel: string | undefined, outcome: "selected" | "switched" | "preserved" | "blocked"): Promise { const active = this.active.get(runId); if (!active) return; if (actualModel) active.model = actualModel; await this.record(active.identity, "root.selection.completed", { plannedModel: selection.model, actualModel: actualModel ?? "unselected", plannedThinkingLevel: selection.thinking, expectedBenefit: selection.expectedBenefit, outcome }); } async recordRootTool(runId: string, eventType: "tool.started" | "tool.completed" | "tool.failed", toolCallId: string, toolName: string, latencyMs: number, nodeId?: string, role?: string): Promise { const active = this.active.get(runId); if (!active) return; const toolIdentity = SAFE_ROOT_TOOLS.has(toolName) ? toolName : await this.traceHash(toolName, "TOOL"); await this.record(active.identity, eventType, { toolIdentity, toolCallHash: await this.traceHash(toolCallId, "TOOL_CALL"), latencyMs, success: eventType !== "tool.failed", ...(nodeId ? { nodeId } : {}), ...(role ? { role } : {}) }); } budgetState() { const active = [...this.active.values()].find((run) => !isTerminal(run.state)); const config = active?.config ?? this.getConfig(); return budgetFor(active?.policy ?? config.policy, active?.spentCredits ?? 0, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget); } async doctor(context: ExtensionContext): Promise { return await runDoctor({ paths: this.paths, agentDir: this.agentDir, ready: () => this.ready() }, context); } close(): void { for (const active of this.active.values()) void closeWarRoomSignalBridge(active); this.telemetry?.close(); } async onPiAgentsEvent(event: RunEvent): Promise { if (event.type === "run_created") return; const queued = { sequence: ++this.piAgentsEventSequence, event }; const delegated = this.delegatedRun(event.runId); if (!delegated) { this.pendingPiAgentsEvents.set(event.runId, [...(this.pendingPiAgentsEvents.get(event.runId) ?? []), queued]); return; } (delegated.active.piAgentsEventQueue ??= []).push(queued); if (delegated.active.piAgentsEventDrain || delegated.active.piAgentsEventGate) return; await this.drainPiAgentsEvents(delegated.active); } private async drainPiAgentsEvents(active: ActiveRun, ignoreGate = false, waitForExisting = true): Promise { const existing = active.piAgentsEventDrain; if (existing) { if (waitForExisting) await existing; return; } if ((active.piAgentsEventGate && !ignoreGate) || (active.unboundDelegatedStarts ?? 0) > 0) return; let drain!: Promise; drain = (async () => { while (active.piAgentsEventQueue?.length && (!active.piAgentsEventGate || ignoreGate) && (active.unboundDelegatedStarts ?? 0) === 0) { const queued = active.piAgentsEventQueue[0]!; const delegated = this.delegatedRun(queued.event.runId); if (!delegated || delegated.active !== active) return; await this.handlePiAgentsEvent(queued.event, queued.sequence, delegated); const index = active.piAgentsEventQueue.indexOf(queued); if (index >= 0) active.piAgentsEventQueue.splice(index, 1); } })(); active.piAgentsEventDrain = drain; try { await drain; } finally { if (active.piAgentsEventDrain === drain) active.piAgentsEventDrain = undefined; } } private async piAgentsEventEffect(active: ActiveRun, sequence: number, effect: string, run: () => Promise): Promise { const key = `${sequence}:${effect}`; if (active.piAgentsEventEffects?.has(key)) return; await run(); (active.piAgentsEventEffects ??= new Set()).add(key); } private async handlePiAgentsEvent(event: DelegatedRunEvent, sequence: number, delegated: { runId: string; active: ActiveRun; execution?: DelegatedExecution }): Promise { const { active, execution } = delegated; const runtimeFields = (instance: string) => execution ? { executionId: execution.executionId, piAgentsRunId: event.runId, piAgentsNodeInstance: instance } : {}; if (event.type === "run_completed") await this.completePiAgentsRun(event.runId, event.status, event.usage, event.error); else if (event.type === "node_started") { setWarRoomParticipantState(active, event.runId, event.instance, "live"); const nodeId = executionNodeId(execution, event.instance); (execution?.nodeKinds ?? (active.nodeKinds ??= new Map())).set(nodeId, event.kind); await this.piAgentsEventEffect(active, sequence, "workflow.node.started", async () => { await this.record(active.identity, "workflow.node.started", { nodeId, ...runtimeFields(event.instance), kind: event.kind, topology: active.topology }); }); if (event.kind === "agent" || event.kind === "reduce") { await this.record(active.identity, "agent.spawned", { nodeId, ...runtimeFields(event.instance), kind: event.kind, topology: active.topology }); await this.recordReceivedHandoffs(active, nodeId); } } else if (event.type === "node_completed") { setWarRoomParticipantState(active, event.runId, event.instance, "completed"); const nodeId = executionNodeId(execution, event.instance); const kind = (execution?.nodeKinds ?? active.nodeKinds)?.get(nodeId); await this.piAgentsEventEffect(active, sequence, "workflow.node.completed", async () => { await this.record(active.identity, "workflow.node.completed", { nodeId, ...runtimeFields(event.instance), ...(kind ? { kind } : {}) }); }); const plannedModel = (execution?.plannedModels ?? active.plannedModels)?.get(nodeId); const usageKey = `${event.runId}:${nodeId}`; const resultRole = kind === "reduce" ? "reducer" : (execution?.resultRoles ?? active.resultRoles)?.get(nodeId); const boardAuthor = active.warRoomPhase === "rounds" ? execution?.boardAuthors.get(nodeId) ?? event.instance.match(/\.branches\.(lead|flow|invariants)$/)?.[1] : undefined; const expectsResult = (kind === "agent" || kind === "reduce" || resultRole) && event.value !== undefined; let boardBatch: BoardEvent[] | undefined; let agentResult: AgentResult | undefined; let resultError: unknown; try { if (boardAuthor && event.value !== undefined) boardBatch = parseBoardBatch(event.value, boardAuthor); else if (expectsResult) { agentResult = active.envelope ? parseAgentResult(event.value, { ...active.envelope, trace: { ...active.envelope.trace, nodeId: event.instance } }, false, resultRole ?? "scout") : event.value as AgentResult; if (agentResult.attributionManifest) { const factsByAgent = new Map(active.agentFacts); if (active.warRoomPhase === "writer") factsByAgent.set("warroom-board", warRoomWriterAttributionFacts(active)); factsByAgent.set(nodeId, agentResult.facts); validateAttributionManifest(agentResult.attributionManifest, [...factsByAgent.values()].flat()); } if (resultRole === "writer" && active.envelope && active.worktree) await assertDeclaredArtifacts(active.envelope, active.worktree.path); } } catch (error) { resultError = error; } const pendingAccounting = active.usageAccounting?.get(usageKey); if (event.usage && plannedModel && (!active.accountedDelegatedNodes?.has(usageKey) || (pendingAccounting && !pendingAccounting.complete))) { const metadata = resultError ? { thinkingLevel: "unknown", totalTokens: event.usage.input + event.usage.output, latencyMs: 0, turnIndex: 0, callIndex: 0, errorClass: "result-contract" } : undefined; (active.accountedDelegatedNodes ??= new Set()).add(usageKey); const spentCredits = active.spentCredits; const spentUsd = active.spentUsd; try { const credits = await this.recordUsage(active, event.usage, plannedModel, nodeId, event.runId, true, metadata, resultRole ?? (kind === "reduce" ? "reducer" : undefined), usageKey); const accounting = active.usageAccounting?.get(usageKey); if (!accounting?.agentCreditsApplied) { (active.agentCredits ??= new Map()).set(nodeId, (active.agentCredits?.get(nodeId) ?? 0) + credits); if (accounting) accounting.agentCreditsApplied = true; } } catch (error) { const charged = active.spentCredits - spentCredits; const accounting = active.usageAccounting?.get(usageKey); if (charged > 0 && !accounting?.agentCreditsApplied) { (active.agentCredits ??= new Map()).set(nodeId, (active.agentCredits?.get(nodeId) ?? 0) + charged); if (accounting) accounting.agentCreditsApplied = true; } if (charged === 0 && active.spentUsd === spentUsd) active.accountedDelegatedNodes.delete(usageKey); throw error; } } if (resultError) { if (boardAuthor || (resultRole && resultRole !== "scout")) active.protocolFailure = true; await this.recordGovernanceViolation(active, resultError, nodeId, resultRole ?? "scout"); await this.recordSafeFailure(active, "agent.failed", nodeId, resultError instanceof Error ? resultError.message : String(resultError), undefined, runtimeFields(event.instance)).catch(() => undefined); } else if (boardBatch) { const batchKey = execution?.phase === "warroom-specialist" ? `${event.runId}:${event.instance}` : event.instance; (active.warRoomBatches ??= new Map()).set(batchKey, boardBatch); await this.storeRawTrace(active, "result", nodeId, "blackboard", JSON.stringify(boardBatch)); await this.record(active.identity, "agent.completed", { nodeId, ...runtimeFields(event.instance), boardEvents: boardBatch.length, costCredits: active.agentCredits?.get(nodeId) ?? 0 }); if (execution?.phase === "warroom-specialist") { try { const room = active.warRoom; if (!room) throw new Error("War Room specialist state is unavailable"); for (const boardEvent of boardBatch) { assertWarRoomTarget(active, boardEvent); const actions = room.accept(boardEvent); this.reserveWarRoomSpecialists(active, actions); const record = room.events.at(-1)!; await this.record(active.identity, "blackboard.event", { boardType: boardEvent.type, author: boardEvent.author, cursor: record.cursor, round: record.round, specialist: true }); await this.executeWarRoomActions(active, actions); } (active.warRoomAcceptedSpecialistRuns ??= new Set()).add(event.runId); } catch (error) { active.protocolFailure = true; await this.recordSafeFailure(active, "agent.failed", nodeId, error instanceof Error ? error.message : String(error), undefined, runtimeFields(event.instance)).catch(() => undefined); } } } else if (agentResult) { try { const factsTotal = agentResult.facts.length; await this.recordEmptyResult(active, nodeId, resultRole ?? "scout", agentResult); (active.agentFacts ??= new Map()).set(nodeId, agentResult.facts); await this.recordAttribution(active, nodeId, agentResult); await this.recordResultHandoff(active, nodeId, agentResult); await this.record(active.identity, "agent.completed", { nodeId, ...runtimeFields(event.instance), factsTotal, costCredits: active.agentCredits?.get(nodeId) ?? 0 }); } catch (error) { if (resultRole && resultRole !== "scout") active.protocolFailure = true; await this.recordSafeFailure(active, "agent.failed", nodeId, error instanceof Error ? error.message : String(error), undefined, runtimeFields(event.instance)).catch(() => undefined); } } else if (kind === "agent" || kind === "reduce") { await this.record(active.identity, "agent.completed", { nodeId, ...runtimeFields(event.instance), costCredits: active.agentCredits?.get(nodeId) ?? 0 }); } } else if (event.type === "node_cancelled") { setWarRoomParticipantState(active, event.runId, event.instance, "completed"); const nodeId = executionNodeId(execution, event.instance); await this.piAgentsEventEffect(active, sequence, "agent.cancelled", async () => { await this.record(active.identity, "agent.cancelled", { nodeId, ...runtimeFields(event.instance), reason: event.reason }); }); } else if (event.type === "node_failed") { setWarRoomParticipantState(active, event.runId, event.instance, "completed"); if (execution) execution.usageIncomplete = true; else active.delegatedUsageIncomplete = true; const status = providerFailureStatus(event.error); if (status) { const health = this.providerResponse(status, 0); await this.recordProviderHealth(active.identity.runId, status, 0, "delegated", health); } await this.recordSafeFailure(active, "agent.failed", executionNodeId(execution, event.instance), event.error, event.partialText, runtimeFields(event.instance)).catch(() => undefined); } } private async replayPendingPiAgentsEvents(piAgentsRunId: string, waitForExisting = true): Promise { const delegated = this.delegatedRun(piAgentsRunId); if (!delegated) return; const pending = this.pendingPiAgentsEvents.get(piAgentsRunId); if (pending?.length) { (delegated.active.piAgentsEventQueue ??= []).push(...pending); delegated.active.piAgentsEventQueue.sort((left, right) => left.sequence - right.sequence); this.pendingPiAgentsEvents.delete(piAgentsRunId); } await this.drainPiAgentsEvents(delegated.active, false, waitForExisting); } private async recordSafeFailure(active: ActiveRun, eventType: UltraEventType, nodeId: string, error: string, partialText?: string, metadata: Record = {}): Promise { const config = active.config ?? await this.ready(); const secret = await loadOrCreateKey(this.paths.hmacKey); let diagnosticStored = false; if (config.telemetry.rawVaultEnabled) { try { await writeRawVault(this.paths.rawVault, this.paths.vaultKey, error); diagnosticStored = true; } catch {} } const traceStored = await this.storeRawTrace(active, "diagnostic", nodeId, "controller", error); const partialTraceStored = partialText ? await this.storeRawTrace(active, "result", nodeId, "controller", partialText) : false; await this.record(active.identity, eventType, { nodeId, ...metadata, errorClass: classifyPiAgentsError(error), errorFingerprint: hmacId(secret, error, "ERROR"), diagnosticStored, traceStored, partialTraceStored, }); } private async finishWarRoomRounds(runId: string, active: ActiveRun, roundAccepted = false): Promise { if (await this.lifecycleSettlementCommitted(active)) return; if (active.warRoomFinalizing) return; const config = active.config ?? this.getConfig(); const room = active.warRoom; if (!room || !active.envelope) throw new Error("War Room state is unavailable"); const members = WAR_ROOM_MEMBERS; const acceptRound = async (): Promise => { for (const member of members) { const nodeId = `$.branches.${member}`; const batch = active.warRoomBatches?.get(nodeId); if (!batch) throw new Error(`War Room member result is missing: ${member}`); for (const boardEvent of batch) { assertWarRoomTarget(active, boardEvent); const actions = room.accept(boardEvent); this.reserveWarRoomSpecialists(active, actions); const record = room.events.at(-1)!; const content = JSON.stringify(boardEvent); const contentHash = await this.traceHash(content, "BOARD"); await this.record(active.identity, "blackboard.event", { boardType: boardEvent.type, author: member, cursor: record.cursor, round: record.round, payloadChars: content.length, payloadEstimatedTokens: record.estimatedTokens, contentHash }); await this.executeWarRoomActions(active, actions); } } }; if (!roundAccepted) await acceptRound(); active.warRoomFinalizing = true; let keepClosed = false; try { while (active.warRoomSpecialistAdmission) await active.warRoomSpecialistAdmission; if (await this.lifecycleSettlementCommitted(active)) return; if (unsettledWarRoomSpecialistRuns(active).size > 0) { active.warRoomRoundCompletion = { runId }; return; } if (await this.lifecycleSettlementCommitted(active)) return; const result = room.finish(warRoomMetrics(room)); const currentRound = active.warRoomRound ?? 1; if (result.status === "continue" && currentRound < (active.warRoomMaxRounds ?? 2)) { const feeds = Object.fromEntries(members.map((member) => [member, JSON.stringify(room.poll(member).events.filter((entry) => entry.author !== member))])) as WarRoomFeeds; const nextRound = (currentRound + 1) as 2 | 3; active.warRoomRound = nextRound; active.warRoomBatches = new Map(); const flow = warRoomRoundFlow(active.envelope, config, active.worktree?.path ?? active.cwd, nextRound, feeds, issueWarRoomRoundCredentials(active)); const assignments = plannedAgents(flow); const execution = await this.recordPlannedAgents(active, assignments, "execution", "warroom-round"); const flowModels = assignments.flatMap((assignment) => assignment.model ? [assignment.model] : []); this.captureModelCosts(active, flowModels); const runtimeConfig = await this.withDelegatedBudget(active, config, flowModels); const piAgentsRunId = await this.startDelegatedRun(active, () => this.agents.start("warroom", active.envelope!, runtimeConfig, active.worktree?.path ?? active.cwd, undefined, flow), (startedRunId) => { this.bindDelegatedExecution(active, startedRunId, execution); registerWarRoomRoundParticipants(active, startedRunId); }); await this.record(active.identity, "workflow.started", { executionId: execution.executionId, piAgentsRunId, topology: "warroom", phase: "warroom-round", round: nextRound, budgets: runtimeConfig.piAgentsBudgets }); await this.replayPendingPiAgentsEvents(piAgentsRunId, false); return; } const model = result.collapse ? config.deep.model : config.root.model; const knownFacts = warRoomWriterFacts(active); const envelope = fitEnvelope({ ...active.envelope!, trace: { runId: active.identity.runId, nodeId: "$" }, lens: result.collapse ? "collapsed Sol deep synthesis" : "War Room synthesis", knownFacts }, config.context.envelopeHardCapTokens); active.envelope = envelope; active.warRoomPhase = "writer"; keepClosed = true; active.model = model; await this.record(active.identity, result.collapse ? "escalation.triggered" : "warroom.converged", { reason: result.collapse ? "warroom-no-convergence" : "warroom-converged", model, round: result.round, openHypothesesReduction: result.openHypothesesReduction, uncertaintyReduction: result.uncertaintyReduction, communicationTokens: result.communicationTokens }); const flow = flowFor("deep", envelope, config, active.worktree?.path ?? active.cwd, model); const assignments = plannedAgents(flow); const execution = await this.recordPlannedAgents(active, assignments, "execution", "writer"); const flowModels = assignments.flatMap((assignment) => assignment.model ? [assignment.model] : []); this.captureModelCosts(active, flowModels); const runtimeConfig = await this.withDelegatedBudget(active, config, flowModels); const piAgentsRunId = await this.startDelegatedRun(active, () => this.agents.start("deep", envelope, runtimeConfig, active.worktree?.path ?? active.cwd, model, flow), (startedRunId) => this.bindDelegatedExecution(active, startedRunId, execution)); await this.transition(runId, "DEEP_RUNNING"); await this.record(active.identity, "workflow.started", { executionId: execution.executionId, piAgentsRunId, topology: "deep", phase: "warroom-writer", budgets: runtimeConfig.piAgentsBudgets }); await this.replayPendingPiAgentsEvents(piAgentsRunId, false); } finally { if (!keepClosed) active.warRoomFinalizing = false; } } private async completePiAgentsRun(piAgentsRunId: string, status: "completed" | "failed" | "stopped", usage: { input: number; output: number; cacheRead: number; cost: number }, error?: string): Promise { const delegated = this.delegatedRun(piAgentsRunId); if (!delegated) return; const { runId, active, execution } = delegated; completeWarRoomParticipantRun(active, piAgentsRunId); const plannedModels = execution?.plannedModels ?? active.plannedModels; const usageIncomplete = () => execution?.usageIncomplete ?? active.delegatedUsageIncomplete ?? false; const markUsageIncomplete = () => { if (execution) execution.usageIncomplete = true; else active.delegatedUsageIncomplete = true; }; if (isTerminal(active.state) || this.settledPiAgentsRuns.has(piAgentsRunId) || this.settlingPiAgentsRuns.has(piAgentsRunId) || (active.terminalIntent && await this.lifecycleSettlementCommitted(active))) { (active.delegatedRunStates ??= new Map()).set(piAgentsRunId, "settled"); // Deliberately no accounting here. `run_completed` carries the delegated run's // *cumulative* usage, and by this point some or all of it has usually been charged // already -- through node_completed, or directly -- without always landing in // `delegatedUsage`, so there is no reliable remainder to compute. Charging anyway // double-bills a run that already hit its ceiling, which "late delegated completion // cannot charge an exhausted run twice" pins down. // // The cost of this is real and worth naming: usage that arrives only with a cancelled // run's completion is never priced, so cancel-and-rerun under-counts the rolling // window. Closing that needs per-node accounting that survives cancellation, not a // remainder guessed at settle time -- under-charging is the safer of the two errors. return; } this.settlingPiAgentsRuns.add(piAgentsRunId); try { if (status === "failed" && error) await this.recordSafeFailure(active, "workflow.failed", "root", error).catch(() => undefined); if ((plannedModels?.size ?? 0) === 0) await this.recordUsage(active, usage); else { const recorded = active.delegatedUsage?.get(piAgentsRunId) ?? { input: 0, output: 0, cacheRead: 0, cost: 0 }; const remainder = { input: Math.max(0, usage.input - recorded.input), output: Math.max(0, usage.output - recorded.output), cacheRead: Math.max(0, usage.cacheRead - recorded.cacheRead), cost: Math.max(0, usage.cost - recorded.cost) }; if (remainder.input + remainder.output + remainder.cacheRead > 0 || remainder.cost > 0) { const models = new Set(plannedModels!.values()); if (models.size === 1 && !usageIncomplete()) await this.recordUsage(active, remainder, [...models][0]!, executionNodeId(execution, "delegated"), piAgentsRunId); else { markUsageIncomplete(); if (remainder.cost > 0) active.spentUsd += remainder.cost; } } if (usageIncomplete()) { const knownCredits = active.spentCredits; const reservedUnknownCredits = Math.max(0, budgetFor(active.policy ?? (active.config ?? this.getConfig()).policy, knownCredits).hardCap - knownCredits); active.spentCredits += reservedUnknownCredits; await this.persistLedger(active); await this.record(active.identity, "budget.accounting_incomplete", { reason: "pi-agents-failed-node-usage-unavailable", knownCredits, knownUsd: active.spentUsd, creditsEstimated: reservedUnknownCredits }); await this.transition(runId, "BUDGET_EXHAUSTED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BUDGET_EXHAUSTED", reason: "budget-accounting-incomplete" }); return; } } active.reservedCredits = 0; if (await this.lifecycleSettlementCommitted(active)) return; if (status === "stopped") { await this.transition(runId, "CANCELLED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "CANCELLED", reason: "stopped-by-user" }); return; } if (status !== "completed") { await this.transition(runId, "FAILED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "FAILED", reason: "delegated-run-failed" }); return; } if (active.protocolFailure) { await this.transition(runId, "FAILED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "FAILED", reason: "required-result-contract-failed" }); return; } if (execution?.phase === "warroom-specialist") { if (!active.warRoomAcceptedSpecialistRuns?.has(piAgentsRunId)) throw new Error("War Room specialist result is missing"); active.warRoomAcceptedSpecialistRuns.delete(piAgentsRunId); (active.delegatedRunStates ??= new Map()).set(piAgentsRunId, "settled"); await this.record(active.identity, "workflow.completed", { executionId: execution.executionId, piAgentsRunId, topology: active.topology, phase: execution.phase }); const pendingRound = active.warRoomRoundCompletion; if (pendingRound && !unsettledWarRoomSpecialistRuns(active).size && !await this.lifecycleSettlementCommitted(active)) { active.warRoomRoundCompletion = undefined; await this.finishWarRoomRounds(pendingRound.runId, active, true); } return; } await this.record(active.identity, "workflow.completed", { ...(execution ? { executionId: execution.executionId } : {}), piAgentsRunId, topology: active.topology, ...(active.warRoomPhase ? { phase: active.warRoomPhase } : {}) }); if (await this.lifecycleSettlementCommitted(active)) return; if (active.topology === "warroom" && active.warRoomPhase === "rounds") { await this.finishWarRoomRounds(runId, active); return; } await this.runVerification(runId, active); } catch (error) { if (!await this.lifecycleSettlementCommitted(active) && !(error instanceof DelegatedStartAbortedError)) { await this.recordSafeFailure(active, "workflow.failed", "root", error instanceof Error ? error.message : String(error)).catch(() => undefined); await this.transition(runId, "FAILED").catch(() => undefined); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "FAILED", reason: "delegated-run-failed" }).catch(() => undefined); } } finally { (active.delegatedRunStates ??= new Map()).set(piAgentsRunId, "settled"); this.settledPiAgentsRuns.add(piAgentsRunId); this.settlingPiAgentsRuns.delete(piAgentsRunId); } } private async runReviewQuorum(active: ActiveRun, patchContext = ""): Promise> { const config = active.config ?? this.getConfig(); if (!active.context || !active.envelope || !active.worktree) throw new Error("Review quorum requires the original Pi context and isolated writer worktree"); const context = active.context; const writerEnvelope = active.envelope; const worktree = active.worktree; const reviewModel = config.profile === "free" ? config.deep.model : config.scout.model; assertModelAllowed(config, reviewModel, true); await this.assertBudgetAdmission(reviewModel, "scout", config, 1, active); this.captureModelCosts(active, [reviewModel]); const results: AgentResult[] = []; const runReviewer = async (role: ReviewRole, reviewerResult?: AgentResult, model = reviewModel, thinking: ThinkingLevel = "high"): Promise => { const envelope = reviewEnvelope(writerEnvelope, role, config.context.envelopeHardCapTokens); const task = reviewPrompt(envelope, role, reviewerResult, patchContext); const worker = new MicroAgentBackend(context, this.agentDir, model, thinking, async (event) => { if (event.type === "model.call.started") { await this.record(active.identity, event.type, { nodeId: role, model: event.model, provider: event.provider, thinkingLevel: event.thinkingLevel, totalTokens: 0, latencyMs: 0, turnIndex: event.turnIndex, callIndex: event.callIndex, role: "reviewer" }); return; } if ("toolCallId" in event) { await this.recordRootTool(active.identity.runId, event.type, event.toolCallId, event.toolName, event.type === "tool.started" ? 0 : event.toolElapsedMs, role, "reviewer"); return; } const errorClass = event.type === "model.call.failed" ? event.errorClass : undefined; await this.recordUsage(active, event, model, role, undefined, true, { thinkingLevel: event.thinkingLevel, totalTokens: event.totalTokens, latencyMs: event.turnElapsedMs, turnIndex: event.turnIndex, callIndex: event.callIndex, role: "reviewer", observedModel: event.model, observedProvider: event.provider, ...(errorClass ? { errorClass } : {}) }); if (isTerminal(active.state)) throw new Error("MicroAgent stopped at the task budget limit"); }, async (failure) => { const health = this.providerResponse(failure.status, 0); await this.recordProviderHealth(active.identity.runId, failure.status, 0, "delegated", health).catch(() => undefined); }, worktree.path, config.scout.allowedSkills ?? []); await this.recordTaskAssignment(active, { nodeId: role, task, model, thinking, tools: ["ultra_read", "ultra_grep", "ultra_find", "ultra_ls"] }, "reviewer"); await this.record(active.identity, "agent.spawned", { nodeId: role, role: "reviewer", reviewStage: role, model, thinkingLevel: thinking, topology: active.topology }); await this.recordReceivedHandoffs(active, role); try { const result = await worker.run(envelope, { prompt: task }); if (isTerminal(active.state)) throw new Error("Review quorum stopped at the task budget limit"); await this.record(active.identity, "agent.completed", { nodeId: role, role: "reviewer", reviewStage: role, findingsTotal: result.facts.length }); await this.recordResultHandoff(active, role, result); return result; } catch (error) { await this.recordSafeFailure(active, "review.failed", role, error instanceof Error ? error.message : String(error)).catch(() => undefined); throw error; } }; const reviewer = await this.reviewerWithOneRefusalRetry(active, runReviewer, config, reviewModel); results.push(reviewer); if (!reviewQuorum(results).approved && reviewer.status === "complete" && reviewer.facts.length > 0) { await this.assertBudgetAdmission(reviewModel, "scout", config, 1, active); results.push(await runReviewer("verifier", reviewer)); } const quorum = reviewQuorum(results); if (quorum.confirmedFindings > 0) { const model = config.arbitration.model; assertModelAllowed(config, model, true); await this.assertBudgetAdmission(model, "scout", config, 1, active); this.captureModelCosts(active, [model]); const decision = await runReviewer("arbiter", reviewer, model, config.arbitration.thinking); if (decision.status !== "complete" || decision.summary !== "BLOCK_INTEGRATION") throw new Error("Arbiter did not issue a bounded integration block"); await this.record(active.identity, "escalation.triggered", { reason: "review-confirmed-findings", model, confirmedFindings: quorum.confirmedFindings }); } return quorum; } private async reviewerWithOneRefusalRetry(active: ActiveRun, runReviewer: (role: ReviewRole, reviewerResult?: AgentResult, model?: string, thinking?: ThinkingLevel) => Promise, config: UltraConfig, reviewModel: string): Promise { try { return await this.withModelRotation(active, { model: reviewModel, nodeId: "reviewer", role: "reviewer", topology: "scout", critical: true }, (model) => runReviewer("reviewer", undefined, model)); } catch (error) { if (classifyMicroAgentFailure(error) !== "result-contract") throw error; await this.recordGovernanceViolation(active, error, "reviewer", "reviewer"); const model = config.arbitration.model; assertModelAllowed(config, model, true); await this.assertBudgetAdmission(model, "scout", config, 1, active); this.captureModelCosts(active, [model]); await this.record(active.identity, "escalation.triggered", { reason: "review-cheap-pass-refusal", model }); return await runReviewer("reviewer", undefined, model, config.arbitration.thinking); } } private async emitVerifiedWarRoomDecision(active: ActiveRun): Promise { const room = active.warRoom; if (active.topology !== "warroom" || !room) return; const targetCursor = [...room.events].reverse().find((event) => event.type === "confirmation")?.targetCursor; if (targetCursor === undefined) return; const decision: BoardEvent = { type: "decision", author: "controller", targetCursor, claim: `Verified War Room hypothesis at cursor ${targetCursor}`, verified: true }; const actions = room.accept(decision); this.reserveWarRoomSpecialists(active, actions); const record = room.events.at(-1)!; await this.record(active.identity, "blackboard.event", { boardType: decision.type, author: decision.author, cursor: record.cursor, round: record.round, verified: true }); await this.executeWarRoomActions(active, actions); } private async runVerification(runId: string, active: ActiveRun, cwd = active.worktree && !active.integrated ? active.worktree.path : active.cwd): Promise { if (!active.acceptanceCommand) { if (active.informational && active.attributionManifest?.changes.length === 0) { await this.transition(runId, "VERIFYING"); await this.transition(runId, "PRESENTING"); await this.transition(runId, "COMPLETED"); await this.record(active.identity, "result.completed", { success: true, verified: false, terminalState: "COMPLETED", reason: "informational-no-executable-check" }); return; } await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "acceptance-command-required" }); return; } if (typeof active.context?.isProjectTrusted === "function" && !active.context.isProjectTrusted()) { await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "project-trust-required" }); return; } await this.transition(runId, "VERIFYING"); const phase = active.worktree?.path === cwd && !active.integrated ? "worktree" : active.integrated ? "shared" : "root"; await this.record(active.identity, "verification.started", { nodeId: "root", commandHash: await this.traceHash(active.acceptanceCommand, "COMMAND"), attempt: active.repairAttempts + 1, phase }); let results; try { results = await verify({ checks: [{ id: "acceptance", command: active.acceptanceCommand, required: true, timeoutMs: 120_000, expectedExitCode: 0 }] }, cwd); } catch (error) { await this.persistLedger(active, { verificationState: "failed" }); await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "verification-command-failed", errorClass: error instanceof Error ? error.name : "VerificationError" }); return; } await this.storeRawTrace(active, "verification", "verifier", "controller", JSON.stringify(results)); for (const result of results) { const fingerprint = result.fingerprint ? await this.traceHash(result.fingerprint.hash, "FAILURE") : undefined; await this.record(active.identity, "verification.completed", { nodeId: "root", commandHash: await this.traceHash(result.command, "COMMAND"), fingerprint, passed: result.passed, durationMs: result.durationMs, attempt: active.repairAttempts + 1, errorClass: result.fingerprint?.class, phase, }); if (!result.passed && result.fingerprint) await this.record(active.identity, "failure.fingerprinted", { nodeId: "root", fingerprint, errorClass: result.fingerprint.class, attempt: active.repairAttempts + 1, phase }); } const verified = isVerified(results); await this.persistLedger(active, { verificationState: verified ? "passed" : "failed" }); if (verified) { if (active.worktree && !active.integrated) { await this.transition(runId, "REVIEWING"); try { const patch = await worktreeDiff(active.worktree.path); const changed = await worktreeChangedPaths(active.worktree.path); assertOwnedPaths(changed, active.writerScope ?? []); if (active.attributionManifest) { const normalize = (path: string) => path.replaceAll("\\", "/").replace(/^\.\//, "").replace(/\/$/, ""); const actual = new Set(changed.map(normalize)); const declared = new Set(active.attributionManifest.changes.map((change) => normalize(change.path))); if (actual.size !== declared.size || [...actual].some((path) => !declared.has(path))) throw new Error("Attribution manifest does not match the changed paths"); } await this.persistLedger(active, { changedFiles: changed }); if (changed.length > 0) { if (patch.length > 16_000 || changed.length > 32) throw new Error("Review input exceeds the bounded patch or path limit"); const patchContext = `Changed paths: ${JSON.stringify(changed)}\nUnified diff:\n${patch}`; const quorum = await this.runReviewQuorum(active, patchContext); await this.record(active.identity, "review.completed", { ...quorum, required: quorum.reviewers, changedFiles: changed.length }); if (!quorum.approved) throw new Error("Independent review quorum did not approve integration"); } await integrateReviewedDiff(active.cwd, patch, true); active.integrated = true; await this.record(active.identity, "worktree.integrated", { changedFiles: changed.length, baseCommit: active.worktree.baseCommit }); } catch (error) { await this.recordSafeFailure(active, "worktree.integration.failed", "writer", error instanceof Error ? error.message : String(error)).catch(() => undefined); if (!isTerminal(active.state)) { await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "worktree-integration-failed" }); } return; } await this.runVerification(runId, active, active.cwd); return; } await this.emitVerifiedWarRoomDecision(active); await this.transition(runId, "PRESENTING"); await this.transition(runId, "COMPLETED"); await this.record(active.identity, "result.completed", { success: true, verified: true, terminalState: "COMPLETED" }); if (active.worktree) { try { await removeWorktree(active.cwd, active.worktree.path); await this.record(active.identity, "worktree.removed", { nodeId: "writer" }); } catch (error) { await this.recordSafeFailure(active, "worktree.cleanup.failed", "writer", error instanceof Error ? error.message : String(error)).catch(() => undefined); } } return; } const failed = results.find((result) => !result.passed); const config = active.config ?? this.getConfig(); if (active.integrated || !failed?.fingerprint) { await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "acceptance-failed-unfingerprinted" }); return; } const previousSeen = active.fingerprints.get(failed.fingerprint.hash) ?? 0; if (previousSeen === 0) active.fingerprints.clear(); const seen = previousSeen + 1; active.fingerprints.set(failed.fingerprint.hash, seen); const repair = repairModel(seen, config.repair.maxAttempts); const policyCap = POLICY_CAPS[active.policy ?? config.policy].maxRepairCycles; const escalating = config.repair.sameFingerprintEscalation && repair.model === "sol" && seen === config.repair.maxAttempts + 1; if ((!escalating && active.repairAttempts >= policyCap) || (repair.model === "sol" && !escalating)) { await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "repair-attempts-exhausted" }); return; } active.repairAttempts += 1; await this.transition(runId, escalating ? "ESCALATING" : "REPAIRING"); const objective = escalating ? `Escalate the unchanged acceptance failure ${failed.fingerprint.hash} after ${config.repair.maxAttempts} distinct Luna repair attempts. Re-evaluate their assumptions, apply the minimal root-cause fix, and re-run the declared acceptance command.` : seen > 1 ? `Repair the unchanged acceptance failure ${failed.fingerprint.hash} with a materially different root-cause hypothesis from attempt ${seen - 1}. Re-run the declared acceptance command.` : `Repair the bounded acceptance failure ${failed.fingerprint.hash}. Diagnose the root cause from exact verification evidence and re-run the declared acceptance command.`; const envelope = fitEnvelope(this.envelope(runId, { objective, paths: active.writerScope, acceptanceCommand: active.acceptanceCommand }, "deep", config), config.context.envelopeHardCapTokens); active.envelope = envelope; const model = escalating ? config.deep.model : config.repair.model; const fingerprint = await this.traceHash(failed.fingerprint.hash, "FAILURE"); if (escalating) await this.record(active.identity, "escalation.triggered", { reason: "same-fingerprint-twice", model, attempt: active.repairAttempts, fingerprint }); await this.record(active.identity, "repair.attempted", { attempt: active.repairAttempts, model, thinkingLevel: repair.thinking, fingerprint }); try { const baseFlow = flowFor("deep", envelope, config, cwd, model); const flow = baseFlow.kind === "agent" ? { ...baseFlow, thinking: repair.thinking } : baseFlow; const assignments = plannedAgents(flow); const execution = await this.recordPlannedAgents(active, assignments, "escalation"); const flowModels = assignments.flatMap((assignment) => assignment.model ? [assignment.model] : []); this.captureModelCosts(active, flowModels); const runtimeConfig = await this.withDelegatedBudget(active, config, flowModels, "escalation"); const piAgentsRunId = await this.startDelegatedRun(active, () => this.agents.start("deep", envelope, runtimeConfig, cwd, model, flow), (startedRunId) => this.bindDelegatedExecution(active, startedRunId, execution)); active.model = model; active.topology = "deep"; await this.transition(runId, "DEEP_RUNNING"); await this.record(active.identity, "workflow.started", { executionId: execution.executionId, piAgentsRunId, topology: "deep", phase: escalating ? "repair-escalation" : "repair", attempt: active.repairAttempts, budgets: runtimeConfig.piAgentsBudgets }); await this.replayPendingPiAgentsEvents(piAgentsRunId, false); } catch (error) { if (await this.lifecycleSettlementCommitted(active) || error instanceof DelegatedStartAbortedError) return; await this.recordSafeFailure(active, "repair.failed", "repair", error instanceof Error ? error.message : String(error)).catch(() => undefined); await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "repair-dispatch-failed" }); } } async finishRootRun(runId: string): Promise<{ runId: string; state: ControllerState; terminal: boolean; verified: boolean; spentUsd: number; changedFiles: string[]; acceptedFacts: Array<{ id: string; claim: string; confidence: number }>; remainingBlockers?: string[]; requiredToContinue?: string } | undefined> { const active = this.active.get(runId); if (!active) return undefined; const outcome = () => ({ runId, state: active.state, terminal: isTerminal(active.state), verified: active.state === "COMPLETED" && active.ledger?.verificationState === "passed", spentUsd: active.spentUsd ?? 0, ...presentationDetails(active) }); if (isTerminal(active.state)) return outcome(); if (!((active.topology === "direct" && active.state === "ROUTED") || (active.topology === "scout" && active.state === "IMPLEMENTING"))) return outcome(); if (!active.attributionManifest) { await this.transition(runId, "BLOCKED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BLOCKED", reason: "root-attribution-required" }); return outcome(); } await this.runVerification(runId, active); return outcome(); } status(): { activeRuns: Array<{ runId: string; state: ControllerState; topology: Topology; spentUsd: number; model?: string; progress: { done: number; total: number }; checklist: ChecklistItem[]; nodes: Array<{ nodeId: string; credits: number; role: string; model?: string; state: string }>; piAgentsRunId?: string }>; health: ProviderHealth; budget: ReturnType; spentUsd: number } { const runs = [...this.active.values()]; const current = runs.find((run) => !isTerminal(run.state)) ?? runs.at(-1); return { activeRuns: [...this.active.entries()].map(([runId, run]) => ({ runId, state: run.state, topology: run.topology, spentUsd: run.spentUsd ?? 0, model: run.model, progress: checklistProgress(run.ledger), checklist: run.ledger?.checklist ?? [], nodes: [...(run.agentCredits ?? new Map())].map(([nodeId, credits]) => ({ nodeId, credits, role: run.resultRoles?.get(nodeId) ?? "root", model: run.plannedModels?.get(nodeId) ?? run.model, state: run.ledger?.checklist?.find((item) => item.owner === nodeId)?.state ?? "active", })).sort((left, right) => left.nodeId.localeCompare(right.nodeId)), ...(run.piAgentsRunId ? { piAgentsRunId: run.piAgentsRunId } : {}), })), health: this.health, budget: this.budgetState(), spentUsd: current?.spentUsd ?? 0, }; } takeDelegatedPresentation(piAgentsRunId: string) { const entry = [...this.active.entries()].find(([, run]) => run.piAgentsRunId === piAgentsRunId); if (!entry) return undefined; const [runId, active] = entry; if (!isTerminal(active.state) || active.presentationQueued) return undefined; active.presentationQueued = true; return { runId, state: active.state, topology: active.topology, verified: active.state === "COMPLETED" && active.ledger?.verificationState === "passed", spentUsd: active.spentUsd ?? 0, ...presentationDetails(active), }; } releaseDelegatedPresentation(piAgentsRunId: string): void { const active = [...this.active.values()].find((run) => run.piAgentsRunId === piAgentsRunId); if (active?.presentationQueued && !active.presentationUsageRecorded) active.presentationQueued = false; } async exportPreview(week: string): Promise { const config = await this.ready(); const secret = await loadOrCreateKey(this.paths.hmacKey); return buildWeeklyExportBundle(week, previewExport(await new EventStore(this.paths.events).all(), secret), config, await loadPriceCatalog(this.paths)); } async exportWeek(week: string, approved: boolean, bundle?: WeeklyExportBundle): Promise { const output = await writeWeeklyExport(this.paths.exports, bundle ?? await this.exportPreview(week), approved); const id = randomUUID(); await this.record({ sessionId: "system", taskId: `export:${id}`, runId: `export:${id}` }, "export.created", { week, approved: true }); return output; } async feedback(kind: "good" | "fixed" | "bad", comment = ""): Promise { const latest = [...this.active.values()].at(-1); if (!latest) throw new Error("No UltraPi run is available for feedback"); const commentProvided = Boolean(comment.trim()); const commentStored = commentProvided && await this.storeRawTrace(latest, "feedback", "user", "controller", comment); await this.record(latest.identity, "user.feedback", { kind, commentProvided, commentStored }); } async recordRootAttribution(runId: string, manifest: AttributionManifest): Promise { const active = this.active.get(runId); if (!active) return; await this.recordAttribution(active, "root", { schemaVersion: 1, status: "complete", summary: "root attribution", facts: [], hypotheses: [], unknowns: [], dependencies: [], attributionManifest: manifest }); } isDirectRootRun(runId: string): boolean { return this.active.get(runId)?.topology === "direct"; } async normalizeRootWritePath(runId: string, path: string): Promise { const active = this.active.get(runId); if (!active?.envelope) throw new Error("UltraPi root scope is unavailable"); await assertWorkerPath(active.cwd, path, active.envelope.scope); const normalized = relative(resolve(active.cwd), resolve(active.cwd, path)).replaceAll("\\", "/"); if (!normalized || normalized === ".." || normalized.startsWith("../")) throw new Error("UltraPi root write is outside its declared scope"); return normalized; } async recordRootUsage(runId: string, model: string, usage: { input?: number; output?: number; cacheRead?: number; totalTokens?: number; cost?: { total?: number } } | undefined, metadata?: { thinkingLevel: string; latencyMs: number; turnIndex: number; callIndex: number; observedModel?: string; observedProvider?: string }): Promise { const active = this.active.get(runId); if (!active) return false; if (!usage) return !isTerminal(active.state); const terminalPresentation = isTerminal(active.state) && active.presentationQueued && !active.presentationUsageRecorded; if (isTerminal(active.state) && !terminalPresentation) return false; if (terminalPresentation) active.presentationUsageRecorded = true; await this.recordUsage(active, { input: usage.input ?? 0, output: usage.output ?? 0, cacheRead: usage.cacheRead ?? 0, cost: usage.cost?.total }, model, "root", undefined, terminalPresentation, { thinkingLevel: metadata?.thinkingLevel ?? "unknown", totalTokens: usage.totalTokens ?? (usage.input ?? 0) + (usage.output ?? 0), latencyMs: metadata?.latencyMs ?? 0, turnIndex: metadata?.turnIndex ?? 0, callIndex: metadata?.callIndex ?? 0, observedModel: metadata?.observedModel, observedProvider: metadata?.observedProvider }); return !isTerminal(active.state); } private async recordUsage(active: ActiveRun, usage: { input: number; output: number; cacheRead: number; cost?: number }, model = active.model, nodeId = "root", delegatedRunId?: string, allowTerminal = false, metadata?: { thinkingLevel: string; totalTokens: number; latencyMs: number; turnIndex: number; callIndex: number; role?: string; errorClass?: string; observedModel?: string; observedProvider?: string }, roleOverride?: string, accountingKey?: string): Promise { if (isTerminal(active.state) && !allowTerminal) return 0; let accounting = accountingKey ? active.usageAccounting?.get(accountingKey) : undefined; if (!accounting) { const catalog = await this.pricing(active, model); const credits = estimateCredits(catalog, model, usage.input, usage.cacheRead, usage.output); const priceSource = active.derivedRates?.has(model) ? "derived" : "catalog"; const usdReported = Number.isFinite(usage.cost) ? usage.cost : undefined; accounting = { credits, priceSource, ...(usdReported === undefined ? {} : { usdReported }), spent: false, modelCallAttempted: false, reservationReleased: false, ledgerPersisted: false, complete: false, agentCreditsApplied: false }; if (accountingKey) (active.usageAccounting ??= new Map()).set(accountingKey, accounting); } if (accounting.complete) return accounting.credits; const { credits, priceSource, usdReported } = accounting; if (!accounting.spent) { active.spentCredits += credits; if (usdReported !== undefined) active.spentUsd = (active.spentUsd ?? 0) + usdReported; if (delegatedRunId) { const total = active.delegatedUsage?.get(delegatedRunId) ?? { input: 0, output: 0, cacheRead: 0, cost: 0 }; (active.delegatedUsage ??= new Map()).set(delegatedRunId, { input: total.input + usage.input, output: total.output + usage.output, cacheRead: total.cacheRead + usage.cacheRead, cost: total.cost + (usdReported ?? 0) }); } accounting.spent = true; } const role = roleOverride ?? metadata?.role ?? (nodeId === "root" ? "root" : active.resultRoles?.get(nodeId) ?? (active.nodeKinds?.get(nodeId) === "reduce" ? "reducer" : "scout")); if (!accounting.modelCallAttempted) { accounting.modelCallAttempted = true; await this.record(active.identity, metadata?.errorClass ? "model.call.failed" : "model.call.completed", { nodeId, model: metadata?.observedModel ?? model, provider: metadata?.observedProvider ?? model.split("/", 1)[0] ?? "unknown", thinkingLevel: metadata?.thinkingLevel ?? "unknown", inputTokens: usage.input, cachedInputTokens: usage.cacheRead, outputTokens: usage.output, totalTokens: metadata?.totalTokens ?? usage.input + usage.output, creditsEstimated: credits, priceSource, latencyMs: metadata?.latencyMs ?? 0, turnIndex: metadata?.turnIndex ?? 0, callIndex: metadata?.callIndex ?? 0, success: !metadata?.errorClass, ...(metadata?.errorClass ? { errorClass: metadata.errorClass } : {}), ...(usdReported === undefined ? {} : { usdReported }), role }); } if (!accounting.reservationReleased) { active.reservedCredits = Math.max(0, (active.reservedCredits ?? 0) - credits); accounting.reservationReleased = true; } if (!accounting.ledgerPersisted) { await this.persistLedger(active); accounting.ledgerPersisted = true; } const config = active.config ?? this.getConfig(); const taskBudget = budgetFor(active.policy ?? config.policy, active.spentCredits, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget); const windows = await this.windowSpend(); const phase = budgetPhase(taskBudget); if (!active.softLimitEmitted && active.spentCredits >= taskBudget.softCap) { active.softLimitEmitted = true; await this.record(active.identity, "budget.soft_limit", { spentCredits: active.spentCredits, softCap: taskBudget.softCap, hardCap: taskBudget.hardCap }); } const weeklyExhausted = config.budgets.weeklyCreditBudget !== undefined && windows.week >= config.budgets.weeklyCreditBudget; const dailyExhausted = config.budgets.dailyCreditBudget !== undefined && windows.day >= config.budgets.dailyCreditBudget; const exhausted = phase === "stop" || weeklyExhausted || dailyExhausted; if (exhausted && !active.hardLimitEmitted) { active.hardLimitEmitted = true; await this.record(active.identity, "budget.hard_limit", { spentCredits: active.spentCredits, hardCap: taskBudget.hardCap, reason: phase === "stop" ? "task-hard-cap" : weeklyExhausted ? "weekly-cap" : "daily-cap" }); } if (exhausted) { const runId = [...this.active.entries()].find(([, run]) => run === active)?.[0]; if (runId && !isTerminal(active.state)) await this.transition(runId, "BUDGET_EXHAUSTED"); if (runId && active.state === "BUDGET_EXHAUSTED" && !active.terminalResultRecorded) await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BUDGET_EXHAUSTED", reason: "budget-exhausted" }); } accounting.complete = true; return credits; } /** * `terminateOnReject` is false for spend the run can do without: refusing to admit an optional * spawn must not end a run that is still delivering. Only spend the run depends on may * transition it to BUDGET_EXHAUSTED. */ private async assertBudgetAdmission(model: string, topology: Topology, config: UltraConfig, agentCount?: number, active?: ActiveRun, priorSpentCredits = 0, terminateOnReject = true): Promise { const catalog = active ? await this.pricing(active, model) : await loadPriceCatalog(this.paths); const policy = active?.policy ?? config.policy; const initial = effectiveInitial(config, policy); const agents = agentCount ?? (topology === "direct" ? 1 : topology === "scout" ? initial : topology === "warroom" ? WAR_ROOM_MEMBERS.length : topology === "swarm" ? initial + 1 : 1); const output = topology === "scout" ? config.scout.maxOutputTokens : config.boundedWriter.maxOutputTokens; if (!catalog.models[model]) throw new Error(unpricedModel(model, this.paths.priceCatalog)); const estimated = estimateCredits(catalog, model, 8_000 * agents, 0, output * agents); const state = budgetFor(policy, active?.spentCredits ?? priorSpentCredits, config.budgets.weeklyCreditBudget, config.budgets.dailyCreditBudget); const windows = await this.windowSpend(); const reserved = [...this.active.values()].reduce((total, run) => total + (isTerminal(run.state) ? 0 : run.reservedCredits ?? 0), 0); const reject = async (message: string, reason: "task-admission" | "weekly-admission" | "daily-admission"): Promise => { if (active && terminateOnReject && !active.hardLimitEmitted) { active.hardLimitEmitted = true; await this.record(active.identity, "budget.hard_limit", { spentCredits: active.spentCredits, hardCap: state.hardCap, estimatedCredits: estimated, reason }); await this.transition(active.identity.runId, "BUDGET_EXHAUSTED"); await this.record(active.identity, "result.completed", { success: false, verified: false, terminalState: "BUDGET_EXHAUSTED", reason: "budget-exhausted" }); } throw new Error(message); }; if (state.spentCredits + (active?.reservedCredits ?? 0) + estimated > state.internalStopTarget) await reject("Budget governor blocks a non-final spawn at the internal stop target", "task-admission"); if (config.budgets.weeklyCreditBudget !== undefined && windows.week + reserved + estimated > config.budgets.weeklyCreditBudget) await reject("Weekly UltraPi credit budget would be exceeded", "weekly-admission"); if (config.budgets.dailyCreditBudget !== undefined && windows.day + reserved + estimated > config.budgets.dailyCreditBudget) await reject("Daily UltraPi credit budget would be exceeded", "daily-admission"); if (active) active.reservedCredits = (active.reservedCredits ?? 0) + estimated; } private async windowSpend(): Promise<{ week: number; day: number }> { const now = new Date(); const day = now.toISOString().slice(0, 10); const weekStart = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() - ((now.getUTCDay() + 6) % 7))).toISOString().slice(0, 10); return (await new EventStore(this.paths.events).all()).reduce((totals, event) => { if (event.eventType !== "model.call.completed" && event.eventType !== "model.call.failed" && event.eventType !== "budget.accounting_incomplete") return totals; if (typeof event.creditsEstimated !== "number") return totals; if (event.timestamp.slice(0, 10) === day) totals.day += event.creditsEstimated; if (event.timestamp.slice(0, 10) >= weekStart) totals.week += event.creditsEstimated; return totals; }, { week: 0, day: 0 }); } }