From 64c2f89c4462ed84fdf5a9e2cdc344337518eaa9 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 06:46:28 +0000 Subject: [PATCH 1/7] fix(agent-gateway): inject KB_AUTO_MCP_DISABLED on headless launch + durable job state Defect 1: HeadlessJobManager.DefaultProcessFactory now sets KB_AUTO_MCP_DISABLED=1 in the child process environment for every headless launch. Council/headless workers on hosted runners no longer fire auto-KB recall or capture hooks by accident. An explicit enableKbHooks=true field on HeadlessLaunchRequest and HeadlessRequest DTO lets individual runs opt back in. Defect 2: Job state is persisted to a JSON sidecar file (headless-.json) in the gateway state dir on every status transition. InitializingBean.afterPropertiesSet reloads sidecar files at startup: completed/failed/cancelled jobs are restored as-is; any job still marked RUNNING is re-surfaced as FAILED because the underlying process cannot have survived the Pod restart. Defect 3 (RunnerPodChatGenerator/AgentGatewayClient alignment) is tracked separately as an agents-api change; this commit does not touch agents-api. Fixes #11 --- .../headless/HeadlessJobManager.kt | 97 ++++++++- .../headless/HeadlessJobSidecar.kt | 188 +++++++++++++++++ .../agentgateway/web/HeadlessController.kt | 1 + .../agentgateway/web/dto/Dtos.kt | 4 + .../headless/HeadlessJobManagerTest.kt | 190 ++++++++++++++++-- .../headless/HeadlessJobSidecarTest.kt | 147 ++++++++++++++ 6 files changed, 606 insertions(+), 21 deletions(-) create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt create mode 100644 services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecarTest.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 d19a867..750e0ca 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 @@ -13,6 +13,7 @@ import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStatusLa 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 @@ -36,9 +37,17 @@ import java.util.concurrent.TimeUnit * `ConcurrentHashMap.compute` so concurrent status reads are safe. * Cancellation kills the OS process and marks the job CANCELLED. * - * The job registry is in-memory — Pod restarts lose running jobs. The - * caller (agents-api) should detect a missing job id and surface it - * as a FAILED status. + * Durability: job metadata is persisted as a JSON sidecar file + * (`headless-.json`) next to the output JSONL file in the state dir. + * On startup the manager reloads any completed sidecar files so callers + * can still read the final status and output of jobs that finished before + * the current process started. Jobs that were RUNNING at the time of a + * previous restart are re-surfaced as FAILED (the process died with them). + * + * KB hooks: `KB_AUTO_MCP_DISABLED=1` is injected into every headless + * process environment by default. Pass `enableKbHooks = true` in + * [HeadlessLaunchRequest] to opt a specific run back in (council workers + * that explicitly want KB recall/capture may do so). */ @Suppress("TooManyFunctions") @Component @@ -46,12 +55,48 @@ class HeadlessJobManager( private val props: GatewayProperties, private val telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, private val processFactory: ProcessFactory = DefaultProcessFactory, -) : DisposableBean { +) : DisposableBean, + InitializingBean { private val log = LoggerFactory.getLogger(HeadlessJobManager::class.java) private val jobs = ConcurrentHashMap() private val processes = ConcurrentHashMap() private val executor: ExecutorService = Executors.newVirtualThreadPerTaskExecutor() + /** 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) + .filter { it.fileName.toString().matches(Regex("headless-[a-f0-9]+\\.json")) } + .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 + } + }.onFailure { ex -> + log.warn("headless job sidecar {} could not be reloaded: {}", sidecar.fileName, ex.message) + } + } + fun launch( kind: AgentKind, prompt: String, @@ -85,6 +130,7 @@ class HeadlessJobManager( createdAt = Instant.now(), ) jobs[id] = job + persistSidecar(job, stateDir) recordJobCounts() recordHeadlessJobEvent( kind = request.kind, @@ -101,10 +147,11 @@ class HeadlessJobManager( cwd = cwd, outputFile = outputFile.toFile(), timeoutSeconds = request.timeoutSeconds, + enableKbHooks = request.enableKbHooks, ), ) } - log.info("launched headless {} job {} in {}", request.kind, id, cwd) + log.info("launched headless {} job {} in {} (kbHooks={})", request.kind, id, cwd, request.enableKbHooks) return job } @@ -157,7 +204,7 @@ class HeadlessJobManager( private fun runJob(context: HeadlessRunContext) { val process = - startProcess(context.id, context.kind, context.command, context.cwd) + startProcess(context.id, context.kind, context.command, context.cwd, context.enableKbHooks) ?: run { recordCleanup(context.kind, GatewayOutcomeLabel.SKIPPED, GatewayFailureReasonLabel.UNKNOWN) return @@ -198,9 +245,10 @@ class HeadlessJobManager( kind: AgentKind, command: List, cwd: File, + enableKbHooks: Boolean, ): Process? = runCatching { - processFactory.start(command, cwd) + processFactory.start(command, cwd, enableKbHooks) }.getOrElse { ex -> log.error("headless job {} failed to start: {}", id, ex.message) val failed = markJob(id, HeadlessJobStatus.FAILED) @@ -279,6 +327,7 @@ class HeadlessJobManager( ?.also { updated = it } ?: job.also { updated = it } } + updated?.let { persistSidecar(it) } recordJobCounts() return updated } @@ -291,10 +340,23 @@ class HeadlessJobManager( 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 -> @@ -402,21 +464,30 @@ class HeadlessJobManager( fun start( command: List, cwd: File, + enableKbHooks: Boolean, ): Process } companion object { const val DEFAULT_TIMEOUT_SECONDS = 600L const val MAX_OUTPUT_CHARS = 65_536 + + /** 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 const val TIMEOUT_EXIT_CODE = -1 private const val GOBBLER_JOIN_MS = 2_000L val DefaultProcessFactory = - ProcessFactory { command, cwd -> + ProcessFactory { command, cwd, enableKbHooks -> ProcessBuilder(command) .directory(cwd) .redirectErrorStream(true) - .start() + .also { pb -> + if (!enableKbHooks) { + pb.environment()[KB_AUTO_MCP_DISABLED_KEY] = KB_AUTO_MCP_DISABLED_VALUE + } + }.start() } } } @@ -428,6 +499,13 @@ data class HeadlessLaunchRequest( val cliSessionId: String? = null, val timeoutSeconds: Long = HeadlessJobManager.DEFAULT_TIMEOUT_SECONDS, val partialMessages: Boolean = false, + /** + * When false (the default) the process environment includes + * `KB_AUTO_MCP_DISABLED=1` so headless/council workers do not fire + * auto-KB recall or capture hooks. Set to true only for runs that + * explicitly need KB hook access. + */ + val enableKbHooks: Boolean = false, ) private data class HeadlessRunContext( @@ -437,4 +515,5 @@ private data class HeadlessRunContext( 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/HeadlessJobSidecar.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt new file mode 100644 index 0000000..9f69ff4 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt @@ -0,0 +1,188 @@ +package com.jorisjonkers.personalstack.agentgateway.headless + +import com.jorisjonkers.personalstack.agentgateway.tmux.AgentKind +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption +import java.time.Instant + +/** + * Simple hand-rolled JSON sidecar format for [HeadlessJob] state. + * + * One file per job (`headless-.json`) in the gateway state dir. + * Written on every status change so the most recent on-disk state is + * always consistent with the last update. Read back at startup to + * 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). + * + * Field order in the serialised object is stable so diff-based tooling + * produces clean output. + */ +internal object HeadlessJobSidecar { + fun write( + job: HeadlessJob, + sidecar: Path, + ) { + val json = buildString { + append('{') + appendStringField("id", job.id) + append(',') + appendStringField("kind", job.kind.name) + append(',') + appendStringField("status", job.status.name) + append(',') + appendStringField("outputFile", job.outputFile.toString()) + append(',') + appendStringField("createdAt", job.createdAt.toString()) + append(',') + if (job.completedAt != null) { + appendStringField("completedAt", job.completedAt.toString()) + append(',') + } + if (job.exitCode != null) { + append('"') + append("exitCode") + append('"') + append(':') + append(job.exitCode) + } else { + append('"') + append("exitCode") + append('"') + append(':') + append("null") + } + append('}') + } + Files.writeString( + sidecar, + json, + Charsets.UTF_8, + StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING, + ) + } + + fun read(sidecar: Path): HeadlessJob { + val text = Files.readString(sidecar, Charsets.UTF_8) + val fields = parseJsonObject(text) + + val id = fields["id"] ?: error("sidecar missing 'id': $sidecar") + val kind = AgentKind.valueOf(fields["kind"] ?: error("sidecar missing 'kind': $sidecar")) + val status = HeadlessJobStatus.valueOf(fields["status"] ?: error("sidecar missing 'status': $sidecar")) + val outputFile = Path.of(fields["outputFile"] ?: error("sidecar missing 'outputFile': $sidecar")) + val createdAt = Instant.parse(fields["createdAt"] ?: error("sidecar missing 'createdAt': $sidecar")) + val completedAt = fields["completedAt"]?.let { Instant.parse(it) } + val exitCode = fields["exitCode"]?.takeIf { it != "null" }?.toIntOrNull() + + return HeadlessJob( + id = id, + kind = kind, + status = status, + outputFile = outputFile, + createdAt = createdAt, + completedAt = completedAt, + exitCode = exitCode, + ) + } + + /** + * Minimal hand-rolled JSON object parser. Handles the small fixed schema + * produced by [write] — string values, integer exitCode, null exitCode. + * Does not handle nested objects, arrays, or escaped characters beyond + * what the stdlib [Instant.toString] and [Path.toString] produce. + */ + private fun parseJsonObject(text: String): Map { + val result = mutableMapOf() + // Strip outer braces + val body = text.trim().removePrefix("{").removeSuffix("}") + var pos = 0 + while (pos < body.length) { + // skip whitespace and commas + while (pos < body.length && (body[pos] == ',' || body[pos].isWhitespace())) pos++ + if (pos >= body.length) break + // read key + val key = readString(body, pos) ?: break + pos += key.second + // skip colon + while (pos < body.length && (body[pos] == ':' || body[pos].isWhitespace())) pos++ + // read value — either a quoted string, null, or an integer + val value: String? + val advance: Int + if (pos < body.length && body[pos] == '"') { + val str = readString(body, pos) + value = str?.first + advance = str?.second ?: 0 + } else if (body.startsWith("null", pos)) { + value = null + advance = 4 + } else { + // integer + val start = pos + while (pos < body.length && (body[pos].isDigit() || body[pos] == '-')) pos++ + value = body.substring(start, pos) + advance = 0 + } + result[key.first] = value + if (advance > 0) pos += advance + } + return result + } + + /** + * Reads a JSON-quoted string at position [start] in [text]. + * Returns the unquoted value and the number of characters consumed, + * or null if there is no quoted string at [start]. + */ + private fun readString( + text: String, + start: Int, + ): Pair? { + if (start >= text.length || text[start] != '"') return null + var pos = start + 1 + val sb = StringBuilder() + while (pos < text.length && text[pos] != '"') { + if (text[pos] == '\\' && pos + 1 < text.length) { + when (text[pos + 1]) { + '"' -> { sb.append('"'); pos += 2 } + '\\' -> { sb.append('\\'); pos += 2 } + 'n' -> { sb.append('\n'); pos += 2 } + 'r' -> { sb.append('\r'); pos += 2 } + 't' -> { sb.append('\t'); pos += 2 } + else -> { sb.append(text[pos + 1]); pos += 2 } + } + } else { + sb.append(text[pos]) + pos++ + } + } + if (pos < text.length) pos++ // consume closing '"' + return sb.toString() to (pos - start) + } +} + +private fun StringBuilder.appendStringField( + key: String, + value: String, +) { + append('"') + append(key) + append('"') + append(':') + append('"') + // Escape backslashes and quotes within the value + for (ch in value) { + when (ch) { + '\\' -> append("\\\\") + '"' -> append("\\\"") + '\n' -> append("\\n") + '\r' -> append("\\r") + '\t' -> append("\\t") + else -> append(ch) + } + } + append('"') +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/HeadlessController.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/HeadlessController.kt index b74febf..32e2b1f 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/HeadlessController.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/HeadlessController.kt @@ -36,6 +36,7 @@ class HeadlessController( cliSessionId = req.cliSessionId, timeoutSeconds = req.timeoutSeconds ?: HeadlessJobManager.DEFAULT_TIMEOUT_SECONDS, partialMessages = req.partialMessages, + enableKbHooks = req.enableKbHooks, ), ) return ResponseEntity.status(HttpStatus.CREATED).body(toResponse(job, null)) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/dto/Dtos.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/dto/Dtos.kt index e1fe592..a0bfceb 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/dto/Dtos.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/web/dto/Dtos.kt @@ -87,6 +87,10 @@ data class HeadlessRequest( // Opt-in token-level streaming (Claude `--include-partial-messages`). // Default off so existing headless callers are unaffected. val partialMessages: Boolean = false, + // When false (the default) KB_AUTO_MCP_DISABLED=1 is injected so the + // headless worker does not fire auto-KB recall/capture hooks. Set to + // true only for runs that explicitly need KB hook access. + val enableKbHooks: Boolean = false, ) data class HeadlessJobResponse( diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt index 4170190..ad42e59 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt @@ -43,7 +43,7 @@ class HeadlessJobManagerTest { props = props, telemetry = telemetry, processFactory = processFactory, - ) + ).also { it.afterPropertiesSet() } } @Test @@ -52,7 +52,7 @@ class HeadlessJobManagerTest { ) { val latch = CountDownLatch(1) val mgr = - manager(tmp) { _, _ -> + manager(tmp) { _, _, _ -> latch.await(5, TimeUnit.SECONDS) processOf(exitCode = 0, output = "done") } @@ -70,7 +70,7 @@ class HeadlessJobManagerTest { fun `launch completes with COMPLETED status on zero exit`( @TempDir tmp: Path, ) { - val mgr = manager(tmp) { _, _ -> processOf(exitCode = 0, output = "result: ok") } + val mgr = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "result: ok") } val job = mgr.launch(AgentKind.CLAUDE, "say hello") @@ -91,7 +91,7 @@ class HeadlessJobManagerTest { fun `launch marks job FAILED on non-zero exit`( @TempDir tmp: Path, ) { - val mgr = manager(tmp) { _, _ -> processOf(exitCode = 1, output = "error: failed") } + val mgr = manager(tmp) { _, _, _ -> processOf(exitCode = 1, output = "error: failed") } val job = mgr.launch(AgentKind.CODEX, "do something") @@ -113,7 +113,7 @@ class HeadlessJobManagerTest { val telemetry = RecordingTelemetry() val rawExitText = "result: ok from private run" val prompt = "summarize secret prompt" - val mgr = manager(tmp, telemetry = telemetry) { _, _ -> processOf(exitCode = 0, output = rawExitText) } + val mgr = manager(tmp, telemetry = telemetry) { _, _, _ -> processOf(exitCode = 0, output = rawExitText) } val job = mgr.launch(AgentKind.CLAUDE, prompt, workspacePath = tmp.resolve("workspace-a").toString()) @@ -168,7 +168,7 @@ class HeadlessJobManagerTest { ) { val telemetry = RecordingTelemetry() val rawExitText = "raw exit text with /private/workspace" - val mgr = manager(tmp, telemetry = telemetry) { _, _ -> processOf(exitCode = 17, output = rawExitText) } + val mgr = manager(tmp, telemetry = telemetry) { _, _, _ -> processOf(exitCode = 17, output = rawExitText) } val job = mgr.launch(AgentKind.CODEX, "prompt should not become a label") @@ -201,7 +201,7 @@ class HeadlessJobManagerTest { val telemetry = RecordingTelemetry() val exceptionMessage = "cannot start in ${tmp.resolve("secret-workspace")}" val mgr = - manager(tmp, telemetry = telemetry) { _, _ -> throw IOException(exceptionMessage) } + manager(tmp, telemetry = telemetry) { _, _, _ -> throw IOException(exceptionMessage) } val job = mgr.launch(AgentKind.SHELL, "secret shell prompt") @@ -231,7 +231,7 @@ class HeadlessJobManagerTest { @TempDir tmp: Path, ) { val telemetry = RecordingTelemetry() - val mgr = manager(tmp, telemetry = telemetry) { _, _ -> sleepingProcess() } + val mgr = manager(tmp, telemetry = telemetry) { _, _, _ -> sleepingProcess() } val job = mgr.launch(AgentKind.CLAUDE, "timeout prompt", timeoutSeconds = 0) @@ -262,7 +262,7 @@ class HeadlessJobManagerTest { val telemetry = RecordingTelemetry() val started = CountDownLatch(1) val mgr = - manager(tmp, telemetry = telemetry) { _, _ -> + manager(tmp, telemetry = telemetry) { _, _, _ -> sleepingProcess().also { started.countDown() } } @@ -306,7 +306,7 @@ class HeadlessJobManagerTest { val latch = CountDownLatch(1) var capturedCommand: List? = null val mgr = - manager(tmp) { command, _ -> + manager(tmp) { command, _, _ -> capturedCommand = command latch.countDown() processOf(exitCode = 0, output = "done") @@ -335,7 +335,7 @@ class HeadlessJobManagerTest { ) { val started = CountDownLatch(1) val mgr = - manager(tmp) { _, _ -> + manager(tmp) { _, _, _ -> sleepingProcess().also { started.countDown() } } @@ -355,7 +355,7 @@ class HeadlessJobManagerTest { fun `readOutput returns captured output after completion`( @TempDir tmp: Path, ) { - val mgr = manager(tmp) { _, _ -> processOf(exitCode = 0, output = "hello from claude") } + val mgr = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "hello from claude") } val job = mgr.launch(AgentKind.CLAUDE, "say hello") @@ -369,6 +369,170 @@ class HeadlessJobManagerTest { mgr.destroy() } + // region: KB_AUTO_MCP_DISABLED injection tests + + @Test + fun `KB_AUTO_MCP_DISABLED is injected by default when enableKbHooks is false`( + @TempDir tmp: Path, + ) { + var capturedKbFlag: Boolean? = null + val latch = CountDownLatch(1) + val mgr = + manager(tmp) { _, _, enableKbHooks -> + capturedKbFlag = enableKbHooks + latch.countDown() + processOf(exitCode = 0, output = "done") + } + + mgr.launch(HeadlessLaunchRequest(kind = AgentKind.CLAUDE, prompt = "test", enableKbHooks = false)) + + assertThat(latch.await(3, TimeUnit.SECONDS)).isTrue() + assertThat(capturedKbFlag).isFalse() + mgr.destroy() + } + + @Test + fun `KB_AUTO_MCP_DISABLED is not injected when enableKbHooks is true`( + @TempDir tmp: Path, + ) { + var capturedKbFlag: Boolean? = null + val latch = CountDownLatch(1) + val mgr = + manager(tmp) { _, _, enableKbHooks -> + capturedKbFlag = enableKbHooks + latch.countDown() + processOf(exitCode = 0, output = "done") + } + + mgr.launch(HeadlessLaunchRequest(kind = AgentKind.CLAUDE, prompt = "test", enableKbHooks = true)) + + assertThat(latch.await(3, TimeUnit.SECONDS)).isTrue() + assertThat(capturedKbFlag).isTrue() + mgr.destroy() + } + + @Test + fun `DefaultProcessFactory injects KB_AUTO_MCP_DISABLED when enableKbHooks is false`( + @TempDir tmp: Path, + ) { + // Use a real process that prints its environment so we can inspect it. + val process = + HeadlessJobManager.DefaultProcessFactory.start( + command = listOf("/bin/sh", "-c", "echo KB=${KB_AUTO_MCP_DISABLED_KEY}"), + cwd = tmp.toFile(), + enableKbHooks = false, + ) + val output = process.inputStream.bufferedReader().use { it.readText() } + process.waitFor(5, TimeUnit.SECONDS) + // The var is in the env — the echo just checks the process starts. + // More importantly: the process env should contain the key. We verify + // that via the factory contract: flag=false → key is set. + assertThat(output).doesNotContain("unbound variable") + // Also verify that default process has the var in its environment via + // a shell that can access it: + val check = + ProcessBuilder(listOf("/bin/sh", "-c", "echo VAL=\$$KB_AUTO_MCP_DISABLED_KEY")) + .redirectErrorStream(true) + .also { pb -> pb.environment()[KB_AUTO_MCP_DISABLED_KEY] = HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE } + .start() + val checkOut = check.inputStream.bufferedReader().use { it.readText() } + check.waitFor(5, TimeUnit.SECONDS) + assertThat(checkOut.trim()).isEqualTo("VAL=${HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE}") + } + + // endregion + + // region: durable sidecar tests + + @Test + fun `completed job is written to sidecar file`( + @TempDir tmp: Path, + ) { + val mgr = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "output data") } + + val job = mgr.launch(AgentKind.CLAUDE, "write code") + waitUntil { mgr.get(job.id)?.status == HeadlessJobStatus.COMPLETED } + + val sidecar = tmp.resolve("headless-${job.id}.json") + assertThat(sidecar).exists() + val content = sidecar.toFile().readText() + assertThat(content).contains("\"id\":\"${job.id}\"") + assertThat(content).contains("\"status\":\"COMPLETED\"") + assertThat(content).contains("\"kind\":\"CLAUDE\"") + mgr.destroy() + } + + @Test + fun `afterPropertiesSet reloads completed jobs from sidecar files`( + @TempDir tmp: Path, + ) { + // First manager launches and completes a job. + val mgr1 = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "output") } + val job = mgr1.launch(AgentKind.CLAUDE, "task") + waitUntil { mgr1.get(job.id)?.status == HeadlessJobStatus.COMPLETED } + mgr1.destroy() + + // Second manager (simulating a Pod restart) reloads the sidecar. + val mgr2 = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "") } + val reloaded = mgr2.get(job.id) + + assertThat(reloaded).isNotNull() + assertThat(reloaded!!.status).isEqualTo(HeadlessJobStatus.COMPLETED) + assertThat(reloaded.kind).isEqualTo(AgentKind.CLAUDE) + mgr2.destroy() + } + + @Test + fun `afterPropertiesSet surfaces RUNNING jobs as FAILED after restart`( + @TempDir tmp: Path, + ) { + // Directly write a sidecar that claims a job was RUNNING when the process died. + val jobId = "abcd1234" + val outputFile = tmp.resolve("headless-$jobId.jsonl") + outputFile.toFile().createNewFile() + val sidecar = tmp.resolve("headless-$jobId.json") + val fakeJob = + HeadlessJob( + id = jobId, + kind = AgentKind.CLAUDE, + status = HeadlessJobStatus.RUNNING, + outputFile = outputFile, + createdAt = java.time.Instant.now(), + ) + HeadlessJobSidecar.write(fakeJob, sidecar) + + val mgr = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "") } + val reloaded = mgr.get(jobId) + + assertThat(reloaded).isNotNull() + assertThat(reloaded!!.status).isEqualTo(HeadlessJobStatus.FAILED) + mgr.destroy() + } + + @Test + fun `afterPropertiesSet skips sidecar when output file is missing`( + @TempDir tmp: Path, + ) { + // Sidecar with no corresponding output file — should be silently ignored. + val jobId = "deadbeef" + val sidecar = tmp.resolve("headless-$jobId.json") + val fakeJob = + HeadlessJob( + id = jobId, + kind = AgentKind.CODEX, + status = HeadlessJobStatus.COMPLETED, + outputFile = tmp.resolve("headless-$jobId-missing.jsonl"), + createdAt = java.time.Instant.now(), + ) + HeadlessJobSidecar.write(fakeJob, sidecar) + + val mgr = manager(tmp) { _, _, _ -> processOf(exitCode = 0, output = "") } + assertThat(mgr.get(jobId)).isNull() + mgr.destroy() + } + + // endregion + /** * Build a fake [Process] that exits immediately with [exitCode] and * writes [output] to its stdout stream. @@ -433,3 +597,5 @@ class HeadlessJobManagerTest { } } } + +private const val KB_AUTO_MCP_DISABLED_KEY = HeadlessJobManager.KB_AUTO_MCP_DISABLED_KEY diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecarTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecarTest.kt new file mode 100644 index 0000000..dac286d --- /dev/null +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecarTest.kt @@ -0,0 +1,147 @@ +package com.jorisjonkers.personalstack.agentgateway.headless + +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 + +class HeadlessJobSidecarTest { + @Test + fun `write and read round-trips a completed job`( + @TempDir tmp: Path, + ) { + val createdAt = Instant.parse("2025-01-15T10:00:00Z") + val completedAt = Instant.parse("2025-01-15T10:05:00Z") + val outputFile = tmp.resolve("headless-abc12345.jsonl") + val job = + HeadlessJob( + id = "abc12345", + kind = AgentKind.CLAUDE, + status = HeadlessJobStatus.COMPLETED, + outputFile = outputFile, + exitCode = 0, + createdAt = createdAt, + completedAt = completedAt, + ) + val sidecar = tmp.resolve("headless-abc12345.json") + + HeadlessJobSidecar.write(job, sidecar) + val reloaded = HeadlessJobSidecar.read(sidecar) + + assertThat(reloaded.id).isEqualTo("abc12345") + assertThat(reloaded.kind).isEqualTo(AgentKind.CLAUDE) + assertThat(reloaded.status).isEqualTo(HeadlessJobStatus.COMPLETED) + assertThat(reloaded.outputFile).isEqualTo(outputFile) + assertThat(reloaded.exitCode).isEqualTo(0) + assertThat(reloaded.createdAt).isEqualTo(createdAt) + assertThat(reloaded.completedAt).isEqualTo(completedAt) + } + + @Test + fun `write and read round-trips a failed job with non-zero exit code`( + @TempDir tmp: Path, + ) { + val createdAt = Instant.parse("2025-03-01T08:30:00Z") + val completedAt = Instant.parse("2025-03-01T08:35:00Z") + val outputFile = tmp.resolve("headless-fail9876.jsonl") + val job = + HeadlessJob( + id = "fail9876", + kind = AgentKind.CODEX, + status = HeadlessJobStatus.FAILED, + outputFile = outputFile, + exitCode = 1, + createdAt = createdAt, + completedAt = completedAt, + ) + val sidecar = tmp.resolve("headless-fail9876.json") + + HeadlessJobSidecar.write(job, sidecar) + val reloaded = HeadlessJobSidecar.read(sidecar) + + assertThat(reloaded.status).isEqualTo(HeadlessJobStatus.FAILED) + assertThat(reloaded.exitCode).isEqualTo(1) + assertThat(reloaded.kind).isEqualTo(AgentKind.CODEX) + } + + @Test + fun `write and read round-trips a running job with null completedAt and exitCode`( + @TempDir tmp: Path, + ) { + val createdAt = Instant.parse("2025-06-01T12:00:00Z") + val outputFile = tmp.resolve("headless-run00001.jsonl") + val job = + HeadlessJob( + id = "run00001", + kind = AgentKind.SHELL, + status = HeadlessJobStatus.RUNNING, + outputFile = outputFile, + exitCode = null, + createdAt = createdAt, + completedAt = null, + ) + val sidecar = tmp.resolve("headless-run00001.json") + + HeadlessJobSidecar.write(job, sidecar) + val reloaded = HeadlessJobSidecar.read(sidecar) + + assertThat(reloaded.status).isEqualTo(HeadlessJobStatus.RUNNING) + assertThat(reloaded.exitCode).isNull() + assertThat(reloaded.completedAt).isNull() + assertThat(reloaded.kind).isEqualTo(AgentKind.SHELL) + } + + @Test + fun `write is idempotent — second write overwrites the first`( + @TempDir tmp: Path, + ) { + val outputFile = tmp.resolve("headless-idem1234.jsonl") + val job = + HeadlessJob( + id = "idem1234", + kind = AgentKind.CLAUDE, + status = HeadlessJobStatus.RUNNING, + outputFile = outputFile, + createdAt = Instant.now(), + ) + val sidecar = tmp.resolve("headless-idem1234.json") + + HeadlessJobSidecar.write(job, sidecar) + val updated = job.copy(status = HeadlessJobStatus.COMPLETED, exitCode = 0, completedAt = Instant.now()) + HeadlessJobSidecar.write(updated, sidecar) + + val reloaded = HeadlessJobSidecar.read(sidecar) + assertThat(reloaded.status).isEqualTo(HeadlessJobStatus.COMPLETED) + assertThat(reloaded.exitCode).isEqualTo(0) + } + + @Test + fun `written JSON contains expected fields`( + @TempDir tmp: Path, + ) { + val outputFile = tmp.resolve("headless-fmtcheck.jsonl") + val job = + HeadlessJob( + id = "fmtcheck", + kind = AgentKind.CLAUDE, + status = HeadlessJobStatus.COMPLETED, + outputFile = outputFile, + exitCode = 0, + createdAt = Instant.parse("2025-01-01T00:00:00Z"), + completedAt = Instant.parse("2025-01-01T00:01:00Z"), + ) + val sidecar = tmp.resolve("headless-fmtcheck.json") + + HeadlessJobSidecar.write(job, sidecar) + val content = sidecar.toFile().readText() + + assertThat(content).contains("\"id\":\"fmtcheck\"") + assertThat(content).contains("\"kind\":\"CLAUDE\"") + assertThat(content).contains("\"status\":\"COMPLETED\"") + assertThat(content).contains("\"exitCode\":0") + assertThat(content).contains("\"createdAt\"") + assertThat(content).contains("\"completedAt\"") + } +} From bfd42185df69a663e2a56346ea1de4ed6bab8901 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 06:54:39 +0000 Subject: [PATCH 2/7] fix(agent-gateway): resolve ktlint violations and test race in headless fixes Fix ktlint style violations in HeadlessJobManager (chain newline) and HeadlessJobSidecar (multiline when branches, buildString newline) flagged by the CI lint pass. Restructure HeadlessJobSidecar.readString to avoid the ComplexMethod suppression by extracting a unescape helper. Fix a timing race in HeadlessJobManagerTest: waitUntil(status==FAILED) could see the status update before the SPAWN/FAILURE telemetry event was recorded. Add a secondary waitUntil for the telemetry event before asserting. --- .../headless/HeadlessJobManager.kt | 3 +- .../headless/HeadlessJobSidecar.kt | 118 ++++++++++-------- .../headless/HeadlessJobManagerTest.kt | 9 +- 3 files changed, 74 insertions(+), 56 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 750e0ca..190062a 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 @@ -67,7 +67,8 @@ class HeadlessJobManager( val stateDir = Path.of(props.tmux.stateDir) if (!Files.isDirectory(stateDir)) return runCatching { - Files.list(stateDir) + Files + .list(stateDir) .filter { it.fileName.toString().matches(Regex("headless-[a-f0-9]+\\.json")) } .forEach { sidecar -> reloadSidecar(sidecar) } }.onFailure { ex -> 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 9f69ff4..292fc4c 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 @@ -26,37 +26,38 @@ internal object HeadlessJobSidecar { job: HeadlessJob, sidecar: Path, ) { - val json = buildString { - append('{') - appendStringField("id", job.id) - append(',') - appendStringField("kind", job.kind.name) - append(',') - appendStringField("status", job.status.name) - append(',') - appendStringField("outputFile", job.outputFile.toString()) - append(',') - appendStringField("createdAt", job.createdAt.toString()) - append(',') - if (job.completedAt != null) { - appendStringField("completedAt", job.completedAt.toString()) + val json = + buildString { + append('{') + appendStringField("id", job.id) append(',') + appendStringField("kind", job.kind.name) + append(',') + appendStringField("status", job.status.name) + append(',') + appendStringField("outputFile", job.outputFile.toString()) + append(',') + appendStringField("createdAt", job.createdAt.toString()) + append(',') + if (job.completedAt != null) { + appendStringField("completedAt", job.completedAt.toString()) + append(',') + } + if (job.exitCode != null) { + append('"') + append("exitCode") + append('"') + append(':') + append(job.exitCode) + } else { + append('"') + append("exitCode") + append('"') + append(':') + append("null") + } + append('}') } - if (job.exitCode != null) { - append('"') - append("exitCode") - append('"') - append(':') - append(job.exitCode) - } else { - append('"') - append("exitCode") - append('"') - append(':') - append("null") - } - append('}') - } Files.writeString( sidecar, json, @@ -110,28 +111,33 @@ internal object HeadlessJobSidecar { // skip colon while (pos < body.length && (body[pos] == ':' || body[pos].isWhitespace())) pos++ // read value — either a quoted string, null, or an integer - val value: String? - val advance: Int - if (pos < body.length && body[pos] == '"') { - val str = readString(body, pos) - value = str?.first - advance = str?.second ?: 0 - } else if (body.startsWith("null", pos)) { - value = null - advance = 4 - } else { - // integer - val start = pos - while (pos < body.length && (body[pos].isDigit() || body[pos] == '-')) pos++ - value = body.substring(start, pos) - advance = 0 - } + val (value, advance) = parseValue(body, pos) result[key.first] = value - if (advance > 0) pos += advance + pos += advance } return result } + private fun parseValue( + body: String, + pos: Int, + ): Pair { + if (pos >= body.length) return null to 0 + return when { + body[pos] == '"' -> { + val str = readString(body, pos) + str?.first to (str?.second ?: 0) + } + body.startsWith("null", pos) -> null to 4 + else -> { + // integer — advance until non-digit + var end = pos + while (end < body.length && (body[end].isDigit() || body[end] == '-')) end++ + body.substring(pos, end) to (end - pos) + } + } + } + /** * Reads a JSON-quoted string at position [start] in [text]. * Returns the unquoted value and the number of characters consumed, @@ -146,14 +152,8 @@ internal object HeadlessJobSidecar { val sb = StringBuilder() while (pos < text.length && text[pos] != '"') { if (text[pos] == '\\' && pos + 1 < text.length) { - when (text[pos + 1]) { - '"' -> { sb.append('"'); pos += 2 } - '\\' -> { sb.append('\\'); pos += 2 } - 'n' -> { sb.append('\n'); pos += 2 } - 'r' -> { sb.append('\r'); pos += 2 } - 't' -> { sb.append('\t'); pos += 2 } - else -> { sb.append(text[pos + 1]); pos += 2 } - } + sb.append(unescape(text[pos + 1])) + pos += 2 } else { sb.append(text[pos]) pos++ @@ -162,6 +162,16 @@ internal object HeadlessJobSidecar { if (pos < text.length) pos++ // consume closing '"' return sb.toString() to (pos - start) } + + private fun unescape(ch: Char): Char = + when (ch) { + '"' -> '"' + '\\' -> '\\' + 'n' -> '\n' + 'r' -> '\r' + 't' -> '\t' + else -> ch + } } private fun StringBuilder.appendStringField( diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt index ad42e59..7ab9e86 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt @@ -206,6 +206,11 @@ class HeadlessJobManagerTest { val job = mgr.launch(AgentKind.SHELL, "secret shell prompt") waitUntil { mgr.get(job.id)?.status == HeadlessJobStatus.FAILED } + waitUntil { + telemetry.operations.any { + it.operation == GatewayOperationLabel.SPAWN && it.outcome == GatewayOutcomeLabel.FAILURE + } + } assertThat(telemetry.operations) .anySatisfy { @@ -433,7 +438,9 @@ class HeadlessJobManagerTest { val check = ProcessBuilder(listOf("/bin/sh", "-c", "echo VAL=\$$KB_AUTO_MCP_DISABLED_KEY")) .redirectErrorStream(true) - .also { pb -> pb.environment()[KB_AUTO_MCP_DISABLED_KEY] = HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE } + .also { pb -> + pb.environment()[KB_AUTO_MCP_DISABLED_KEY] = HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE + } .start() val checkOut = check.inputStream.bufferedReader().use { it.readText() } check.waitFor(5, TimeUnit.SECONDS) From 0ccd7f76bdd6470edf21333c891a76e1e5c4238d Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 07:05:39 +0000 Subject: [PATCH 3/7] fix(agent-gateway): fix newline-before-dot ktlint violation in test --- .../agentgateway/headless/HeadlessJobManagerTest.kt | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt index 7ab9e86..706a527 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt @@ -440,8 +440,7 @@ class HeadlessJobManagerTest { .redirectErrorStream(true) .also { pb -> pb.environment()[KB_AUTO_MCP_DISABLED_KEY] = HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE - } - .start() + }.start() val checkOut = check.inputStream.bufferedReader().use { it.readText() } check.waitFor(5, TimeUnit.SECONDS) assertThat(checkOut.trim()).isEqualTo("VAL=${HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE}") From c55490e662a5567a5550d07c11bb12698c6114a6 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 07:17:08 +0000 Subject: [PATCH 4/7] fix(agent-gateway): resolve detekt violations in HeadlessJobSidecar --- .../headless/HeadlessJobSidecar.kt | 25 +++++++++++++------ 1 file changed, 18 insertions(+), 7 deletions(-) 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 292fc4c..2f18af6 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 @@ -22,6 +22,8 @@ import java.time.Instant * produces clean output. */ internal object HeadlessJobSidecar { + private const val NULL_LITERAL = "null" + fun write( job: HeadlessJob, sidecar: Path, @@ -101,12 +103,8 @@ internal object HeadlessJobSidecar { // Strip outer braces val body = text.trim().removePrefix("{").removeSuffix("}") var pos = 0 - while (pos < body.length) { - // skip whitespace and commas - while (pos < body.length && (body[pos] == ',' || body[pos].isWhitespace())) pos++ - if (pos >= body.length) break - // read key - val key = readString(body, pos) ?: break + var key = nextKey(body, pos) + while (key != null) { pos += key.second // skip colon while (pos < body.length && (body[pos] == ':' || body[pos].isWhitespace())) pos++ @@ -114,10 +112,23 @@ internal object HeadlessJobSidecar { val (value, advance) = parseValue(body, pos) result[key.first] = value pos += advance + key = nextKey(body, pos) } return result } + private fun nextKey( + body: String, + fromPos: Int, + ): Pair? { + var pos = fromPos + // skip whitespace and commas + while (pos < body.length && (body[pos] == ',' || body[pos].isWhitespace())) pos++ + if (pos >= body.length) return null + val key = readString(body, pos) ?: return null + return key.first to (key.second + (pos - fromPos)) + } + private fun parseValue( body: String, pos: Int, @@ -128,7 +139,7 @@ internal object HeadlessJobSidecar { val str = readString(body, pos) str?.first to (str?.second ?: 0) } - body.startsWith("null", pos) -> null to 4 + body.startsWith(NULL_LITERAL, pos) -> null to NULL_LITERAL.length else -> { // integer — advance until non-digit var end = pos From c6f516637f102e1b3325062d14d0fb458cb81dca Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 07:28:08 +0000 Subject: [PATCH 5/7] fix(agent-gateway): fix ReturnCount detekt violation in HeadlessJobSidecar --- .../agentgateway/headless/HeadlessJobSidecar.kt | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 2f18af6..3c3df56 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 @@ -124,9 +124,8 @@ internal object HeadlessJobSidecar { var pos = fromPos // skip whitespace and commas while (pos < body.length && (body[pos] == ',' || body[pos].isWhitespace())) pos++ - if (pos >= body.length) return null - val key = readString(body, pos) ?: return null - return key.first to (key.second + (pos - fromPos)) + val key = if (pos < body.length) readString(body, pos) else null + return key?.let { it.first to (it.second + (pos - fromPos)) } } private fun parseValue( From 20a614922e42be2043f3e465161cfc07c6676b1f Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 07:35:48 +0000 Subject: [PATCH 6/7] fix(agent-gateway): apply cross-review findings on headless durability Close the Files.list directory stream with use{} to avoid leaking a file descriptor on every startup reload. Persist the recovered FAILED sidecar back to disk so a RUNNING->FAILED recovery converges instead of being re-applied with a fresh completedAt on every restart. Widen the sidecar filename pattern to accept uppercase hex. Replace the DefaultProcessFactory KB test with assertions that actually launch through the factory and verify the injected (or absent) variable in the child process environment. --- .../headless/HeadlessJobManager.kt | 15 +++++-- .../headless/HeadlessJobManagerTest.kt | 41 ++++++++++--------- 2 files changed, 33 insertions(+), 23 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 190062a..5293059 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 @@ -67,10 +67,11 @@ class HeadlessJobManager( val stateDir = Path.of(props.tmux.stateDir) if (!Files.isDirectory(stateDir)) return runCatching { - Files - .list(stateDir) - .filter { it.fileName.toString().matches(Regex("headless-[a-f0-9]+\\.json")) } - .forEach { sidecar -> reloadSidecar(sidecar) } + 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) } @@ -92,6 +93,11 @@ class HeadlessJobManager( // 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) @@ -476,6 +482,7 @@ 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 diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt index 706a527..c6ce27d 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobManagerTest.kt @@ -420,30 +420,35 @@ class HeadlessJobManagerTest { fun `DefaultProcessFactory injects KB_AUTO_MCP_DISABLED when enableKbHooks is false`( @TempDir tmp: Path, ) { - // Use a real process that prints its environment so we can inspect it. val process = HeadlessJobManager.DefaultProcessFactory.start( - command = listOf("/bin/sh", "-c", "echo KB=${KB_AUTO_MCP_DISABLED_KEY}"), + command = listOf("/bin/sh", "-c", "echo VAL=\$${HeadlessJobManager.KB_AUTO_MCP_DISABLED_KEY}"), cwd = tmp.toFile(), enableKbHooks = false, ) val output = process.inputStream.bufferedReader().use { it.readText() } process.waitFor(5, TimeUnit.SECONDS) - // The var is in the env — the echo just checks the process starts. - // More importantly: the process env should contain the key. We verify - // that via the factory contract: flag=false → key is set. - assertThat(output).doesNotContain("unbound variable") - // Also verify that default process has the var in its environment via - // a shell that can access it: - val check = - ProcessBuilder(listOf("/bin/sh", "-c", "echo VAL=\$$KB_AUTO_MCP_DISABLED_KEY")) - .redirectErrorStream(true) - .also { pb -> - pb.environment()[KB_AUTO_MCP_DISABLED_KEY] = HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE - }.start() - val checkOut = check.inputStream.bufferedReader().use { it.readText() } - check.waitFor(5, TimeUnit.SECONDS) - assertThat(checkOut.trim()).isEqualTo("VAL=${HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE}") + + assertThat(output.trim()).isEqualTo("VAL=${HeadlessJobManager.KB_AUTO_MCP_DISABLED_VALUE}") + } + + @Test + fun `DefaultProcessFactory leaves KB_AUTO_MCP_DISABLED unset when enableKbHooks is true`( + @TempDir tmp: Path, + ) { + val process = + HeadlessJobManager.DefaultProcessFactory.start( + command = listOf("/bin/sh", "-c", "echo VAL=\$${HeadlessJobManager.KB_AUTO_MCP_DISABLED_KEY}"), + cwd = tmp.toFile(), + enableKbHooks = true, + ) + val output = process.inputStream.bufferedReader().use { it.readText() } + process.waitFor(5, TimeUnit.SECONDS) + + // With hooks enabled the factory must not inject the switch; the child + // only sees whatever the test JVM itself inherited (usually nothing). + val inherited = System.getenv(HeadlessJobManager.KB_AUTO_MCP_DISABLED_KEY).orEmpty() + assertThat(output.trim()).isEqualTo("VAL=$inherited") } // endregion @@ -603,5 +608,3 @@ class HeadlessJobManagerTest { } } } - -private const val KB_AUTO_MCP_DISABLED_KEY = HeadlessJobManager.KB_AUTO_MCP_DISABLED_KEY From 154f66f6d93b91a19ea4ef8c80848180fa311fcb Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 07:40:07 +0000 Subject: [PATCH 7/7] fix(agent-gateway): await executor termination in destroy to avoid sidecar write race shutdownNow interrupts in-flight job threads whose interrupt path persists a final sidecar state file. Without awaiting termination the write can race external cleanup of the state directory (observed as a JUnit TempDir deletion failure in CI). destroy now waits up to five seconds for job threads to finish their final transition. --- .../agentgateway/headless/HeadlessJobManager.kt | 6 ++++++ 1 file changed, 6 insertions(+) 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 5293059..a252d33 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 @@ -207,6 +207,11 @@ class HeadlessJobManager( 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) { @@ -485,6 +490,7 @@ class HeadlessJobManager( 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 ->