Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
6.0 kB · 162 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163import { randomUUID } from "node:crypto";import type { Socket } from "node:net";import type { AgentSessionRuntime } from "../../core/agent-session-runtime.js";import type { AgentStatus } from "../../core/session-manager.js";import type { DaemonClientCapability, DaemonEventSequence, DaemonExtensionUIResponse, DaemonOutbound,} from "./daemon-protocol.js";import { formatSessionDisplayId, matchesSessionIdSuffix } from "./daemon-session-id.js";
export interface DaemonSocketClient { id: string; socket: Socket; attachedActiveSessionIds: Set<string>; /** Session events are dropped while the socket is blocked and replaced with one catch-up snapshot on drain. */ catchupActiveSessionIds?: Set<string>; /** A real runtime replacement takes precedence over an ordinary resync for the same queued catch-up. */ catchupPurposes?: Map<string, "replacement" | "resync">; /** The single catch-up drain currently serving this client. */ catchupPromise?: Promise<void>; /** Delayed retry after transient catch-up snapshot preparation failure. */ catchupRetryTimer?: NodeJS.Timeout; backpressured?: boolean; rosterSubscribed?: boolean; /** A push hit backpressure; one full-roster resync goes out on drain. */ rosterResyncPending?: boolean; authenticated?: boolean; authenticationRole?: "supervisor" | "session_client"; transport?: "jsonl" | "private-framed"; snapshotStreaming?: boolean; snapshotActiveSessionIds?: Set<string>; /** * Session frames withheld while a snapshot stream is active for that session * (worker side: DaemonOutbound). They replay when the stream completes * instead of each one triggering a full snapshot re-transfer. */ deferredSessionOutbounds?: Map<string, { frames: DaemonOutbound[]; bytes: number }>; /** Worker-side sessions whose deferral buffer overflowed this stream. */ deferredSessionFramesDropped?: Set<string>; /** * Relay payloads withheld while a snapshot stream is active for that session * (supervisor side: serialized buffers). Same replay-on-completion policy. */ deferredSessionPayloads?: Map<string, { payloads: Buffer[]; bytes: number }>; /** Supervisor-side sessions whose deferral buffer overflowed this stream. */ deferredSessionPayloadsDropped?: Set<string>; snapshotActiveSessionCounts?: Map<string, number>; snapshotTransferAbortControllers?: Map<string, AbortController>; snapshotTransferTails?: Map<string, Promise<void>>; detachInput: () => void; supportsExtensionUi: boolean; capabilities: Set<DaemonClientCapability>; capabilitiesByActiveSessionId?: Map<string, Set<DaemonClientCapability>>;}
export interface ActiveSessionState { activeSessionId: string; runtime: AgentSessionRuntime; clients: Set<DaemonSocketClient>; /** Attach snapshots in flight: reserved for passivation busyness, but not yet event recipients. */ pendingAttaches: number; extensionUiRequests: Map<string, ActiveSessionExtensionUiRequest>; eventGeneration: string; lastEventSequence: DaemonEventSequence; inFlightBash?: Promise<void>; unsubscribe?: () => void; /** Latest background status summary, surfaced in the agents view. */ summaryState?: AgentStatus; /** * Client env (e.g. herdr pane identity), merged over process.env for this * session's pi.exec() subprocesses. Bound when the runtime is created (or * adopted from the first env-carrying create that reuses an env-less * session); never overwritten after that — watchers also attach, and * extensions capture identity at load. Subagents inherit the parent's. */ clientEnv?: Record<string, string>;}
export interface ActiveSessionExtensionUiRequest { resolve: (response: DaemonExtensionUIResponse) => void;}
interface ActiveSessionIdIndex { has(activeSessionId: string): boolean;}
export function createActiveSessionId(existingIds?: ActiveSessionIdIndex): string { while (true) { const activeSessionId = formatSessionDisplayId(randomUUID()); if (!existingIds?.has(activeSessionId)) { return activeSessionId; } }}
export class AmbiguousActiveSessionError extends Error { constructor(message: string) { super(message); this.name = "AmbiguousActiveSessionError"; }}
export function resolveActiveSessionState( sessions: ReadonlyMap<string, ActiveSessionState>, selector: string,): ActiveSessionState { const direct = sessions.get(selector); if (direct) { return direct; }
const exactMatches = uniqueStates( [...sessions.values()].filter((state) => { const session = state.runtime.session; return session.sessionId === selector || session.sessionName === selector; }), ); if (exactMatches.length === 1) { return exactMatches[0]!; } if (exactMatches.length > 1) { throw new AmbiguousActiveSessionError(formatAmbiguousSessionError(selector, exactMatches)); }
const suffixMatches = uniqueStates( [...sessions.values()].filter((state) => { const session = state.runtime.session; return ( matchesSessionIdSuffix(state.activeSessionId, selector) || matchesSessionIdSuffix(session.sessionId, selector) ); }), ); if (suffixMatches.length === 1) { return suffixMatches[0]!; } if (suffixMatches.length > 1) { throw new AmbiguousActiveSessionError(formatAmbiguousSessionError(selector, suffixMatches)); }
throw new Error(`Unknown active session: ${selector}`);}
function uniqueStates(states: ActiveSessionState[]): ActiveSessionState[] { const unique = new Map<string, ActiveSessionState>(); for (const state of states) { unique.set(state.activeSessionId, state); } return [...unique.values()];}
function formatAmbiguousSessionError(selector: string, states: readonly ActiveSessionState[]): string { return `Ambiguous active session "${selector}": matches ${states.map(formatSessionMatch).join(", ")}`;}
function formatSessionMatch(state: ActiveSessionState): string { const session = state.runtime.session; const name = session.sessionName ? ` (${session.sessionName})` : ""; return `${state.activeSessionId}/${session.sessionId}${name}`;}