AgentSession Lifecycle Management in pi-web's rpc-manager.ts

AgentSession lifecycle management in pi-web relies on a global registry of wrapped sessions that survive Next.js hot-reloads, enforce 10-minute idle timeouts, and broadcast real-time state changes to keep the UI and SSE streams synchronized.

The agegr/pi-web repository implements a robust AgentSession lifecycle management system in lib/rpc-manager.ts that wraps the underlying pi-coding-agent SDK. This architecture ensures sessions persist across development server reloads while providing clean resource cleanup, concurrency protection, and real-time UI updates.

Global Session Registry and Wrapper Architecture

The Global Registry Pattern

The system uses a global Map attached to globalThis to store active sessions. In lib/rpc-manager.ts, the getRegistry() function initializes globalThis.__piSessions on first access and registers process cleanup handlers.

declare global {
  var __piSessions: Map<string, AgentSessionWrapper> | undefined;
}

function getRegistry(): Map<string, AgentSessionWrapper> {
  if (!globalThis.__piSessions) {
    globalThis.__piSessions = new Map();
    const cleanup = () => globalThis.__piSessions?.forEach((s) => s.destroy());
    process.once("exit", cleanup);
    process.once("SIGINT", cleanup);
    process.once("SIGTERM", cleanup);
  }
  return globalThis.__piSessions;
}

This pattern ensures that AgentSession instances survive Next.js hot-reloads during development. The cleanup handlers guarantee that all sessions call destroy() when the process exits or receives termination signals.

AgentSessionWrapper Design

The AgentSessionWrapper class encapsulates the raw SDK session (inner) and adds lifecycle bookkeeping. It tracks event listeners, pending UI responses, custom UI components, and idle timers.

export class AgentSessionWrapper {
  private listeners: EventListener[] = [];
  private _alive = true;
  private idleTimer: NodeJS.Timeout | undefined;
  
  constructor(public readonly inner: AgentSessionLike) {}
  
  isAlive(): boolean { return this._alive; }
  // ... additional lifecycle methods
}

The wrapper provides the bridge between the low-level SDK and pi-web's UI requirements, managing state broadcasts and resource cleanup independently of the core agent logic.

Starting and Initializing Sessions

The startRpcSession Function

The startRpcSession() function in lib/rpc-manager.ts orchestrates session creation. It checks the registry for existing sessions, handles concurrent start requests, initializes services, and binds extensions.

export async function startRpcSession(
  sessionId: string,
  sessionFile: string,
  cwd: string,
  options: RpcSessionStartOptions = {}
): Promise<{ session: AgentSessionWrapper; realSessionId: string }> {
  const registry = getRegistry();
  const locks = getLocks();

  // Return existing alive wrapper if present
  const existing = registry.get(sessionId);
  if (existing?.isAlive()) return { session: existing, realSessionId: sessionId };

  // Guard against concurrent starts
  const inflight = locks.get(sessionId);
  if (inflight) return inflight;

  const starting = (async () => {
    initTheme();
    const services = await createAgentSessionServices({ cwd, agentDir: getAgentDir() });
    const sessionManager = sessionFile
      ? SessionManager.open(sessionFile, undefined)
      : SessionManager.create(cwd, undefined);

    const scope = await resolveVisibleModels(services.modelRuntime, services.settingsManager.getEnabledModels());
    const { session: inner } = await createAgentSessionFromServices({
      services,
      sessionManager,
      scopedModels: scope.scopedModels,
    });

    if (options.toolNames?.length) {
      inner.setActiveToolsByName(withExtensionTools(inner, options.toolNames));
    }

    const wrapper = new AgentSessionWrapper(inner);
    if (options.toolNames?.length === 0) wrapper.setForceEmptySystemPrompt(true);
    
    wrapper.start();
    const realSessionId = inner.sessionId as string;
    wrapper.onDestroy(() => registry.delete(realSessionId));
    registry.set(realSessionId, wrapper);
    wrapper.beginExtensionBinding({ forceEmptySystemPrompt: options.toolNames?.length === 0 });
    
    return { session: wrapper, realSessionId };
  })().finally(() => {
    locks.delete(sessionId);
  });

  locks.set(sessionId, starting);
  return starting;
}

This function resolves visible models before SDK initialization, applies tool allow-lists, and registers the wrapper in the global registry before returning.

Concurrency and Lock Management

To prevent race conditions during session initialization, the implementation uses a separate lock map (getLocks()) that tracks in-flight start promises by sessionId. If a second request arrives while a session is still initializing, it returns the existing promise rather than creating duplicate sessions.

Runtime Management and Event Broadcasting

Idle Timeout Handling

The wrapper implements automatic cleanup through an idle timer. The resetIdleTimer() method schedules destruction after 10 minutes of inactivity, clearing any previous timeout first.

start(): void {
  this.unsubscribe = this.inner.subscribe((event: AgentEvent) => {
    this.resetIdleTimer();
    if (event.type === "agent_end") invalidateSessionListCache();
    this.emit(event);
    notifyRunningChange();
  });
  this.resetIdleTimer();
  notifyRunningChange();
}

Every incoming event resets the timer, ensuring that only truly idle sessions are terminated. This prevents resource leaks from abandoned browser tabs or disconnected clients.

Running-State Broadcasting

State synchronization relies on notifyRunningChange(), which collects all running session IDs and broadcasts them to subscribed listeners.

export function notifyRunningChange(): void {
  const ids = getRunningRpcSessionIds();
  const snapshot = JSON.stringify([...ids].sort());
  if (snapshot === lastRunningSnapshot) return;
  lastRunningSnapshot = snapshot;
  for (const listener of getRunningListeners()) {
    try { listener(ids); } catch { /* ignore */ }
  }
}

