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.
11 kB · 280 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281import { closeSync, readFileSync } from "node:fs";import type { AgentSessionMessageDeliveryMode, AgentSessionMessageSender } from "../../core/agent-messages.js";import type { IdleEvictionMinutes } from "../../core/session-action-store.js";
export { SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV } from "../../core/session-lease.js";
import type { WorkerRosterEntry } from "./agent-roster.js";import type { DaemonClientCapability, DaemonCommand, DaemonOutbound } from "./daemon-protocol.js";
export const DAEMON_WORKER_ROLE_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER";export const DAEMON_WORKER_TOKEN_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_TOKEN";export const DAEMON_WORKER_INSTANCE_ID_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_INSTANCE_ID";export const DAEMON_WORKER_ACTIVE_SESSION_ID_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_ACTIVE_SESSION_ID";export const DAEMON_WORKER_SUPERVISOR_SOCKET_ENV = "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_SOCKET";export const DAEMON_WORKER_RECOVERY_JOURNAL_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_RECOVERY_JOURNAL";export const DAEMON_WORKER_STARTUP_GATE_FD_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_STARTUP_GATE_FD";export const DAEMON_WORKER_STARTUP_GATE_COMMIT = "start\n";export type DaemonWorkerLifecycle = "starting" | "ready" | "recovering" | "stopping" | "failed";
// Worker->supervisor roster frames live outside the client-facing DaemonOutbound schema.export type DaemonWorkerRosterOutbound = | { type: "roster_delta"; entries: WorkerRosterEntry[]; removedAgentIds?: string[]; snapshot?: true; } | { type: "roster_heartbeat" };
/** Advertised by new workers in the worker_auth response; absent on legacy workers. */export const DAEMON_WORKER_ROSTER_CAPABILITY = "agent_roster";
/** Advertised in the worker_auth response by workers that accept peer transport grants. */export const DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY = "peer_transport";
/** Idle keepalive cadence for worker->supervisor roster frames; the supervisor staleness threshold derives from it. */export const ROSTER_HEARTBEAT_INTERVAL_MS = 15_000;
export type DaemonWorkerFrameHeader = | { kind: "command"; requestId: string; commandType: string; } | { kind: "outbound"; requestId?: string; outboundType: DaemonOutbound["type"] | DaemonWorkerRosterOutbound["type"]; activeSessionId?: string; snapshotId?: string; sessionEventType?: string; payloadEncoding?: "jsonl" | "assistant-delta"; snapshotPurpose?: "attach" | "replacement" | "catchup"; };
export type DaemonCreateCommand = Extract<DaemonCommand, { type: "create" }>;
export interface DurableDaemonCreateCommand { type: "create"; sessionPath?: string; noSession?: boolean;}
export function durableDaemonCreateCommand(command: DaemonCreateCommand): DurableDaemonCreateCommand { return { type: "create", ...(command.sessionPath !== undefined ? { sessionPath: command.sessionPath } : {}), ...(command.noSession !== undefined ? { noSession: command.noSession } : {}), };}
/** * A single-use, worker-memory-only admission for one direct peer, scoped to one active-session * slot (stable across switch_session/new_session/fork) of one worker incarnation. The session pin * is a routing/accident guard, not a privilege boundary: ticket holders already hold supervisor * access, which can attach to, switch, or kill any session. */export interface DaemonWorkerPeerGrant { grantId: string; token: string; expiresAt: string; purpose: "session_client"; workerInstanceId: string; activeSessionId: string; issuerGeneration: string;}
/** Commands a direct peer may send before it holds an authenticated session role. */export type DaemonPeerCommand = { id?: string; type: "peer_auth"; grantId: string; token: string; workerInstanceId: string; purpose: "session_client";};
export type DaemonPeerCommandBody = Omit<DaemonPeerCommand, "id">;
export type DaemonWorkerCommand = | { id?: string; type: "worker_auth"; token: string; workerInstanceId?: string; supervisorGeneration: string; supervisorPid: number; supervisorProcessStartId?: string; supervisorSocketPath: string; } | { id?: string; type: "worker_subscribe"; activeSessionId: string; capabilities?: readonly DaemonClientCapability[]; supportsExtensionUi?: boolean; } | { id?: string; type: "worker_unsubscribe"; activeSessionId: string } | { id?: string; type: "worker_register_peer_transport"; grant: DaemonWorkerPeerGrant } | { id?: string; type: "worker_archive_and_shutdown" } | { id?: string; type: "worker_passivate_idle_children"; idleEvictionMinutes: IdleEvictionMinutes; now: number; limit: number; } | { id?: string; type: "worker_deliver_message"; targetActiveSessionId: string; message: string; sender: AgentSessionMessageSender; deliveryMode?: AgentSessionMessageDeliveryMode; } | { id?: string; type: "worker_prepare_update" } | { id?: string; type: "worker_commit_update" } | { id?: string; type: "worker_cancel_update" };
export type DaemonWorkerCommandBody = DaemonWorkerCommand extends infer TCommand ? TCommand extends { id?: string } ? Omit<TCommand, "id"> : never : never;
export interface DaemonWorkerDescriptor { version: 1 | 2; workerId: string; pid: number; processStartId?: string; socketPath: string; recoveryJournalPath: string; orphanProcessJournalPath?: string; supervisorSocketPath: string; authenticationToken: string; workerInstanceId?: string; rootActiveSessionId: string; /** Stable protocol client that owns this worker. Omitted for resident sessions. */ ownerClientId?: string; rootSessionId?: string; sessionFile?: string; sessionDir?: string; telemetryDisabled?: true; createdAt: string; updatedAt: string; lifecycle: DaemonWorkerLifecycle; createCommand: DurableDaemonCreateCommand; consecutiveFailures: number; /** Durable intent written before root termination so replacement supervisors never recover it. */ stopRequestedAt?: string; /** Complete the root's archived lifecycle state after its process has stopped. */ archiveOnStop?: boolean; lastFailureAt?: string; lastError?: string;}
export function durableDaemonWorkerDescriptor(descriptor: DaemonWorkerDescriptor): DaemonWorkerDescriptor { const versionOneCreateCommand = descriptor.createCommand as unknown as { config?: unknown }; const versionOneConfig = descriptor.version === 1 && typeof versionOneCreateCommand.config === "object" && versionOneCreateCommand.config !== null ? (versionOneCreateCommand.config as Record<string, unknown>) : undefined; const sessionDir = descriptor.sessionDir ?? (typeof versionOneConfig?.sessionDir === "string" ? versionOneConfig.sessionDir : undefined); const telemetryDisabled = descriptor.telemetryDisabled === true || versionOneConfig?.telemetryDisabled === true; return { version: 2, workerId: descriptor.workerId, pid: descriptor.pid, ...(descriptor.processStartId !== undefined ? { processStartId: descriptor.processStartId } : {}), socketPath: descriptor.socketPath, recoveryJournalPath: descriptor.recoveryJournalPath, ...(descriptor.orphanProcessJournalPath !== undefined ? { orphanProcessJournalPath: descriptor.orphanProcessJournalPath } : {}), supervisorSocketPath: descriptor.supervisorSocketPath, authenticationToken: descriptor.authenticationToken, ...(descriptor.workerInstanceId !== undefined ? { workerInstanceId: descriptor.workerInstanceId } : {}), rootActiveSessionId: descriptor.rootActiveSessionId, ...(descriptor.ownerClientId !== undefined ? { ownerClientId: descriptor.ownerClientId } : {}), ...(descriptor.rootSessionId !== undefined ? { rootSessionId: descriptor.rootSessionId } : {}), ...(descriptor.sessionFile !== undefined ? { sessionFile: descriptor.sessionFile } : {}), ...(sessionDir !== undefined ? { sessionDir } : {}), ...(telemetryDisabled ? { telemetryDisabled: true as const } : {}), createdAt: descriptor.createdAt, updatedAt: descriptor.updatedAt, lifecycle: descriptor.lifecycle, createCommand: durableDaemonCreateCommand(descriptor.createCommand), consecutiveFailures: descriptor.consecutiveFailures, ...(descriptor.stopRequestedAt !== undefined ? { stopRequestedAt: descriptor.stopRequestedAt } : {}), ...(descriptor.archiveOnStop !== undefined ? { archiveOnStop: descriptor.archiveOnStop } : {}), ...(descriptor.lastFailureAt !== undefined ? { lastFailureAt: descriptor.lastFailureAt } : {}), ...(descriptor.lifecycle === "failed" ? { lastError: "Waiting for a client with fresh runtime context" } : {}), };}
export function isDaemonWorkerProcess(environment: NodeJS.ProcessEnv = process.env): boolean { return environment[DAEMON_WORKER_ROLE_ENV] === "1";}
export function waitForDaemonWorkerStartupGate(environment: NodeJS.ProcessEnv = process.env): void { const rawFd = environment[DAEMON_WORKER_STARTUP_GATE_FD_ENV]; if (rawFd === undefined) { return; } delete environment[DAEMON_WORKER_STARTUP_GATE_FD_ENV]; const fd = Number(rawFd); if (!Number.isInteger(fd) || fd < 3) { throw new Error("Daemon session worker has an invalid startup gate"); } let marker: string; try { marker = readFileSync(fd, "utf8"); } finally { closeSync(fd); } if (marker !== DAEMON_WORKER_STARTUP_GATE_COMMIT) { throw new Error("Daemon session worker startup was cancelled"); }}
export function daemonWorkerInstanceId(environment: NodeJS.ProcessEnv = process.env): string | undefined { return environment[DAEMON_WORKER_INSTANCE_ID_ENV] || undefined;}
export function requireDaemonWorkerAuthenticationToken(environment: NodeJS.ProcessEnv = process.env): string { const token = environment[DAEMON_WORKER_TOKEN_ENV]; if (!token) { throw new Error("Daemon session worker is missing its authentication token"); } return token;}
export function isDaemonWorkerFrameHeader(value: unknown): value is DaemonWorkerFrameHeader { if (!value || typeof value !== "object") { return false; } const candidate = value as Record<string, unknown>; if (candidate.kind === "command") { return typeof candidate.requestId === "string" && typeof candidate.commandType === "string"; } return ( candidate.kind === "outbound" && typeof candidate.outboundType === "string" && (candidate.requestId === undefined || typeof candidate.requestId === "string") && (candidate.activeSessionId === undefined || typeof candidate.activeSessionId === "string") && (candidate.snapshotId === undefined || typeof candidate.snapshotId === "string") && (candidate.sessionEventType === undefined || typeof candidate.sessionEventType === "string") && (candidate.snapshotPurpose === undefined || candidate.snapshotPurpose === "attach" || candidate.snapshotPurpose === "replacement" || candidate.snapshotPurpose === "catchup") && (candidate.payloadEncoding === undefined || candidate.payloadEncoding === "jsonl" || candidate.payloadEncoding === "assistant-delta") );}