From 02a2ff73fc3b73c5cc0154b3df9b6d68f7ab4249 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:27:44 +0000 Subject: [PATCH 1/4] refactor: extract job registry and process lifecycle from HeadlessJobManager HeadlessJobManager carried 20 functions behind a documented TooManyFunctions suppression. A durable job registry (with cancelled-transition preservation and RUNNING sidecar recovery) and a process-lifecycle collaborator split out behind the unchanged public API; the suppression is gone and the registry is unit-tested. --- .../headless/HeadlessJobManager.kt | 384 ++---------------- .../headless/HeadlessJobRegistry.kt | 128 ++++++ .../headless/HeadlessJobSidecar.kt | 4 +- .../headless/HeadlessJobTelemetry.kt | 78 ++++ .../headless/HeadlessProcessLifecycle.kt | 181 +++++++++ .../headless/HeadlessJobRegistryTest.kt | 104 +++++ 6 files changed, 523 insertions(+), 356 deletions(-) create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistry.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobTelemetry.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt create mode 100644 services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt index a252d33..2233dab 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt @@ -2,30 +2,19 @@ package com.jorisjonkers.personalstack.agentgateway.headless 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.tmux.AgentKind import org.slf4j.LoggerFactory import org.springframework.beans.factory.DisposableBean import org.springframework.beans.factory.InitializingBean import org.springframework.stereotype.Component import java.io.File -import java.io.IOException import java.nio.file.Files import java.nio.file.Path import java.time.Duration import java.time.Instant import java.util.UUID -import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.ExecutorService -import java.util.concurrent.Executors -import java.util.concurrent.TimeUnit /** * Async registry for one-shot (headless) agent runs. Each job launches @@ -33,8 +22,8 @@ import java.util.concurrent.TimeUnit * to a JSONL capture file in the gateway state dir, and updates status * when the process exits or times out. * - * Concurrency: one virtual thread per job; job state is updated via - * `ConcurrentHashMap.compute` so concurrent status reads are safe. + * Concurrency: one virtual thread per job; job transitions are serialized + * by the registry so concurrent status reads are safe. * Cancellation kills the OS process and marks the job CANCELLED. * * Durability: job metadata is persisted as a JSON sidecar file @@ -49,59 +38,22 @@ import java.util.concurrent.TimeUnit * [HeadlessLaunchRequest] to opt a specific run back in (council workers * that explicitly want KB recall/capture may do so). */ -@Suppress("TooManyFunctions") @Component class HeadlessJobManager( private val props: GatewayProperties, - private val telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, - private val processFactory: ProcessFactory = DefaultProcessFactory, + telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, + processFactory: ProcessFactory = DefaultProcessFactory, ) : DisposableBean, InitializingBean { private val log = LoggerFactory.getLogger(HeadlessJobManager::class.java) - private val jobs = ConcurrentHashMap() - private val processes = ConcurrentHashMap() - private val executor: ExecutorService = Executors.newVirtualThreadPerTaskExecutor() + private val stateDir: Path = Path.of(props.tmux.stateDir) + private val jobTelemetry = HeadlessJobTelemetry(telemetry) + private val registry = HeadlessJobRegistry(stateDir, jobTelemetry) + private val lifecycle = HeadlessProcessLifecycle(registry, jobTelemetry, processFactory) /** Reload durable job state from the state directory at startup. */ override fun afterPropertiesSet() { - val stateDir = Path.of(props.tmux.stateDir) - if (!Files.isDirectory(stateDir)) return - runCatching { - Files.list(stateDir).use { entries -> - entries - .filter { it.fileName.toString().matches(SIDECAR_FILE_PATTERN) } - .forEach { sidecar -> reloadSidecar(sidecar) } - } - }.onFailure { ex -> - log.warn("headless job state reload failed: {}", ex.message) - } - log.info("headless job registry reloaded {} entries from {}", jobs.size, stateDir) - } - - private fun reloadSidecar(sidecar: Path) { - runCatching { - val job = HeadlessJobSidecar.read(sidecar) - // Jobs that were RUNNING when the previous process died will never - // complete — surface them as FAILED so callers aren't stuck polling. - val recovered = - if (job.status == HeadlessJobStatus.RUNNING) { - job.copy(status = HeadlessJobStatus.FAILED, completedAt = job.completedAt ?: Instant.now()) - } else { - job - } - // Only register if the output file still exists; orphaned sidecars - // whose output was cleaned up separately are silently skipped. - if (Files.exists(recovered.outputFile)) { - jobs[recovered.id] = recovered - // Converge the on-disk state so a RUNNING->FAILED recovery is - // not re-applied (with a fresh completedAt) on every restart. - if (recovered != job) { - persistSidecar(recovered) - } - } - }.onFailure { ex -> - log.warn("headless job sidecar {} could not be reloaded: {}", sidecar.fileName, ex.message) - } + registry.reload() } fun launch( @@ -124,316 +76,55 @@ class HeadlessJobManager( fun launch(request: HeadlessLaunchRequest): HeadlessJob { val id = UUID.randomUUID().toString().substring(0, 8) val cwd = File(request.workspacePath ?: props.workspaceRoot) - val stateDir = Path.of(props.tmux.stateDir).also { Files.createDirectories(it) } + Files.createDirectories(stateDir) val outputFile = stateDir.resolve("headless-$id.jsonl") Files.createFile(outputFile) val command = headlessCommandFor(request.kind, request.prompt, request.cliSessionId, request.partialMessages) - val job = + val job = registry.register( HeadlessJob( id = id, kind = request.kind, status = HeadlessJobStatus.RUNNING, outputFile = outputFile, createdAt = Instant.now(), - ) - jobs[id] = job - persistSidecar(job, stateDir) - recordJobCounts() - recordHeadlessJobEvent( + ), + ) + jobTelemetry.recordEvent( kind = request.kind, operation = GatewayOperationLabel.SPAWN, outcome = GatewayOutcomeLabel.SUCCESS, duration = Duration.between(job.createdAt, Instant.now()), ) - executor.submit { - runJob( - HeadlessRunContext( - id = id, - kind = request.kind, - command = command, - cwd = cwd, - outputFile = outputFile.toFile(), - timeoutSeconds = request.timeoutSeconds, - enableKbHooks = request.enableKbHooks, - ), - ) - } + lifecycle.submit( + HeadlessRunContext( + id = id, + kind = request.kind, + command = command, + cwd = cwd, + outputFile = outputFile.toFile(), + timeoutSeconds = request.timeoutSeconds, + enableKbHooks = request.enableKbHooks, + ), + ) log.info("launched headless {} job {} in {} (kbHooks={})", request.kind, id, cwd, request.enableKbHooks) return job } - fun get(id: String): HeadlessJob? = jobs[id] + fun get(id: String): HeadlessJob? = registry.get(id) - fun list(): List = jobs.values.sortedBy { it.createdAt } + fun list(): List = registry.list() - fun cancel(id: String): Boolean { - val process = - processes.remove(id) - ?: return jobs[id]?.let { job -> - recordHeadlessJobEvent( - kind = job.kind, - operation = GatewayOperationLabel.STOP, - outcome = GatewayOutcomeLabel.SKIPPED, - duration = Duration.ZERO, - ) - true - } ?: false - process.destroyForcibly() - val cancelled = markJob(id, HeadlessJobStatus.CANCELLED) - cancelled?.let { - recordHeadlessJobEvent( - kind = it.kind, - operation = GatewayOperationLabel.STOP, - outcome = GatewayOutcomeLabel.CANCELLED, - reason = GatewayFailureReasonLabel.CANCELLED, - duration = jobDuration(it), - ) - } - log.info("cancelled headless job {}", id) - return true - } + fun cancel(id: String): Boolean = lifecycle.cancel(id) fun readOutput( id: String, maxChars: Int = MAX_OUTPUT_CHARS, - ): String { - val job = jobs[id] ?: return "" - return runCatching { - val bytes = Files.readAllBytes(job.outputFile) - val text = String(bytes, Charsets.UTF_8) - if (text.length <= maxChars) text else "…" + text.takeLast(maxChars) - }.getOrDefault("") - } + ): String = registry.readOutput(id, maxChars) override fun destroy() { - executor.shutdownNow() - // Wait for interrupted job threads to finish their final state - // transition (which persists a sidecar file) so shutdown does not - // race their writes — e.g. against a temp-dir cleanup in tests or - // a Pod teardown removing the state dir. - runCatching { executor.awaitTermination(SHUTDOWN_GRACE_SECONDS, TimeUnit.SECONDS) } - } - - private fun runJob(context: HeadlessRunContext) { - val process = - startProcess(context.id, context.kind, context.command, context.cwd, context.enableKbHooks) - ?: run { - recordCleanup(context.kind, GatewayOutcomeLabel.SKIPPED, GatewayFailureReasonLabel.UNKNOWN) - return - } - processes[context.id] = process - try { - awaitAndCapture(context.id, process, context.outputFile, context.timeoutSeconds) - } catch (ex: InterruptedException) { - process.destroyForcibly() - val cancelled = markJob(context.id, HeadlessJobStatus.CANCELLED) - cancelled?.let { - recordHeadlessJobEvent( - kind = it.kind, - operation = GatewayOperationLabel.HEADLESS_JOB, - outcome = GatewayOutcomeLabel.CANCELLED, - reason = GatewayFailureReasonLabel.CANCELLED, - duration = jobDuration(it), - ) - } - } finally { - val removed = processes.remove(context.id) - val reason = - if (jobs[context.id]?.status == HeadlessJobStatus.CANCELLED) { - GatewayFailureReasonLabel.CANCELLED - } else { - GatewayFailureReasonLabel.NONE - } - recordCleanup( - kind = context.kind, - outcome = if (removed == null) GatewayOutcomeLabel.SKIPPED else GatewayOutcomeLabel.SUCCESS, - reason = reason, - ) - } - } - - private fun startProcess( - id: String, - kind: AgentKind, - command: List, - cwd: File, - enableKbHooks: Boolean, - ): Process? = - runCatching { - processFactory.start(command, cwd, enableKbHooks) - }.getOrElse { ex -> - log.error("headless job {} failed to start: {}", id, ex.message) - val failed = markJob(id, HeadlessJobStatus.FAILED) - failed?.let { - recordHeadlessJobEvent( - kind = kind, - operation = GatewayOperationLabel.SPAWN, - outcome = GatewayOutcomeLabel.FAILURE, - reason = startFailureReason(ex), - duration = jobDuration(it), - ) - } - null - } - - private fun awaitAndCapture( - id: String, - process: Process, - outputFile: File, - timeoutSeconds: Long, - ) { - // Async gobbler keeps the pipe drained so waitFor never hangs on a full buffer. - val gobbler = Thread.ofVirtual().start { process.inputStream.copyTo(outputFile.outputStream()) } - val finished = process.waitFor(timeoutSeconds, TimeUnit.SECONDS) - gobbler.join(GOBBLER_JOIN_MS) - if (!finished) { - process.destroyForcibly() - val failed = updateJob(id, HeadlessJobStatus.FAILED, TIMEOUT_EXIT_CODE) - failed?.takeUnless { it.status == HeadlessJobStatus.CANCELLED }?.let { - recordHeadlessJobEvent( - kind = it.kind, - operation = GatewayOperationLabel.HEADLESS_JOB, - outcome = GatewayOutcomeLabel.FAILURE, - reason = GatewayFailureReasonLabel.TIMEOUT, - duration = jobDuration(it), - ) - } - log.warn("headless job {} timed out after {}s", id, timeoutSeconds) - } else { - val exitCode = process.exitValue() - val status = if (exitCode == 0) HeadlessJobStatus.COMPLETED else HeadlessJobStatus.FAILED - val updated = updateJob(id, status, exitCode) - updated?.takeUnless { it.status == HeadlessJobStatus.CANCELLED }?.let { - recordHeadlessJobEvent( - kind = it.kind, - operation = GatewayOperationLabel.HEADLESS_JOB, - outcome = - if (status == HeadlessJobStatus.COMPLETED) { - GatewayOutcomeLabel.SUCCESS - } else { - GatewayOutcomeLabel.FAILURE - }, - reason = - if (status == HeadlessJobStatus.COMPLETED) { - GatewayFailureReasonLabel.NONE - } else { - GatewayFailureReasonLabel.PROCESS_EXITED - }, - duration = jobDuration(it), - ) - } - log.info("headless job {} finished status={} exitCode={}", id, status, exitCode) - } - } - - private fun updateJob( - id: String, - status: HeadlessJobStatus, - exitCode: Int, - ): HeadlessJob? { - var updated: HeadlessJob? = null - jobs.compute(id) { _, job -> - job - ?.takeUnless { it.status == HeadlessJobStatus.CANCELLED } - ?.copy(status = status, exitCode = exitCode, completedAt = Instant.now()) - ?.also { updated = it } - ?: job.also { updated = it } - } - updated?.let { persistSidecar(it) } - recordJobCounts() - return updated - } - - private fun markJob( - id: String, - status: HeadlessJobStatus, - ): HeadlessJob? { - var updated: HeadlessJob? = null - jobs.compute(id) { _, job -> - job?.copy(status = status, completedAt = Instant.now()).also { updated = it } - } - updated?.let { persistSidecar(it) } - recordJobCounts() - return updated - } - - private fun persistSidecar( - job: HeadlessJob, - stateDir: Path = Path.of(props.tmux.stateDir), - ) { - runCatching { - val sidecar = stateDir.resolve("headless-${job.id}.json") - HeadlessJobSidecar.write(job, sidecar) - }.onFailure { ex -> - log.warn("headless job {} sidecar write failed: {}", job.id, ex.message) - } - } - - private fun recordJobCounts() { - val counts = jobs.values.groupingBy { it.kind to it.status }.eachCount() - AgentKind.values().forEach { kind -> - HeadlessJobStatus.values().forEach { status -> - telemetry.recordActiveSessions( - GatewayActiveSessionsSample( - status = status.toTelemetryStatus(), - kind = kind.toTelemetryKind(), - mode = GatewayModeLabel.HEADLESS, - count = counts[kind to status]?.toLong() ?: 0, - ), - ) - } - } - } - - private fun recordHeadlessJobEvent( - kind: AgentKind, - operation: GatewayOperationLabel, - outcome: GatewayOutcomeLabel, - reason: GatewayFailureReasonLabel = GatewayFailureReasonLabel.NONE, - duration: Duration, - ) { - telemetry.recordOperation( - GatewayOperationTelemetry( - operation = operation, - kind = kind.toTelemetryKind(), - mode = GatewayModeLabel.HEADLESS, - outcome = outcome, - reason = reason, - duration = duration, - ), - ) - } - - private fun recordCleanup( - kind: AgentKind, - outcome: GatewayOutcomeLabel, - reason: GatewayFailureReasonLabel, - ) { - recordHeadlessJobEvent( - kind = kind, - operation = GatewayOperationLabel.STOP, - outcome = outcome, - reason = reason, - duration = Duration.ZERO, - ) - } - - private fun jobDuration(job: HeadlessJob): Duration { - val completedAt = job.completedAt ?: Instant.now() - return Duration.between(job.createdAt, completedAt) + lifecycle.destroy() } - private fun startFailureReason(ex: Throwable): GatewayFailureReasonLabel = - when (ex) { - is IOException -> GatewayFailureReasonLabel.IO_ERROR - is SecurityException -> GatewayFailureReasonLabel.PERMISSION_DENIED - else -> GatewayFailureReasonLabel.UNKNOWN - } - - private fun AgentKind.toTelemetryKind(): GatewayAgentKindLabel = GatewayAgentKindLabel.fromRaw(name) - - private fun HeadlessJobStatus.toTelemetryStatus(): GatewayStatusLabel = GatewayStatusLabel.fromRaw(name) - /** * Build the one-shot CLI command for a headless run. Claude uses * `-p` (print mode) with `--output-format stream-json` for machine- @@ -487,11 +178,6 @@ class HeadlessJobManager( /** Environment variable that suppresses auto-KB recall/capture hooks in agent-kit. */ const val KB_AUTO_MCP_DISABLED_KEY = "KB_AUTO_MCP_DISABLED" const val KB_AUTO_MCP_DISABLED_VALUE = "1" - private val SIDECAR_FILE_PATTERN = Regex("headless-[0-9a-fA-F]+\\.json") - private const val TIMEOUT_EXIT_CODE = -1 - private const val GOBBLER_JOIN_MS = 2_000L - private const val SHUTDOWN_GRACE_SECONDS = 5L - val DefaultProcessFactory = ProcessFactory { command, cwd, enableKbHooks -> ProcessBuilder(command) @@ -521,13 +207,3 @@ data class HeadlessLaunchRequest( */ val enableKbHooks: Boolean = false, ) - -private data class HeadlessRunContext( - val id: String, - val kind: AgentKind, - val command: List, - val cwd: File, - val outputFile: File, - val timeoutSeconds: Long, - val enableKbHooks: Boolean = false, -) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistry.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistry.kt new file mode 100644 index 0000000..1ced7c6 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistry.kt @@ -0,0 +1,128 @@ +package com.jorisjonkers.personalstack.agentgateway.headless + +import org.slf4j.LoggerFactory +import java.nio.file.Files +import java.nio.file.Path +import java.time.Duration +import java.time.Instant +import java.util.concurrent.ConcurrentHashMap + +internal class HeadlessJobRegistry( + private val stateDir: Path, + private val telemetry: HeadlessJobTelemetry, +) { + private val log = LoggerFactory.getLogger(HeadlessJobRegistry::class.java) + private val jobs = ConcurrentHashMap() + + fun reload() { + if (!Files.isDirectory(stateDir)) return + runCatching { + Files.list(stateDir).use { entries -> + entries + .filter { it.fileName.toString().matches(SIDECAR_FILE_PATTERN) } + .forEach { sidecar -> reloadSidecar(sidecar) } + } + }.onFailure { ex -> + log.warn("headless job state reload failed: {}", ex.message) + } + log.info("headless job registry reloaded {} entries from {}", jobs.size, stateDir) + } + + @Synchronized + fun register(job: HeadlessJob): HeadlessJob { + jobs[job.id] = job + persistSidecar(job) + telemetry.recordCounts(jobs.values) + return job + } + + fun get(id: String): HeadlessJob? = jobs[id] + + fun list(): List = jobs.values.sortedBy { it.createdAt } + + fun readOutput( + id: String, + maxChars: Int, + ): String { + val job = jobs[id] ?: return "" + return runCatching { + val text = Files.readString(job.outputFile, Charsets.UTF_8) + if (text.length <= maxChars) text else "\u2026" + text.takeLast(maxChars) + }.getOrDefault("") + } + + @Synchronized + fun updateUnlessCancelled( + id: String, + status: HeadlessJobStatus, + exitCode: Int, + ): HeadlessJob? { + var updated: HeadlessJob? = null + jobs.compute(id) { _, job -> + val next = + job + ?.takeUnless { it.status == HeadlessJobStatus.CANCELLED } + ?.copy(status = status, exitCode = exitCode, completedAt = Instant.now()) + ?: job + updated = next + next + } + afterTransition(updated) + return updated + } + + @Synchronized + fun markStatus( + id: String, + status: HeadlessJobStatus, + ): HeadlessJob? { + var updated: HeadlessJob? = null + jobs.compute(id) { _, job -> + job?.copy(status = status, completedAt = Instant.now()).also { updated = it } + } + afterTransition(updated) + return updated + } + + fun duration(job: HeadlessJob): Duration { + val completedAt = job.completedAt ?: Instant.now() + return Duration.between(job.createdAt, completedAt) + } + + private fun reloadSidecar(sidecar: Path) { + runCatching { + val job = HeadlessJobSidecar.read(sidecar) + val recovered = + if (job.status == HeadlessJobStatus.RUNNING) { + job.copy(status = HeadlessJobStatus.FAILED, completedAt = job.completedAt ?: Instant.now()) + } else { + job + } + if (Files.exists(recovered.outputFile)) { + jobs[recovered.id] = recovered + if (recovered != job) { + persistSidecar(recovered) + } + } + }.onFailure { ex -> + log.warn("headless job sidecar {} could not be reloaded: {}", sidecar.fileName, ex.message) + } + } + + private fun afterTransition(job: HeadlessJob?) { + job?.let { persistSidecar(it) } + telemetry.recordCounts(jobs.values) + } + + private fun persistSidecar(job: HeadlessJob) { + runCatching { + HeadlessJobSidecar.write(job, stateDir.resolve("headless-${job.id}.json")) + }.onFailure { ex -> + log.warn("headless job {} sidecar write failed: {}", job.id, ex.message) + } + } + + private companion object { + val SIDECAR_FILE_PATTERN = Regex("headless-[0-9a-fA-F]+\\.json") + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt index 3c3df56..770c238 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt @@ -15,8 +15,8 @@ import java.time.Instant * reconstitute the job registry after a Pod restart. * * The format is deliberately minimal — no Jackson dependency — - * because the only consumers are [HeadlessJobManager.afterPropertiesSet] - * (read) and [HeadlessJobManager.persistSidecar] (write). + * because the only consumers are [HeadlessJobRegistry.reload] + * (read) and [HeadlessJobRegistry] state transitions (write). * * Field order in the serialised object is stable so diff-based tooling * produces clean output. diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobTelemetry.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobTelemetry.kt new file mode 100644 index 0000000..99f5ec6 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobTelemetry.kt @@ -0,0 +1,78 @@ +package com.jorisjonkers.personalstack.agentgateway.headless + +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.tmux.AgentKind +import java.io.IOException +import java.time.Duration + +internal class HeadlessJobTelemetry( + private val telemetry: AgentGatewayTelemetry, +) { + fun recordCounts(jobs: Collection) { + val counts = jobs.groupingBy { it.kind to it.status }.eachCount() + AgentKind.values().forEach { kind -> + HeadlessJobStatus.values().forEach { status -> + telemetry.recordActiveSessions( + GatewayActiveSessionsSample( + status = status.toTelemetryStatus(), + kind = kind.toTelemetryKind(), + mode = GatewayModeLabel.HEADLESS, + count = counts[kind to status]?.toLong() ?: 0, + ), + ) + } + } + } + + fun recordEvent( + kind: AgentKind, + operation: GatewayOperationLabel, + outcome: GatewayOutcomeLabel, + reason: GatewayFailureReasonLabel = GatewayFailureReasonLabel.NONE, + duration: Duration, + ) { + telemetry.recordOperation( + GatewayOperationTelemetry( + operation = operation, + kind = kind.toTelemetryKind(), + mode = GatewayModeLabel.HEADLESS, + outcome = outcome, + reason = reason, + duration = duration, + ), + ) + } + + fun recordCleanup( + kind: AgentKind, + outcome: GatewayOutcomeLabel, + reason: GatewayFailureReasonLabel, + ) { + recordEvent( + kind = kind, + operation = GatewayOperationLabel.STOP, + outcome = outcome, + reason = reason, + duration = Duration.ZERO, + ) + } + + fun startFailureReason(ex: Throwable): GatewayFailureReasonLabel = + when (ex) { + is IOException -> GatewayFailureReasonLabel.IO_ERROR + is SecurityException -> GatewayFailureReasonLabel.PERMISSION_DENIED + else -> GatewayFailureReasonLabel.UNKNOWN + } + + private fun AgentKind.toTelemetryKind(): GatewayAgentKindLabel = GatewayAgentKindLabel.fromRaw(name) + + private fun HeadlessJobStatus.toTelemetryStatus(): GatewayStatusLabel = GatewayStatusLabel.fromRaw(name) +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt new file mode 100644 index 0000000..1876531 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt @@ -0,0 +1,181 @@ +package com.jorisjonkers.personalstack.agentgateway.headless + +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayFailureReasonLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOperationLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayOutcomeLabel +import com.jorisjonkers.personalstack.agentgateway.tmux.AgentKind +import org.slf4j.LoggerFactory +import java.io.File +import java.time.Duration +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + +internal class HeadlessProcessLifecycle( + private val registry: HeadlessJobRegistry, + private val telemetry: HeadlessJobTelemetry, + private val processFactory: HeadlessJobManager.ProcessFactory, +) { + private val log = LoggerFactory.getLogger(HeadlessProcessLifecycle::class.java) + private val processes = ConcurrentHashMap() + private val executor: ExecutorService = Executors.newVirtualThreadPerTaskExecutor() + + fun submit(context: HeadlessRunContext) { + executor.submit { runJob(context) } + } + + fun cancel(id: String): Boolean { + val process = + processes.remove(id) + ?: return registry.get(id)?.let { job -> + telemetry.recordEvent( + kind = job.kind, + operation = GatewayOperationLabel.STOP, + outcome = GatewayOutcomeLabel.SKIPPED, + duration = Duration.ZERO, + ) + true + } ?: false + process.destroyForcibly() + val cancelled = registry.markStatus(id, HeadlessJobStatus.CANCELLED) + cancelled?.let { + telemetry.recordEvent( + kind = it.kind, + operation = GatewayOperationLabel.STOP, + outcome = GatewayOutcomeLabel.CANCELLED, + reason = GatewayFailureReasonLabel.CANCELLED, + duration = registry.duration(it), + ) + } + log.info("cancelled headless job {}", id) + return true + } + + fun destroy() { + executor.shutdownNow() + runCatching { executor.awaitTermination(SHUTDOWN_GRACE_SECONDS, TimeUnit.SECONDS) } + } + + private fun runJob(context: HeadlessRunContext) { + val process = + startProcess(context) + ?: run { + telemetry.recordCleanup( + context.kind, + GatewayOutcomeLabel.SKIPPED, + GatewayFailureReasonLabel.UNKNOWN, + ) + return + } + processes[context.id] = process + try { + awaitAndCapture(context, process) + } catch (ex: InterruptedException) { + process.destroyForcibly() + registry.markStatus(context.id, HeadlessJobStatus.CANCELLED)?.let { + telemetry.recordEvent( + kind = it.kind, + operation = GatewayOperationLabel.HEADLESS_JOB, + outcome = GatewayOutcomeLabel.CANCELLED, + reason = GatewayFailureReasonLabel.CANCELLED, + duration = registry.duration(it), + ) + } + } finally { + val removed = processes.remove(context.id) + val reason = + if (registry.get(context.id)?.status == HeadlessJobStatus.CANCELLED) { + GatewayFailureReasonLabel.CANCELLED + } else { + GatewayFailureReasonLabel.NONE + } + val outcome = if (removed == null) GatewayOutcomeLabel.SKIPPED else GatewayOutcomeLabel.SUCCESS + telemetry.recordCleanup(context.kind, outcome, reason) + } + } + + private fun startProcess(context: HeadlessRunContext): Process? = + runCatching { + processFactory.start(context.command, context.cwd, context.enableKbHooks) + }.getOrElse { ex -> + log.error("headless job {} failed to start: {}", context.id, ex.message) + registry.markStatus(context.id, HeadlessJobStatus.FAILED)?.let { + telemetry.recordEvent( + kind = context.kind, + operation = GatewayOperationLabel.SPAWN, + outcome = GatewayOutcomeLabel.FAILURE, + reason = telemetry.startFailureReason(ex), + duration = registry.duration(it), + ) + } + null + } + + private fun awaitAndCapture( + context: HeadlessRunContext, + process: Process, + ) { + val gobbler = Thread.ofVirtual().start { process.inputStream.copyTo(context.outputFile.outputStream()) } + val finished = process.waitFor(context.timeoutSeconds, TimeUnit.SECONDS) + gobbler.join(GOBBLER_JOIN_MS) + if (!finished) { + process.destroyForcibly() + registry.updateUnlessCancelled(context.id, HeadlessJobStatus.FAILED, TIMEOUT_EXIT_CODE) + ?.takeUnless { it.status == HeadlessJobStatus.CANCELLED } + ?.let { + telemetry.recordEvent( + kind = it.kind, + operation = GatewayOperationLabel.HEADLESS_JOB, + outcome = GatewayOutcomeLabel.FAILURE, + reason = GatewayFailureReasonLabel.TIMEOUT, + duration = registry.duration(it), + ) + } + log.warn("headless job {} timed out after {}s", context.id, context.timeoutSeconds) + } else { + val exitCode = process.exitValue() + val status = if (exitCode == 0) HeadlessJobStatus.COMPLETED else HeadlessJobStatus.FAILED + registry.updateUnlessCancelled(context.id, status, exitCode) + ?.takeUnless { it.status == HeadlessJobStatus.CANCELLED } + ?.let { + val outcome = + if (status == HeadlessJobStatus.COMPLETED) { + GatewayOutcomeLabel.SUCCESS + } else { + GatewayOutcomeLabel.FAILURE + } + val reason = + if (status == HeadlessJobStatus.COMPLETED) { + GatewayFailureReasonLabel.NONE + } else { + GatewayFailureReasonLabel.PROCESS_EXITED + } + telemetry.recordEvent( + kind = it.kind, + operation = GatewayOperationLabel.HEADLESS_JOB, + outcome = outcome, + reason = reason, + duration = registry.duration(it), + ) + } + log.info("headless job {} finished status={} exitCode={}", context.id, status, exitCode) + } + } + + private companion object { + private const val TIMEOUT_EXIT_CODE = -1 + private const val GOBBLER_JOIN_MS = 2_000L + private const val SHUTDOWN_GRACE_SECONDS = 5L + } +} + +internal data class HeadlessRunContext( + val id: String, + val kind: AgentKind, + val command: List, + val cwd: File, + val outputFile: File, + val timeoutSeconds: Long, + val enableKbHooks: Boolean = false, +) diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt new file mode 100644 index 0000000..b49f16a --- /dev/null +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt @@ -0,0 +1,104 @@ +package com.jorisjonkers.personalstack.agentgateway.headless + +import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry +import com.jorisjonkers.personalstack.agentgateway.tmux.AgentKind +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.nio.file.Path +import java.time.Instant +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + +class HeadlessJobRegistryTest { + @Test + fun `updateUnlessCancelled does not overwrite cancelled jobs`( + @TempDir tmp: Path, + ) { + val registry = registry(tmp) + val job = runningJob(tmp, "cace1111") + registry.register(job) + + val cancelled = registry.markStatus(job.id, HeadlessJobStatus.CANCELLED) + val completed = registry.updateUnlessCancelled(job.id, HeadlessJobStatus.COMPLETED, 0) + + assertThat(cancelled!!.status).isEqualTo(HeadlessJobStatus.CANCELLED) + assertThat(completed!!.status).isEqualTo(HeadlessJobStatus.CANCELLED) + assertThat(registry.get(job.id)!!.status).isEqualTo(HeadlessJobStatus.CANCELLED) + assertThat(registry.get(job.id)!!.exitCode).isNull() + } + + @Test + fun `concurrent terminal updates leave one consistent persisted state`( + @TempDir tmp: Path, + ) { + val registry = registry(tmp) + val job = runningJob(tmp, "face2222") + registry.register(job) + val ready = CountDownLatch(2) + val start = CountDownLatch(1) + val executor = Executors.newFixedThreadPool(2) + + val completion = executor.submit { + ready.countDown() + start.await(3, TimeUnit.SECONDS) + registry.updateUnlessCancelled(job.id, HeadlessJobStatus.COMPLETED, 0) + } + val cancellation = executor.submit { + ready.countDown() + start.await(3, TimeUnit.SECONDS) + registry.markStatus(job.id, HeadlessJobStatus.CANCELLED) + } + assertThat(ready.await(3, TimeUnit.SECONDS)).isTrue() + start.countDown() + executor.shutdown() + assertThat(executor.awaitTermination(3, TimeUnit.SECONDS)).isTrue() + completion.get() + cancellation.get() + + val current = registry.get(job.id)!! + val persisted = HeadlessJobSidecar.read(tmp.resolve("headless-${job.id}.json")) + assertThat(persisted.status).isEqualTo(current.status) + assertThat(persisted.exitCode).isEqualTo(current.exitCode) + assertThat(current.completedAt).isNotNull() + assertThat(current.status).isEqualTo(HeadlessJobStatus.CANCELLED) + } + + @Test + fun `reload converts running sidecars to failed jobs`( + @TempDir tmp: Path, + ) { + val job = runningJob(tmp, "dead3333") + HeadlessJobSidecar.write(job, tmp.resolve("headless-${job.id}.json")) + + val registry = registry(tmp) + registry.reload() + + val reloaded = registry.get(job.id) + assertThat(reloaded).isNotNull() + assertThat(reloaded!!.status).isEqualTo(HeadlessJobStatus.FAILED) + assertThat(reloaded.completedAt).isNotNull() + } + + private fun registry(tmp: Path): HeadlessJobRegistry = + HeadlessJobRegistry( + stateDir = tmp, + telemetry = HeadlessJobTelemetry(AgentGatewayTelemetry.NOOP), + ) + + private fun runningJob( + tmp: Path, + id: String, + ): HeadlessJob { + val outputFile = tmp.resolve("headless-$id.jsonl") + outputFile.toFile().createNewFile() + return HeadlessJob( + id = id, + kind = AgentKind.CLAUDE, + status = HeadlessJobStatus.RUNNING, + outputFile = outputFile, + createdAt = Instant.now(), + ) + } +} From fa4655b125c093d283731751a85825b88051c2af Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:47:02 +0000 Subject: [PATCH 2/4] fix: address detekt and ktlint findings from CI --- .../headless/HeadlessJobRegistryTest.kt | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt index b49f16a..3699959 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobRegistryTest.kt @@ -40,16 +40,18 @@ class HeadlessJobRegistryTest { val start = CountDownLatch(1) val executor = Executors.newFixedThreadPool(2) - val completion = executor.submit { - ready.countDown() - start.await(3, TimeUnit.SECONDS) - registry.updateUnlessCancelled(job.id, HeadlessJobStatus.COMPLETED, 0) - } - val cancellation = executor.submit { - ready.countDown() - start.await(3, TimeUnit.SECONDS) - registry.markStatus(job.id, HeadlessJobStatus.CANCELLED) - } + val completion = + executor.submit { + ready.countDown() + start.await(3, TimeUnit.SECONDS) + registry.updateUnlessCancelled(job.id, HeadlessJobStatus.COMPLETED, 0) + } + val cancellation = + executor.submit { + ready.countDown() + start.await(3, TimeUnit.SECONDS) + registry.markStatus(job.id, HeadlessJobStatus.CANCELLED) + } assertThat(ready.await(3, TimeUnit.SECONDS)).isTrue() start.countDown() executor.shutdown() From b14e103864c089310b22062c01485e361810abcb Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:53:42 +0000 Subject: [PATCH 3/4] fix: remaining ktlint style and test reference findings --- .../agentgateway/headless/HeadlessJobManager.kt | 3 ++- .../agentgateway/headless/HeadlessProcessLifecycle.kt | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt index 2233dab..31cdfd6 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt @@ -79,7 +79,8 @@ class HeadlessJobManager( Files.createDirectories(stateDir) val outputFile = stateDir.resolve("headless-$id.jsonl") Files.createFile(outputFile) - val command = headlessCommandFor(request.kind, request.prompt, request.cliSessionId, request.partialMessages) + val command = + headlessCommandFor(request.kind, request.prompt, request.cliSessionId, request.partialMessages) val job = registry.register( HeadlessJob( id = id, diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt index 1876531..5c3af17 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessProcessLifecycle.kt @@ -121,7 +121,8 @@ internal class HeadlessProcessLifecycle( gobbler.join(GOBBLER_JOIN_MS) if (!finished) { process.destroyForcibly() - registry.updateUnlessCancelled(context.id, HeadlessJobStatus.FAILED, TIMEOUT_EXIT_CODE) + registry + .updateUnlessCancelled(context.id, HeadlessJobStatus.FAILED, TIMEOUT_EXIT_CODE) ?.takeUnless { it.status == HeadlessJobStatus.CANCELLED } ?.let { telemetry.recordEvent( @@ -136,7 +137,8 @@ internal class HeadlessProcessLifecycle( } else { val exitCode = process.exitValue() val status = if (exitCode == 0) HeadlessJobStatus.COMPLETED else HeadlessJobStatus.FAILED - registry.updateUnlessCancelled(context.id, status, exitCode) + registry + .updateUnlessCancelled(context.id, status, exitCode) ?.takeUnless { it.status == HeadlessJobStatus.CANCELLED } ?.let { val outcome = From 4964afc82f26da4de6554beafafc5179856fa830 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 08:01:58 +0000 Subject: [PATCH 4/4] fix: slim facade below function threshold and settle multiline style --- .../headless/HeadlessJobManager.kt | 26 ++++++++++++------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt index 31cdfd6..80d002b 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManager.kt @@ -80,16 +80,22 @@ class HeadlessJobManager( val outputFile = stateDir.resolve("headless-$id.jsonl") Files.createFile(outputFile) val command = - headlessCommandFor(request.kind, request.prompt, request.cliSessionId, request.partialMessages) - val job = registry.register( - HeadlessJob( - id = id, - kind = request.kind, - status = HeadlessJobStatus.RUNNING, - outputFile = outputFile, - createdAt = Instant.now(), - ), - ) + headlessCommandFor( + request.kind, + request.prompt, + request.cliSessionId, + request.partialMessages, + ) + val job = + registry.register( + HeadlessJob( + id = id, + kind = request.kind, + status = HeadlessJobStatus.RUNNING, + outputFile = outputFile, + createdAt = Instant.now(), + ), + ) jobTelemetry.recordEvent( kind = request.kind, operation = GatewayOperationLabel.SPAWN,