This function triggers on every significant state change—session start, prompt submission, tool execution, and destruction—ensuring the sidebar and SSE endpoints in app/api/agent/running/events/route.ts display accurate live data.

Session Termination and Forking

The destroy() Method

Clean shutdown is handled by AgentSessionWrapper.destroy(), which performs comprehensive resource cleanup:

destroy(): void {
  if (!this._alive) return;
  this._alive = false;
  
  if (this.idleTimer) clearTimeout(this.idleTimer);
  if (this.inner.isBashRunning) this.inner.abortBash();
  this.unsubscribe?.();
  
  for (const pending of this.pendingUiResponses.values()) pending.cancel();
  for (const id of Array.from(this.activeCustomUis.keys())) {
    this.closeCustomUi(id, undefined);
  }
  
  this.pendingUiResponses.clear();
  this.pendingUiRequests.clear();
  this.onDestroyCallback?.();  // Removes from global registry
  notifyRunningChange();
}

The method clears timers, aborts running Bash commands, cancels pending UI promises, closes custom UI components, and removes the session from globalThis.__piSessions.

Forking Strategy

When users fork a conversation, the system creates a new session file and destroys the original wrapper to prevent stale state. The fork logic in the message handler creates a branched .jsonl file, registers the new ID, and immediately terminates the current wrapper:

case "fork": {
  const sessionDir = this.inner.sessionManager.getSessionDir();
  let newSessionFile: string;
  
  if (!entry.parentId) {
    const newManager = SessionManager.create(sessionDir, sessionDir);
    newManager.newSession({ parentSession: currentSessionFile });
    newSessionFile = newManager.getSessionFile() as string;
  } else {
    const sourceManager = SessionManager.open(currentSessionFile, sessionDir);
    newSessionFile = sourceManager.createBranchedSession(entry.parentId);
  }
  
  const newSessionId = SessionManager.open(newSessionFile, sessionDir).getSessionId();
  cacheSessionPath(newSessionId, newSessionFile);
  invalidateSessionListCache();
  
  this.destroy();  // Critical: prevent stale state reuse
  return { cancelled: false, newSessionId };
}

This ensures that the mutated internal state from the original session does not leak into the new forked session.

Practical Implementation Examples

Starting a New Session

API routes use startRpcSession to initialize agent instances:

// app/api/agent/new/route.ts
import { startRpcSession } from "@/lib/rpc-manager";

export async function POST(req: Request) {
  const { cwd, toolNames, initialModel, thinkingLevel } = await req.json();
  const { session } = await startRpcSession(
    "tmp",  // Temporary ID; SDK generates real ID
    "",     // Empty string creates new session file
    cwd,
    { toolNames, initialModel, thinkingLevel }
  );
  return Response.json({ sessionId: session.sessionId });
}

Sending Messages and Handling State

Client-side hooks interact with the wrapper to send prompts:

// hooks/useAgentSession.ts
await session.send({
  type: "prompt",
  message: "Explain the algorithm",
  streamingBehavior: "steer",
});

The wrapper sets internal flags during prompt execution and guarantees notifyRunningChange() updates the UI when processing begins and ends.

Manual Cleanup

Explicit destruction removes sessions immediately:

if (session) session.destroy();

This triggers the full cleanup sequence, removing the session from the global registry and broadcasting the state change to all connected clients.

Summary

• Global registry (globalThis.__piSessions) stores wrapped sessions that persist across Next.js hot-reloads in lib/rpc-manager.ts.
• startRpcSession handles concurrency via locks, initializes SDK services, resolves model scopes, and registers new wrappers with the registry.
• Idle timeout management automatically destroys sessions after 10 minutes of inactivity, preventing resource leaks.
• Real-time synchronization through notifyRunningChange() broadcasts running session IDs to SSE streams and UI components.
• Clean destruction via destroy() aborts commands, cancels promises, clears timers, and removes registry entries.
• Forking safety requires destroying the original wrapper after creating a new session file to prevent state contamination.

Frequently Asked Questions

How does pi-web handle AgentSession persistence during Next.js hot-reloads?

The system attaches the session registry to globalThis.__piSessions rather than module-level variables. Since Next.js preserves globalThis across hot-reloads but resets module state, the getRegistry() function in lib/rpc-manager.ts can retrieve existing sessions after a reload. Process exit handlers ensure graceful cleanup when the development server actually terminates.

What happens when an AgentSession remains idle for too long?

The AgentSessionWrapper starts a 10-minute idle timer whenever a session initializes or receives events. If resetIdleTimer() is not called within that window—meaning no agent events, prompts, or tool results occurred—the timer triggers destroy(). This automatically clears resources, aborts running Bash commands, and removes the session from the global registry.

Why does the fork command destroy the original session wrapper?

Forking creates a new .jsonl session file with branched history, but the underlying AgentSession instance contains mutable state that could leak between conversations. By calling this.destroy() immediately after registering the new session ID, lib/rpc-manager.ts ensures the old wrapper cannot be reused, forcing the client to load the fresh, isolated session state from the new file.

How does pi-web prevent duplicate session creation during concurrent requests?

The startRpcSession function uses a separate lock map (getLocks()) to track in-flight initialization promises by sessionId. When a second request arrives for the same ID before the first completes, the function returns the existing promise from locks.get(sessionId) rather than creating a second AgentSession. Once initialization finishes, the .finally() block removes the lock entry.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}