From ad50a2bd0be18165a23b2cf4bcb027e027945e23 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:27:44 +0000 Subject: [PATCH 1/2] refactor: split AgentSessionManager into session collaborators AgentSessionManager carried 29 functions plus spawn/maintainTranscripts complexity behind documented suppressions. It becomes a facade over registry, lifecycle, controls, transcript admin/maintenance, telemetry, command, Codex-home, and spawn-workflow collaborators; the three @Suppress annotations are gone and collaborators are unit-tested. --- .../agentgateway/tmux/AgentCommandFactory.kt | 85 ++ .../agentgateway/tmux/AgentSessionControls.kt | 142 ++++ .../tmux/AgentSessionLifecycle.kt | 67 ++ .../agentgateway/tmux/AgentSessionManager.kt | 779 ++---------------- .../tmux/AgentSessionOperations.kt | 62 ++ .../agentgateway/tmux/AgentSessionRegistry.kt | 41 + .../tmux/AgentSessionSpawnWorkflow.kt | 187 +++++ .../tmux/AgentSessionTelemetry.kt | 102 +++ .../agentgateway/tmux/AgentTranscriptAdmin.kt | 46 ++ .../tmux/AgentTranscriptMaintenance.kt | 105 +++ .../agentgateway/tmux/CodexSessionHomes.kt | 71 ++ .../tmux/AgentSessionCollaboratorsTest.kt | 173 ++++ 12 files changed, 1156 insertions(+), 704 deletions(-) create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentCommandFactory.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionLifecycle.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionOperations.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionRegistry.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionTelemetry.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptAdmin.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/CodexSessionHomes.kt create mode 100644 services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentCommandFactory.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentCommandFactory.kt new file mode 100644 index 0000000..8dbe395 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentCommandFactory.kt @@ -0,0 +1,85 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import org.slf4j.LoggerFactory +import java.nio.file.Path +import java.util.UUID + +internal class AgentCommandFactory( + private val props: GatewayProperties, + private val claudeTranscriptLocator: ClaudeTranscriptLocator, +) { + private val log = LoggerFactory.getLogger(AgentCommandFactory::class.java) + + /** + * Build the CLI command and return the native session id alongside it. + * + * Claude gets an explicit `--session-id` on fresh starts and `--resume` + * only when the matching local transcript still exists. Codex runs under + * an isolated CODEX_HOME so fresh rollout capture has no cross-session + * ambiguity; shell has no native session id. + */ + fun commandAndSessionIdFor( + kind: AgentKind, + cwd: String, + resumeCliSessionId: String? = null, + codexHome: Path? = null, + ): AgentCommand = + when (kind) { + AgentKind.CLAUDE -> claudeCommand(cwd, resumeCliSessionId) + AgentKind.CODEX -> codexCommand(cwd, resumeCliSessionId, codexHome) + AgentKind.SHELL -> AgentCommand(command = listOf("/bin/bash", "-l"), cliSessionId = null, cwd = cwd) + } + + private fun claudeCommand( + cwd: String, + resumeCliSessionId: String?, + ): AgentCommand { + val cliSessionId = resumeCliSessionId ?: UUID.randomUUID().toString() + val transcript = resumeCliSessionId?.let { claudeTranscriptLocator.findTranscript(cwd, it) } + val sessionArgs = + if (resumeCliSessionId == null) { + listOf("--session-id", cliSessionId) + } else if (transcript != null) { + log.info( + "claude revival selected resume sessionId={} reason=transcript-exists " + + "requestedCwd={} transcriptCwd={}", + resumeCliSessionId, + cwd, + transcript.cwd, + ) + listOf("--resume", resumeCliSessionId) + } else { + log.info( + "claude revival selected fresh-with-stable-id sessionId={} reason=transcript-missing", + resumeCliSessionId, + ) + listOf("--session-id", resumeCliSessionId) + } + return AgentCommand( + command = listOf(props.cli.claude) + props.cli.claudeArgs + sessionArgs, + cliSessionId = cliSessionId, + cwd = transcript?.cwd ?: cwd, + ) + } + + private fun codexCommand( + cwd: String, + resumeCliSessionId: String?, + codexHome: Path?, + ): AgentCommand { + val envPrefix = codexHome?.let { listOf("env", "CODEX_HOME=$it") }.orEmpty() + val resumeArgs = resumeCliSessionId?.let { listOf("resume", it) }.orEmpty() + return AgentCommand( + command = envPrefix + listOf(props.cli.codex) + props.cli.codexArgs + resumeArgs, + cliSessionId = resumeCliSessionId, + cwd = cwd, + ) + } +} + +internal data class AgentCommand( + val command: List, + val cliSessionId: String?, + val cwd: String, +) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt new file mode 100644 index 0000000..6f4f5e5 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt @@ -0,0 +1,142 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayAgentKindLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import io.micrometer.observation.Observation +import org.slf4j.LoggerFactory +import java.nio.file.Files +import java.nio.file.StandardOpenOption +import java.time.Instant +import java.time.ZoneOffset +import java.time.format.DateTimeFormatter +import java.util.UUID +import kotlin.io.path.Path + +internal class AgentSessionControls( + private val tmux: TmuxClient, + private val props: GatewayProperties, + private val registry: AgentSessionRegistry, + private val telemetry: AgentSessionTelemetry, +) : AgentSessionControlOperations { + private val log = LoggerFactory.getLogger(AgentSessionControls::class.java) + + override fun send( + id: String, + input: String, + enter: Boolean, + ) = withSessionOperation(id, GatewayOperationLabel.INPUT, observed = true) { session -> + tmux.sendKeys(session.tmuxSession, input, enter = enter) + } + + override fun stageInput( + id: String, + content: String, + requestedName: String?, + ): StagedInput = + withSessionOperation(id, GatewayOperationLabel.INPUT, observed = false) { session -> + val bytes = content.toByteArray(Charsets.UTF_8) + require(bytes.isNotEmpty()) { "staged input content is empty" } + require(bytes.size.toLong() <= props.stagedInputs.maxBytes) { + "staged input exceeds ${props.stagedInputs.maxBytes} bytes" + } + + val root = Path(session.cwd).toAbsolutePath().normalize() + val dir = root.resolve(props.stagedInputs.dirName).normalize() + require(dir.startsWith(root)) { "staged input directory must stay inside the workspace" } + + Files.createDirectories(dir) + val safeName = safeFileName(requestedName) + val fileName = "${timestamp()}-${UUID.randomUUID().toString().take(ID_PREVIEW_CHARS)}-$safeName" + val target = dir.resolve(fileName).normalize() + require(target.startsWith(dir)) { "staged input path must stay inside the staging directory" } + + Files.write(target, bytes, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE) + log.info("staged {} bytes for agent {} at {}", bytes.size, id, target) + StagedInput(path = target.toString(), bytes = bytes.size.toLong(), name = safeName) + } + + override fun capture( + id: String, + historyLines: Int, + ): String = + withSessionOperation(id, GatewayOperationLabel.REPLAY, observed = false) { session -> + tmux.capture(session.tmuxSession, historyLines) + } + + override fun captureWithEscapes(id: String): String = + withSessionOperation(id, GatewayOperationLabel.REPLAY, observed = false) { session -> + tmux.captureWithEscapes(session.tmuxSession) + } + + override fun resize( + id: String, + cols: Int, + rows: Int, + ) = withSessionOperation(id, GatewayOperationLabel.RESIZE, observed = true) { session -> + tmux.resize(session.tmuxSession, cols, rows) + } + + private fun withSessionOperation( + id: String, + operation: GatewayOperationLabel, + observed: Boolean, + block: (AgentSession) -> T, + ): T { + val startedAt = Instant.now() + var kind = GatewayAgentKindLabel.OTHER + var outcome = GatewayOutcomeLabel.SUCCESS + var reason = GatewayFailureReasonLabel.NONE + var failure: Throwable? = null + var observation: Observation? = null + try { + return runCatching { + val session = registry.get(id) ?: error("unknown agent: $id") + kind = with(telemetry) { session.kind.toTelemetryKind() } + if (observed) observation = telemetry.startSessionObservation(operation, kind) + block(session) + }.getOrElse { error -> + outcome = GatewayOutcomeLabel.FAILURE + reason = telemetry.failureReasonLabel(error) + failure = error + throw error + } + } finally { + telemetry.recordSessionOperation( + SessionOperationRecord( + operation = operation, + kind = kind, + outcome = outcome, + reason = reason, + startedAt = startedAt, + observed = observed, + observation = observation, + error = failure, + ), + ) + } + } + + private fun safeFileName(requestedName: String?): String { + val raw = requestedName?.trim()?.takeIf { it.isNotBlank() } ?: DEFAULT_STAGED_INPUT_NAME + val leaf = raw.replace('\\', '/').substringAfterLast('/') + val safe = + SAFE_NAME_CHARS + .replace(leaf, "-") + .trim('.', '-', '_') + .take(MAX_STAGED_INPUT_NAME_CHARS) + return safe.takeIf { it.isNotBlank() } ?: DEFAULT_STAGED_INPUT_NAME + } + + private fun timestamp(): String = STAGED_INPUT_TIMESTAMP.format(Instant.now()) + + private companion object { + const val DEFAULT_STAGED_INPUT_NAME = "input.txt" + const val ID_PREVIEW_CHARS = 8 + const val MAX_STAGED_INPUT_NAME_CHARS = 80 + val SAFE_NAME_CHARS = Regex("[^A-Za-z0-9._-]+") + val STAGED_INPUT_TIMESTAMP = DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss-SSS").withZone(ZoneOffset.UTC) + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionLifecycle.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionLifecycle.kt new file mode 100644 index 0000000..56665dd --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionLifecycle.kt @@ -0,0 +1,67 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayAgentKindLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import io.micrometer.observation.Observation +import org.slf4j.LoggerFactory +import java.time.Instant + +internal class AgentSessionLifecycle( + private val tmux: TmuxClient, + private val transcriptStore: TranscriptStore, + private val registry: AgentSessionRegistry, + private val telemetry: AgentSessionTelemetry, +) : AgentSessionLifecycleOperations { + private val log = LoggerFactory.getLogger(AgentSessionLifecycle::class.java) + + override fun stop(id: String): Boolean { + val startedAt = Instant.now() + var kind = GatewayAgentKindLabel.OTHER + var outcome = GatewayOutcomeLabel.FAILURE + var reason = GatewayFailureReasonLabel.NOT_FOUND + var failure: Throwable? = null + var observation: Observation? = null + try { + return runCatching { + val session = registry.remove(id) ?: return@runCatching false + kind = with(telemetry) { session.kind.toTelemetryKind() } + observation = telemetry.startSessionObservation(GatewayOperationLabel.STOP, kind) + tmux.killSession(session.tmuxSession) + session.stableSessionId?.let { + transcriptStore.seal(it) + session.transcriptLease?.let(transcriptStore::releaseLease) + } + outcome = GatewayOutcomeLabel.SUCCESS + reason = GatewayFailureReasonLabel.NONE + log.info("stopped agent {}", id) + true + }.getOrElse { error -> + outcome = GatewayOutcomeLabel.FAILURE + reason = telemetry.failureReasonLabel(error) + failure = error + throw error + } + } finally { + telemetry.recordActiveSessionCounts() + telemetry.recordSessionOperation( + SessionOperationRecord( + operation = GatewayOperationLabel.STOP, + kind = kind, + outcome = outcome, + reason = reason, + startedAt = startedAt, + observation = observation, + error = failure, + ), + ) + } + } + + fun stopAll(onFailure: (String, Throwable) -> Unit) { + registry.ids().forEach { id -> + runCatching { stop(id) }.onFailure { onFailure(id, it) } + } + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt index 8029138..7c6e708 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt @@ -2,70 +2,57 @@ package com.jorisjonkers.personalstack.agentgateway.tmux import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayActiveSessionsSample -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayAgentKindLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayModeLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationTelemetry -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStatusLabel -import com.jorisjonkers.personalstack.agentgateway.process.ProcessFailedException -import com.jorisjonkers.personalstack.agentgateway.process.ProcessTimeoutException -import io.micrometer.observation.Observation import io.micrometer.observation.ObservationRegistry import jakarta.annotation.PreDestroy import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Component -import java.io.IOException -import java.nio.channels.FileChannel -import java.nio.file.Files -import java.nio.file.LinkOption -import java.nio.file.Path -import java.nio.file.StandardOpenOption -import java.time.Duration -import java.time.Instant -import java.time.ZoneOffset -import java.time.format.DateTimeFormatter -import java.util.UUID -import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.Executors import java.util.concurrent.TimeUnit -import kotlin.io.path.Path /** - * In-memory registry of active agents on this Pod. The Pod is the - * unit of restart, and agents-api owns the source-of-truth state, - * so persisting here would double-bookkeeper for no win. - * - * Concurrency: ConcurrentHashMap is enough — the only racy operation - * is spawn-vs-stop on the same id, and that's a caller bug worth - * surfacing as a 409. + * Spring-facing facade for active agent sessions on this Pod. The Pod is the + * unit of restart, and agents-api owns the source-of-truth state, so this class + * delegates registry, spawning, transcript, and tmux operations to focused + * collaborators. */ @Component -@Suppress("TooManyFunctions") -class AgentSessionManager( - private val tmux: TmuxClient, - private val props: GatewayProperties, - private val transcriptStore: TranscriptStore, - private val telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, - private val observationRegistry: ObservationRegistry = ObservationRegistry.NOOP, - private val claudeTranscriptLocator: ClaudeTranscriptLocator = ClaudeTranscriptLocator.fromEnvironment(), -) { - private val log = LoggerFactory.getLogger(AgentSessionManager::class.java) - private val sessions = ConcurrentHashMap() +class AgentSessionManager private constructor( + private val components: AgentSessionComponents, +) : AgentSessionSpawnOperations by components.spawn, + AgentSessionLifecycleOperations by components.lifecycle, + AgentSessionRegistryOperations by components.registry, + AgentSessionControlOperations by components.controls, + AgentTranscriptAdminOperations by components.transcripts { + @Autowired + constructor( + tmux: TmuxClient, + props: GatewayProperties, + transcriptStore: TranscriptStore, + telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, + observationRegistry: ObservationRegistry = ObservationRegistry.NOOP, + claudeTranscriptLocator: ClaudeTranscriptLocator = ClaudeTranscriptLocator.fromEnvironment(), + ) : this( + AgentSessionComponents.create( + tmux = tmux, + props = props, + transcriptStore = transcriptStore, + telemetry = telemetry, + observationRegistry = observationRegistry, + claudeTranscriptLocator = claudeTranscriptLocator, + ), + ) - // Durable transcripts are capped by deleting old closed segments. - // Legacy non-durable pipe logs are still truncated in place. + private val log = LoggerFactory.getLogger(AgentSessionManager::class.java) private val trimmer = Executors.newSingleThreadScheduledExecutor { r -> Thread(r, "agent-transcript-maintenance").apply { isDaemon = true } } init { - val period = props.transcripts.trimIntervalSeconds - trimmer.scheduleWithFixedDelay(::maintainTranscripts, period, period, TimeUnit.SECONDS) - recordActiveSessionCounts() + val period = components.props.transcripts.trimIntervalSeconds + trimmer.scheduleWithFixedDelay(components.maintenance::maintain, period, period, TimeUnit.SECONDS) + components.telemetry.recordActiveSessionCounts() } @PreDestroy @@ -76,668 +63,52 @@ class AgentSessionManager( } catch (e: InterruptedException) { Thread.currentThread().interrupt() } - sessions.keys.sorted().forEach { id -> - runCatching { stop(id) } - .onFailure { log.warn("shutdown cleanup of agent {} failed: {}", id, it.message) } + components.lifecycle.stopAll { id, error -> + log.warn("shutdown cleanup of agent {} failed: {}", id, error.message) } - recordActiveSessionCounts() + components.telemetry.recordActiveSessionCounts() } +} - @Suppress("LongMethod") - private fun maintainTranscripts() { - val cap = props.tmux.logCapBytes - sessions.values.forEach { session -> - val startedAt = Instant.now() - runCatching { - val stableSessionId = session.stableSessionId - if (stableSessionId != null) { - val beforeTrim = transcriptStore.recoverMetadata(stableSessionId).logicalStart - val current = - session.transcriptLease - ?.let(transcriptStore::renewLease) - ?.let { session.copy(transcriptLease = it) } - ?: session - if (current !== session) sessions[session.id] = current - transcriptStore.rotateIfNeeded(stableSessionId) - val active = transcriptStore.activeSegmentPath(stableSessionId) - if (active != current.logFile) { - tmux.startPipeToFile(current.tmuxSession, active) - sessions[current.id] = current.copy(logFile = active, transcriptFile = active) - log.info("rotated agent {} transcript pipe to {}", current.id, active) - } - val trimmed = transcriptStore.trimIfNeeded(stableSessionId) - if (trimmed.logicalStart > beforeTrim) { - recordSessionOperation( - SessionOperationRecord( - operation = GatewayOperationLabel.REPLAY, - kind = current.kind.toTelemetryKind(), - outcome = GatewayOutcomeLabel.SUCCESS, - reason = GatewayFailureReasonLabel.NONE, - startedAt = startedAt, - observed = false, - ), - ) - } - } else { - val file = session.logFile - if (Files.exists(file) && Files.size(file) > cap) { - FileChannel.open(file, StandardOpenOption.WRITE).use { it.truncate(0) } - log.info("trimmed agent {} log past {} bytes", session.id, cap) - recordSessionOperation( - SessionOperationRecord( - operation = GatewayOperationLabel.REPLAY, - kind = session.kind.toTelemetryKind(), - outcome = GatewayOutcomeLabel.SUCCESS, - reason = GatewayFailureReasonLabel.NONE, - startedAt = startedAt, - observed = false, - ), - ) - } - } - }.onFailure { - recordSessionOperation( - SessionOperationRecord( - operation = GatewayOperationLabel.REPLAY, - kind = session.kind.toTelemetryKind(), - outcome = GatewayOutcomeLabel.FAILURE, - reason = failureReasonLabel(it), - startedAt = startedAt, - observed = false, - error = it, +private data class AgentSessionComponents( + val props: GatewayProperties, + val registry: AgentSessionRegistry, + val telemetry: AgentSessionTelemetry, + val spawn: AgentSessionSpawnWorkflow, + val lifecycle: AgentSessionLifecycle, + val controls: AgentSessionControls, + val transcripts: AgentTranscriptAdmin, + val maintenance: AgentTranscriptMaintenance, +) { + companion object { + fun create( + tmux: TmuxClient, + props: GatewayProperties, + transcriptStore: TranscriptStore, + telemetry: AgentGatewayTelemetry, + observationRegistry: ObservationRegistry, + claudeTranscriptLocator: ClaudeTranscriptLocator, + ): AgentSessionComponents { + val registry = AgentSessionRegistry() + val sessionTelemetry = AgentSessionTelemetry(registry, telemetry, observationRegistry) + return AgentSessionComponents( + props = props, + registry = registry, + telemetry = sessionTelemetry, + spawn = + AgentSessionSpawnWorkflow( + tmux = tmux, + props = props, + transcriptStore = transcriptStore, + registry = registry, + telemetry = sessionTelemetry, + claudeTranscriptLocator = claudeTranscriptLocator, ), - ) - log.warn("trim of {} failed: {}", session.logFile, it.message) - } - } - } - - fun spawn( - kind: AgentKind, - workspacePath: String? = null, - stableSessionId: String? = null, - epoch: Long? = null, - continuation: AgentContinuation? = null, - ): AgentSession = - spawn( - AgentSpawnRequest( - kind = kind, - workspacePath = workspacePath, - stableSessionId = stableSessionId, - epoch = epoch, - continuation = continuation, - ), - ) - - fun spawn( - kind: AgentKind, - workspacePath: String? = null, - resumeCliSessionId: String, - ): AgentSession = - spawn( - AgentSpawnRequest( - kind = kind, - workspacePath = workspacePath, - resumeCliSessionId = resumeCliSessionId, - ), - ) - - // Lease cleanup must stay adjacent to transcript and tmux startup ordering. - @Suppress("CyclomaticComplexMethod", "LongMethod", "ThrowsCount") - fun spawn(request: AgentSpawnRequest): AgentSession { - val startedAt = Instant.now() - val telemetryKind = request.kind.toTelemetryKind() - var outcome = GatewayOutcomeLabel.SUCCESS - var reason = GatewayFailureReasonLabel.NONE - var failure: Throwable? = null - val observation = startSessionObservation(GatewayOperationLabel.SPAWN, telemetryKind) - try { - return runCatching { - val id = UUID.randomUUID().toString().substring(0, 8) - val tmuxSession = "agent-$id" - val requestedCwd = request.workspacePath ?: props.workspaceRoot - val durableStableSessionId = - request.stableSessionId?.let(transcriptStore::validateStableSessionId) - ?: UUID.randomUUID().toString() - val durableEpoch = request.epoch ?: 1 - require(durableEpoch > 0) { "epoch must be positive" } - val lease = transcriptStore.acquireLease(durableStableSessionId, tmuxSession, durableEpoch) - val logFile: Path = - try { - transcriptStore.open(durableStableSessionId, durableEpoch) - if (durableEpoch > 1 || request.continuation != null) { - transcriptStore.appendContinuationDelimiter( - durableStableSessionId, - durableEpoch, - request.continuation, - ) - } - transcriptStore.activeSegmentPath(durableStableSessionId) - } catch (e: IOException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: NumberFormatException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: IllegalArgumentException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: IllegalStateException) { - transcriptStore.releaseLease(lease) - throw e - } - - val codexHome = if (request.kind == AgentKind.CODEX) codexSessionHome(durableStableSessionId) else null - codexHome?.let(::prepareCodexHome) - val launch = commandAndSessionIdFor(request.kind, requestedCwd, request.resumeCliSessionId, codexHome) - try { - tmux.newSession(tmuxSession, launch.command, launch.cwd) - tmux.startPipeToFile(tmuxSession, logFile) - } catch (e: IOException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: InterruptedException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: ProcessFailedException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: ProcessTimeoutException) { - transcriptStore.releaseLease(lease) - throw e - } catch (e: SecurityException) { - transcriptStore.releaseLease(lease) - throw e - } - - // Codex's native id only exists once it writes the rollout at - // session start; capture it from the isolated home (fresh case). - // Claude and Codex-resume already know their id at build time. - val cliSessionId = - launch.cliSessionId - ?: codexHome?.takeIf { request.kind == AgentKind.CODEX }?.let(::captureCodexSessionId) - - val session = - AgentSession( - id = id, - kind = request.kind, - tmuxSession = tmuxSession, - logFile = logFile, - cwd = launch.cwd, - createdAt = Instant.now(), - cliSessionId = cliSessionId, - stableSessionId = durableStableSessionId, - epoch = durableEpoch, - continuation = request.continuation, - transcriptFile = logFile, - transcriptLease = lease, - ) - sessions[id] = session - recordActiveSessionCounts() - log.info( - "spawned {} agent {} ({}) in {} cliSessionId={} stableSessionId={} epoch={}", - request.kind, - id, - tmuxSession, - launch.cwd, - cliSessionId, - durableStableSessionId, - durableEpoch, - ) - session - }.getOrElse { error -> - outcome = GatewayOutcomeLabel.FAILURE - reason = failureReasonLabel(error) - failure = error - throw error - } - } finally { - recordSessionOperation( - SessionOperationRecord( - operation = GatewayOperationLabel.SPAWN, - kind = telemetryKind, - outcome = outcome, - reason = reason, - startedAt = startedAt, - observation = observation, - error = failure, - ), - ) - } - } - - fun stop(id: String): Boolean { - val startedAt = Instant.now() - var kind = GatewayAgentKindLabel.OTHER - var outcome = GatewayOutcomeLabel.FAILURE - var reason = GatewayFailureReasonLabel.NOT_FOUND - var failure: Throwable? = null - var observation: Observation? = null - try { - return runCatching { - val session = sessions.remove(id) ?: return@runCatching false - kind = session.kind.toTelemetryKind() - observation = startSessionObservation(GatewayOperationLabel.STOP, kind) - tmux.killSession(session.tmuxSession) - session.stableSessionId?.let { - transcriptStore.seal(it) - session.transcriptLease?.let(transcriptStore::releaseLease) - } - outcome = GatewayOutcomeLabel.SUCCESS - reason = GatewayFailureReasonLabel.NONE - log.info("stopped agent {}", id) - true - }.getOrElse { error -> - outcome = GatewayOutcomeLabel.FAILURE - reason = failureReasonLabel(error) - failure = error - throw error - } - } finally { - recordActiveSessionCounts() - recordSessionOperation( - SessionOperationRecord( - operation = GatewayOperationLabel.STOP, - kind = kind, - outcome = outcome, - reason = reason, - startedAt = startedAt, - observation = observation, - error = failure, - ), - ) - } - } - - fun get(id: String): AgentSession? = sessions[id] - - fun list(): List = sessions.values.sortedBy { it.createdAt } - - /** - * Milliseconds since the agent last produced output, derived from the - * mtime of the pane's append-only log (tmux `pipe-pane` touches it on - * every write, even with no client attached). This is the only signal of - * "the agent is between turns / idle" that survives a disconnected client, - * so the control plane can hold off recycling a runner until its agent has - * gone quiet. Null when the session or its log is unknown — callers treat - * an unknown as "not safe to recycle". - */ - fun idleMillis(id: String): Long? { - val session = sessions[id] ?: return null - return runCatching { - val lastWrite = Files.getLastModifiedTime(session.logFile).toMillis() - (Instant.now().toEpochMilli() - lastWrite).coerceAtLeast(0L) - }.getOrNull() - } - - fun cleanupTranscript(stableSessionId: String): Boolean { - val startedAt = Instant.now() - var outcome = GatewayOutcomeLabel.SUCCESS - var reason = GatewayFailureReasonLabel.NONE - var failure: Throwable? = null - try { - return runCatching { - val ok = transcriptStore.cleanup(transcriptStore.validateStableSessionId(stableSessionId)) - if (!ok) { - outcome = GatewayOutcomeLabel.FAILURE - reason = GatewayFailureReasonLabel.NOT_FOUND - } - ok - }.getOrElse { error -> - outcome = GatewayOutcomeLabel.FAILURE - reason = failureReasonLabel(error) - failure = error - throw error - } - } finally { - recordSessionOperation( - SessionOperationRecord( - operation = GatewayOperationLabel.REPLAY, - kind = GatewayAgentKindLabel.OTHER, - outcome = outcome, - reason = reason, - startedAt = startedAt, - observed = false, - error = failure, - ), - ) - } - } - - fun send( - id: String, - input: String, - enter: Boolean = true, - ) = withSessionOperation(id, GatewayOperationLabel.INPUT, observed = true) { session -> - tmux.sendKeys(session.tmuxSession, input, enter = enter) - } - - fun stageInput( - id: String, - content: String, - requestedName: String?, - ): StagedInput = - withSessionOperation(id, GatewayOperationLabel.INPUT, observed = false) { session -> - val bytes = content.toByteArray(Charsets.UTF_8) - require(bytes.isNotEmpty()) { "staged input content is empty" } - require(bytes.size.toLong() <= props.stagedInputs.maxBytes) { - "staged input exceeds ${props.stagedInputs.maxBytes} bytes" - } - - val root = Path(session.cwd).toAbsolutePath().normalize() - val dir = root.resolve(props.stagedInputs.dirName).normalize() - require(dir.startsWith(root)) { "staged input directory must stay inside the workspace" } - - Files.createDirectories(dir) - val safeName = safeFileName(requestedName) - val fileName = "${timestamp()}-${UUID.randomUUID().toString().take(ID_PREVIEW_CHARS)}-$safeName" - val target = dir.resolve(fileName).normalize() - require(target.startsWith(dir)) { "staged input path must stay inside the staging directory" } - - Files.write(target, bytes, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE) - log.info("staged {} bytes for agent {} at {}", bytes.size, id, target) - StagedInput(path = target.toString(), bytes = bytes.size.toLong(), name = safeName) - } - - fun capture( - id: String, - historyLines: Int = 1_000, - ): String = - withSessionOperation(id, GatewayOperationLabel.REPLAY, observed = false) { session -> - tmux.capture(session.tmuxSession, historyLines) - } - - fun captureWithEscapes(id: String): String = - withSessionOperation(id, GatewayOperationLabel.REPLAY, observed = false) { session -> - tmux.captureWithEscapes(session.tmuxSession) - } - - fun resize( - id: String, - cols: Int, - rows: Int, - ) = withSessionOperation(id, GatewayOperationLabel.RESIZE, observed = true) { session -> - tmux.resize(session.tmuxSession, cols, rows) - } - - private fun withSessionOperation( - id: String, - operation: GatewayOperationLabel, - observed: Boolean, - block: (AgentSession) -> T, - ): T { - val startedAt = Instant.now() - var kind = GatewayAgentKindLabel.OTHER - var outcome = GatewayOutcomeLabel.SUCCESS - var reason = GatewayFailureReasonLabel.NONE - var failure: Throwable? = null - var observation: Observation? = null - try { - return runCatching { - val session = sessions[id] ?: error("unknown agent: $id") - kind = session.kind.toTelemetryKind() - if (observed) observation = startSessionObservation(operation, kind) - block(session) - }.getOrElse { error -> - outcome = GatewayOutcomeLabel.FAILURE - reason = failureReasonLabel(error) - failure = error - throw error - } - } finally { - recordSessionOperation( - SessionOperationRecord( - operation = operation, - kind = kind, - outcome = outcome, - reason = reason, - startedAt = startedAt, - observed = observed, - observation = observation, - error = failure, - ), - ) - } - } - - private fun recordActiveSessionCounts() { - val counts = sessions.values.groupingBy { it.kind }.eachCount() - AgentKind.values().forEach { kind -> - telemetry.recordActiveSessions( - GatewayActiveSessionsSample( - status = GatewayStatusLabel.RUNNING, - kind = kind.toTelemetryKind(), - mode = GatewayModeLabel.INTERACTIVE, - count = counts[kind]?.toLong() ?: 0, - ), + lifecycle = AgentSessionLifecycle(tmux, transcriptStore, registry, sessionTelemetry), + controls = AgentSessionControls(tmux, props, registry, sessionTelemetry), + transcripts = AgentTranscriptAdmin(transcriptStore, sessionTelemetry), + maintenance = AgentTranscriptMaintenance(tmux, props, transcriptStore, registry, sessionTelemetry), ) } } - - private fun startSessionObservation( - operation: GatewayOperationLabel, - kind: GatewayAgentKindLabel, - ): Observation = - Observation - .start(SESSION_OBSERVATION_NAME, observationRegistry) - .lowCardinalityKeyValue("operation", operation.label) - .lowCardinalityKeyValue("kind", kind.label) - .lowCardinalityKeyValue("mode", GatewayModeLabel.INTERACTIVE.label) - - private fun recordSessionOperation(record: SessionOperationRecord) { - val activeObservation = - record.observation ?: if (record.observed) startSessionObservation(record.operation, record.kind) else null - activeObservation - ?.lowCardinalityKeyValue("outcome", record.outcome.label) - ?.lowCardinalityKeyValue("reason", record.reason.label) - record.error?.let { activeObservation?.error(it) } - activeObservation?.stop() - telemetry.recordOperation( - GatewayOperationTelemetry( - operation = record.operation, - kind = record.kind, - mode = GatewayModeLabel.INTERACTIVE, - outcome = record.outcome, - reason = record.reason, - duration = Duration.between(record.startedAt, Instant.now()), - ), - ) - } - - private fun failureReasonLabel(error: Throwable): GatewayFailureReasonLabel = - when (error) { - is IllegalArgumentException -> GatewayFailureReasonLabel.INVALID_REQUEST - is IllegalStateException -> - if (error.message?.startsWith("unknown agent:") == true) { - GatewayFailureReasonLabel.NOT_FOUND - } else { - GatewayFailureReasonLabel.OTHER - } - is IOException -> GatewayFailureReasonLabel.IO_ERROR - is ProcessFailedException -> GatewayFailureReasonLabel.PROCESS_EXITED - is ProcessTimeoutException -> GatewayFailureReasonLabel.TIMEOUT - is SecurityException -> GatewayFailureReasonLabel.PERMISSION_DENIED - is InterruptedException -> GatewayFailureReasonLabel.CANCELLED - else -> GatewayFailureReasonLabel.UNKNOWN - } - - private fun AgentKind.toTelemetryKind(): GatewayAgentKindLabel = GatewayAgentKindLabel.fromRaw(name) - - /** - * Build the CLI command and return the native session id alongside it. - * - * For Claude: on a fresh start a new UUID is passed as `--session-id - * ` so the CLI process has a stable native identity without - * inheriting another conversation. On revival ([resumeCliSessionId] - * set) the prior conversation is resumed via `--resume ` only when - * Claude's local transcript still exists under ~/.claude/projects for - * this cwd; otherwise `--session-id ` starts fresh while preserving - * the persisted native id across epochs. - * - * For Codex: Codex has no `--session-id` to pre-set, so each session - * runs under its own [codexHome] (auth/config symlinked in by - * [prepareCodexHome]). That isolation makes the rollout this session - * creates the only one in its sessions dir, so its id is captured after - * launch ([captureCodexSessionId]) and `codex resume ` resumes that - * exact rollout on revival — avoiding the `codex resume --last` - * cross-session risk on the shared credentials PVC. The id is unknown at - * command-build time, so the fresh case returns null here and the caller - * captures it post-launch. - * - * Shell has no session id. - */ - private fun commandAndSessionIdFor( - kind: AgentKind, - cwd: String, - resumeCliSessionId: String? = null, - codexHome: Path? = null, - ): AgentCommand = - when (kind) { - AgentKind.CLAUDE -> { - val cliSessionId = resumeCliSessionId ?: UUID.randomUUID().toString() - val transcript = resumeCliSessionId?.let { claudeTranscriptLocator.findTranscript(cwd, it) } - val sessionArgs = - if (resumeCliSessionId == null) { - listOf("--session-id", cliSessionId) - } else if (transcript != null) { - log.info( - "claude revival selected resume sessionId={} reason=transcript-exists " + - "requestedCwd={} transcriptCwd={}", - resumeCliSessionId, - cwd, - transcript.cwd, - ) - listOf("--resume", resumeCliSessionId) - } else { - log.info( - "claude revival selected fresh-with-stable-id sessionId={} reason=transcript-missing", - resumeCliSessionId, - ) - listOf("--session-id", resumeCliSessionId) - } - AgentCommand( - command = listOf(props.cli.claude) + props.cli.claudeArgs + sessionArgs, - cliSessionId = cliSessionId, - cwd = transcript?.cwd ?: cwd, - ) - } - AgentKind.CODEX -> { - // `env CODEX_HOME=` scopes this session's rollouts - // independently of the shared tmux server environment. - val envPrefix = codexHome?.let { listOf("env", "CODEX_HOME=$it") }.orEmpty() - val resumeArgs = resumeCliSessionId?.let { listOf("resume", it) }.orEmpty() - AgentCommand( - command = envPrefix + listOf(props.cli.codex) + props.cli.codexArgs + resumeArgs, - cliSessionId = resumeCliSessionId, - cwd = cwd, - ) - } - AgentKind.SHELL -> - AgentCommand( - command = listOf("/bin/bash", "-l"), - cliSessionId = null, - cwd = cwd, - ) - } - - private fun codexSessionHome(stableSessionId: String): Path = - Path(props.codex.home).resolve(props.codex.sessionHomesSubdir).resolve(stableSessionId) - - // Isolate this session's Codex state: its own sessions/ dir, with the - // shared OAuth + config symlinked back to the credentials home so token - // refresh still writes through to the one canonical auth.json. Best - // effort — if it fails, Codex still launches and id capture just misses. - private fun prepareCodexHome(home: Path) { - runCatching { - Files.createDirectories(home.resolve("sessions")) - for (name in CODEX_SHARED_FILES) { - val src = Path(props.codex.home).resolve(name) - val link = home.resolve(name) - if (Files.exists(link, LinkOption.NOFOLLOW_LINKS) || !Files.exists(src)) continue - Files.createSymbolicLink(link, src) - } - }.onFailure { log.warn("could not prepare codex home {}: {}", home, it.message) } - } - - // Poll the isolated sessions dir for the rollout Codex writes at session - // start and read its UUID from the filename. Isolation guarantees a single - // rollout, so there is no cross-session ambiguity; on timeout we return - // null and the session simply starts fresh on its next revival. - private fun captureCodexSessionId(codexHome: Path): String? { - val sessionsDir = codexHome.resolve("sessions") - val deadline = Instant.now().plusMillis(props.codex.captureTimeoutMs) - while (Instant.now().isBefore(deadline)) { - newestRolloutId(sessionsDir)?.let { return it } - try { - Thread.sleep(props.codex.capturePollMs) - } catch (e: InterruptedException) { - Thread.currentThread().interrupt() - break - } - } - log.warn("codex session id not captured within {}ms under {}", props.codex.captureTimeoutMs, sessionsDir) - return null - } - - private fun newestRolloutId(sessionsDir: Path): String? { - val newest = - if (Files.isDirectory(sessionsDir)) { - newestRolloutFile(sessionsDir) - } else { - null - } - return newest?.let { ROLLOUT_ID.find(it.fileName.toString())?.groupValues?.get(1) } - } - - private fun newestRolloutFile(sessionsDir: Path): Path? = - Files.walk(sessionsDir).use { stream -> - stream - .filter { Files.isRegularFile(it) && ROLLOUT_FILE.matches(it.fileName.toString()) } - .max(compareBy { runCatching { Files.getLastModifiedTime(it).toMillis() }.getOrDefault(0L) }) - .orElse(null) - } - - private fun safeFileName(requestedName: String?): String { - val raw = requestedName?.trim()?.takeIf { it.isNotBlank() } ?: DEFAULT_STAGED_INPUT_NAME - val leaf = raw.replace('\\', '/').substringAfterLast('/') - val safe = - SAFE_NAME_CHARS - .replace(leaf, "-") - .trim('.', '-', '_') - .take(MAX_STAGED_INPUT_NAME_CHARS) - return safe.takeIf { it.isNotBlank() } ?: DEFAULT_STAGED_INPUT_NAME - } - - private fun timestamp(): String = STAGED_INPUT_TIMESTAMP.format(Instant.now()) - - companion object { - private const val DEFAULT_STAGED_INPUT_NAME = "input.txt" - private const val ID_PREVIEW_CHARS = 8 - private const val MAX_STAGED_INPUT_NAME_CHARS = 80 - private const val SESSION_OBSERVATION_NAME = "agent.gateway.session.operation" - private val SAFE_NAME_CHARS = Regex("[^A-Za-z0-9._-]+") - private val ROLLOUT_FILE = Regex("""^rollout-.*\.jsonl$""") - private val ROLLOUT_ID = - Regex("""([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})\.jsonl$""") - private val CODEX_SHARED_FILES = listOf("auth.json", "config.toml") - private val STAGED_INPUT_TIMESTAMP = - DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss-SSS").withZone(ZoneOffset.UTC) - } - - private data class AgentCommand( - val command: List, - val cliSessionId: String?, - val cwd: String, - ) - - private data class SessionOperationRecord( - val operation: GatewayOperationLabel, - val kind: GatewayAgentKindLabel, - val outcome: GatewayOutcomeLabel, - val reason: GatewayFailureReasonLabel, - val startedAt: Instant, - val observed: Boolean = true, - val observation: Observation? = null, - val error: Throwable? = null, - ) } diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionOperations.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionOperations.kt new file mode 100644 index 0000000..1777a55 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionOperations.kt @@ -0,0 +1,62 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +interface AgentSessionSpawnOperations { + fun spawn( + kind: AgentKind, + workspacePath: String? = null, + stableSessionId: String? = null, + epoch: Long? = null, + continuation: AgentContinuation? = null, + ): AgentSession + + fun spawn( + kind: AgentKind, + workspacePath: String? = null, + resumeCliSessionId: String, + ): AgentSession + + fun spawn(request: AgentSpawnRequest): AgentSession +} + +interface AgentSessionLifecycleOperations { + fun stop(id: String): Boolean +} + +interface AgentSessionRegistryOperations { + fun get(id: String): AgentSession? + + fun list(): List + + fun idleMillis(id: String): Long? +} + +interface AgentSessionControlOperations { + fun send( + id: String, + input: String, + enter: Boolean = true, + ) + + fun stageInput( + id: String, + content: String, + requestedName: String?, + ): StagedInput + + fun capture( + id: String, + historyLines: Int = 1_000, + ): String + + fun captureWithEscapes(id: String): String + + fun resize( + id: String, + cols: Int, + rows: Int, + ) +} + +interface AgentTranscriptAdminOperations { + fun cleanupTranscript(stableSessionId: String): Boolean +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionRegistry.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionRegistry.kt new file mode 100644 index 0000000..7421391 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionRegistry.kt @@ -0,0 +1,41 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import java.nio.file.Files +import java.time.Instant +import java.util.concurrent.ConcurrentHashMap + +/** + * In-memory registry of active agents on this Pod. ConcurrentHashMap is enough: + * spawn-vs-stop on the same id is a caller conflict, not shared state recovery. + */ +internal class AgentSessionRegistry : AgentSessionRegistryOperations { + private val sessions = ConcurrentHashMap() + + override fun get(id: String): AgentSession? = sessions[id] + + override fun list(): List = sessions.values.sortedBy { it.createdAt } + + override fun idleMillis(id: String): Long? { + val session = sessions[id] ?: return null + return runCatching { + val lastWrite = Files.getLastModifiedTime(session.logFile).toMillis() + (Instant.now().toEpochMilli() - lastWrite).coerceAtLeast(0L) + }.getOrNull() + } + + fun put(session: AgentSession) { + sessions[session.id] = session + } + + fun update(session: AgentSession) { + sessions[session.id] = session + } + + fun remove(id: String): AgentSession? = sessions.remove(id) + + fun values(): List = sessions.values.toList() + + fun ids(): List = sessions.keys.sorted() + + fun countsByKind(): Map = sessions.values.groupingBy { it.kind }.eachCount() +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt new file mode 100644 index 0000000..e2d251e --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt @@ -0,0 +1,187 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import org.slf4j.LoggerFactory +import java.nio.file.Path +import java.time.Instant +import java.util.UUID + +internal class AgentSessionSpawnWorkflow( + private val tmux: TmuxClient, + private val props: GatewayProperties, + private val transcriptStore: TranscriptStore, + private val registry: AgentSessionRegistry, + private val telemetry: AgentSessionTelemetry, + claudeTranscriptLocator: ClaudeTranscriptLocator, +) : AgentSessionSpawnOperations { + private val log = LoggerFactory.getLogger(AgentSessionSpawnWorkflow::class.java) + private val codexHomes = CodexSessionHomes(props) + private val commandFactory = AgentCommandFactory(props, claudeTranscriptLocator) + + override fun spawn( + kind: AgentKind, + workspacePath: String?, + stableSessionId: String?, + epoch: Long?, + continuation: AgentContinuation?, + ): AgentSession = + spawn( + AgentSpawnRequest( + kind = kind, + workspacePath = workspacePath, + stableSessionId = stableSessionId, + epoch = epoch, + continuation = continuation, + ), + ) + + override fun spawn( + kind: AgentKind, + workspacePath: String?, + resumeCliSessionId: String, + ): AgentSession = + spawn( + AgentSpawnRequest( + kind = kind, + workspacePath = workspacePath, + resumeCliSessionId = resumeCliSessionId, + ), + ) + + override fun spawn(request: AgentSpawnRequest): AgentSession { + val startedAt = Instant.now() + val telemetryKind = with(telemetry) { request.kind.toTelemetryKind() } + var outcome = GatewayOutcomeLabel.SUCCESS + var reason = GatewayFailureReasonLabel.NONE + var failure: Throwable? = null + val observation = telemetry.startSessionObservation(GatewayOperationLabel.SPAWN, telemetryKind) + try { + return runCatching { createSession(request) } + .getOrElse { error -> + outcome = GatewayOutcomeLabel.FAILURE + reason = telemetry.failureReasonLabel(error) + failure = error + throw error + } + } finally { + telemetry.recordSessionOperation( + SessionOperationRecord( + operation = GatewayOperationLabel.SPAWN, + kind = telemetryKind, + outcome = outcome, + reason = reason, + startedAt = startedAt, + observation = observation, + error = failure, + ), + ) + } + } + + private fun createSession(request: AgentSpawnRequest): AgentSession { + val id = UUID.randomUUID().toString().substring(0, ID_PREVIEW_CHARS) + val tmuxSession = "agent-$id" + val requestedCwd = request.workspacePath ?: props.workspaceRoot + val stableSessionId = + request.stableSessionId?.let(transcriptStore::validateStableSessionId) ?: UUID.randomUUID().toString() + val epoch = request.epoch ?: 1 + require(epoch > 0) { "epoch must be positive" } + val lease = transcriptStore.acquireLease(stableSessionId, tmuxSession, epoch) + val logFile = openTranscript(request, stableSessionId, epoch, lease) + val codexHome = if (request.kind == AgentKind.CODEX) codexHomes.homeFor(stableSessionId) else null + codexHome?.let(codexHomes::prepare) + val launch = + commandFactory.commandAndSessionIdFor( + kind = request.kind, + cwd = requestedCwd, + resumeCliSessionId = request.resumeCliSessionId, + codexHome = codexHome, + ) + startTmux(tmuxSession, launch, logFile, lease) + val session = sessionFrom(id, request, tmuxSession, launch, logFile, lease, stableSessionId, epoch, codexHome) + registry.put(session) + telemetry.recordActiveSessionCounts() + log.info( + "spawned {} agent {} ({}) in {} cliSessionId={} stableSessionId={} epoch={}", + request.kind, + id, + tmuxSession, + launch.cwd, + session.cliSessionId, + stableSessionId, + epoch, + ) + return session + } + + private fun openTranscript( + request: AgentSpawnRequest, + stableSessionId: String, + epoch: Long, + lease: TranscriptLease, + ): Path { + var opened = false + try { + transcriptStore.open(stableSessionId, epoch) + if (epoch > 1 || request.continuation != null) { + transcriptStore.appendContinuationDelimiter(stableSessionId, epoch, request.continuation) + } + return transcriptStore.activeSegmentPath(stableSessionId).also { opened = true } + } finally { + if (!opened) transcriptStore.releaseLease(lease) + } + } + + private fun startTmux( + tmuxSession: String, + launch: AgentCommand, + logFile: Path, + lease: TranscriptLease, + ) { + var started = false + try { + tmux.newSession(tmuxSession, launch.command, launch.cwd) + tmux.startPipeToFile(tmuxSession, logFile) + started = true + } finally { + if (!started) transcriptStore.releaseLease(lease) + } + } + + private fun sessionFrom( + id: String, + request: AgentSpawnRequest, + tmuxSession: String, + launch: AgentCommand, + logFile: Path, + lease: TranscriptLease, + stableSessionId: String, + epoch: Long, + codexHome: Path?, + ): AgentSession { + val cliSessionId = + launch.cliSessionId + ?: codexHome?.takeIf { request.kind == AgentKind.CODEX }?.let(codexHomes::captureSessionId) + return AgentSession( + id = id, + kind = request.kind, + tmuxSession = tmuxSession, + logFile = logFile, + cwd = launch.cwd, + createdAt = Instant.now(), + cliSessionId = cliSessionId, + stableSessionId = stableSessionId, + epoch = epoch, + continuation = request.continuation, + transcriptFile = logFile, + transcriptLease = lease, + ) + } + + private companion object { + const val ID_PREVIEW_CHARS = 8 + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionTelemetry.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionTelemetry.kt new file mode 100644 index 0000000..e07d8c1 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionTelemetry.kt @@ -0,0 +1,102 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayActiveSessionsSample +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayAgentKindLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayModeLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationTelemetry +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStatusLabel +import com.jorisjonkers.personalstack.agentgateway.process.ProcessFailedException +import com.jorisjonkers.personalstack.agentgateway.process.ProcessTimeoutException +import io.micrometer.observation.Observation +import io.micrometer.observation.ObservationRegistry +import java.io.IOException +import java.time.Duration +import java.time.Instant + +internal class AgentSessionTelemetry( + private val registry: AgentSessionRegistry, + private val telemetry: AgentGatewayTelemetry, + private val observationRegistry: ObservationRegistry, +) { + fun recordActiveSessionCounts() { + val counts = registry.countsByKind() + AgentKind.values().forEach { kind -> + telemetry.recordActiveSessions( + GatewayActiveSessionsSample( + status = GatewayStatusLabel.RUNNING, + kind = kind.toTelemetryKind(), + mode = GatewayModeLabel.INTERACTIVE, + count = counts[kind]?.toLong() ?: 0, + ), + ) + } + } + + fun startSessionObservation( + operation: GatewayOperationLabel, + kind: GatewayAgentKindLabel, + ): Observation = + Observation + .start(SESSION_OBSERVATION_NAME, observationRegistry) + .lowCardinalityKeyValue("operation", operation.label) + .lowCardinalityKeyValue("kind", kind.label) + .lowCardinalityKeyValue("mode", GatewayModeLabel.INTERACTIVE.label) + + fun recordSessionOperation(record: SessionOperationRecord) { + val activeObservation = + record.observation ?: if (record.observed) startSessionObservation(record.operation, record.kind) else null + activeObservation + ?.lowCardinalityKeyValue("outcome", record.outcome.label) + ?.lowCardinalityKeyValue("reason", record.reason.label) + record.error?.let { activeObservation?.error(it) } + activeObservation?.stop() + telemetry.recordOperation( + GatewayOperationTelemetry( + operation = record.operation, + kind = record.kind, + mode = GatewayModeLabel.INTERACTIVE, + outcome = record.outcome, + reason = record.reason, + duration = Duration.between(record.startedAt, Instant.now()), + ), + ) + } + + fun failureReasonLabel(error: Throwable): GatewayFailureReasonLabel = + when (error) { + is IllegalArgumentException -> GatewayFailureReasonLabel.INVALID_REQUEST + is IllegalStateException -> + if (error.message?.startsWith("unknown agent:") == true) { + GatewayFailureReasonLabel.NOT_FOUND + } else { + GatewayFailureReasonLabel.OTHER + } + is IOException -> GatewayFailureReasonLabel.IO_ERROR + is ProcessFailedException -> GatewayFailureReasonLabel.PROCESS_EXITED + is ProcessTimeoutException -> GatewayFailureReasonLabel.TIMEOUT + is SecurityException -> GatewayFailureReasonLabel.PERMISSION_DENIED + is InterruptedException -> GatewayFailureReasonLabel.CANCELLED + else -> GatewayFailureReasonLabel.UNKNOWN + } + + fun AgentKind.toTelemetryKind(): GatewayAgentKindLabel = GatewayAgentKindLabel.fromRaw(name) + + private companion object { + const val SESSION_OBSERVATION_NAME = "agent.gateway.session.operation" + } +} + +internal data class SessionOperationRecord( + val operation: GatewayOperationLabel, + val kind: GatewayAgentKindLabel, + val outcome: GatewayOutcomeLabel, + val reason: GatewayFailureReasonLabel, + val startedAt: Instant, + val observed: Boolean = true, + val observation: Observation? = null, + val error: Throwable? = null, +) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptAdmin.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptAdmin.kt new file mode 100644 index 0000000..0a7bbb2 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptAdmin.kt @@ -0,0 +1,46 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayAgentKindLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import java.time.Instant + +internal class AgentTranscriptAdmin( + private val transcriptStore: TranscriptStore, + private val telemetry: AgentSessionTelemetry, +) : AgentTranscriptAdminOperations { + override fun cleanupTranscript(stableSessionId: String): Boolean { + val startedAt = Instant.now() + var outcome = GatewayOutcomeLabel.SUCCESS + var reason = GatewayFailureReasonLabel.NONE + var failure: Throwable? = null + try { + return runCatching { + val ok = transcriptStore.cleanup(transcriptStore.validateStableSessionId(stableSessionId)) + if (!ok) { + outcome = GatewayOutcomeLabel.FAILURE + reason = GatewayFailureReasonLabel.NOT_FOUND + } + ok + }.getOrElse { error -> + outcome = GatewayOutcomeLabel.FAILURE + reason = telemetry.failureReasonLabel(error) + failure = error + throw error + } + } finally { + telemetry.recordSessionOperation( + SessionOperationRecord( + operation = GatewayOperationLabel.REPLAY, + kind = GatewayAgentKindLabel.OTHER, + outcome = outcome, + reason = reason, + startedAt = startedAt, + observed = false, + error = failure, + ), + ) + } + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt new file mode 100644 index 0000000..2f4c3d2 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt @@ -0,0 +1,105 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import org.slf4j.LoggerFactory +import java.nio.channels.FileChannel +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption +import java.time.Instant + +internal class AgentTranscriptMaintenance( + private val tmux: TmuxClient, + private val props: GatewayProperties, + private val transcriptStore: TranscriptStore, + private val registry: AgentSessionRegistry, + private val telemetry: AgentSessionTelemetry, +) { + private val log = LoggerFactory.getLogger(AgentTranscriptMaintenance::class.java) + + fun maintain() { + registry.values().forEach(::maintainSession) + } + + private fun maintainSession(session: AgentSession) { + val startedAt = Instant.now() + runCatching { + if (session.stableSessionId != null) { + maintainDurableSession(session, startedAt) + } else { + maintainLegacyLog(session, startedAt) + } + }.onFailure { + recordMaintenance(session, startedAt, GatewayOutcomeLabel.FAILURE, it) + log.warn("trim of {} failed: {}", session.logFile, it.message) + } + } + + private fun maintainDurableSession( + session: AgentSession, + startedAt: Instant, + ) { + val stableSessionId = requireNotNull(session.stableSessionId) + val beforeTrim = transcriptStore.recoverMetadata(stableSessionId).logicalStart + val current = renewLease(session) + transcriptStore.rotateIfNeeded(stableSessionId) + rotatePipeIfNeeded(current, transcriptStore.activeSegmentPath(stableSessionId)) + val trimmed = transcriptStore.trimIfNeeded(stableSessionId) + if (trimmed.logicalStart > beforeTrim) { + recordMaintenance(current, startedAt, GatewayOutcomeLabel.SUCCESS) + } + } + + private fun renewLease(session: AgentSession): AgentSession { + val current = + session.transcriptLease + ?.let(transcriptStore::renewLease) + ?.let { session.copy(transcriptLease = it) } + ?: session + if (current !== session) registry.update(current) + return current + } + + private fun rotatePipeIfNeeded( + session: AgentSession, + active: Path, + ) { + if (active == session.logFile) return + tmux.startPipeToFile(session.tmuxSession, active) + registry.update(session.copy(logFile = active, transcriptFile = active)) + log.info("rotated agent {} transcript pipe to {}", session.id, active) + } + + private fun maintainLegacyLog( + session: AgentSession, + startedAt: Instant, + ) { + val file = session.logFile + if (!Files.exists(file) || Files.size(file) <= props.tmux.logCapBytes) return + FileChannel.open(file, StandardOpenOption.WRITE).use { it.truncate(0) } + log.info("trimmed agent {} log past {} bytes", session.id, props.tmux.logCapBytes) + recordMaintenance(session, startedAt, GatewayOutcomeLabel.SUCCESS) + } + + private fun recordMaintenance( + session: AgentSession, + startedAt: Instant, + outcome: GatewayOutcomeLabel, + error: Throwable? = null, + ) { + telemetry.recordSessionOperation( + SessionOperationRecord( + operation = GatewayOperationLabel.REPLAY, + kind = with(telemetry) { session.kind.toTelemetryKind() }, + outcome = outcome, + reason = error?.let(telemetry::failureReasonLabel) ?: GatewayFailureReasonLabel.NONE, + startedAt = startedAt, + observed = false, + error = error, + ), + ) + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/CodexSessionHomes.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/CodexSessionHomes.kt new file mode 100644 index 0000000..54a78cd --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/CodexSessionHomes.kt @@ -0,0 +1,71 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import org.slf4j.LoggerFactory +import java.nio.file.Files +import java.nio.file.LinkOption +import java.nio.file.Path +import java.time.Instant +import kotlin.io.path.Path + +internal class CodexSessionHomes( + private val props: GatewayProperties, +) { + private val log = LoggerFactory.getLogger(CodexSessionHomes::class.java) + + fun homeFor(stableSessionId: String): Path = + Path(props.codex.home).resolve(props.codex.sessionHomesSubdir).resolve(stableSessionId) + + fun prepare(home: Path) { + runCatching { + Files.createDirectories(home.resolve("sessions")) + for (name in CODEX_SHARED_FILES) { + val src = Path(props.codex.home).resolve(name) + val link = home.resolve(name) + if (Files.exists(link, LinkOption.NOFOLLOW_LINKS) || !Files.exists(src)) continue + Files.createSymbolicLink(link, src) + } + }.onFailure { log.warn("could not prepare codex home {}: {}", home, it.message) } + } + + fun captureSessionId(codexHome: Path): String? { + val sessionsDir = codexHome.resolve("sessions") + val deadline = Instant.now().plusMillis(props.codex.captureTimeoutMs) + while (Instant.now().isBefore(deadline)) { + newestRolloutId(sessionsDir)?.let { return it } + try { + Thread.sleep(props.codex.capturePollMs) + } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + break + } + } + log.warn("codex session id not captured within {}ms under {}", props.codex.captureTimeoutMs, sessionsDir) + return null + } + + private fun newestRolloutId(sessionsDir: Path): String? { + val newest = + if (Files.isDirectory(sessionsDir)) { + newestRolloutFile(sessionsDir) + } else { + null + } + return newest?.let { ROLLOUT_ID.find(it.fileName.toString())?.groupValues?.get(1) } + } + + private fun newestRolloutFile(sessionsDir: Path): Path? = + Files.walk(sessionsDir).use { stream -> + stream + .filter { Files.isRegularFile(it) && ROLLOUT_FILE.matches(it.fileName.toString()) } + .max(compareBy { runCatching { Files.getLastModifiedTime(it).toMillis() }.getOrDefault(0L) }) + .orElse(null) + } + + private companion object { + val ROLLOUT_FILE = Regex("""^rollout-.*\.jsonl$""") + val ROLLOUT_ID = + Regex("""([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})\.jsonl$""") + val CODEX_SHARED_FILES = listOf("auth.json", "config.toml") + } +} diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt new file mode 100644 index 0000000..efbc0b7 --- /dev/null +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt @@ -0,0 +1,173 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry +import io.micrometer.observation.ObservationRegistry +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assertions.assertThatThrownBy +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.io.IOException +import java.nio.file.Files +import java.nio.file.Path +import java.time.Instant + +class AgentSessionCollaboratorsTest { + private val tmux = mockk(relaxed = true) + + @Test + fun `registry lists sessions by creation time and reports idle milliseconds`( + @TempDir tmp: Path, + ) { + val registry = AgentSessionRegistry() + val olderLog = tmp.resolve("older.log") + val newerLog = tmp.resolve("newer.log") + Files.writeString(olderLog, "older") + Files.writeString(newerLog, "newer") + registry.put(session("newer", AgentKind.CODEX, newerLog, Instant.parse("2026-01-02T00:00:00Z"))) + registry.put(session("older", AgentKind.CLAUDE, olderLog, Instant.parse("2026-01-01T00:00:00Z"))) + + assertThat(registry.list().map { it.id }).containsExactly("older", "newer") + assertThat(registry.idleMillis("older")).isNotNull().isGreaterThanOrEqualTo(0L) + assertThat(registry.idleMillis("missing")).isNull() + assertThat(registry.countsByKind()).containsEntry(AgentKind.CLAUDE, 1) + } + + @Test + fun `command factory resumes claude only when a matching transcript exists`( + @TempDir tmp: Path, + ) { + val workspace = tmp.resolve("repo") + Files.createDirectories(workspace) + val prior = "11111111-2222-4333-8444-555555555555" + val locator = ClaudeTranscriptLocator(tmp.resolve("claude/projects")) + val transcript = locator.transcriptPath(workspace.toString(), prior) + Files.createDirectories(transcript.parent) + Files.writeString(transcript, """{"sessionId":"$prior"}""" + "\n") + val factory = AgentCommandFactory(props(tmp), locator) + + val launch = factory.commandAndSessionIdFor(AgentKind.CLAUDE, workspace.toString(), prior) + + assertThat(launch.cliSessionId).isEqualTo(prior) + assertThat(launch.command).containsSequence("--resume", prior) + assertThat(launch.command).doesNotContain("--session-id") + } + + @Test + fun `codex session homes capture rollout id from isolated sessions directory`( + @TempDir tmp: Path, + ) { + val stable = "33333333-3333-4333-8333-333333333333" + val rolloutId = "019edf83-d3a0-7c73-8572-fec943c6091f" + val homes = CodexSessionHomes(props(tmp)) + val sessionsDir = homes.homeFor(stable).resolve("sessions/2026/06/19") + Files.createDirectories(sessionsDir) + Files.writeString( + sessionsDir.resolve("rollout-2026-06-19T10-53-39-$rolloutId.jsonl"), + """{"type":"session_meta"}""", + ) + + assertThat(homes.captureSessionId(homes.homeFor(stable))).isEqualTo(rolloutId) + } + + @Test + fun `maintenance rotates durable transcript pipe when active segment changes`( + @TempDir tmp: Path, + ) { + val props = props(tmp, transcripts = GatewayProperties.Transcripts(segmentBytes = 8, capBytes = 8)) + val store = TranscriptStore(props) + val registry = AgentSessionRegistry() + val telemetry = AgentSessionTelemetry(registry, AgentGatewayTelemetry.NOOP, ObservationRegistry.NOOP) + val stable = "11111111-1111-1111-1111-111111111111" + val lease = store.acquireLease(stable, "agent-one", 1) + store.open(stable, 1) + val firstSegment = store.activeSegmentPath(stable) + Files.write(firstSegment, ByteArray(16) { 'x'.code.toByte() }) + registry.put( + session("one", AgentKind.SHELL, firstSegment).copy( + tmuxSession = "agent-one", + stableSessionId = stable, + transcriptFile = firstSegment, + transcriptLease = lease, + ), + ) + + AgentTranscriptMaintenance(tmux, props, store, registry, telemetry).maintain() + + val current = requireNotNull(registry.get("one")) + assertThat(current.logFile).isEqualTo(store.activeSegmentPath(stable)) + assertThat(current.logFile).isNotEqualTo(firstSegment) + verify { tmux.startPipeToFile("agent-one", current.logFile) } + } + + @Test + fun `spawn workflow releases transcript lease when tmux startup fails`( + @TempDir tmp: Path, + ) { + val props = props(tmp) + val store = TranscriptStore(props) + val registry = AgentSessionRegistry() + val telemetry = AgentSessionTelemetry(registry, AgentGatewayTelemetry.NOOP, ObservationRegistry.NOOP) + val workflow = + AgentSessionSpawnWorkflow( + tmux = tmux, + props = props, + transcriptStore = store, + registry = registry, + telemetry = telemetry, + claudeTranscriptLocator = ClaudeTranscriptLocator(tmp.resolve("claude/projects")), + ) + val stable = "11111111-1111-1111-1111-111111111111" + every { tmux.newSession(any(), any(), any()) } throws IOException("tmux failed") + + assertThatThrownBy { workflow.spawn(AgentKind.SHELL, stableSessionId = stable) } + .isInstanceOf(IOException::class.java) + + val replacement = store.acquireLease(stable, "replacement", 1) + assertThat(replacement.owner).isEqualTo("replacement") + } + + private fun props( + tmp: Path, + transcripts: GatewayProperties.Transcripts = GatewayProperties.Transcripts(trimIntervalSeconds = 600), + ): GatewayProperties { + val workspace = tmp.resolve("workspace") + Files.createDirectories(workspace) + return GatewayProperties( + workspaceRoot = workspace.toString(), + tmux = GatewayProperties.Tmux(socketName = "agent-gw", stateDir = tmp.toString()), + cli = + GatewayProperties.Cli( + claude = "/usr/local/bin/claude", + codex = "/usr/local/bin/codex", + claudeArgs = listOf("--dangerously-skip-permissions"), + codexArgs = listOf("--dangerously-bypass-approvals-and-sandbox"), + ), + transcripts = transcripts, + codex = + GatewayProperties.Codex( + home = tmp.resolve("codex-home").toString(), + captureTimeoutMs = 50, + capturePollMs = 5, + ), + ) + } + + private fun session( + id: String, + kind: AgentKind, + logFile: Path, + createdAt: Instant = Instant.now(), + ): AgentSession = + AgentSession( + id = id, + kind = kind, + tmuxSession = "agent-$id", + logFile = logFile, + cwd = logFile.parent.toString(), + createdAt = createdAt, + ) +} From 7fd9873127cc77a053fe00675a4b24a7b2681314 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:47:02 +0000 Subject: [PATCH 2/2] fix: address detekt and ktlint findings from CI --- .../agentgateway/tmux/AgentSessionControls.kt | 3 +- .../agentgateway/tmux/AgentSessionManager.kt | 83 +++++++++++++------ .../tmux/AgentSessionSpawnWorkflow.kt | 75 ++++++++++++----- 3 files changed, 116 insertions(+), 45 deletions(-) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt index 6f4f5e5..a76f19b 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionControls.kt @@ -137,6 +137,7 @@ internal class AgentSessionControls( const val ID_PREVIEW_CHARS = 8 const val MAX_STAGED_INPUT_NAME_CHARS = 80 val SAFE_NAME_CHARS = Regex("[^A-Za-z0-9._-]+") - val STAGED_INPUT_TIMESTAMP = DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss-SSS").withZone(ZoneOffset.UTC) + val STAGED_INPUT_TIMESTAMP: DateTimeFormatter = + DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss-SSS").withZone(ZoneOffset.UTC) } } diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt index 7c6e708..d9969af 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManager.kt @@ -34,12 +34,17 @@ class AgentSessionManager private constructor( claudeTranscriptLocator: ClaudeTranscriptLocator = ClaudeTranscriptLocator.fromEnvironment(), ) : this( AgentSessionComponents.create( - tmux = tmux, - props = props, - transcriptStore = transcriptStore, - telemetry = telemetry, - observationRegistry = observationRegistry, - claudeTranscriptLocator = claudeTranscriptLocator, + AgentSessionComponentDependencies( + tmux = tmux, + props = props, + transcriptStore = transcriptStore, + telemetry = + AgentTelemetryDependencies( + gateway = telemetry, + observationRegistry = observationRegistry, + ), + claudeTranscriptLocator = claudeTranscriptLocator, + ), ), ) @@ -81,34 +86,64 @@ private data class AgentSessionComponents( val maintenance: AgentTranscriptMaintenance, ) { companion object { - fun create( - tmux: TmuxClient, - props: GatewayProperties, - transcriptStore: TranscriptStore, - telemetry: AgentGatewayTelemetry, - observationRegistry: ObservationRegistry, - claudeTranscriptLocator: ClaudeTranscriptLocator, - ): AgentSessionComponents { + fun create(dependencies: AgentSessionComponentDependencies): AgentSessionComponents { val registry = AgentSessionRegistry() - val sessionTelemetry = AgentSessionTelemetry(registry, telemetry, observationRegistry) + val sessionTelemetry = + AgentSessionTelemetry( + registry, + dependencies.telemetry.gateway, + dependencies.telemetry.observationRegistry, + ) return AgentSessionComponents( - props = props, + props = dependencies.props, registry = registry, telemetry = sessionTelemetry, spawn = AgentSessionSpawnWorkflow( - tmux = tmux, - props = props, - transcriptStore = transcriptStore, + tmux = dependencies.tmux, + props = dependencies.props, + transcriptStore = dependencies.transcriptStore, registry = registry, telemetry = sessionTelemetry, - claudeTranscriptLocator = claudeTranscriptLocator, + claudeTranscriptLocator = dependencies.claudeTranscriptLocator, + ), + lifecycle = + AgentSessionLifecycle( + dependencies.tmux, + dependencies.transcriptStore, + registry, + sessionTelemetry, + ), + controls = + AgentSessionControls( + dependencies.tmux, + dependencies.props, + registry, + sessionTelemetry, + ), + transcripts = AgentTranscriptAdmin(dependencies.transcriptStore, sessionTelemetry), + maintenance = + AgentTranscriptMaintenance( + dependencies.tmux, + dependencies.props, + dependencies.transcriptStore, + registry, + sessionTelemetry, ), - lifecycle = AgentSessionLifecycle(tmux, transcriptStore, registry, sessionTelemetry), - controls = AgentSessionControls(tmux, props, registry, sessionTelemetry), - transcripts = AgentTranscriptAdmin(transcriptStore, sessionTelemetry), - maintenance = AgentTranscriptMaintenance(tmux, props, transcriptStore, registry, sessionTelemetry), ) } } } + +private data class AgentSessionComponentDependencies( + val tmux: TmuxClient, + val props: GatewayProperties, + val transcriptStore: TranscriptStore, + val telemetry: AgentTelemetryDependencies, + val claudeTranscriptLocator: ClaudeTranscriptLocator, +) + +private data class AgentTelemetryDependencies( + val gateway: AgentGatewayTelemetry, + val observationRegistry: ObservationRegistry, +) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt index e2d251e..3f8bd69 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt @@ -101,7 +101,26 @@ internal class AgentSessionSpawnWorkflow( codexHome = codexHome, ) startTmux(tmuxSession, launch, logFile, lease) - val session = sessionFrom(id, request, tmuxSession, launch, logFile, lease, stableSessionId, epoch, codexHome) + val session = + sessionFrom( + SpawnedSessionContext( + identity = + SpawnedSessionIdentity( + id = id, + stableSessionId = stableSessionId, + epoch = epoch, + ), + request = request, + tmuxSession = tmuxSession, + launch = launch, + resources = + SpawnedSessionResources( + logFile = logFile, + lease = lease, + codexHome = codexHome, + ), + ), + ) registry.put(session) telemetry.recordActiveSessionCounts() log.info( @@ -151,36 +170,52 @@ internal class AgentSessionSpawnWorkflow( } } - private fun sessionFrom( - id: String, - request: AgentSpawnRequest, - tmuxSession: String, - launch: AgentCommand, - logFile: Path, - lease: TranscriptLease, - stableSessionId: String, - epoch: Long, - codexHome: Path?, - ): AgentSession { + private fun sessionFrom(context: SpawnedSessionContext): AgentSession { + val request = context.request + val launch = context.launch + val identity = context.identity + val resources = context.resources val cliSessionId = launch.cliSessionId - ?: codexHome?.takeIf { request.kind == AgentKind.CODEX }?.let(codexHomes::captureSessionId) + ?: resources.codexHome + ?.takeIf { request.kind == AgentKind.CODEX } + ?.let(codexHomes::captureSessionId) return AgentSession( - id = id, + id = identity.id, kind = request.kind, - tmuxSession = tmuxSession, - logFile = logFile, + tmuxSession = context.tmuxSession, + logFile = resources.logFile, cwd = launch.cwd, createdAt = Instant.now(), cliSessionId = cliSessionId, - stableSessionId = stableSessionId, - epoch = epoch, + stableSessionId = identity.stableSessionId, + epoch = identity.epoch, continuation = request.continuation, - transcriptFile = logFile, - transcriptLease = lease, + transcriptFile = resources.logFile, + transcriptLease = resources.lease, ) } + private data class SpawnedSessionContext( + val identity: SpawnedSessionIdentity, + val request: AgentSpawnRequest, + val tmuxSession: String, + val launch: AgentCommand, + val resources: SpawnedSessionResources, + ) + + private data class SpawnedSessionIdentity( + val id: String, + val stableSessionId: String, + val epoch: Long, + ) + + private data class SpawnedSessionResources( + val logFile: Path, + val lease: TranscriptLease, + val codexHome: Path?, + ) + private companion object { const val ID_PREVIEW_CHARS = 8 }