diff --git a/src/features/tmux-subagent/action-executor-core.ts b/src/features/tmux-subagent/action-executor-core.ts index 70a8f463e..75cc345c6 100644 --- a/src/features/tmux-subagent/action-executor-core.ts +++ b/src/features/tmux-subagent/action-executor-core.ts @@ -10,6 +10,7 @@ export interface ActionResult { export interface ExecuteContext { config: TmuxConfig + directory: string serverUrl: string windowState: WindowState } @@ -55,6 +56,7 @@ export async function executeActionWithDeps( action.description, ctx.config, ctx.serverUrl, + ctx.directory, ) return { success: result.success, @@ -67,6 +69,7 @@ export async function executeActionWithDeps( action.description, ctx.config, ctx.serverUrl, + ctx.directory, action.targetPaneId, action.splitDirection, ) diff --git a/src/features/tmux-subagent/action-executor.ts b/src/features/tmux-subagent/action-executor.ts index 9635ff7cb..0ebbd4378 100644 --- a/src/features/tmux-subagent/action-executor.ts +++ b/src/features/tmux-subagent/action-executor.ts @@ -10,10 +10,7 @@ import { import { getTmuxPath } from "../../tools/interactive-bash/tmux-path-resolver" import { queryWindowState } from "./pane-state-querier" import { log } from "../../shared" -import type { - ActionResult, - ActionExecutorDeps, -} from "./action-executor-core" +import type { ActionResult } from "./action-executor-core" export type { ActionExecutorDeps, ActionResult } from "./action-executor-core" @@ -25,6 +22,7 @@ export interface ExecuteActionsResult { export interface ExecuteContext { config: TmuxConfig + directory: string serverUrl: string windowState: WindowState sourcePaneId?: string @@ -79,10 +77,11 @@ export async function executeAction( const result = await replaceTmuxPane( action.paneId, action.newSessionId, - action.description, - ctx.config, - ctx.serverUrl - ) + action.description, + ctx.config, + ctx.serverUrl, + ctx.directory, + ) if (result.success) { await enforceLayoutAndMainPane(ctx) } @@ -94,12 +93,13 @@ export async function executeAction( const result = await spawnTmuxPane( action.sessionId, - action.description, - ctx.config, - ctx.serverUrl, - action.targetPaneId, - action.splitDirection - ) + action.description, + ctx.config, + ctx.serverUrl, + ctx.directory, + action.targetPaneId, + action.splitDirection + ) if (result.success) { await enforceLayoutAndMainPane(ctx) diff --git a/src/features/tmux-subagent/attachable-session-status.ts b/src/features/tmux-subagent/attachable-session-status.ts new file mode 100644 index 000000000..22dc89770 --- /dev/null +++ b/src/features/tmux-subagent/attachable-session-status.ts @@ -0,0 +1,11 @@ +const ATTACHABLE_SESSION_STATUSES = ["idle", "running"] as const + +export type AttachableSessionStatus = (typeof ATTACHABLE_SESSION_STATUSES)[number] + +export function isAttachableSessionStatus( + status: string | undefined, +): status is AttachableSessionStatus { + return ATTACHABLE_SESSION_STATUSES.some( + (attachableSessionStatus) => attachableSessionStatus === status, + ) +} diff --git a/src/features/tmux-subagent/manager.ts b/src/features/tmux-subagent/manager.ts index 353bdffec..0da723a1c 100644 --- a/src/features/tmux-subagent/manager.ts +++ b/src/features/tmux-subagent/manager.ts @@ -1,26 +1,33 @@ import type { PluginInput } from "@opencode-ai/plugin" import type { TmuxConfig } from "../../config/schema" import type { TrackedSession, CapacityConfig, WindowState } from "./types" -import { log, normalizeSDKResponse } from "../../shared" +import { log } from "../../shared" import { isInsideTmux as defaultIsInsideTmux, getCurrentPaneId as defaultGetCurrentPaneId, POLL_INTERVAL_BACKGROUND_MS, - SESSION_READY_POLL_INTERVAL_MS, - SESSION_READY_TIMEOUT_MS, spawnTmuxWindow, spawnTmuxSession, killTmuxSessionIfExists, getIsolatedSessionName, sweepStaleOmoAgentSessions, } from "../../shared/tmux" -import { queryWindowState } from "./pane-state-querier" +import { queryWindowState as defaultQueryWindowState } from "./pane-state-querier" import { decideSpawnActions, decideCloseAction, type SessionMapping } from "./decision-engine" import { executeActions, executeAction } from "./action-executor" import { TmuxPollingManager } from "./polling-manager" import { createTrackedSession, markTrackedSessionClosePending } from "./tracked-session-state" +import { waitForSessionReady } from "./session-ready-waiter" +import { isAttachableSessionStatus } from "./attachable-session-status" +import { parseSessionStatusMap } from "./session-status-parser" type OpencodeClient = PluginInput["client"] +type SpawnStage = + | "deferred.attach" + | "deferred.isolated-container" + | "session.created" + | "session.idle.retry" + interface SessionCreatedEvent { type: string properties?: { info?: { id?: string; parentID?: string; title?: string } } @@ -33,17 +40,30 @@ interface DeferredSession { retryIsolatedContainer: boolean } +interface FailedReadinessSessionSeed { + sessionId: string + title: string +} + +interface FailedReadinessSession extends FailedReadinessSessionSeed { + rememberedAt: number +} + export interface TmuxUtilDeps { isInsideTmux: () => boolean getCurrentPaneId: () => string | undefined + queryWindowState: (paneId: string) => Promise } const defaultTmuxDeps: TmuxUtilDeps = { isInsideTmux: defaultIsInsideTmux, getCurrentPaneId: defaultGetCurrentPaneId, + queryWindowState: defaultQueryWindowState, } const DEFERRED_SESSION_TTL_MS = 5 * 60 * 1000 +const FAILED_READINESS_SESSION_TTL_MS = 5 * 60 * 1000 +const FAILED_READINESS_SWEEP_INTERVAL_MS = 60 * 1000 const MAX_DEFERRED_QUEUE_SIZE = 20 const MAX_CLOSE_RETRY_COUNT = 3 const MAX_ISOLATED_CONTAINER_NULL_STATE_COUNT = 2 @@ -51,10 +71,14 @@ const MAX_ISOLATED_CONTAINER_NULL_STATE_COUNT = 2 export class TmuxSessionManager { private client: OpencodeClient private tmuxConfig: TmuxConfig + private projectDirectory: string private serverUrl: string private sourcePaneId: string | undefined private sessions = new Map() private pendingSessions = new Set() + private failedReadinessSessions = new Map() + private closedByPolling = new Set() + private failedReadinessSweepInterval?: ReturnType private spawnQueue: Promise = Promise.resolve() private deferredSessions = new Map() private deferredQueue: string[] = [] @@ -71,6 +95,7 @@ export class TmuxSessionManager { constructor(ctx: PluginInput, tmuxConfig: TmuxConfig, deps: TmuxUtilDeps = defaultTmuxDeps) { this.client = ctx.client this.tmuxConfig = tmuxConfig + this.projectDirectory = ctx.directory || process.cwd() this.deps = deps const configuredPort = process.env.OPENCODE_PORT const parsedPort = configuredPort ? Number(configuredPort) : 4096 @@ -98,12 +123,13 @@ export class TmuxSessionManager { this.pollingManager = new TmuxPollingManager( this.client, this.sessions, - this.closeSessionById.bind(this), + this.closeSessionFromPolling.bind(this), this.retryPendingCloses.bind(this) ) log("[tmux-session-manager] initialized", { configEnabled: this.tmuxConfig.enabled, tmuxConfig: this.tmuxConfig, + projectDirectory: this.projectDirectory, serverUrl: this.serverUrl, sourcePaneId: this.sourcePaneId, }) @@ -129,7 +155,7 @@ export class TmuxSessionManager { ): Promise { if (!this.isIsolated()) return null if (this.isolatedWindowPaneId) { - const state = await queryWindowState(this.isolatedWindowPaneId).catch((error) => { + const state = await this.deps.queryWindowState(this.isolatedWindowPaneId).catch((error) => { log("[tmux-session-manager] failed to query isolated window state", { paneId: this.isolatedWindowPaneId, error: String(error), @@ -158,8 +184,8 @@ export class TmuxSessionManager { log("[tmux-session-manager] creating isolated tmux container", { isolation, sessionId, title }) const result = isolation === "session" - ? await spawnTmuxSession(sessionId, title, this.tmuxConfig, this.serverUrl, this.sourcePaneId) - : await spawnTmuxWindow(sessionId, title, this.tmuxConfig, this.serverUrl) + ? await spawnTmuxSession(sessionId, title, this.tmuxConfig, this.serverUrl, this.projectDirectory, this.sourcePaneId) + : await spawnTmuxWindow(sessionId, title, this.tmuxConfig, this.serverUrl, this.projectDirectory) if (result.success && result.paneId) { this.isolatedContainerPaneId = result.paneId @@ -196,6 +222,10 @@ export class TmuxSessionManager { return this.sessions.get(sessionId)?.paneId } + getServerUrl(): string { + return this.serverUrl + } + private removeTrackedSession(sessionId: string): void { this.sessions.delete(sessionId) @@ -250,6 +280,7 @@ export class TmuxSessionManager { { type: "close", paneId: isolatedContainerPaneId, sessionId: tracked.sessionId }, { config: this.tmuxConfig, + directory: this.projectDirectory, serverUrl: this.serverUrl, windowState: state, sourcePaneId: this.sourcePaneId ?? tracked.paneId, @@ -288,7 +319,7 @@ export class TmuxSessionManager { if (!paneId) return null try { - return await queryWindowState(paneId) + return await this.deps.queryWindowState(paneId) } catch (error) { log("[tmux-session-manager] failed to query window state for close", { error: String(error), @@ -297,6 +328,47 @@ export class TmuxSessionManager { } } + private windowStateContainsPane(state: WindowState, paneId: string): boolean { + return state.mainPane?.paneId === paneId + || state.agentPanes.some((pane) => pane.paneId === paneId) + } + + private async finalizeForceRemoveCandidate( + tracked: TrackedSession, + source: string, + ): Promise { + const state = await this.queryWindowStateSafely() + if (!state) { + log("[tmux-session-manager] unable to verify pane after max close retries; keeping session tracked", { + sessionId: tracked.sessionId, + paneId: tracked.paneId, + source, + }) + return false + } + + if (this.windowStateContainsPane(state, tracked.paneId)) { + log("[tmux-session-manager] pane still exists after max close retries; manual intervention required", { + sessionId: tracked.sessionId, + paneId: tracked.paneId, + source, + }) + return false + } + + log("[tmux-session-manager] pane already gone after max close retries; finalizing tracked close", { + sessionId: tracked.sessionId, + paneId: tracked.paneId, + source, + }) + await this.finalizeTrackedSessionClose({ + tracked, + state, + isolatedPaneAlreadyClosed: true, + }) + return true + } + private async closeTrackedSessionPane(args: { tracked: TrackedSession state: WindowState @@ -308,6 +380,7 @@ export class TmuxSessionManager { { type: "close", paneId: tracked.paneId, sessionId: tracked.sessionId }, { config: this.tmuxConfig, + directory: this.projectDirectory, serverUrl: this.serverUrl, windowState: state, sourcePaneId: this.getEffectiveSourcePaneId(), @@ -365,12 +438,7 @@ export class TmuxSessionManager { if (!this.sessions.has(tracked.sessionId)) continue if (tracked.closeRetryCount >= MAX_CLOSE_RETRY_COUNT) { - log("[tmux-session-manager] force removing close-pending session after max retries", { - sessionId: tracked.sessionId, - paneId: tracked.paneId, - closeRetryCount: tracked.closeRetryCount, - }) - this.removeTrackedSession(tracked.sessionId) + await this.finalizeForceRemoveCandidate(tracked, "retryPendingCloses.max-retries") continue } @@ -391,12 +459,7 @@ export class TmuxSessionManager { const nextRetryCount = currentTracked.closeRetryCount + 1 if (nextRetryCount >= MAX_CLOSE_RETRY_COUNT) { - log("[tmux-session-manager] force removing close-pending session after failed retry", { - sessionId: currentTracked.sessionId, - paneId: currentTracked.paneId, - closeRetryCount: nextRetryCount, - }) - this.removeTrackedSession(currentTracked.sessionId) + await this.finalizeForceRemoveCandidate(currentTracked, "retryPendingCloses.failed-retry") continue } @@ -418,6 +481,11 @@ export class TmuxSessionManager { title: string, retryIsolatedContainer = false, ): void { + if (this.shouldSkipRespawnAfterPollingClose(sessionId, "deferred enqueue")) { + this.clearFailedReadinessSession(sessionId) + return + } + const existingDeferredSession = this.deferredSessions.get(sessionId) if (existingDeferredSession) { if (retryIsolatedContainer && !existingDeferredSession.retryIsolatedContainer) { @@ -490,6 +558,376 @@ export class TmuxSessionManager { log("[tmux-session-manager] deferred attach polling stopped") } + private beginPendingSession( + sessionId: string, + options?: { allowDeferredSession?: boolean }, + ): boolean { + if ( + this.sessions.has(sessionId) + || this.pendingSessions.has(sessionId) + || (!options?.allowDeferredSession && this.deferredSessions.has(sessionId)) + ) { + log("[tmux-session-manager] session already tracked or pending", { sessionId }) + return false + } + + this.pendingSessions.add(sessionId) + return true + } + + private async ensureSessionReadyBeforeSpawn( + sessionId: string, + stage: SpawnStage, + ): Promise { + try { + const ready = await waitForSessionReady({ + client: this.client, + sessionId, + }) + + if (ready) { + return true + } + + const readinessError = new Error("Session readiness timed out") + log("[tmux-session-manager] session readiness failed before spawn", { + sessionId, + stage, + error: String(readinessError), + }) + return false + } catch (error) { + log("[tmux-session-manager] session readiness failed before spawn", { + sessionId, + stage, + error: String(error), + }) + return false + } + } + + private async getSessionStatusType(sessionId: string): Promise { + try { + const statusResult = await this.client.session.status({ path: undefined }) + const allStatuses = parseSessionStatusMap(statusResult.data) + return allStatuses[sessionId]?.type + } catch (error) { + log("[tmux-session-manager] failed to read session status before spawn", { + sessionId, + error: String(error), + }) + return undefined + } + } + + private rememberFailedReadinessSession( + session: FailedReadinessSessionSeed, + ): void { + this.failedReadinessSessions.set(session.sessionId, { + ...session, + rememberedAt: Date.now(), + }) + this.startFailedReadinessSweep() + } + + private clearFailedReadinessSession(sessionId: string): void { + this.failedReadinessSessions.delete(sessionId) + if (this.failedReadinessSessions.size === 0) { + this.stopFailedReadinessSweep() + } + } + + private startFailedReadinessSweep(): void { + if (this.failedReadinessSweepInterval) { + return + } + + this.failedReadinessSweepInterval = setInterval(() => { + this.sweepExpiredFailedReadinessSessions() + }, FAILED_READINESS_SWEEP_INTERVAL_MS) + } + + private stopFailedReadinessSweep(): void { + if (!this.failedReadinessSweepInterval) { + return + } + + clearInterval(this.failedReadinessSweepInterval) + this.failedReadinessSweepInterval = undefined + } + + private isFailedReadinessSessionExpired( + session: FailedReadinessSession, + now: number, + ): boolean { + return now - session.rememberedAt >= FAILED_READINESS_SESSION_TTL_MS + } + + private sweepExpiredFailedReadinessSessions(): void { + const now = Date.now() + + for (const [sessionId, failedReadinessSession] of this.failedReadinessSessions.entries()) { + if (!this.isFailedReadinessSessionExpired(failedReadinessSession, now)) { + continue + } + + this.failedReadinessSessions.delete(sessionId) + log("[tmux-session-manager] expired failed readiness session", { + sessionId, + ttlMs: FAILED_READINESS_SESSION_TTL_MS, + }) + } + + if (this.failedReadinessSessions.size === 0) { + this.stopFailedReadinessSweep() + } + } + + private getFailedReadinessSession(sessionId: string): FailedReadinessSession | undefined { + const failedReadinessSession = this.failedReadinessSessions.get(sessionId) + if (!failedReadinessSession) { + return undefined + } + + if (!this.isFailedReadinessSessionExpired(failedReadinessSession, Date.now())) { + return failedReadinessSession + } + + this.failedReadinessSessions.delete(sessionId) + log("[tmux-session-manager] expired failed readiness session on access", { + sessionId, + ttlMs: FAILED_READINESS_SESSION_TTL_MS, + }) + + if (this.failedReadinessSessions.size === 0) { + this.stopFailedReadinessSweep() + } + + return undefined + } + + private async spawnPendingSession(args: { + session: FailedReadinessSessionSeed + stage: SpawnStage + rememberReadinessFailure: boolean + }): Promise { + const { session, stage, rememberReadinessFailure } = args + const { sessionId, title } = session + + const readyForSpawn = await this.ensureSessionReadyBeforeSpawn(sessionId, stage) + if (!readyForSpawn) { + if (rememberReadinessFailure) { + this.rememberFailedReadinessSession(session) + } + return + } + + const sessionStatus = await this.getSessionStatusType(sessionId) + if (!isAttachableSessionStatus(sessionStatus)) { + log("[tmux-session-manager] session not attachable for pane spawn", { + sessionId, + stage, + status: sessionStatus, + }) + if (rememberReadinessFailure) { + this.rememberFailedReadinessSession(session) + } + return + } + + this.clearFailedReadinessSession(sessionId) + + const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, title) + if (isolatedPaneId) { + this.sessions.set( + sessionId, + createTrackedSession({ sessionId, paneId: isolatedPaneId, description: title }), + ) + this.pollingManager.startPolling() + log("[tmux-session-manager] first subagent spawned in isolated window", { + sessionId, + paneId: isolatedPaneId, + }) + return + } + + if (this.isIsolated() && !this.isolatedWindowPaneId) { + log("[tmux-session-manager] isolated container failed, deferring session for retry", { sessionId }) + this.enqueueDeferredSession(sessionId, title, true) + return + } + const sourcePaneId = this.getEffectiveSourcePaneId() + if (!sourcePaneId) { + log("[tmux-session-manager] no effective source pane id") + return + } + + const state = await this.deps.queryWindowState(sourcePaneId) + if (!state) { + log("[tmux-session-manager] failed to query window state, deferring session") + this.enqueueDeferredSession(sessionId, title) + return + } + + log("[tmux-session-manager] window state queried", { + windowWidth: state.windowWidth, + mainPane: state.mainPane?.paneId, + agentPaneCount: state.agentPanes.length, + agentPanes: state.agentPanes.map((pane) => pane.paneId), + }) + + const decision = decideSpawnActions( + state, + sessionId, + title, + this.getCapacityConfig(), + this.getSessionMappings(), + ) + + log("[tmux-session-manager] spawn decision", { + canSpawn: decision.canSpawn, + reason: decision.reason, + actionCount: decision.actions.length, + actions: decision.actions.map((action) => { + if (action.type === "close") return { type: "close", paneId: action.paneId } + if (action.type === "replace") { + return { + type: "replace", + paneId: action.paneId, + newSessionId: action.newSessionId, + } + } + return { type: "spawn", sessionId: action.sessionId } + }), + }) + + if (!decision.canSpawn) { + log("[tmux-session-manager] cannot spawn", { reason: decision.reason }) + this.enqueueDeferredSession(sessionId, title) + return + } + + const result = await executeActions( + decision.actions, + { + config: this.tmuxConfig, + directory: this.projectDirectory, + serverUrl: this.serverUrl, + windowState: state, + sourcePaneId, + }, + ) + + for (const { action, result: actionResult } of result.results) { + if (action.type === "close" && actionResult.success) { + this.sessions.delete(action.sessionId) + log("[tmux-session-manager] removed closed session from cache", { + sessionId: action.sessionId, + }) + } + if (action.type === "replace" && actionResult.success) { + this.sessions.delete(action.oldSessionId) + log("[tmux-session-manager] removed replaced session from cache", { + oldSessionId: action.oldSessionId, + newSessionId: action.newSessionId, + }) + } + } + + if (result.success && result.spawnedPaneId) { + this.sessions.set( + sessionId, + createTrackedSession({ + sessionId, + paneId: result.spawnedPaneId, + description: title, + }), + ) + this.clearFailedReadinessSession(sessionId) + log("[tmux-session-manager] pane spawned and tracked", { + sessionId, + paneId: result.spawnedPaneId, + }) + this.pollingManager.startPolling() + return + } + + log("[tmux-session-manager] spawn failed", { + success: result.success, + results: result.results.map((resultEntry) => ({ + type: resultEntry.action.type, + success: resultEntry.result.success, + error: resultEntry.result.error, + })), + }) + + log("[tmux-session-manager] re-queueing deferred session after spawn failure", { + sessionId, + }) + this.enqueueDeferredSession(sessionId, title) + + if (result.spawnedPaneId) { + await executeAction( + { type: "close", paneId: result.spawnedPaneId, sessionId }, + { + config: this.tmuxConfig, + directory: this.projectDirectory, + serverUrl: this.serverUrl, + windowState: state, + }, + ) + } + } + + private getEventSessionId(event: { + type: string + properties?: Record + }): string | undefined { + const sessionId = event.properties?.sessionID + return typeof sessionId === "string" ? sessionId : undefined + } + + private async retryFailedReadinessSession(sessionId: string): Promise { + if (this.shouldSkipRespawnAfterPollingClose(sessionId, "session.idle retry")) { + return + } + + const failedReadinessSession = this.getFailedReadinessSession(sessionId) + if (!failedReadinessSession) { + return + } + + if (!this.beginPendingSession(sessionId)) { + return + } + + try { + await this.enqueueSpawn(async () => { + try { + const sessionStatus = await this.getSessionStatusType(sessionId) + if (!isAttachableSessionStatus(sessionStatus)) { + log("[tmux-session-manager] session.idle retry skipped because session is not attachable", { + sessionId, + status: sessionStatus, + }) + return + } + + this.clearFailedReadinessSession(sessionId) + await this.spawnPendingSession({ + session: failedReadinessSession, + stage: "session.idle.retry", + rememberReadinessFailure: false, + }) + } finally { + this.pendingSessions.delete(sessionId) + } + }) + } finally { + this.pendingSessions.delete(sessionId) + } + } + private async tryAttachDeferredSession(): Promise { const sessionId = this.deferredQueue[0] if (!sessionId) { @@ -503,151 +941,142 @@ export class TmuxSessionManager { return } - if (Date.now() - deferred.queuedAt.getTime() > DEFERRED_SESSION_TTL_MS) { - this.deferredQueue.shift() - this.deferredSessions.delete(sessionId) - log("[tmux-session-manager] deferred session expired", { - sessionId, - queuedAt: deferred.queuedAt.toISOString(), - ttlMs: DEFERRED_SESSION_TTL_MS, - queueLength: this.deferredQueue.length, - }) - if (this.deferredQueue.length === 0) { - this.stopDeferredAttachLoop() - } + if (this.shouldSkipRespawnAfterPollingClose(sessionId, "deferred attach")) { + this.removeDeferredSession(sessionId) return } - if (deferred.retryIsolatedContainer) { - const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, deferred.title) - if (isolatedPaneId) { - this.sessions.set( + if (!this.beginPendingSession(sessionId, { allowDeferredSession: true })) { + return + } + + try { + if (Date.now() - deferred.queuedAt.getTime() > DEFERRED_SESSION_TTL_MS) { + this.deferredQueue.shift() + this.deferredSessions.delete(sessionId) + log("[tmux-session-manager] deferred session expired", { sessionId, - createTrackedSession({ - sessionId, - paneId: isolatedPaneId, - description: deferred.title, - }), - ) - this.removeDeferredSession(sessionId) - this.pollingManager.startPolling() - log("[tmux-session-manager] deferred session attached in isolated window", { - sessionId, - paneId: isolatedPaneId, + queuedAt: deferred.queuedAt.toISOString(), + ttlMs: DEFERRED_SESSION_TTL_MS, + queueLength: this.deferredQueue.length, }) - this.logSessionReadinessInBackground(sessionId) + if (this.deferredQueue.length === 0) { + this.stopDeferredAttachLoop() + } return } - } - const effectiveSourcePaneId = this.getEffectiveSourcePaneId() - if (!effectiveSourcePaneId) return + if (deferred.retryIsolatedContainer) { + const readyForIsolatedContainer = await this.ensureSessionReadyBeforeSpawn( + sessionId, + "deferred.isolated-container", + ) + if (!readyForIsolatedContainer) { + this.removeDeferredSession(sessionId) + return + } - const state = await queryWindowState(effectiveSourcePaneId) - if (!state) { - this.nullStateCount += 1 - log("[tmux-session-manager] deferred attach window state is null", { - nullStateCount: this.nullStateCount, - }) - if (this.nullStateCount >= 3) { - log("[tmux-session-manager] stopping deferred attach loop after consecutive null states", { + const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, deferred.title) + if (isolatedPaneId) { + this.sessions.set( + sessionId, + createTrackedSession({ + sessionId, + paneId: isolatedPaneId, + description: deferred.title, + }), + ) + this.removeDeferredSession(sessionId) + this.pollingManager.startPolling() + log("[tmux-session-manager] deferred session attached in isolated window", { + sessionId, + paneId: isolatedPaneId, + }) + return + } + } + + const effectiveSourcePaneId = this.getEffectiveSourcePaneId() + if (!effectiveSourcePaneId) return + + const state = await this.deps.queryWindowState(effectiveSourcePaneId) + if (!state) { + this.nullStateCount += 1 + log("[tmux-session-manager] deferred attach window state is null", { nullStateCount: this.nullStateCount, }) - this.stopDeferredAttachLoop() + if (this.nullStateCount >= 3) { + log("[tmux-session-manager] stopping deferred attach loop after consecutive null states", { + nullStateCount: this.nullStateCount, + }) + this.stopDeferredAttachLoop() + } + return } - return - } - this.nullStateCount = 0 + this.nullStateCount = 0 - const decision = decideSpawnActions( - state, - sessionId, - deferred.title, - this.getCapacityConfig(), - this.getSessionMappings(), - ) - - if (!decision.canSpawn || decision.actions.length === 0) { - log("[tmux-session-manager] deferred session still waiting for capacity", { + const decision = decideSpawnActions( + state, sessionId, - reason: decision.reason, - }) - return - } + deferred.title, + this.getCapacityConfig(), + this.getSessionMappings(), + ) - const result = await executeActions(decision.actions, { - config: this.tmuxConfig, - serverUrl: this.serverUrl, - windowState: state, - sourcePaneId: effectiveSourcePaneId, - }) + if (!decision.canSpawn || decision.actions.length === 0) { + log("[tmux-session-manager] deferred session still waiting for capacity", { + sessionId, + reason: decision.reason, + }) + return + } - if (!result.success || !result.spawnedPaneId) { - log("[tmux-session-manager] deferred session attach failed", { + const readyForDeferredAttach = await this.ensureSessionReadyBeforeSpawn( sessionId, - results: result.results.map((r) => ({ - type: r.action.type, - success: r.result.success, - error: r.result.error, - })), - }) - return - } + "deferred.attach", + ) + if (!readyForDeferredAttach) { + this.removeDeferredSession(sessionId) + return + } - this.sessions.set( - sessionId, - createTrackedSession({ + const result = await executeActions(decision.actions, { + config: this.tmuxConfig, + directory: this.projectDirectory, + serverUrl: this.serverUrl, + windowState: state, + sourcePaneId: effectiveSourcePaneId, + }) + + if (!result.success || !result.spawnedPaneId) { + log("[tmux-session-manager] deferred session attach failed", { + sessionId, + results: result.results.map((r) => ({ + type: r.action.type, + success: r.result.success, + error: r.result.error, + })), + }) + return + } + + this.sessions.set( + sessionId, + createTrackedSession({ + sessionId, + paneId: result.spawnedPaneId, + description: deferred.title, + }), + ) + this.removeDeferredSession(sessionId) + this.pollingManager.startPolling() + log("[tmux-session-manager] deferred session attached", { sessionId, paneId: result.spawnedPaneId, - description: deferred.title, - }), - ) - this.removeDeferredSession(sessionId) - this.pollingManager.startPolling() - log("[tmux-session-manager] deferred session attached", { - sessionId, - paneId: result.spawnedPaneId, - }) - this.logSessionReadinessInBackground(sessionId) - } - - private logSessionReadinessInBackground(sessionId: string): void { - void this.waitForSessionReady(sessionId).catch((error) => { - log("[tmux-session-manager] background readiness probe failed", { - sessionId, - error: String(error), }) - }) - } - - private async waitForSessionReady(sessionId: string): Promise { - const startTime = Date.now() - - while (Date.now() - startTime < SESSION_READY_TIMEOUT_MS) { - try { - const statusResult = await this.client.session.status({ path: undefined }) - const allStatuses = normalizeSDKResponse(statusResult, {} as Record) - - if (allStatuses[sessionId]) { - log("[tmux-session-manager] session ready", { - sessionId, - status: allStatuses[sessionId].type, - waitedMs: Date.now() - startTime, - }) - return true - } - } catch (err) { - log("[tmux-session-manager] session status check error", { error: String(err) }) - } - - await new Promise((resolve) => setTimeout(resolve, SESSION_READY_POLL_INTERVAL_MS)) + } finally { + this.pendingSessions.delete(sessionId) } - - log("[tmux-session-manager] session ready timeout", { - sessionId, - timeoutMs: SESSION_READY_TIMEOUT_MS, - }) - return false } async onSessionCreated(event: SessionCreatedEvent): Promise { @@ -675,156 +1104,30 @@ export class TmuxSessionManager { return } - await this.sweepStaleIsolatedSessionsOnce() - await this.retryPendingCloses() - - if ( - this.sessions.has(sessionId) || - this.pendingSessions.has(sessionId) || - this.deferredSessions.has(sessionId) - ) { - log("[tmux-session-manager] session already tracked or pending", { sessionId }) + if (!this.beginPendingSession(sessionId)) { return } - this.pendingSessions.add(sessionId) + try { + await this.sweepStaleIsolatedSessionsOnce() + await this.retryPendingCloses() - await this.enqueueSpawn(async () => { - try { - const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, title) - if (isolatedPaneId) { - this.sessions.set( - sessionId, - createTrackedSession({ sessionId, paneId: isolatedPaneId, description: title }), - ) - this.pollingManager.startPolling() - log("[tmux-session-manager] first subagent spawned in isolated window", { - sessionId, - paneId: isolatedPaneId, + const session = { sessionId, title } + + await this.enqueueSpawn(async () => { + try { + await this.spawnPendingSession({ + session, + stage: "session.created", + rememberReadinessFailure: true, }) - this.logSessionReadinessInBackground(sessionId) - return + } finally { + this.pendingSessions.delete(sessionId) } - - if (this.isIsolated() && !this.isolatedWindowPaneId) { - log("[tmux-session-manager] isolated container failed, deferring session for retry", { sessionId }) - this.enqueueDeferredSession(sessionId, title, true) - return - } - const sourcePaneId = this.getEffectiveSourcePaneId() - if (!sourcePaneId) { - log("[tmux-session-manager] no effective source pane id") - return - } - - const state = await queryWindowState(sourcePaneId) - if (!state) { - log("[tmux-session-manager] failed to query window state, deferring session") - this.enqueueDeferredSession(sessionId, title) - return - } - - log("[tmux-session-manager] window state queried", { - windowWidth: state.windowWidth, - mainPane: state.mainPane?.paneId, - agentPaneCount: state.agentPanes.length, - agentPanes: state.agentPanes.map((p) => p.paneId), }) - - const decision = decideSpawnActions( - state, - sessionId, - title, - this.getCapacityConfig(), - this.getSessionMappings() - ) - - log("[tmux-session-manager] spawn decision", { - canSpawn: decision.canSpawn, - reason: decision.reason, - actionCount: decision.actions.length, - actions: decision.actions.map((a) => { - if (a.type === "close") return { type: "close", paneId: a.paneId } - if (a.type === "replace") return { type: "replace", paneId: a.paneId, newSessionId: a.newSessionId } - return { type: "spawn", sessionId: a.sessionId } - }), - }) - - if (!decision.canSpawn) { - log("[tmux-session-manager] cannot spawn", { reason: decision.reason }) - this.enqueueDeferredSession(sessionId, title) - return - } - - const result = await executeActions( - decision.actions, - { - config: this.tmuxConfig, - serverUrl: this.serverUrl, - windowState: state, - sourcePaneId, - } - ) - - for (const { action, result: actionResult } of result.results) { - if (action.type === "close" && actionResult.success) { - this.sessions.delete(action.sessionId) - log("[tmux-session-manager] removed closed session from cache", { - sessionId: action.sessionId, - }) - } - if (action.type === "replace" && actionResult.success) { - this.sessions.delete(action.oldSessionId) - log("[tmux-session-manager] removed replaced session from cache", { - oldSessionId: action.oldSessionId, - newSessionId: action.newSessionId, - }) - } - } - - if (result.success && result.spawnedPaneId) { - this.sessions.set( - sessionId, - createTrackedSession({ - sessionId, - paneId: result.spawnedPaneId, - description: title, - }), - ) - log("[tmux-session-manager] pane spawned and tracked", { - sessionId, - paneId: result.spawnedPaneId, - }) - this.pollingManager.startPolling() - this.logSessionReadinessInBackground(sessionId) - } else { - log("[tmux-session-manager] spawn failed", { - success: result.success, - results: result.results.map((r) => ({ - type: r.action.type, - success: r.result.success, - error: r.result.error, - })), - }) - - log("[tmux-session-manager] re-queueing deferred session after spawn failure", { - sessionId, - }) - this.enqueueDeferredSession(sessionId, title) - - if (result.spawnedPaneId) { - await executeAction( - { type: "close", paneId: result.spawnedPaneId, sessionId }, - { config: this.tmuxConfig, serverUrl: this.serverUrl, windowState: state } - ) - } - - return - } - } finally { - this.pendingSessions.delete(sessionId) - } - }) + } finally { + this.pendingSessions.delete(sessionId) + } } private async enqueueSpawn(run: () => Promise): Promise { @@ -845,10 +1148,13 @@ export class TmuxSessionManager { async onSessionDeleted(event: { sessionID: string }): Promise { if (!this.isEnabled()) return - if (!this.getEffectiveSourcePaneId()) return + this.closedByPolling.delete(event.sessionID) + this.clearFailedReadinessSession(event.sessionID) this.removeDeferredSession(event.sessionID) + if (!this.getEffectiveSourcePaneId()) return + const tracked = this.sessions.get(event.sessionID) if (!tracked) return @@ -876,6 +1182,7 @@ export class TmuxSessionManager { try { const result = await executeAction(closeAction, { config: this.tmuxConfig, + directory: this.projectDirectory, serverUrl: this.serverUrl, windowState: state, sourcePaneId: this.getEffectiveSourcePaneId(), @@ -907,12 +1214,7 @@ export class TmuxSessionManager { if (!tracked) return if (tracked.closePending && tracked.closeRetryCount >= MAX_CLOSE_RETRY_COUNT) { - log("[tmux-session-manager] force removing close-pending session after max retries", { - sessionId, - paneId: tracked.paneId, - closeRetryCount: tracked.closeRetryCount, - }) - this.removeTrackedSession(sessionId) + await this.finalizeForceRemoveCandidate(tracked, "closeSessionById.max-retries") return } @@ -928,8 +1230,37 @@ export class TmuxSessionManager { } } + private async closeSessionFromPolling(sessionId: string): Promise { + this.closedByPolling.add(sessionId) + await this.closeSessionById(sessionId) + } + + private shouldSkipRespawnAfterPollingClose(sessionId: string, source: string): boolean { + if (!this.closedByPolling.has(sessionId)) { + return false + } + + log("[tmux-session-manager] skipping tmux respawn because polling already closed the session", { + sessionId, + source, + }) + return true + } + onEvent(event: { type: string; properties?: Record }): void { this.pollingManager.handleEvent(event) + + const sessionId = this.getEventSessionId(event) + if (event.type !== "session.idle" || !sessionId) { + return + } + + void this.retryFailedReadinessSession(sessionId).catch((error) => { + log("[tmux-session-manager] session.idle retry failed", { + sessionId, + error: String(error), + }) + }) } createEventHandler(): (input: { event: { type: string; properties?: unknown } }) => Promise { @@ -942,6 +1273,9 @@ export class TmuxSessionManager { this.stopDeferredAttachLoop() this.deferredQueue = [] this.deferredSessions.clear() + this.failedReadinessSessions.clear() + this.closedByPolling.clear() + this.stopFailedReadinessSweep() this.pollingManager.stopPolling() if (this.sessions.size > 0) { diff --git a/src/features/tmux-subagent/pane-state-querier.ts b/src/features/tmux-subagent/pane-state-querier.ts index 0d01b03c0..add949bc3 100644 --- a/src/features/tmux-subagent/pane-state-querier.ts +++ b/src/features/tmux-subagent/pane-state-querier.ts @@ -1,4 +1,3 @@ -import { spawn } from "../../shared/bun-spawn-shim" import type { WindowState, TmuxPaneInfo } from "./types" import { parsePaneStateOutput } from "./pane-state-parser" import { getTmuxPath } from "../../tools/interactive-bash/tmux-path-resolver" @@ -7,28 +6,22 @@ import { log } from "../../shared" export async function queryWindowState(sourcePaneId: string): Promise { const tmux = await getTmuxPath() if (!tmux) return null + const { runTmuxCommand } = await import("../../shared/tmux") - const proc = spawn( - [ - tmux, - "list-panes", - "-t", - sourcePaneId, - "-F", - "#{pane_id}\t#{pane_width}\t#{pane_height}\t#{pane_left}\t#{pane_top}\t#{pane_active}\t#{window_width}\t#{window_height}\t#{pane_title}", - ], - { stdout: "pipe", stderr: "pipe" } - ) + const result = await runTmuxCommand(tmux, [ + "list-panes", + "-t", + sourcePaneId, + "-F", + "#{pane_id}\t#{pane_width}\t#{pane_height}\t#{pane_left}\t#{pane_top}\t#{pane_active}\t#{window_width}\t#{window_height}\t#{pane_title}", + ]) - const exitCode = await proc.exited - const stdout = await new Response(proc.stdout).text() + if (result.exitCode !== 0) { + log("[pane-state-querier] list-panes failed", { exitCode: result.exitCode }) + return null + } - if (exitCode !== 0) { - log("[pane-state-querier] list-panes failed", { exitCode }) - return null - } - - const parsedPaneState = parsePaneStateOutput(stdout) + const parsedPaneState = parsePaneStateOutput(result.output) if (!parsedPaneState) { log("[pane-state-querier] failed to parse pane state output", { sourcePaneId, diff --git a/src/features/tmux-subagent/polling-manager.ts b/src/features/tmux-subagent/polling-manager.ts index 1a74be801..74e017c5e 100644 --- a/src/features/tmux-subagent/polling-manager.ts +++ b/src/features/tmux-subagent/polling-manager.ts @@ -1,11 +1,13 @@ import type { OpencodeClient } from "../../tools/delegate-task/types" -import { POLL_INTERVAL_BACKGROUND_MS } from "../../shared/tmux" +import { + POLL_INTERVAL_BACKGROUND_MS, + SESSION_MISSING_GRACE_MS, + SESSION_TIMEOUT_MS, +} from "../../shared/tmux" import type { TrackedSession } from "./types" -import { SESSION_MISSING_GRACE_MS } from "../../shared/tmux" import { log } from "../../shared" import { normalizeSDKResponse } from "../../shared" -const SESSION_TIMEOUT_MS = 10 * 60 * 1000 const MIN_STABILITY_TIME_MS = 10 * 1000 const STABLE_POLLS_REQUIRED = 3 @@ -86,30 +88,42 @@ export class TmuxPollingManager { if (isIdle && elapsedMs >= MIN_STABILITY_TIME_MS) { const activityVersion = tracked.activityVersion ?? 0 - if (tracked.observedIdleActivityVersion === activityVersion) { - tracked.stableIdlePolls = (tracked.stableIdlePolls ?? 0) + 1 - - if (tracked.stableIdlePolls >= STABLE_POLLS_REQUIRED) { - const recheckResult = await this.client.session.status({ path: undefined }) - const recheckStatuses = normalizeSDKResponse(recheckResult, {} as Record) - const recheckStatus = recheckStatuses[sessionId] - - if (recheckStatus?.type === "idle") { - shouldCloseViaStability = true - } else { - tracked.stableIdlePolls = 0 - log("[tmux-session-manager] stability reached but session not idle on recheck, resetting", { - sessionId, - recheckStatus: recheckStatus?.type, - }) - } - } - } else { - tracked.stableIdlePolls = 0 + if (tracked.observedIdleActivityVersion !== activityVersion) { + tracked.stableIdlePolls = 1 tracked.observedIdleActivityVersion = activityVersion + } else { + tracked.stableIdlePolls = (tracked.stableIdlePolls ?? 0) + 1 + } + + if ((tracked.stableIdlePolls ?? 0) >= STABLE_POLLS_REQUIRED) { + const stableWindowActivityVersion = tracked.observedIdleActivityVersion ?? activityVersion + const recheckResult = await this.client.session.status({ path: undefined }) + const recheckStatuses = normalizeSDKResponse(recheckResult, {} as Record) + const recheckStatus = recheckStatuses[sessionId] + const latestTracked = this.sessions.get(sessionId) ?? tracked + const recheckActivityVersion = latestTracked.activityVersion ?? 0 + + if (recheckActivityVersion !== stableWindowActivityVersion) { + latestTracked.stableIdlePolls = 0 + latestTracked.observedIdleActivityVersion = recheckActivityVersion + log("[tmux-session-manager] stability recheck aborted after new activity", { + sessionId, + stableWindowActivityVersion, + recheckActivityVersion, + }) + } else if (recheckStatus?.type === "idle") { + shouldCloseViaStability = true + } else { + latestTracked.stableIdlePolls = 0 + log("[tmux-session-manager] stability reached but session not idle on recheck, resetting", { + sessionId, + recheckStatus: recheckStatus?.type, + }) + } } } else if (!isIdle) { tracked.stableIdlePolls = 0 + tracked.observedIdleActivityVersion = undefined } log("[tmux-session-manager] session check", { @@ -126,7 +140,8 @@ export class TmuxPollingManager { shouldCloseViaStability, }) - if (shouldCloseViaStability || missingTooLong || isTimedOut) { + if (!tracked.closePending && (shouldCloseViaStability || missingTooLong || isTimedOut)) { + tracked.closePending = true sessionsToClose.push(sessionId) } } diff --git a/src/features/tmux-subagent/polling.ts b/src/features/tmux-subagent/polling.ts index a438be488..a8b3dd925 100644 --- a/src/features/tmux-subagent/polling.ts +++ b/src/features/tmux-subagent/polling.ts @@ -30,6 +30,7 @@ export interface SessionPollingController { export function createSessionPollingController(params: { client: OpencodeClient tmuxConfig: TmuxConfig + directory: string serverUrl: string sourcePaneId: string | undefined sessions: Map @@ -49,7 +50,12 @@ export function createSessionPollingController(params: { if (state) { await executeAction( { type: "close", paneId: tracked.paneId, sessionId }, - { config: params.tmuxConfig, serverUrl: params.serverUrl, windowState: state }, + { + config: params.tmuxConfig, + directory: params.directory, + serverUrl: params.serverUrl, + windowState: state, + }, ) } diff --git a/src/features/tmux-subagent/session-ready-waiter.ts b/src/features/tmux-subagent/session-ready-waiter.ts index d98757c5d..a9f802bc2 100644 --- a/src/features/tmux-subagent/session-ready-waiter.ts +++ b/src/features/tmux-subagent/session-ready-waiter.ts @@ -4,6 +4,7 @@ import { SESSION_READY_TIMEOUT_MS, } from "../../shared/tmux" import { log } from "../../shared" +import { isAttachableSessionStatus } from "./attachable-session-status" import { parseSessionStatusMap } from "./session-status-parser" type OpencodeClient = PluginInput["client"] @@ -18,11 +19,12 @@ export async function waitForSessionReady(params: { try { const statusResult = await params.client.session.status({ path: undefined }) const allStatuses = parseSessionStatusMap(statusResult.data) + const sessionStatus = allStatuses[params.sessionId]?.type - if (allStatuses[params.sessionId]) { + if (isAttachableSessionStatus(sessionStatus)) { log("[tmux-session-manager] session ready", { sessionId: params.sessionId, - status: allStatuses[params.sessionId].type, + status: sessionStatus, waitedMs: Date.now() - startTime, }) return true