refactor(tmux-subagent): stabilize polling, execution and session lifecycle with runner

This commit is contained in:
YeonGyu-Kim
2026-04-28 10:45:17 +09:00
parent 90e6e0cdce
commit c8b225099c
8 changed files with 719 additions and 355 deletions
@@ -10,6 +10,7 @@ export interface ActionResult {
export interface ExecuteContext { export interface ExecuteContext {
config: TmuxConfig config: TmuxConfig
directory: string
serverUrl: string serverUrl: string
windowState: WindowState windowState: WindowState
} }
@@ -55,6 +56,7 @@ export async function executeActionWithDeps(
action.description, action.description,
ctx.config, ctx.config,
ctx.serverUrl, ctx.serverUrl,
ctx.directory,
) )
return { return {
success: result.success, success: result.success,
@@ -67,6 +69,7 @@ export async function executeActionWithDeps(
action.description, action.description,
ctx.config, ctx.config,
ctx.serverUrl, ctx.serverUrl,
ctx.directory,
action.targetPaneId, action.targetPaneId,
action.splitDirection, action.splitDirection,
) )
@@ -10,10 +10,7 @@ import {
import { getTmuxPath } from "../../tools/interactive-bash/tmux-path-resolver" import { getTmuxPath } from "../../tools/interactive-bash/tmux-path-resolver"
import { queryWindowState } from "./pane-state-querier" import { queryWindowState } from "./pane-state-querier"
import { log } from "../../shared" import { log } from "../../shared"
import type { import type { ActionResult } from "./action-executor-core"
ActionResult,
ActionExecutorDeps,
} from "./action-executor-core"
export type { ActionExecutorDeps, ActionResult } from "./action-executor-core" export type { ActionExecutorDeps, ActionResult } from "./action-executor-core"
@@ -25,6 +22,7 @@ export interface ExecuteActionsResult {
export interface ExecuteContext { export interface ExecuteContext {
config: TmuxConfig config: TmuxConfig
directory: string
serverUrl: string serverUrl: string
windowState: WindowState windowState: WindowState
sourcePaneId?: string sourcePaneId?: string
@@ -81,7 +79,8 @@ export async function executeAction(
action.newSessionId, action.newSessionId,
action.description, action.description,
ctx.config, ctx.config,
ctx.serverUrl ctx.serverUrl,
ctx.directory,
) )
if (result.success) { if (result.success) {
await enforceLayoutAndMainPane(ctx) await enforceLayoutAndMainPane(ctx)
@@ -97,6 +96,7 @@ export async function executeAction(
action.description, action.description,
ctx.config, ctx.config,
ctx.serverUrl, ctx.serverUrl,
ctx.directory,
action.targetPaneId, action.targetPaneId,
action.splitDirection action.splitDirection
) )
@@ -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,
)
}
+542 -208
View File
@@ -1,26 +1,33 @@
import type { PluginInput } from "@opencode-ai/plugin" import type { PluginInput } from "@opencode-ai/plugin"
import type { TmuxConfig } from "../../config/schema" import type { TmuxConfig } from "../../config/schema"
import type { TrackedSession, CapacityConfig, WindowState } from "./types" import type { TrackedSession, CapacityConfig, WindowState } from "./types"
import { log, normalizeSDKResponse } from "../../shared" import { log } from "../../shared"
import { import {
isInsideTmux as defaultIsInsideTmux, isInsideTmux as defaultIsInsideTmux,
getCurrentPaneId as defaultGetCurrentPaneId, getCurrentPaneId as defaultGetCurrentPaneId,
POLL_INTERVAL_BACKGROUND_MS, POLL_INTERVAL_BACKGROUND_MS,
SESSION_READY_POLL_INTERVAL_MS,
SESSION_READY_TIMEOUT_MS,
spawnTmuxWindow, spawnTmuxWindow,
spawnTmuxSession, spawnTmuxSession,
killTmuxSessionIfExists, killTmuxSessionIfExists,
getIsolatedSessionName, getIsolatedSessionName,
sweepStaleOmoAgentSessions, sweepStaleOmoAgentSessions,
} from "../../shared/tmux" } 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 { decideSpawnActions, decideCloseAction, type SessionMapping } from "./decision-engine"
import { executeActions, executeAction } from "./action-executor" import { executeActions, executeAction } from "./action-executor"
import { TmuxPollingManager } from "./polling-manager" import { TmuxPollingManager } from "./polling-manager"
import { createTrackedSession, markTrackedSessionClosePending } from "./tracked-session-state" 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 OpencodeClient = PluginInput["client"]
type SpawnStage =
| "deferred.attach"
| "deferred.isolated-container"
| "session.created"
| "session.idle.retry"
interface SessionCreatedEvent { interface SessionCreatedEvent {
type: string type: string
properties?: { info?: { id?: string; parentID?: string; title?: string } } properties?: { info?: { id?: string; parentID?: string; title?: string } }
@@ -33,17 +40,30 @@ interface DeferredSession {
retryIsolatedContainer: boolean retryIsolatedContainer: boolean
} }
interface FailedReadinessSessionSeed {
sessionId: string
title: string
}
interface FailedReadinessSession extends FailedReadinessSessionSeed {
rememberedAt: number
}
export interface TmuxUtilDeps { export interface TmuxUtilDeps {
isInsideTmux: () => boolean isInsideTmux: () => boolean
getCurrentPaneId: () => string | undefined getCurrentPaneId: () => string | undefined
queryWindowState: (paneId: string) => Promise<WindowState | null>
} }
const defaultTmuxDeps: TmuxUtilDeps = { const defaultTmuxDeps: TmuxUtilDeps = {
isInsideTmux: defaultIsInsideTmux, isInsideTmux: defaultIsInsideTmux,
getCurrentPaneId: defaultGetCurrentPaneId, getCurrentPaneId: defaultGetCurrentPaneId,
queryWindowState: defaultQueryWindowState,
} }
const DEFERRED_SESSION_TTL_MS = 5 * 60 * 1000 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_DEFERRED_QUEUE_SIZE = 20
const MAX_CLOSE_RETRY_COUNT = 3 const MAX_CLOSE_RETRY_COUNT = 3
const MAX_ISOLATED_CONTAINER_NULL_STATE_COUNT = 2 const MAX_ISOLATED_CONTAINER_NULL_STATE_COUNT = 2
@@ -51,10 +71,14 @@ const MAX_ISOLATED_CONTAINER_NULL_STATE_COUNT = 2
export class TmuxSessionManager { export class TmuxSessionManager {
private client: OpencodeClient private client: OpencodeClient
private tmuxConfig: TmuxConfig private tmuxConfig: TmuxConfig
private projectDirectory: string
private serverUrl: string private serverUrl: string
private sourcePaneId: string | undefined private sourcePaneId: string | undefined
private sessions = new Map<string, TrackedSession>() private sessions = new Map<string, TrackedSession>()
private pendingSessions = new Set<string>() private pendingSessions = new Set<string>()
private failedReadinessSessions = new Map<string, FailedReadinessSession>()
private closedByPolling = new Set<string>()
private failedReadinessSweepInterval?: ReturnType<typeof setInterval>
private spawnQueue: Promise<void> = Promise.resolve() private spawnQueue: Promise<void> = Promise.resolve()
private deferredSessions = new Map<string, DeferredSession>() private deferredSessions = new Map<string, DeferredSession>()
private deferredQueue: string[] = [] private deferredQueue: string[] = []
@@ -71,6 +95,7 @@ export class TmuxSessionManager {
constructor(ctx: PluginInput, tmuxConfig: TmuxConfig, deps: TmuxUtilDeps = defaultTmuxDeps) { constructor(ctx: PluginInput, tmuxConfig: TmuxConfig, deps: TmuxUtilDeps = defaultTmuxDeps) {
this.client = ctx.client this.client = ctx.client
this.tmuxConfig = tmuxConfig this.tmuxConfig = tmuxConfig
this.projectDirectory = ctx.directory || process.cwd()
this.deps = deps this.deps = deps
const configuredPort = process.env.OPENCODE_PORT const configuredPort = process.env.OPENCODE_PORT
const parsedPort = configuredPort ? Number(configuredPort) : 4096 const parsedPort = configuredPort ? Number(configuredPort) : 4096
@@ -98,12 +123,13 @@ export class TmuxSessionManager {
this.pollingManager = new TmuxPollingManager( this.pollingManager = new TmuxPollingManager(
this.client, this.client,
this.sessions, this.sessions,
this.closeSessionById.bind(this), this.closeSessionFromPolling.bind(this),
this.retryPendingCloses.bind(this) this.retryPendingCloses.bind(this)
) )
log("[tmux-session-manager] initialized", { log("[tmux-session-manager] initialized", {
configEnabled: this.tmuxConfig.enabled, configEnabled: this.tmuxConfig.enabled,
tmuxConfig: this.tmuxConfig, tmuxConfig: this.tmuxConfig,
projectDirectory: this.projectDirectory,
serverUrl: this.serverUrl, serverUrl: this.serverUrl,
sourcePaneId: this.sourcePaneId, sourcePaneId: this.sourcePaneId,
}) })
@@ -129,7 +155,7 @@ export class TmuxSessionManager {
): Promise<string | null> { ): Promise<string | null> {
if (!this.isIsolated()) return null if (!this.isIsolated()) return null
if (this.isolatedWindowPaneId) { 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", { log("[tmux-session-manager] failed to query isolated window state", {
paneId: this.isolatedWindowPaneId, paneId: this.isolatedWindowPaneId,
error: String(error), error: String(error),
@@ -158,8 +184,8 @@ export class TmuxSessionManager {
log("[tmux-session-manager] creating isolated tmux container", { isolation, sessionId, title }) log("[tmux-session-manager] creating isolated tmux container", { isolation, sessionId, title })
const result = isolation === "session" const result = isolation === "session"
? await spawnTmuxSession(sessionId, title, this.tmuxConfig, this.serverUrl, this.sourcePaneId) ? await spawnTmuxSession(sessionId, title, this.tmuxConfig, this.serverUrl, this.projectDirectory, this.sourcePaneId)
: await spawnTmuxWindow(sessionId, title, this.tmuxConfig, this.serverUrl) : await spawnTmuxWindow(sessionId, title, this.tmuxConfig, this.serverUrl, this.projectDirectory)
if (result.success && result.paneId) { if (result.success && result.paneId) {
this.isolatedContainerPaneId = result.paneId this.isolatedContainerPaneId = result.paneId
@@ -196,6 +222,10 @@ export class TmuxSessionManager {
return this.sessions.get(sessionId)?.paneId return this.sessions.get(sessionId)?.paneId
} }
getServerUrl(): string {
return this.serverUrl
}
private removeTrackedSession(sessionId: string): void { private removeTrackedSession(sessionId: string): void {
this.sessions.delete(sessionId) this.sessions.delete(sessionId)
@@ -250,6 +280,7 @@ export class TmuxSessionManager {
{ type: "close", paneId: isolatedContainerPaneId, sessionId: tracked.sessionId }, { type: "close", paneId: isolatedContainerPaneId, sessionId: tracked.sessionId },
{ {
config: this.tmuxConfig, config: this.tmuxConfig,
directory: this.projectDirectory,
serverUrl: this.serverUrl, serverUrl: this.serverUrl,
windowState: state, windowState: state,
sourcePaneId: this.sourcePaneId ?? tracked.paneId, sourcePaneId: this.sourcePaneId ?? tracked.paneId,
@@ -288,7 +319,7 @@ export class TmuxSessionManager {
if (!paneId) return null if (!paneId) return null
try { try {
return await queryWindowState(paneId) return await this.deps.queryWindowState(paneId)
} catch (error) { } catch (error) {
log("[tmux-session-manager] failed to query window state for close", { log("[tmux-session-manager] failed to query window state for close", {
error: String(error), 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<boolean> {
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: { private async closeTrackedSessionPane(args: {
tracked: TrackedSession tracked: TrackedSession
state: WindowState state: WindowState
@@ -308,6 +380,7 @@ export class TmuxSessionManager {
{ type: "close", paneId: tracked.paneId, sessionId: tracked.sessionId }, { type: "close", paneId: tracked.paneId, sessionId: tracked.sessionId },
{ {
config: this.tmuxConfig, config: this.tmuxConfig,
directory: this.projectDirectory,
serverUrl: this.serverUrl, serverUrl: this.serverUrl,
windowState: state, windowState: state,
sourcePaneId: this.getEffectiveSourcePaneId(), sourcePaneId: this.getEffectiveSourcePaneId(),
@@ -365,12 +438,7 @@ export class TmuxSessionManager {
if (!this.sessions.has(tracked.sessionId)) continue if (!this.sessions.has(tracked.sessionId)) continue
if (tracked.closeRetryCount >= MAX_CLOSE_RETRY_COUNT) { if (tracked.closeRetryCount >= MAX_CLOSE_RETRY_COUNT) {
log("[tmux-session-manager] force removing close-pending session after max retries", { await this.finalizeForceRemoveCandidate(tracked, "retryPendingCloses.max-retries")
sessionId: tracked.sessionId,
paneId: tracked.paneId,
closeRetryCount: tracked.closeRetryCount,
})
this.removeTrackedSession(tracked.sessionId)
continue continue
} }
@@ -391,12 +459,7 @@ export class TmuxSessionManager {
const nextRetryCount = currentTracked.closeRetryCount + 1 const nextRetryCount = currentTracked.closeRetryCount + 1
if (nextRetryCount >= MAX_CLOSE_RETRY_COUNT) { if (nextRetryCount >= MAX_CLOSE_RETRY_COUNT) {
log("[tmux-session-manager] force removing close-pending session after failed retry", { await this.finalizeForceRemoveCandidate(currentTracked, "retryPendingCloses.failed-retry")
sessionId: currentTracked.sessionId,
paneId: currentTracked.paneId,
closeRetryCount: nextRetryCount,
})
this.removeTrackedSession(currentTracked.sessionId)
continue continue
} }
@@ -418,6 +481,11 @@ export class TmuxSessionManager {
title: string, title: string,
retryIsolatedContainer = false, retryIsolatedContainer = false,
): void { ): void {
if (this.shouldSkipRespawnAfterPollingClose(sessionId, "deferred enqueue")) {
this.clearFailedReadinessSession(sessionId)
return
}
const existingDeferredSession = this.deferredSessions.get(sessionId) const existingDeferredSession = this.deferredSessions.get(sessionId)
if (existingDeferredSession) { if (existingDeferredSession) {
if (retryIsolatedContainer && !existingDeferredSession.retryIsolatedContainer) { if (retryIsolatedContainer && !existingDeferredSession.retryIsolatedContainer) {
@@ -490,6 +558,376 @@ export class TmuxSessionManager {
log("[tmux-session-manager] deferred attach polling stopped") 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<boolean> {
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<string | undefined> {
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<void> {
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, unknown>
}): string | undefined {
const sessionId = event.properties?.sessionID
return typeof sessionId === "string" ? sessionId : undefined
}
private async retryFailedReadinessSession(sessionId: string): Promise<void> {
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<void> { private async tryAttachDeferredSession(): Promise<void> {
const sessionId = this.deferredQueue[0] const sessionId = this.deferredQueue[0]
if (!sessionId) { if (!sessionId) {
@@ -503,6 +941,16 @@ export class TmuxSessionManager {
return return
} }
if (this.shouldSkipRespawnAfterPollingClose(sessionId, "deferred attach")) {
this.removeDeferredSession(sessionId)
return
}
if (!this.beginPendingSession(sessionId, { allowDeferredSession: true })) {
return
}
try {
if (Date.now() - deferred.queuedAt.getTime() > DEFERRED_SESSION_TTL_MS) { if (Date.now() - deferred.queuedAt.getTime() > DEFERRED_SESSION_TTL_MS) {
this.deferredQueue.shift() this.deferredQueue.shift()
this.deferredSessions.delete(sessionId) this.deferredSessions.delete(sessionId)
@@ -519,6 +967,15 @@ export class TmuxSessionManager {
} }
if (deferred.retryIsolatedContainer) { if (deferred.retryIsolatedContainer) {
const readyForIsolatedContainer = await this.ensureSessionReadyBeforeSpawn(
sessionId,
"deferred.isolated-container",
)
if (!readyForIsolatedContainer) {
this.removeDeferredSession(sessionId)
return
}
const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, deferred.title) const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, deferred.title)
if (isolatedPaneId) { if (isolatedPaneId) {
this.sessions.set( this.sessions.set(
@@ -535,7 +992,6 @@ export class TmuxSessionManager {
sessionId, sessionId,
paneId: isolatedPaneId, paneId: isolatedPaneId,
}) })
this.logSessionReadinessInBackground(sessionId)
return return
} }
} }
@@ -543,7 +999,7 @@ export class TmuxSessionManager {
const effectiveSourcePaneId = this.getEffectiveSourcePaneId() const effectiveSourcePaneId = this.getEffectiveSourcePaneId()
if (!effectiveSourcePaneId) return if (!effectiveSourcePaneId) return
const state = await queryWindowState(effectiveSourcePaneId) const state = await this.deps.queryWindowState(effectiveSourcePaneId)
if (!state) { if (!state) {
this.nullStateCount += 1 this.nullStateCount += 1
log("[tmux-session-manager] deferred attach window state is null", { log("[tmux-session-manager] deferred attach window state is null", {
@@ -575,8 +1031,18 @@ export class TmuxSessionManager {
return return
} }
const readyForDeferredAttach = await this.ensureSessionReadyBeforeSpawn(
sessionId,
"deferred.attach",
)
if (!readyForDeferredAttach) {
this.removeDeferredSession(sessionId)
return
}
const result = await executeActions(decision.actions, { const result = await executeActions(decision.actions, {
config: this.tmuxConfig, config: this.tmuxConfig,
directory: this.projectDirectory,
serverUrl: this.serverUrl, serverUrl: this.serverUrl,
windowState: state, windowState: state,
sourcePaneId: effectiveSourcePaneId, sourcePaneId: effectiveSourcePaneId,
@@ -608,46 +1074,9 @@ export class TmuxSessionManager {
sessionId, sessionId,
paneId: result.spawnedPaneId, paneId: result.spawnedPaneId,
}) })
this.logSessionReadinessInBackground(sessionId) } finally {
this.pendingSessions.delete(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<boolean> {
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<string, { type: string }>)
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))
}
log("[tmux-session-manager] session ready timeout", {
sessionId,
timeoutMs: SESSION_READY_TIMEOUT_MS,
})
return false
} }
async onSessionCreated(event: SessionCreatedEvent): Promise<void> { async onSessionCreated(event: SessionCreatedEvent): Promise<void> {
@@ -675,156 +1104,30 @@ export class TmuxSessionManager {
return return
} }
if (!this.beginPendingSession(sessionId)) {
return
}
try {
await this.sweepStaleIsolatedSessionsOnce() await this.sweepStaleIsolatedSessionsOnce()
await this.retryPendingCloses() await this.retryPendingCloses()
if ( const session = { sessionId, title }
this.sessions.has(sessionId) ||
this.pendingSessions.has(sessionId) ||
this.deferredSessions.has(sessionId)
) {
log("[tmux-session-manager] session already tracked or pending", { sessionId })
return
}
this.pendingSessions.add(sessionId)
await this.enqueueSpawn(async () => { await this.enqueueSpawn(async () => {
try { try {
const isolatedPaneId = await this.spawnInIsolatedContainer(sessionId, title) await this.spawnPendingSession({
if (isolatedPaneId) { session,
this.sessions.set( stage: "session.created",
sessionId, rememberReadinessFailure: true,
createTrackedSession({ sessionId, paneId: isolatedPaneId, description: title }),
)
this.pollingManager.startPolling()
log("[tmux-session-manager] first subagent spawned in isolated window", {
sessionId,
paneId: isolatedPaneId,
}) })
this.logSessionReadinessInBackground(sessionId)
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 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 { } finally {
this.pendingSessions.delete(sessionId) this.pendingSessions.delete(sessionId)
} }
}) })
} finally {
this.pendingSessions.delete(sessionId)
}
} }
private async enqueueSpawn(run: () => Promise<void>): Promise<void> { private async enqueueSpawn(run: () => Promise<void>): Promise<void> {
@@ -845,10 +1148,13 @@ export class TmuxSessionManager {
async onSessionDeleted(event: { sessionID: string }): Promise<void> { async onSessionDeleted(event: { sessionID: string }): Promise<void> {
if (!this.isEnabled()) return if (!this.isEnabled()) return
if (!this.getEffectiveSourcePaneId()) return
this.closedByPolling.delete(event.sessionID)
this.clearFailedReadinessSession(event.sessionID)
this.removeDeferredSession(event.sessionID) this.removeDeferredSession(event.sessionID)
if (!this.getEffectiveSourcePaneId()) return
const tracked = this.sessions.get(event.sessionID) const tracked = this.sessions.get(event.sessionID)
if (!tracked) return if (!tracked) return
@@ -876,6 +1182,7 @@ export class TmuxSessionManager {
try { try {
const result = await executeAction(closeAction, { const result = await executeAction(closeAction, {
config: this.tmuxConfig, config: this.tmuxConfig,
directory: this.projectDirectory,
serverUrl: this.serverUrl, serverUrl: this.serverUrl,
windowState: state, windowState: state,
sourcePaneId: this.getEffectiveSourcePaneId(), sourcePaneId: this.getEffectiveSourcePaneId(),
@@ -907,12 +1214,7 @@ export class TmuxSessionManager {
if (!tracked) return if (!tracked) return
if (tracked.closePending && tracked.closeRetryCount >= MAX_CLOSE_RETRY_COUNT) { if (tracked.closePending && tracked.closeRetryCount >= MAX_CLOSE_RETRY_COUNT) {
log("[tmux-session-manager] force removing close-pending session after max retries", { await this.finalizeForceRemoveCandidate(tracked, "closeSessionById.max-retries")
sessionId,
paneId: tracked.paneId,
closeRetryCount: tracked.closeRetryCount,
})
this.removeTrackedSession(sessionId)
return return
} }
@@ -928,8 +1230,37 @@ export class TmuxSessionManager {
} }
} }
private async closeSessionFromPolling(sessionId: string): Promise<void> {
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<string, unknown> }): void { onEvent(event: { type: string; properties?: Record<string, unknown> }): void {
this.pollingManager.handleEvent(event) 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<void> { createEventHandler(): (input: { event: { type: string; properties?: unknown } }) => Promise<void> {
@@ -942,6 +1273,9 @@ export class TmuxSessionManager {
this.stopDeferredAttachLoop() this.stopDeferredAttachLoop()
this.deferredQueue = [] this.deferredQueue = []
this.deferredSessions.clear() this.deferredSessions.clear()
this.failedReadinessSessions.clear()
this.closedByPolling.clear()
this.stopFailedReadinessSweep()
this.pollingManager.stopPolling() this.pollingManager.stopPolling()
if (this.sessions.size > 0) { if (this.sessions.size > 0) {
@@ -1,4 +1,3 @@
import { spawn } from "../../shared/bun-spawn-shim"
import type { WindowState, TmuxPaneInfo } from "./types" import type { WindowState, TmuxPaneInfo } from "./types"
import { parsePaneStateOutput } from "./pane-state-parser" import { parsePaneStateOutput } from "./pane-state-parser"
import { getTmuxPath } from "../../tools/interactive-bash/tmux-path-resolver" import { getTmuxPath } from "../../tools/interactive-bash/tmux-path-resolver"
@@ -7,28 +6,22 @@ import { log } from "../../shared"
export async function queryWindowState(sourcePaneId: string): Promise<WindowState | null> { export async function queryWindowState(sourcePaneId: string): Promise<WindowState | null> {
const tmux = await getTmuxPath() const tmux = await getTmuxPath()
if (!tmux) return null if (!tmux) return null
const { runTmuxCommand } = await import("../../shared/tmux")
const proc = spawn( const result = await runTmuxCommand(tmux, [
[
tmux,
"list-panes", "list-panes",
"-t", "-t",
sourcePaneId, sourcePaneId,
"-F", "-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}", "#{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 exitCode = await proc.exited if (result.exitCode !== 0) {
const stdout = await new Response(proc.stdout).text() log("[pane-state-querier] list-panes failed", { exitCode: result.exitCode })
if (exitCode !== 0) {
log("[pane-state-querier] list-panes failed", { exitCode })
return null return null
} }
const parsedPaneState = parsePaneStateOutput(stdout) const parsedPaneState = parsePaneStateOutput(result.output)
if (!parsedPaneState) { if (!parsedPaneState) {
log("[pane-state-querier] failed to parse pane state output", { log("[pane-state-querier] failed to parse pane state output", {
sourcePaneId, sourcePaneId,
+27 -12
View File
@@ -1,11 +1,13 @@
import type { OpencodeClient } from "../../tools/delegate-task/types" 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 type { TrackedSession } from "./types"
import { SESSION_MISSING_GRACE_MS } from "../../shared/tmux"
import { log } from "../../shared" import { log } from "../../shared"
import { normalizeSDKResponse } from "../../shared" import { normalizeSDKResponse } from "../../shared"
const SESSION_TIMEOUT_MS = 10 * 60 * 1000
const MIN_STABILITY_TIME_MS = 10 * 1000 const MIN_STABILITY_TIME_MS = 10 * 1000
const STABLE_POLLS_REQUIRED = 3 const STABLE_POLLS_REQUIRED = 3
@@ -86,30 +88,42 @@ export class TmuxPollingManager {
if (isIdle && elapsedMs >= MIN_STABILITY_TIME_MS) { if (isIdle && elapsedMs >= MIN_STABILITY_TIME_MS) {
const activityVersion = tracked.activityVersion ?? 0 const activityVersion = tracked.activityVersion ?? 0
if (tracked.observedIdleActivityVersion === activityVersion) { if (tracked.observedIdleActivityVersion !== activityVersion) {
tracked.stableIdlePolls = 1
tracked.observedIdleActivityVersion = activityVersion
} else {
tracked.stableIdlePolls = (tracked.stableIdlePolls ?? 0) + 1 tracked.stableIdlePolls = (tracked.stableIdlePolls ?? 0) + 1
}
if (tracked.stableIdlePolls >= STABLE_POLLS_REQUIRED) { if ((tracked.stableIdlePolls ?? 0) >= STABLE_POLLS_REQUIRED) {
const stableWindowActivityVersion = tracked.observedIdleActivityVersion ?? activityVersion
const recheckResult = await this.client.session.status({ path: undefined }) const recheckResult = await this.client.session.status({ path: undefined })
const recheckStatuses = normalizeSDKResponse(recheckResult, {} as Record<string, { type: string }>) const recheckStatuses = normalizeSDKResponse(recheckResult, {} as Record<string, { type: string }>)
const recheckStatus = recheckStatuses[sessionId] const recheckStatus = recheckStatuses[sessionId]
const latestTracked = this.sessions.get(sessionId) ?? tracked
const recheckActivityVersion = latestTracked.activityVersion ?? 0
if (recheckStatus?.type === "idle") { 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 shouldCloseViaStability = true
} else { } else {
tracked.stableIdlePolls = 0 latestTracked.stableIdlePolls = 0
log("[tmux-session-manager] stability reached but session not idle on recheck, resetting", { log("[tmux-session-manager] stability reached but session not idle on recheck, resetting", {
sessionId, sessionId,
recheckStatus: recheckStatus?.type, recheckStatus: recheckStatus?.type,
}) })
} }
} }
} else {
tracked.stableIdlePolls = 0
tracked.observedIdleActivityVersion = activityVersion
}
} else if (!isIdle) { } else if (!isIdle) {
tracked.stableIdlePolls = 0 tracked.stableIdlePolls = 0
tracked.observedIdleActivityVersion = undefined
} }
log("[tmux-session-manager] session check", { log("[tmux-session-manager] session check", {
@@ -126,7 +140,8 @@ export class TmuxPollingManager {
shouldCloseViaStability, shouldCloseViaStability,
}) })
if (shouldCloseViaStability || missingTooLong || isTimedOut) { if (!tracked.closePending && (shouldCloseViaStability || missingTooLong || isTimedOut)) {
tracked.closePending = true
sessionsToClose.push(sessionId) sessionsToClose.push(sessionId)
} }
} }
+7 -1
View File
@@ -30,6 +30,7 @@ export interface SessionPollingController {
export function createSessionPollingController(params: { export function createSessionPollingController(params: {
client: OpencodeClient client: OpencodeClient
tmuxConfig: TmuxConfig tmuxConfig: TmuxConfig
directory: string
serverUrl: string serverUrl: string
sourcePaneId: string | undefined sourcePaneId: string | undefined
sessions: Map<string, TrackedSession> sessions: Map<string, TrackedSession>
@@ -49,7 +50,12 @@ export function createSessionPollingController(params: {
if (state) { if (state) {
await executeAction( await executeAction(
{ type: "close", paneId: tracked.paneId, sessionId }, { 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,
},
) )
} }
@@ -4,6 +4,7 @@ import {
SESSION_READY_TIMEOUT_MS, SESSION_READY_TIMEOUT_MS,
} from "../../shared/tmux" } from "../../shared/tmux"
import { log } from "../../shared" import { log } from "../../shared"
import { isAttachableSessionStatus } from "./attachable-session-status"
import { parseSessionStatusMap } from "./session-status-parser" import { parseSessionStatusMap } from "./session-status-parser"
type OpencodeClient = PluginInput["client"] type OpencodeClient = PluginInput["client"]
@@ -18,11 +19,12 @@ export async function waitForSessionReady(params: {
try { try {
const statusResult = await params.client.session.status({ path: undefined }) const statusResult = await params.client.session.status({ path: undefined })
const allStatuses = parseSessionStatusMap(statusResult.data) const allStatuses = parseSessionStatusMap(statusResult.data)
const sessionStatus = allStatuses[params.sessionId]?.type
if (allStatuses[params.sessionId]) { if (isAttachableSessionStatus(sessionStatus)) {
log("[tmux-session-manager] session ready", { log("[tmux-session-manager] session ready", {
sessionId: params.sessionId, sessionId: params.sessionId,
status: allStatuses[params.sessionId].type, status: sessionStatus,
waitedMs: Date.now() - startTime, waitedMs: Date.now() - startTime,
}) })
return true return true