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..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 @@ -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,55 @@ 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).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) + } + } + fun launch( kind: AgentKind, prompt: String, @@ -85,6 +137,7 @@ class HeadlessJobManager( createdAt = Instant.now(), ) jobs[id] = job + persistSidecar(job, stateDir) recordJobCounts() recordHeadlessJobEvent( kind = request.kind, @@ -101,10 +154,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 } @@ -153,11 +207,16 @@ 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) { 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 +257,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 +339,7 @@ class HeadlessJobManager( ?.also { updated = it } ?: job.also { updated = it } } + updated?.let { persistSidecar(it) } recordJobCounts() return updated } @@ -291,10 +352,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 +476,32 @@ 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 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 -> + 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 +513,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 +529,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..3c3df56 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/headless/HeadlessJobSidecar.kt @@ -0,0 +1,208 @@ +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 { + private const val NULL_LITERAL = "null" + + 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 + var key = nextKey(body, pos) + while (key != null) { + 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, 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++ + val key = if (pos < body.length) readString(body, pos) else null + return key?.let { it.first to (it.second + (pos - fromPos)) } + } + + 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_LITERAL, pos) -> null to NULL_LITERAL.length + 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, + * 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) { + sb.append(unescape(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 unescape(ch: Char): Char = + when (ch) { + '"' -> '"' + '\\' -> '\\' + 'n' -> '\n' + 'r' -> '\r' + 't' -> '\t' + else -> ch + } +} + +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..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 @@ -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,11 +201,16 @@ 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") waitUntil { mgr.get(job.id)?.status == HeadlessJobStatus.FAILED } + waitUntil { + telemetry.operations.any { + it.operation == GatewayOperationLabel.SPAWN && it.outcome == GatewayOutcomeLabel.FAILURE + } + } assertThat(telemetry.operations) .anySatisfy { @@ -231,7 +236,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 +267,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 +311,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 +340,7 @@ class HeadlessJobManagerTest { ) { val started = CountDownLatch(1) val mgr = - manager(tmp) { _, _ -> + manager(tmp) { _, _, _ -> sleepingProcess().also { started.countDown() } } @@ -355,7 +360,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 +374,176 @@ 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, + ) { + val process = + HeadlessJobManager.DefaultProcessFactory.start( + 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) + + 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 + + // 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. 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\"") + } +}