From 6606ae441aee9f10c7124c248fd9305366ced05c Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:27:44 +0000 Subject: [PATCH 1/6] refactor: carve TranscriptStore into lease, metadata, segment and stats stores TranscriptStore held 39 functions behind a documented TooManyFunctions suppression. Lease, metadata, segment IO, continuation, and storage-stats stores split out over a shared context that keeps the lock namespace and atomic properties writes; the suppression is gone and the new stores are unit-tested. --- .../tmux/TranscriptContinuationStore.kt | 83 +++ .../agentgateway/tmux/TranscriptLeaseStore.kt | 92 +++ .../tmux/TranscriptMetadataStore.kt | 134 ++++ .../tmux/TranscriptSegmentStore.kt | 161 +++++ .../tmux/TranscriptStorageStatsStore.kt | 107 +++ .../agentgateway/tmux/TranscriptStore.kt | 664 ++---------------- .../tmux/TranscriptStoreContext.kt | 52 ++ .../agentgateway/tmux/TranscriptStorePaths.kt | 48 ++ .../tmux/TranscriptStoreCollaboratorsTest.kt | 115 +++ 9 files changed, 856 insertions(+), 600 deletions(-) create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptContinuationStore.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptSegmentStore.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorageStatsStore.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreContext.kt create mode 100644 services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorePaths.kt create mode 100644 services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreCollaboratorsTest.kt diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptContinuationStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptContinuationStore.kt new file mode 100644 index 0000000..c117e2e --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptContinuationStore.kt @@ -0,0 +1,83 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +internal class TranscriptContinuationStore( + private val context: TranscriptStoreContext, + private val metadataStore: TranscriptMetadataStore, + private val segmentStore: TranscriptSegmentStore, +) { + fun appendDelimiter( + stableSessionId: String, + epoch: Long, + continuation: AgentContinuation?, + ): TranscriptMetadata { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + val metadata = metadataStore.recover(id) + if (epoch in metadata.delimiterEpochs) return@synchronized metadata + + segmentStore.writeContinuationMarker(id, marker(epoch, continuation)) + metadataStore + .recover(id) + .copy(delimiterEpochs = metadata.delimiterEpochs + epoch, sealed = false) + .also { metadataStore.write(id, it) } + } + } + + private fun marker( + epoch: Long, + continuation: AgentContinuation?, + ): ByteArray = + buildString { + append("\r\n") + append("[agent-gateway continuation epoch=") + append(epoch) + append(' ') + append(continuation.summary()) + continuation?.previousEpoch?.let { append(" previousEpoch=").append(it) } + continuation?.reason?.takeIf { it.isNotBlank() }?.let { + append(" reason=") + append(redactSecrets(it.replace(Regex("\\s+"), " ")).take(MAX_REASON_CHARS)) + } + append("]\r\n") + }.toByteArray(Charsets.UTF_8) + + private fun AgentContinuation?.summary(): String { + val reason = this?.reason?.trim()?.lowercase() + val from = this?.fromSetupLabel.redactedLabel() + val to = this?.toSetupLabel.redactedLabel() + if (from != null || to != null) { + return buildString { + append("setup transition") + from?.let { append(" fromSetup=").append(it) } + to?.let { append(" toSetup=").append(it) } + } + } + return when (reason) { + "rebind" -> "agent rebound" + else -> "agent restarted" + } + } + + private fun String?.redactedLabel(): String? = + this + ?.replace(Regex("\\s+"), " ") + ?.trim() + ?.takeIf { it.isNotBlank() } + ?.let(::redactSecrets) + ?.take(MAX_SETUP_LABEL_CHARS) + ?.let { "\"${it.replace("\\", "\\\\").replace("\"", "\\\"")}\"" } + + private fun redactSecrets(value: String): String = + SECRET_ASSIGNMENT.replace(SECRET_TOKEN.replace(value, "[redacted]")) { + "${it.groupValues[1]}=[redacted]" + } + + companion object { + private const val MAX_REASON_CHARS = 160 + private const val MAX_SETUP_LABEL_CHARS = 160 + private val SECRET_TOKEN = + Regex("""(?i)\b(?:sk-[A-Za-z0-9_-]{12,}|gh[pousr]_[A-Za-z0-9_]{12,}|github_pat_[A-Za-z0-9_]{12,})""") + private val SECRET_ASSIGNMENT = + Regex("""(?i)\b(password|passwd|secret|token|api[_-]?key|bearer|credential)=\S+""") + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt new file mode 100644 index 0000000..be138f8 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt @@ -0,0 +1,92 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import java.nio.file.Files +import java.nio.file.Path +import java.util.Properties +import java.util.UUID + +internal class TranscriptLeaseStore( + private val context: TranscriptStoreContext, +) { + fun acquire( + stableSessionId: String, + owner: String, + epoch: Long, + ): TranscriptLease { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + require(epoch > 0) { "epoch must be positive" } + Files.createDirectories(context.paths.sessionDir(id)) + val leaseFile = context.paths.leaseFile(id) + val now = context.clock.millis() + val existing = read(leaseFile) + if (existing != null && existing.expiresAtMillis > now && existing.owner != owner) { + error("transcript $id is leased by ${existing.owner}") + } + write( + TranscriptLease( + stableSessionId = id, + owner = owner, + token = UUID.randomUUID().toString(), + epoch = epoch, + expiresAtMillis = now + context.props.transcripts.leaseTtlSeconds * MILLIS_PER_SECOND, + ), + ) + } + } + + fun release(lease: TranscriptLease) { + val id = context.validateStableSessionId(lease.stableSessionId) + synchronized(context.lockFor(id)) { + val existing = read(context.paths.leaseFile(id)) ?: return + if (existing.token == lease.token) Files.deleteIfExists(context.paths.leaseFile(id)) + } + } + + fun renew(lease: TranscriptLease): TranscriptLease? { + val id = context.validateStableSessionId(lease.stableSessionId) + return synchronized(context.lockFor(id)) { + read(context.paths.leaseFile(id)) + ?.takeIf { it.token == lease.token } + ?.let { current -> + write( + current.copy( + expiresAtMillis = context.clock.millis() + + context.props.transcripts.leaseTtlSeconds * MILLIS_PER_SECOND, + ), + ) + } + } + } + + fun read(file: Path): TranscriptLease? { + if (!Files.exists(file)) return null + val properties = Properties() + Files.newInputStream(file).use(properties::load) + return TranscriptLease( + stableSessionId = properties.getProperty("stableSessionId"), + owner = properties.getProperty("owner"), + token = properties.getProperty("token"), + epoch = properties.getProperty("epoch", "1").toLong(), + expiresAtMillis = properties.getProperty("expiresAtMillis", "0").toLong(), + ) + } + + private fun write(lease: TranscriptLease): TranscriptLease { + context.writePropertiesAtomic( + context.paths.leaseFile(lease.stableSessionId), + Properties().apply { + this["stableSessionId"] = lease.stableSessionId + this["owner"] = lease.owner + this["token"] = lease.token + this["epoch"] = lease.epoch.toString() + this["expiresAtMillis"] = lease.expiresAtMillis.toString() + }, + ) + return lease + } + + companion object { + private const val MILLIS_PER_SECOND = 1_000L + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt new file mode 100644 index 0000000..993ff4c --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt @@ -0,0 +1,134 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import java.nio.file.Files +import java.time.Instant +import java.util.Comparator +import java.util.Properties + +internal class TranscriptMetadataStore( + private val context: TranscriptStoreContext, +) { + fun open( + stableSessionId: String, + epoch: Long, + ): TranscriptMetadata { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + require(epoch > 0) { "epoch must be positive" } + Files.createDirectories(context.paths.segmentsDir(id)) + recover(id).copy(epoch = epoch, sealed = false).also { write(id, it) } + } + } + + fun recover(stableSessionId: String): TranscriptMetadata { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + Files.createDirectories(context.paths.segmentsDir(id)) + val existing = read(id) + val segments = context.paths.segmentFiles(id) + if (segments.isEmpty()) { + val metadata = + (existing ?: TranscriptMetadata(stableSessionId = id)).copy( + logicalEnd = existing?.logicalStart ?: 0L, + activeSegment = 0, + ) + Files.createFile(context.paths.segmentPath(id, 0)) + write(id, metadata) + return@synchronized metadata + } + recoverFromSegments(id, existing, segments) + } + } + + fun seal(stableSessionId: String): TranscriptMetadata { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + recover(id) + .copy(sealed = true, updatedAt = Instant.now(context.clock)) + .also { write(id, it) } + } + } + + fun cleanup( + stableSessionId: String, + leaseStore: TranscriptLeaseStore, + ): Boolean { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + val dir = context.paths.sessionDir(id) + if (!Files.exists(dir)) return@synchronized false + val metadata = read(id) + val lease = leaseStore.read(context.paths.leaseFile(id)) + if (lease != null && lease.expiresAtMillis > context.clock.millis()) return@synchronized false + if (metadata != null && !metadata.sealed) return@synchronized false + val cutoff = Instant.now(context.clock).minusSeconds(context.props.transcripts.retentionSeconds) + if (metadata != null && metadata.updatedAt.isAfter(cutoff)) return@synchronized false + Files.walk(dir).use { paths -> + paths.sorted(Comparator.reverseOrder()).forEach { Files.deleteIfExists(it) } + } + true + } + } + + fun read(stableSessionId: String): TranscriptMetadata? { + val file = context.paths.metadataFile(stableSessionId) + if (!Files.exists(file)) return null + val properties = Properties() + Files.newInputStream(file).use(properties::load) + return TranscriptMetadata( + stableSessionId = properties.getProperty("stableSessionId", stableSessionId), + epoch = properties.getProperty("epoch", "1").toLong(), + logicalStart = properties.getProperty("logicalStart", "0").toLong(), + logicalEnd = properties.getProperty("logicalEnd", "0").toLong(), + activeSegment = properties.getProperty("activeSegment", "0").toInt(), + sealed = properties.getProperty("sealed", "false").toBoolean(), + delimiterEpochs = delimiterEpochs(properties), + updatedAt = Instant.ofEpochMilli(properties.getProperty("updatedAtMillis", "0").toLong()), + ) + } + + fun write( + stableSessionId: String, + metadata: TranscriptMetadata, + ) { + val now = Instant.now(context.clock) + context.writePropertiesAtomic( + context.paths.metadataFile(stableSessionId), + Properties().apply { + this["stableSessionId"] = metadata.stableSessionId + this["epoch"] = metadata.epoch.toString() + this["logicalStart"] = metadata.logicalStart.toString() + this["logicalEnd"] = metadata.logicalEnd.toString() + this["activeSegment"] = metadata.activeSegment.toString() + this["sealed"] = metadata.sealed.toString() + this["delimiterEpochs"] = metadata.delimiterEpochs.sorted().joinToString(",") + this["updatedAtMillis"] = now.toEpochMilli().toString() + }, + ) + } + + private fun recoverFromSegments( + stableSessionId: String, + existing: TranscriptMetadata?, + segments: List, + ): TranscriptMetadata { + val start = existing?.logicalStart ?: 0L + val end = start + segments.sumOf { Files.size(it) } + val active = context.paths.segmentIndex(segments.last()) + val recovered = + (existing ?: TranscriptMetadata(stableSessionId = stableSessionId)).copy( + stableSessionId = stableSessionId, + logicalEnd = end, + activeSegment = active, + ) + write(stableSessionId, recovered) + return recovered + } + + private fun delimiterEpochs(properties: Properties): Set = + properties + .getProperty("delimiterEpochs", "") + .split(',') + .mapNotNull { it.trim().takeIf(String::isNotBlank)?.toLong() } + .toSet() +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptSegmentStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptSegmentStore.kt new file mode 100644 index 0000000..1431104 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptSegmentStore.kt @@ -0,0 +1,161 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import java.io.InputStream +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption + +internal class TranscriptSegmentStore( + private val context: TranscriptStoreContext, + private val metadataStore: TranscriptMetadataStore, +) { + fun activeSegmentPath(stableSessionId: String): Path { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + val metadata = metadataStore.recover(id) + context.paths.segmentPath(id, metadata.activeSegment).also { + Files.createDirectories(it.parent) + if (!Files.exists(it)) Files.createFile(it) + } + } + } + + fun rotateIfNeeded(stableSessionId: String): TranscriptMetadata { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + var metadata = metadataStore.recover(id) + val active = context.paths.segmentPath(id, metadata.activeSegment) + if (Files.exists(active) && Files.size(active) >= context.props.transcripts.segmentBytes) { + metadata = metadata.copy(activeSegment = metadata.activeSegment + 1) + val next = context.paths.segmentPath(id, metadata.activeSegment) + if (!Files.exists(next)) Files.createFile(next) + metadataStore.write(id, metadata) + } + metadata + } + } + + fun trimIfNeeded(stableSessionId: String): TranscriptMetadata { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + var metadata = metadataStore.recover(id) + val cap = context.props.transcripts.capBytes + if (metadata.byteCount <= cap) return@synchronized metadata + + val segments = context.paths.segmentFiles(id).toMutableList() + while (segments.size > 1 && metadata.byteCount > cap) { + val victim = segments.removeAt(0) + val bytes = Files.size(victim) + Files.deleteIfExists(victim) + metadata = metadata.copy(logicalStart = metadata.logicalStart + bytes) + } + metadata = + metadata.copy( + logicalEnd = metadata.logicalStart + segments.sumOf { Files.size(it) }, + activeSegment = context.paths.segmentIndex(segments.last()), + ) + metadataStore.write(id, metadata) + metadata + } + } + + fun readRaw( + stableSessionId: String, + fromOffset: Long, + maxBytes: Int = DEFAULT_READ_BYTES, + ): TranscriptRawRead { + val id = context.validateStableSessionId(stableSessionId) + return synchronized(context.lockFor(id)) { + val metadata = metadataStore.recover(id) + val start = fromOffset.coerceAtLeast(metadata.logicalStart) + val bytes = + if (start >= metadata.logicalEnd) { + ByteArray(0) + } else { + readSegments(id, start, maxBytes, metadata.logicalStart) + } + TranscriptRawRead(startOffset = start, bytes = bytes, metadata = metadata) + } + } + + fun writeContinuationMarker( + stableSessionId: String, + marker: ByteArray, + ) { + Files.write(activeSegmentPath(stableSessionId), marker, StandardOpenOption.APPEND) + } + + private fun readSegments( + stableSessionId: String, + start: Long, + maxBytes: Int, + logicalStart: Long, + ): ByteArray { + val out = ByteArrayBuilder(maxBytes) + var segmentStart = logicalStart + val segments = context.paths.segmentFiles(stableSessionId) + var index = 0 + while (index < segments.size && out.size < maxBytes) { + val segment = segments[index] + val segmentSize = Files.size(segment) + val segmentEnd = segmentStart + segmentSize + if (segmentEnd > start) { + readFromSegment(segment, start + out.size - segmentStart, segmentSize, out, maxBytes) + } + segmentStart = segmentEnd + index++ + } + return out.toByteArray() + } + + private fun readFromSegment( + segment: Path, + offset: Long, + segmentSize: Long, + out: ByteArrayBuilder, + maxBytes: Int, + ) { + val offsetInSegment = offset.coerceAtLeast(0) + if (offsetInSegment >= segmentSize) return + Files.newInputStream(segment).use { input -> + input.skipFully(offsetInSegment) + out.readFrom(input, maxBytes - out.size) + } + } + + private fun InputStream.skipFully(bytes: Long) { + var remaining = bytes + while (remaining > 0) { + val skipped = skip(remaining) + if (skipped <= 0) break + remaining -= skipped + } + } + + private class ByteArrayBuilder( + capacity: Int, + ) { + private val bytes = ByteArray(capacity) + var size: Int = 0 + private set + + fun readFrom( + input: InputStream, + limit: Int, + ) { + var remaining = limit + while (remaining > 0) { + val read = input.read(bytes, size, remaining) + if (read <= 0) return + size += read + remaining -= read + } + } + + fun toByteArray(): ByteArray = bytes.copyOf(size) + } + + companion object { + private const val DEFAULT_READ_BYTES = 64 * 1024 + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorageStatsStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorageStatsStore.kt new file mode 100644 index 0000000..ab60700 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorageStatsStore.kt @@ -0,0 +1,107 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayModeLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStorageObjectLabel +import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStorageTelemetrySample +import java.nio.file.Files +import java.nio.file.LinkOption +import java.nio.file.Path +import java.time.Instant + +internal class TranscriptStorageStatsStore( + private val context: TranscriptStoreContext, +) { + @Volatile + private var cached = + TranscriptStorageStats( + usedBytes = 0, + capBytes = context.props.transcripts.capBytes, + refreshedAt = Instant.EPOCH, + ) + + fun current(): TranscriptStorageStats = cached + + fun refresh(): TranscriptStorageStats { + val stats = scan() + cached = stats + context.telemetry.recordStorage( + GatewayStorageTelemetrySample( + storageObject = GatewayStorageObjectLabel.TRANSCRIPT, + mode = GatewayModeLabel.DURABLE, + bytes = stats.usedBytes, + ), + ) + context.telemetry.recordStorageLimit( + GatewayStorageTelemetrySample( + storageObject = GatewayStorageObjectLabel.TRANSCRIPT, + mode = GatewayModeLabel.DURABLE, + bytes = stats.capBytes, + ), + ) + return stats + } + + private fun scan(): TranscriptStorageStats { + val transcriptRoot = context.root() + if (!Files.isDirectory(transcriptRoot, LinkOption.NOFOLLOW_LINKS)) { + return stats(0) + } + val usedBytes = + Files.list(transcriptRoot).use { paths -> + paths + .filter { TranscriptStorePaths.UUID_SESSION_DIR.matches(it.fileName.toString()) } + .filter { safeDirectoryInside(transcriptRoot, it) } + .mapToLong { sessionStorageBytes(transcriptRoot, it) } + .sum() + } + return stats(usedBytes) + } + + private fun sessionStorageBytes( + transcriptRoot: Path, + sessionDir: Path, + ): Long { + val segments = sessionDir.resolve("segments") + if (!safeDirectoryInside(transcriptRoot, segments)) return 0 + return runCatching { + Files.list(segments).use { paths -> + paths + .filter { TranscriptStorePaths.SEGMENT_FILE.matches(it.fileName.toString()) } + .filter { safeRegularFileInside(transcriptRoot, it) } + .mapToLong { runCatching { Files.size(it) }.getOrDefault(0L) } + .sum() + } + }.getOrDefault(0L) + } + + private fun stats(usedBytes: Long): TranscriptStorageStats = + TranscriptStorageStats( + usedBytes = usedBytes, + capBytes = context.props.transcripts.capBytes, + refreshedAt = Instant.now(context.clock), + ) + + private fun safeDirectoryInside( + transcriptRoot: Path, + path: Path, + ): Boolean = pathInside(transcriptRoot, path) && Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS) + + private fun safeRegularFileInside( + transcriptRoot: Path, + path: Path, + ): Boolean = pathInside(transcriptRoot, path) && Files.isRegularFile(path, LinkOption.NOFOLLOW_LINKS) + + private fun pathInside( + transcriptRoot: Path, + path: Path, + ): Boolean { + val normalizedRoot = transcriptRoot.toAbsolutePath().normalize() + val normalizedPath = path.toAbsolutePath().normalize() + if (!normalizedPath.startsWith(normalizedRoot)) return false + return runCatching { + path.toRealPath(LinkOption.NOFOLLOW_LINKS).startsWith( + transcriptRoot.toRealPath(LinkOption.NOFOLLOW_LINKS), + ) + }.getOrDefault(false) + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt index 6e1ba9c..9e49fe9 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt @@ -2,629 +2,93 @@ package com.jorisjonkers.personalstack.agentgateway.tmux import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayModeLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStorageObjectLabel -import com.jorisjonkers.personalstack.agentgateway.observability.GatewayStorageTelemetrySample import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Component -import java.io.InputStream -import java.nio.file.Files -import java.nio.file.LinkOption import java.nio.file.Path -import java.nio.file.StandardCopyOption -import java.nio.file.StandardOpenOption import java.time.Clock import java.time.Instant -import java.util.Comparator -import java.util.Locale -import java.util.Properties -import java.util.UUID -import kotlin.io.path.Path -import kotlin.streams.toList @Component -// Transcript metadata, leases, and segment paths share one lock namespace. -@Suppress("TooManyFunctions") class TranscriptStore + private constructor( + private val context: TranscriptStoreContext, + ) { @Autowired constructor( - private val props: GatewayProperties, - private val telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, - ) { - private var clock: Clock = Clock.systemUTC() - - @Volatile - private var cachedStorageStats = - TranscriptStorageStats( - usedBytes = 0, - capBytes = props.transcripts.capBytes, - refreshedAt = Instant.EPOCH, - ) - - internal constructor( - props: GatewayProperties, - clock: Clock, - telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, - ) : this(props, telemetry) { - this.clock = clock - } - - fun validateStableSessionId(value: String): String = UUID.fromString(value).toString() - - fun root(): Path { - val workspace = Path(props.workspaceRoot).toAbsolutePath().normalize() - val root = workspace.resolve(props.transcripts.dirName).normalize() - require(root.startsWith(workspace)) { "transcript directory must stay inside the workspace" } - Files.createDirectories(root) - return root - } - - fun open( - stableSessionId: String, - epoch: Long, - ): TranscriptMetadata { - val id = validateStableSessionId(stableSessionId) - val metadata = - synchronized(lockFor(id)) { - require(epoch > 0) { "epoch must be positive" } - Files.createDirectories(segmentsDir(id)) - val metadata = recoverMetadata(id).copy(epoch = epoch, sealed = false) - writeMetadata(id, metadata) - metadata - } - refreshStorageStats() - return metadata - } - - // Lease validation and atomic persistence are kept together under one lock. - fun acquireLease( - stableSessionId: String, - owner: String, - epoch: Long, - ): TranscriptLease { - val id = validateStableSessionId(stableSessionId) - return synchronized(lockFor(id)) { - require(epoch > 0) { "epoch must be positive" } - Files.createDirectories(sessionDir(id)) - val leaseFile = leaseFile(id) - val now = clock.millis() - val existing = readLease(leaseFile) - if (existing != null && existing.expiresAtMillis > now && existing.owner != owner) { - error("transcript $id is leased by ${existing.owner}") - } - val lease = - TranscriptLease( - stableSessionId = id, - owner = owner, - token = UUID.randomUUID().toString(), - epoch = epoch, - expiresAtMillis = now + props.transcripts.leaseTtlSeconds * MILLIS_PER_SECOND, - ) - writePropertiesAtomic( - leaseFile, - Properties().apply { - this["stableSessionId"] = lease.stableSessionId - this["owner"] = lease.owner - this["token"] = lease.token - this["epoch"] = lease.epoch.toString() - this["expiresAtMillis"] = lease.expiresAtMillis.toString() - }, - ) - lease - } - } - - fun releaseLease(lease: TranscriptLease) { - val id = validateStableSessionId(lease.stableSessionId) - synchronized(lockFor(id)) { - val existing = readLease(leaseFile(id)) ?: return - if (existing.token == lease.token) Files.deleteIfExists(leaseFile(id)) - } - } - - fun renewLease(lease: TranscriptLease): TranscriptLease? { - val id = validateStableSessionId(lease.stableSessionId) - return synchronized(lockFor(id)) { - readLease(leaseFile(id)) - ?.takeIf { it.token == lease.token } - ?.let(::renewExistingLease) - } - } - - private fun renewExistingLease(existing: TranscriptLease): TranscriptLease { - val renewed = - existing.copy( - expiresAtMillis = clock.millis() + props.transcripts.leaseTtlSeconds * MILLIS_PER_SECOND, - ) - writePropertiesAtomic( - leaseFile(renewed.stableSessionId), - Properties().apply { - this["stableSessionId"] = renewed.stableSessionId - this["owner"] = renewed.owner - this["token"] = renewed.token - this["epoch"] = renewed.epoch.toString() - this["expiresAtMillis"] = renewed.expiresAtMillis.toString() - }, - ) - return renewed - } - - fun activeSegmentPath(stableSessionId: String): Path { - val id = validateStableSessionId(stableSessionId) - return synchronized(lockFor(id)) { - val metadata = recoverMetadata(id) - segmentPath(id, metadata.activeSegment).also { - Files.createDirectories(it.parent) - if (!Files.exists(it)) Files.createFile(it) - } - } - } - - fun appendContinuationDelimiter( - stableSessionId: String, - epoch: Long, - continuation: AgentContinuation?, - ): TranscriptMetadata { - val id = validateStableSessionId(stableSessionId) - return synchronized(lockFor(id)) { - var metadata = recoverMetadata(id) - if (epoch in metadata.delimiterEpochs) return metadata - - val marker = - buildString { - append("\r\n") - append("[agent-gateway continuation epoch=") - append(epoch) - append(' ') - append(continuation.summary()) - continuation?.previousEpoch?.let { append(" previousEpoch=").append(it) } - continuation?.reason?.takeIf { it.isNotBlank() }?.let { - append(" reason=") - append(redactSecrets(it.replace(Regex("\\s+"), " ")).take(MAX_REASON_CHARS)) - } - append("]\r\n") - }.toByteArray(Charsets.UTF_8) - Files.write(activeSegmentPath(id), marker, StandardOpenOption.APPEND) - metadata = - recoverMetadata(id).copy( - delimiterEpochs = metadata.delimiterEpochs + epoch, - sealed = false, - ) - writeMetadata(id, metadata) - metadata - } - } - - fun recoverMetadata(stableSessionId: String): TranscriptMetadata { - val id = validateStableSessionId(stableSessionId) - return synchronized(lockFor(id)) { - Files.createDirectories(segmentsDir(id)) - val existing = readMetadata(id) - val segments = segmentFiles(id) - if (segments.isEmpty()) { - val metadata = - (existing ?: TranscriptMetadata(stableSessionId = id)).copy( - logicalEnd = existing?.logicalStart ?: 0L, - activeSegment = 0, - ) - Files.createFile(segmentPath(id, 0)) - writeMetadata(id, metadata) - return metadata - } - val start = existing?.logicalStart ?: 0L - val end = start + segments.sumOf { Files.size(it) } - val active = segmentIndex(segments.last()) - val recovered = - (existing ?: TranscriptMetadata(stableSessionId = id)).copy( - stableSessionId = id, - logicalEnd = end, - activeSegment = active, - ) - writeMetadata(id, recovered) - recovered - } - } - - fun rotateIfNeeded(stableSessionId: String): TranscriptMetadata { - val id = validateStableSessionId(stableSessionId) - val metadata = - synchronized(lockFor(id)) { - var metadata = recoverMetadata(id) - val active = segmentPath(id, metadata.activeSegment) - if (Files.exists(active) && Files.size(active) >= props.transcripts.segmentBytes) { - metadata = metadata.copy(activeSegment = metadata.activeSegment + 1) - val next = segmentPath(id, metadata.activeSegment) - if (!Files.exists(next)) Files.createFile(next) - writeMetadata(id, metadata) - } - metadata - } - refreshStorageStats() - return metadata - } - - fun trimIfNeeded(stableSessionId: String): TranscriptMetadata { - val id = validateStableSessionId(stableSessionId) - val metadata = - synchronized(lockFor(id)) { - var metadata = recoverMetadata(id) - val cap = props.transcripts.capBytes - if (metadata.byteCount <= cap) return@synchronized metadata - - val segments = segmentFiles(id).toMutableList() - while (segments.size > 1 && metadata.byteCount > cap) { - val victim = segments.removeAt(0) - val bytes = Files.size(victim) - Files.deleteIfExists(victim) - metadata = metadata.copy(logicalStart = metadata.logicalStart + bytes) - } - metadata = - metadata.copy( - logicalEnd = metadata.logicalStart + segments.sumOf { Files.size(it) }, - activeSegment = segmentIndex(segments.last()), - ) - writeMetadata(id, metadata) - metadata - } - refreshStorageStats() - return metadata - } - - fun seal(stableSessionId: String): TranscriptMetadata { - val id = validateStableSessionId(stableSessionId) - val metadata = - synchronized(lockFor(id)) { - val metadata = recoverMetadata(id).copy(sealed = true, updatedAt = Instant.now(clock)) - writeMetadata(id, metadata) - metadata - } - refreshStorageStats() - return metadata - } - - // Early exits map directly to cleanup safety gates. - fun cleanup(stableSessionId: String): Boolean { - val id = validateStableSessionId(stableSessionId) - val removed = - synchronized(lockFor(id)) { - val dir = sessionDir(id) - if (!Files.exists(dir)) return@synchronized false - val metadata = readMetadata(id) - val lease = readLease(leaseFile(id)) - if (lease != null && lease.expiresAtMillis > clock.millis()) return@synchronized false - if (metadata != null && !metadata.sealed) return@synchronized false - val cutoff = Instant.now(clock).minusSeconds(props.transcripts.retentionSeconds) - if (metadata != null && metadata.updatedAt.isAfter(cutoff)) return@synchronized false - Files.walk(dir).use { paths -> - paths.sorted(Comparator.reverseOrder()).forEach { Files.deleteIfExists(it) } - } - true - } - refreshStorageStats() - return removed - } - - fun storageStats(): TranscriptStorageStats = cachedStorageStats - - fun refreshStorageStats(): TranscriptStorageStats { - val stats = scanStorageStats() - cachedStorageStats = stats - telemetry.recordStorage( - GatewayStorageTelemetrySample( - storageObject = GatewayStorageObjectLabel.TRANSCRIPT, - mode = GatewayModeLabel.DURABLE, - bytes = stats.usedBytes, - ), - ) - telemetry.recordStorageLimit( - GatewayStorageTelemetrySample( - storageObject = GatewayStorageObjectLabel.TRANSCRIPT, - mode = GatewayModeLabel.DURABLE, - bytes = stats.capBytes, - ), - ) - return stats - } - - // The segment scan advances or stops at precise byte boundaries. - fun readRaw( - stableSessionId: String, - fromOffset: Long, - maxBytes: Int = DEFAULT_READ_BYTES, - ): TranscriptRawRead { - val id = validateStableSessionId(stableSessionId) - return synchronized(lockFor(id)) { - val metadata = recoverMetadata(id) - val start = fromOffset.coerceAtLeast(metadata.logicalStart) - val bytes = - if (start >= metadata.logicalEnd) { - ByteArray(0) - } else { - readSegments(id, start, maxBytes, metadata.logicalStart) - } - TranscriptRawRead(startOffset = start, bytes = bytes, metadata = metadata) - } - } - - private fun readSegments( - stableSessionId: String, - start: Long, - maxBytes: Int, - logicalStart: Long, - ): ByteArray { - val out = ByteArrayBuilder(maxBytes) - var segmentStart = logicalStart - val segments = segmentFiles(stableSessionId) - var index = 0 - while (index < segments.size && out.size < maxBytes) { - val segment = segments[index] - val segmentSize = Files.size(segment) - val segmentEnd = segmentStart + segmentSize - if (segmentEnd > start) { - val offsetInSegment = (start + out.size - segmentStart).coerceAtLeast(0) - if (offsetInSegment < segmentSize) { - Files.newInputStream(segment).use { input -> - input.skipFully(offsetInSegment) - out.readFrom(input, maxBytes - out.size) - } - } - } - segmentStart = segmentEnd - index++ - } - return out.toByteArray() - } - - private fun sessionDir(stableSessionId: String): Path = root().resolve(stableSessionId) - - private fun segmentsDir(stableSessionId: String): Path = sessionDir(stableSessionId).resolve("segments") - - private fun metadataFile(stableSessionId: String): Path { - val dir = sessionDir(stableSessionId) - return dir.resolve("metadata.properties") - } - - private fun leaseFile(stableSessionId: String): Path = sessionDir(stableSessionId).resolve("lease.properties") - - private fun segmentPath( - stableSessionId: String, - index: Int, - ): Path = segmentsDir(stableSessionId).resolve("segment-%06d.log".format(Locale.ROOT, index)) - - private fun segmentFiles(stableSessionId: String): List { - val dir = segmentsDir(stableSessionId) - if (!Files.exists(dir)) return emptyList() - return Files.list(dir).use { paths -> - paths - .filter { SEGMENT_FILE.matches(it.fileName.toString()) } - .sorted(Comparator.comparingInt { segmentIndex(it) }) - .toList() - } - } - - private fun segmentIndex(path: Path): Int = - SEGMENT_FILE - .matchEntire(path.fileName.toString()) - ?.groupValues - ?.get(1) - ?.toInt() - ?: error("invalid segment file: $path") - - private fun scanStorageStats(): TranscriptStorageStats { - val transcriptRoot = root() - if (!Files.isDirectory(transcriptRoot, LinkOption.NOFOLLOW_LINKS)) { - return TranscriptStorageStats( - usedBytes = 0, - capBytes = props.transcripts.capBytes, - refreshedAt = Instant.now(clock), - ) - } - val usedBytes = - Files.list(transcriptRoot).use { paths -> - paths - .filter { UUID_SESSION_DIR.matches(it.fileName.toString()) } - .filter { safeDirectoryInside(transcriptRoot, it) } - .mapToLong { sessionStorageBytes(transcriptRoot, it) } - .sum() - } - return TranscriptStorageStats( - usedBytes = usedBytes, - capBytes = props.transcripts.capBytes, - refreshedAt = Instant.now(clock), - ) - } - - private fun sessionStorageBytes( - transcriptRoot: Path, - sessionDir: Path, - ): Long { - val segments = sessionDir.resolve("segments") - if (!safeDirectoryInside(transcriptRoot, segments)) return 0 - return runCatching { - Files.list(segments).use { paths -> - paths - .filter { SEGMENT_FILE.matches(it.fileName.toString()) } - .filter { safeRegularFileInside(transcriptRoot, it) } - .mapToLong { runCatching { Files.size(it) }.getOrDefault(0L) } - .sum() - } - }.getOrDefault(0L) - } - - private fun safeDirectoryInside( - transcriptRoot: Path, - path: Path, - ): Boolean = pathInside(transcriptRoot, path) && Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS) - - private fun safeRegularFileInside( - transcriptRoot: Path, - path: Path, - ): Boolean = pathInside(transcriptRoot, path) && Files.isRegularFile(path, LinkOption.NOFOLLOW_LINKS) - - private fun pathInside( - transcriptRoot: Path, - path: Path, - ): Boolean { - val normalizedRoot = transcriptRoot.toAbsolutePath().normalize() - val normalizedPath = path.toAbsolutePath().normalize() - if (!normalizedPath.startsWith(normalizedRoot)) return false - return runCatching { - path.toRealPath(LinkOption.NOFOLLOW_LINKS).startsWith( - transcriptRoot.toRealPath(LinkOption.NOFOLLOW_LINKS), - ) - }.getOrDefault(false) - } - - private fun AgentContinuation?.summary(): String { - val reason = this?.reason?.trim()?.lowercase() - val from = this?.fromSetupLabel.redactedLabel() - val to = this?.toSetupLabel.redactedLabel() - if (from != null || to != null) { - return buildString { - append("setup transition") - from?.let { append(" fromSetup=").append(it) } - to?.let { append(" toSetup=").append(it) } - } - } - return when (reason) { - "rebind" -> "agent rebound" - else -> "agent restarted" - } - } + props: GatewayProperties, + telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, + ) : this(TranscriptStoreContext(props, Clock.systemUTC(), telemetry)) + + internal constructor( + props: GatewayProperties, + clock: Clock, + telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, + ) : this(TranscriptStoreContext(props, clock, telemetry)) + + private val leaseStore = TranscriptLeaseStore(context) + private val metadataStore = TranscriptMetadataStore(context) + private val segmentStore = TranscriptSegmentStore(context, metadataStore) + private val continuationStore = TranscriptContinuationStore(context, metadataStore, segmentStore) + private val storageStatsStore = TranscriptStorageStatsStore(context) + + fun validateStableSessionId(value: String): String = context.validateStableSessionId(value) + + fun root(): Path = context.root() + + fun open( + stableSessionId: String, + epoch: Long, + ): TranscriptMetadata = + metadataStore.open(stableSessionId, epoch).also { refreshStorageStats() } + + fun acquireLease( + stableSessionId: String, + owner: String, + epoch: Long, + ): TranscriptLease = leaseStore.acquire(stableSessionId, owner, epoch) + + fun releaseLease(lease: TranscriptLease) { + leaseStore.release(lease) + } - private fun String?.redactedLabel(): String? = - this - ?.replace(Regex("\\s+"), " ") - ?.trim() - ?.takeIf { it.isNotBlank() } - ?.let(::redactSecrets) - ?.take(MAX_SETUP_LABEL_CHARS) - ?.let { "\"${it.replace("\\", "\\\\").replace("\"", "\\\"")}\"" } + fun renewLease(lease: TranscriptLease): TranscriptLease? = leaseStore.renew(lease) - private fun redactSecrets(value: String): String = - SECRET_ASSIGNMENT.replace(SECRET_TOKEN.replace(value, "[redacted]")) { - "${it.groupValues[1]}=[redacted]" - } + fun activeSegmentPath(stableSessionId: String): Path = segmentStore.activeSegmentPath(stableSessionId) - private fun readMetadata(stableSessionId: String): TranscriptMetadata? { - val file = metadataFile(stableSessionId) - if (!Files.exists(file)) return null - val p = Properties() - Files.newInputStream(file).use(p::load) - return TranscriptMetadata( - stableSessionId = p.getProperty("stableSessionId", stableSessionId), - epoch = p.getProperty("epoch", "1").toLong(), - logicalStart = p.getProperty("logicalStart", "0").toLong(), - logicalEnd = p.getProperty("logicalEnd", "0").toLong(), - activeSegment = p.getProperty("activeSegment", "0").toInt(), - sealed = p.getProperty("sealed", "false").toBoolean(), - delimiterEpochs = - p - .getProperty("delimiterEpochs", "") - .split(',') - .mapNotNull { it.trim().takeIf(String::isNotBlank)?.toLong() } - .toSet(), - updatedAt = Instant.ofEpochMilli(p.getProperty("updatedAtMillis", "0").toLong()), - ) - } + fun appendContinuationDelimiter( + stableSessionId: String, + epoch: Long, + continuation: AgentContinuation?, + ): TranscriptMetadata = continuationStore.appendDelimiter(stableSessionId, epoch, continuation) - private fun writeMetadata( - stableSessionId: String, - metadata: TranscriptMetadata, - ) { - val now = Instant.now(clock) - val p = - Properties().apply { - this["stableSessionId"] = metadata.stableSessionId - this["epoch"] = metadata.epoch.toString() - this["logicalStart"] = metadata.logicalStart.toString() - this["logicalEnd"] = metadata.logicalEnd.toString() - this["activeSegment"] = metadata.activeSegment.toString() - this["sealed"] = metadata.sealed.toString() - this["delimiterEpochs"] = metadata.delimiterEpochs.sorted().joinToString(",") - this["updatedAtMillis"] = now.toEpochMilli().toString() - } - writePropertiesAtomic(metadataFile(stableSessionId), p) - } + fun recoverMetadata(stableSessionId: String): TranscriptMetadata = metadataStore.recover(stableSessionId) - private fun readLease(file: Path): TranscriptLease? { - if (!Files.exists(file)) return null - val p = Properties() - Files.newInputStream(file).use(p::load) - return TranscriptLease( - stableSessionId = p.getProperty("stableSessionId"), - owner = p.getProperty("owner"), - token = p.getProperty("token"), - epoch = p.getProperty("epoch", "1").toLong(), - expiresAtMillis = p.getProperty("expiresAtMillis", "0").toLong(), - ) - } + fun rotateIfNeeded(stableSessionId: String): TranscriptMetadata = + segmentStore.rotateIfNeeded(stableSessionId).also { refreshStorageStats() } - private fun writePropertiesAtomic( - file: Path, - properties: Properties, - ) { - Files.createDirectories(file.parent) - val tmp = file.resolveSibling("${file.fileName}.tmp-${UUID.randomUUID()}") - Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE).use { out -> - properties.store(out, null) - } - runCatching { - Files.move(tmp, file, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING) - }.getOrElse { - Files.move(tmp, file, StandardCopyOption.REPLACE_EXISTING) - } - } + fun trimIfNeeded(stableSessionId: String): TranscriptMetadata = + segmentStore.trimIfNeeded(stableSessionId).also { refreshStorageStats() } - private fun lockFor(stableSessionId: String): Any = locks.computeIfAbsent(stableSessionId) { Any() } + fun seal(stableSessionId: String): TranscriptMetadata = + metadataStore.seal(stableSessionId).also { refreshStorageStats() } - private fun InputStream.skipFully(bytes: Long) { - var remaining = bytes - while (remaining > 0) { - val skipped = skip(remaining) - if (skipped <= 0) break - remaining -= skipped - } - } + fun cleanup(stableSessionId: String): Boolean = + metadataStore.cleanup(stableSessionId, leaseStore).also { refreshStorageStats() } - private class ByteArrayBuilder( - capacity: Int, - ) { - private val bytes = ByteArray(capacity) - var size: Int = 0 - private set + fun storageStats(): TranscriptStorageStats = storageStatsStore.current() - fun readFrom( - input: InputStream, - limit: Int, - ) { - var remaining = limit - while (remaining > 0) { - val read = input.read(bytes, size, remaining) - if (read <= 0) return - size += read - remaining -= read - } - } + fun refreshStorageStats(): TranscriptStorageStats = storageStatsStore.refresh() - fun toByteArray(): ByteArray = bytes.copyOf(size) - } + fun readRaw( + stableSessionId: String, + fromOffset: Long, + maxBytes: Int = DEFAULT_READ_BYTES, + ): TranscriptRawRead = segmentStore.readRaw(stableSessionId, fromOffset, maxBytes) - companion object { - private const val DEFAULT_READ_BYTES = 64 * 1024 - private const val MILLIS_PER_SECOND = 1_000L - private const val MAX_REASON_CHARS = 160 - private const val MAX_SETUP_LABEL_CHARS = 160 - private val UUID_SESSION_DIR = - Regex("[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") - private val SEGMENT_FILE = Regex("segment-(\\d{6})\\.log") - private val SECRET_TOKEN = - Regex("""(?i)\b(?:sk-[A-Za-z0-9_-]{12,}|gh[pousr]_[A-Za-z0-9_]{12,}|github_pat_[A-Za-z0-9_]{12,})""") - private val SECRET_ASSIGNMENT = - Regex("""(?i)\b(password|passwd|secret|token|api[_-]?key|bearer|credential)=\S+""") - private val locks = java.util.concurrent.ConcurrentHashMap() - } + companion object { + private const val DEFAULT_READ_BYTES = 64 * 1024 } +} data class TranscriptMetadata( val stableSessionId: String, diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreContext.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreContext.kt new file mode 100644 index 0000000..976b07d --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreContext.kt @@ -0,0 +1,52 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardCopyOption +import java.nio.file.StandardOpenOption +import java.time.Clock +import java.util.Properties +import java.util.UUID +import kotlin.io.path.Path + +internal class TranscriptStoreContext( + val props: GatewayProperties, + val clock: Clock, + val telemetry: AgentGatewayTelemetry, +) { + val paths = TranscriptStorePaths(this) + + fun validateStableSessionId(value: String): String = UUID.fromString(value).toString() + + fun root(): Path { + val workspace = Path(props.workspaceRoot).toAbsolutePath().normalize() + val root = workspace.resolve(props.transcripts.dirName).normalize() + require(root.startsWith(workspace)) { "transcript directory must stay inside the workspace" } + Files.createDirectories(root) + return root + } + + fun lockFor(stableSessionId: String): Any = locks.computeIfAbsent(stableSessionId) { Any() } + + fun writePropertiesAtomic( + file: Path, + properties: Properties, + ) { + Files.createDirectories(file.parent) + val tmp = file.resolveSibling("${file.fileName}.tmp-${UUID.randomUUID()}") + Files.newOutputStream(tmp, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE).use { out -> + properties.store(out, null) + } + runCatching { + Files.move(tmp, file, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING) + }.getOrElse { + Files.move(tmp, file, StandardCopyOption.REPLACE_EXISTING) + } + } + + companion object { + private val locks = java.util.concurrent.ConcurrentHashMap() + } +} diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorePaths.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorePaths.kt new file mode 100644 index 0000000..0866e15 --- /dev/null +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStorePaths.kt @@ -0,0 +1,48 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import java.nio.file.Files +import java.nio.file.Path +import java.util.Comparator +import java.util.Locale +import kotlin.streams.toList + +internal class TranscriptStorePaths( + private val context: TranscriptStoreContext, +) { + fun sessionDir(stableSessionId: String): Path = context.root().resolve(stableSessionId) + + fun segmentsDir(stableSessionId: String): Path = sessionDir(stableSessionId).resolve("segments") + + fun metadataFile(stableSessionId: String): Path = sessionDir(stableSessionId).resolve("metadata.properties") + + fun leaseFile(stableSessionId: String): Path = sessionDir(stableSessionId).resolve("lease.properties") + + fun segmentPath( + stableSessionId: String, + index: Int, + ): Path = segmentsDir(stableSessionId).resolve("segment-%06d.log".format(Locale.ROOT, index)) + + fun segmentFiles(stableSessionId: String): List { + val dir = segmentsDir(stableSessionId) + if (!Files.exists(dir)) return emptyList() + return Files.list(dir).use { paths -> + paths + .filter { SEGMENT_FILE.matches(it.fileName.toString()) } + .sorted(Comparator.comparingInt { segmentIndex(it) }) + .toList() + } + } + + fun segmentIndex(path: Path): Int = + SEGMENT_FILE + .matchEntire(path.fileName.toString()) + ?.groupValues + ?.get(1) + ?.toInt() + ?: error("invalid segment file: $path") + + companion object { + val UUID_SESSION_DIR = Regex("[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") + val SEGMENT_FILE = Regex("segment-(\\d{6})\\.log") + } +} diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreCollaboratorsTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreCollaboratorsTest.kt new file mode 100644 index 0000000..c8b7b7c --- /dev/null +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreCollaboratorsTest.kt @@ -0,0 +1,115 @@ +package com.jorisjonkers.personalstack.agentgateway.tmux + +import com.jorisjonkers.personalstack.agentgateway.config.GatewayProperties +import com.jorisjonkers.personalstack.agentgateway.observability.AgentGatewayTelemetry +import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assertions.assertThatThrownBy +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption +import java.time.Clock +import java.time.Instant +import java.time.ZoneOffset + +class TranscriptStoreCollaboratorsTest { + private fun context( + tmp: Path, + segmentBytes: Long = 8, + capBytes: Long = 16, + leaseTtlSeconds: Long = 60, + clock: Clock = Clock.fixed(Instant.parse("2026-06-12T09:00:00Z"), ZoneOffset.UTC), + ): TranscriptStoreContext = + TranscriptStoreContext( + GatewayProperties( + workspaceRoot = tmp.toString(), + tmux = GatewayProperties.Tmux(socketName = "agent-gw", stateDir = tmp.resolve("tmux").toString()), + cli = GatewayProperties.Cli(claude = "claude", codex = "codex"), + transcripts = + GatewayProperties.Transcripts( + segmentBytes = segmentBytes, + capBytes = capBytes, + leaseTtlSeconds = leaseTtlSeconds, + ), + ), + clock, + telemetry = AgentGatewayTelemetry.NOOP, + ) + + @Test + fun `lease store rejects a competing live owner`( + @TempDir tmp: Path, + ) { + val leaseStore = TranscriptLeaseStore(context(tmp, leaseTtlSeconds = 30)) + val stable = "11111111-1111-1111-1111-111111111111" + + val lease = leaseStore.acquire(stable, "owner-a", 1) + + assertThat(lease.owner).isEqualTo("owner-a") + assertThatThrownBy { leaseStore.acquire(stable, "owner-b", 1) } + .isInstanceOf(IllegalStateException::class.java) + .hasMessageContaining("leased by owner-a") + } + + @Test + fun `metadata store rebuilds logical end from segment files`( + @TempDir tmp: Path, + ) { + val context = context(tmp) + val metadataStore = TranscriptMetadataStore(context) + val stable = "11111111-1111-1111-1111-111111111111" + metadataStore.open(stable, 7) + Files.writeString(context.paths.segmentPath(stable, 0), "hello", StandardOpenOption.APPEND) + + val metadata = metadataStore.recover(stable) + + assertThat(metadata.epoch).isEqualTo(7) + assertThat(metadata.logicalStart).isZero() + assertThat(metadata.logicalEnd).isEqualTo(5) + } + + @Test + fun `segment store reads across retained segments after trim`( + @TempDir tmp: Path, + ) { + val context = context(tmp, segmentBytes = 4, capBytes = 8) + val metadataStore = TranscriptMetadataStore(context) + val segmentStore = TranscriptSegmentStore(context, metadataStore) + val stable = "11111111-1111-1111-1111-111111111111" + metadataStore.open(stable, 1) + Files.writeString(segmentStore.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) + segmentStore.rotateIfNeeded(stable) + Files.writeString(segmentStore.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) + segmentStore.rotateIfNeeded(stable) + Files.writeString(segmentStore.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) + + val metadata = segmentStore.trimIfNeeded(stable) + val read = segmentStore.readRaw(stable, 0) + + assertThat(metadata.logicalStart).isEqualTo(4) + assertThat(read.startOffset).isEqualTo(4) + assertThat(String(read.bytes, Charsets.UTF_8)).isEqualTo("22223333") + } + + @Test + fun `storage stats store counts only direct uuid segment files`( + @TempDir tmp: Path, + ) { + val context = context(tmp) + val statsStore = TranscriptStorageStatsStore(context) + val root = context.root() + val stable = "11111111-1111-1111-1111-111111111111" + val segments = root.resolve(stable).resolve("segments") + Files.createDirectories(segments.resolve("nested")) + Files.writeString(segments.resolve("segment-000000.log"), "ok") + Files.writeString(segments.resolve("nested").resolve("segment-000001.log"), "nested") + Files.createDirectories(root.resolve("not-a-session").resolve("segments")) + Files.writeString(root.resolve("not-a-session").resolve("segments").resolve("segment-000000.log"), "ignored") + + val stats = statsStore.refresh() + + assertThat(stats.usedBytes).isEqualTo(2) + assertThat(statsStore.current()).isEqualTo(stats) + } +} From a96a34ee624cf1a90ed218a3c4e9b83abdeadd30 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:47:02 +0000 Subject: [PATCH 2/6] fix: address detekt and ktlint findings from CI --- .../agentgateway/tmux/TranscriptLeaseStore.kt | 5 +++-- .../agentgateway/tmux/TranscriptMetadataStore.kt | 2 +- .../personalstack/agentgateway/tmux/TranscriptStore.kt | 7 +++---- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt index be138f8..3b3df2f 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt @@ -51,8 +51,9 @@ internal class TranscriptLeaseStore( ?.let { current -> write( current.copy( - expiresAtMillis = context.clock.millis() + - context.props.transcripts.leaseTtlSeconds * MILLIS_PER_SECOND, + expiresAtMillis = + context.clock.millis() + + context.props.transcripts.leaseTtlSeconds * MILLIS_PER_SECOND, ), ) } diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt index 993ff4c..4edb8f7 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptMetadataStore.kt @@ -31,7 +31,7 @@ internal class TranscriptMetadataStore( (existing ?: TranscriptMetadata(stableSessionId = id)).copy( logicalEnd = existing?.logicalStart ?: 0L, activeSegment = 0, - ) + ) Files.createFile(context.paths.segmentPath(id, 0)) write(id, metadata) return@synchronized metadata diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt index 9e49fe9..624f8b9 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt @@ -9,10 +9,9 @@ import java.time.Clock import java.time.Instant @Component -class TranscriptStore - private constructor( - private val context: TranscriptStoreContext, - ) { +class TranscriptStore private constructor( + private val context: TranscriptStoreContext, +) { @Autowired constructor( props: GatewayProperties, From 500ecef6fc5d923ad986ad76b149414362be652d Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 07:53:42 +0000 Subject: [PATCH 3/6] fix: remaining ktlint style and test reference findings --- .../personalstack/agentgateway/tmux/TranscriptStore.kt | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt index 624f8b9..03b7b8e 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt @@ -37,8 +37,7 @@ class TranscriptStore private constructor( fun open( stableSessionId: String, epoch: Long, - ): TranscriptMetadata = - metadataStore.open(stableSessionId, epoch).also { refreshStorageStats() } + ): TranscriptMetadata = metadataStore.open(stableSessionId, epoch).also { refreshStorageStats() } fun acquireLease( stableSessionId: String, From 9fc0713d3efacf95863d515603492a14940e4c85 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 08:01:58 +0000 Subject: [PATCH 4/6] fix: slim facade below function threshold and settle multiline style --- .../tmux/AgentSessionSpawnWorkflow.kt | 4 +- .../tmux/AgentTranscriptMaintenance.kt | 8 +-- .../agentgateway/tmux/TranscriptStore.kt | 34 ++++------- .../tmux/AgentSessionCollaboratorsTest.kt | 4 +- .../tmux/AgentSessionManagerTest.kt | 6 +- .../agentgateway/tmux/LogTailerTest.kt | 10 +++- .../agentgateway/tmux/TranscriptStoreTest.kt | 60 +++++++++---------- .../agentgateway/tmux/TranscriptTailerTest.kt | 2 +- .../agentgateway/ws/AgentAttachHandlerTest.kt | 40 ++++++------- 9 files changed, 79 insertions(+), 89 deletions(-) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt index 3f8bd69..d927930 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionSpawnWorkflow.kt @@ -146,9 +146,9 @@ internal class AgentSessionSpawnWorkflow( try { transcriptStore.open(stableSessionId, epoch) if (epoch > 1 || request.continuation != null) { - transcriptStore.appendContinuationDelimiter(stableSessionId, epoch, request.continuation) + transcriptStore.continuationStore.appendDelimiter(stableSessionId, epoch, request.continuation) } - return transcriptStore.activeSegmentPath(stableSessionId).also { opened = true } + return transcriptStore.segmentStore.activeSegmentPath(stableSessionId).also { opened = true } } finally { if (!opened) transcriptStore.releaseLease(lease) } diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt index 2f4c3d2..0cc7594 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentTranscriptMaintenance.kt @@ -44,19 +44,19 @@ internal class AgentTranscriptMaintenance( ) { val stableSessionId = requireNotNull(session.stableSessionId) val beforeTrim = transcriptStore.recoverMetadata(stableSessionId).logicalStart - val current = renewLease(session) + val current = renewTranscriptLease(session) transcriptStore.rotateIfNeeded(stableSessionId) - rotatePipeIfNeeded(current, transcriptStore.activeSegmentPath(stableSessionId)) + rotatePipeIfNeeded(current, transcriptStore.segmentStore.activeSegmentPath(stableSessionId)) val trimmed = transcriptStore.trimIfNeeded(stableSessionId) if (trimmed.logicalStart > beforeTrim) { recordMaintenance(current, startedAt, GatewayOutcomeLabel.SUCCESS) } } - private fun renewLease(session: AgentSession): AgentSession { + private fun renewTranscriptLease(session: AgentSession): AgentSession { val current = session.transcriptLease - ?.let(transcriptStore::renewLease) + ?.let(transcriptStore.leaseStore::renew) ?.let { session.copy(transcriptLease = it) } ?: session if (current !== session) registry.update(current) diff --git a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt index 03b7b8e..664c533 100644 --- a/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt +++ b/services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStore.kt @@ -24,11 +24,11 @@ class TranscriptStore private constructor( telemetry: AgentGatewayTelemetry = AgentGatewayTelemetry.NOOP, ) : this(TranscriptStoreContext(props, clock, telemetry)) - private val leaseStore = TranscriptLeaseStore(context) - private val metadataStore = TranscriptMetadataStore(context) - private val segmentStore = TranscriptSegmentStore(context, metadataStore) - private val continuationStore = TranscriptContinuationStore(context, metadataStore, segmentStore) - private val storageStatsStore = TranscriptStorageStatsStore(context) + internal val leaseStore = TranscriptLeaseStore(context) + internal val metadataStore = TranscriptMetadataStore(context) + internal val segmentStore = TranscriptSegmentStore(context, metadataStore) + internal val continuationStore = TranscriptContinuationStore(context, metadataStore, segmentStore) + internal val storageStatsStore = TranscriptStorageStatsStore(context) fun validateStableSessionId(value: String): String = context.validateStableSessionId(value) @@ -37,7 +37,7 @@ class TranscriptStore private constructor( fun open( stableSessionId: String, epoch: Long, - ): TranscriptMetadata = metadataStore.open(stableSessionId, epoch).also { refreshStorageStats() } + ): TranscriptMetadata = metadataStore.open(stableSessionId, epoch).also { storageStatsStore.refresh() } fun acquireLease( stableSessionId: String, @@ -49,33 +49,19 @@ class TranscriptStore private constructor( leaseStore.release(lease) } - fun renewLease(lease: TranscriptLease): TranscriptLease? = leaseStore.renew(lease) - - fun activeSegmentPath(stableSessionId: String): Path = segmentStore.activeSegmentPath(stableSessionId) - - fun appendContinuationDelimiter( - stableSessionId: String, - epoch: Long, - continuation: AgentContinuation?, - ): TranscriptMetadata = continuationStore.appendDelimiter(stableSessionId, epoch, continuation) - fun recoverMetadata(stableSessionId: String): TranscriptMetadata = metadataStore.recover(stableSessionId) fun rotateIfNeeded(stableSessionId: String): TranscriptMetadata = - segmentStore.rotateIfNeeded(stableSessionId).also { refreshStorageStats() } + segmentStore.rotateIfNeeded(stableSessionId).also { storageStatsStore.refresh() } fun trimIfNeeded(stableSessionId: String): TranscriptMetadata = - segmentStore.trimIfNeeded(stableSessionId).also { refreshStorageStats() } + segmentStore.trimIfNeeded(stableSessionId).also { storageStatsStore.refresh() } fun seal(stableSessionId: String): TranscriptMetadata = - metadataStore.seal(stableSessionId).also { refreshStorageStats() } + metadataStore.seal(stableSessionId).also { storageStatsStore.refresh() } fun cleanup(stableSessionId: String): Boolean = - metadataStore.cleanup(stableSessionId, leaseStore).also { refreshStorageStats() } - - fun storageStats(): TranscriptStorageStats = storageStatsStore.current() - - fun refreshStorageStats(): TranscriptStorageStats = storageStatsStore.refresh() + metadataStore.cleanup(stableSessionId, leaseStore).also { storageStatsStore.refresh() } fun readRaw( stableSessionId: String, diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt index efbc0b7..92a9c54 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionCollaboratorsTest.kt @@ -84,7 +84,7 @@ class AgentSessionCollaboratorsTest { val stable = "11111111-1111-1111-1111-111111111111" val lease = store.acquireLease(stable, "agent-one", 1) store.open(stable, 1) - val firstSegment = store.activeSegmentPath(stable) + val firstSegment = store.segmentStore.activeSegmentPath(stable) Files.write(firstSegment, ByteArray(16) { 'x'.code.toByte() }) registry.put( session("one", AgentKind.SHELL, firstSegment).copy( @@ -98,7 +98,7 @@ class AgentSessionCollaboratorsTest { AgentTranscriptMaintenance(tmux, props, store, registry, telemetry).maintain() val current = requireNotNull(registry.get("one")) - assertThat(current.logFile).isEqualTo(store.activeSegmentPath(stable)) + assertThat(current.logFile).isEqualTo(store.segmentStore.activeSegmentPath(stable)) assertThat(current.logFile).isNotEqualTo(firstSegment) verify { tmux.startPipeToFile("agent-one", current.logFile) } } diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManagerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManagerTest.kt index 576fbf4..e937059 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManagerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/AgentSessionManagerTest.kt @@ -99,7 +99,7 @@ class AgentSessionManagerTest { val session = mgr.spawn(AgentKind.SHELL, stableSessionId = stable) - assertThat(session.logFile).isEqualTo(store.activeSegmentPath(stable)) + assertThat(session.logFile).isEqualTo(store.segmentStore.activeSegmentPath(stable)) verify { store.acquireLease(stable, session.tmuxSession, 1) } } @@ -607,7 +607,7 @@ class AgentSessionManagerTest { val mgr = AgentSessionManager(tmux, props, store).also(managers::add) try { val s = mgr.spawn(AgentKind.SHELL, stableSessionId = stable) - val firstSegment = store.activeSegmentPath(requireNotNull(s.stableSessionId)) + val firstSegment = store.segmentStore.activeSegmentPath(requireNotNull(s.stableSessionId)) val payload = ByteArray(96) { 'x'.code.toByte() } assertThat(firstSegment).isEqualTo(s.logFile) @@ -622,7 +622,7 @@ class AgentSessionManagerTest { } val current = requireNotNull(mgr.get(s.id)) val metadata = store.recoverMetadata(stable) - assertThat(current.logFile).isEqualTo(store.activeSegmentPath(stable)) + assertThat(current.logFile).isEqualTo(store.segmentStore.activeSegmentPath(stable)) assertThat(current.transcriptFile).isEqualTo(current.logFile) assertThat(metadata.logicalStart).isEqualTo(payload.size.toLong()) assertThat(metadata.byteCount).isLessThanOrEqualTo(props.transcripts.capBytes) diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/LogTailerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/LogTailerTest.kt index b721fc7..1fd452c 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/LogTailerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/LogTailerTest.kt @@ -124,7 +124,7 @@ class LogTailerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "abéZ", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "abéZ", StandardOpenOption.APPEND) store.recoverMetadata(stable) val frames = CopyOnWriteArrayList() @@ -159,12 +159,16 @@ class LogTailerTest { ).use { tailer -> tailer.start() Files.write( - store.activeSegmentPath(stable), + store.segmentStore.activeSegmentPath(stable), byteArrayOf(0xE2.toByte(), 0x94.toByte()), StandardOpenOption.APPEND, ) Thread.sleep(40) - Files.write(store.activeSegmentPath(stable), byteArrayOf(0x80.toByte(), 0x41), StandardOpenOption.APPEND) + Files.write( + store.segmentStore.activeSegmentPath(stable), + byteArrayOf(0x80.toByte(), 0x41), + StandardOpenOption.APPEND, + ) await().atMost(Duration.ofSeconds(2)).until { received.joinToString("") { it.output } == "─A" } assertThat(received.last().off).isEqualTo(4) } diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt index 3fe3a99..5fe1e58 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt @@ -55,7 +55,7 @@ class TranscriptStoreTest { val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - val active = store.activeSegmentPath(stable) + val active = store.segmentStore.activeSegmentPath(stable) assertThat(active.toString()).startsWith(tmp.resolve(".agent-transcripts").resolve(stable).toString()) assertThat(active.fileName.toString()).isEqualTo("segment-000000.log") @@ -68,7 +68,7 @@ class TranscriptStoreTest { val store = store(tmp) val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) val recovered = store.recoverMetadata(stable) @@ -83,12 +83,12 @@ class TranscriptStoreTest { val store = store(tmp, StoreOptions(segmentBytes = 4)) val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "1234", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "1234", StandardOpenOption.APPEND) val metadata = store.rotateIfNeeded(stable) assertThat(metadata.activeSegment).isEqualTo(1) - assertThat(store.activeSegmentPath(stable).fileName.toString()).isEqualTo("segment-000001.log") + assertThat(store.segmentStore.activeSegmentPath(stable).fileName.toString()).isEqualTo("segment-000001.log") } @Test @@ -98,11 +98,11 @@ class TranscriptStoreTest { val store = store(tmp, StoreOptions(segmentBytes = 4, capBytes = 8)) val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) store.rotateIfNeeded(stable) - Files.writeString(store.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) store.rotateIfNeeded(stable) - Files.writeString(store.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) val trimmed = store.trimIfNeeded(stable) @@ -141,7 +141,7 @@ class TranscriptStoreTest { store(tmp, StoreOptions(leaseTtlSeconds = 30, clock = firstClock)) .acquireLease(stable, "owner-a", 1) - val renewed = store(tmp, StoreOptions(leaseTtlSeconds = 30, clock = secondClock)).renewLease(lease) + val renewed = store(tmp, StoreOptions(leaseTtlSeconds = 30, clock = secondClock)).leaseStore.renew(lease) assertThat(renewed).isNotNull assertThat(renewed!!.token).isEqualTo(lease.token) @@ -155,7 +155,7 @@ class TranscriptStoreTest { val store = store(tmp, StoreOptions(retentionSeconds = 0)) val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - val path = store.activeSegmentPath(stable) + val path = store.segmentStore.activeSegmentPath(stable) Files.writeString(path, "bytes", StandardOpenOption.APPEND) store.seal(stable) @@ -170,11 +170,11 @@ class TranscriptStoreTest { val registry = SimpleMeterRegistry() val store = store(tmp, StoreOptions(capBytes = 123, telemetry = MicrometerAgentGatewayTelemetry(registry))) - val stats = store.refreshStorageStats() + val stats = store.storageStatsStore.refresh() assertThat(stats.usedBytes).isZero() assertThat(stats.capBytes).isEqualTo(123) - assertThat(store.storageStats()).isEqualTo(stats) + assertThat(store.storageStatsStore.current()).isEqualTo(stats) assertThat(registry.find("agent.gateway.storage.bytes").gauge()!!.value()).isZero() assertThat(registry.find("agent.gateway.storage.limit.bytes").gauge()!!.value()).isEqualTo(123.0) } @@ -187,18 +187,18 @@ class TranscriptStoreTest { val store = store(tmp, StoreOptions(telemetry = MicrometerAgentGatewayTelemetry(registry))) val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "one", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "one", StandardOpenOption.APPEND) - val first = store.refreshStorageStats() - Files.writeString(store.activeSegmentPath(stable), "two", StandardOpenOption.APPEND) + val first = store.storageStatsStore.refresh() + Files.writeString(store.segmentStore.activeSegmentPath(stable), "two", StandardOpenOption.APPEND) - assertThat(store.storageStats().usedBytes).isEqualTo(first.usedBytes) + assertThat(store.storageStatsStore.current().usedBytes).isEqualTo(first.usedBytes) assertThat(registry.find("agent.gateway.storage.bytes").gauge()!!.value()).isEqualTo(first.usedBytes.toDouble()) val refreshed = store.trimIfNeeded(stable) assertThat(refreshed.byteCount).isEqualTo(6) - assertThat(store.storageStats().usedBytes).isEqualTo(6) + assertThat(store.storageStatsStore.current().usedBytes).isEqualTo(6) assertThat(registry.find("agent.gateway.storage.bytes").gauge()!!.value()).isEqualTo(6.0) } @@ -210,12 +210,12 @@ class TranscriptStoreTest { val active = "11111111-1111-1111-1111-111111111111" val retained = "22222222-2222-2222-2222-222222222222" store.open(active, 1) - Files.writeString(store.activeSegmentPath(active), "active", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(active), "active", StandardOpenOption.APPEND) store.open(retained, 1) - Files.writeString(store.activeSegmentPath(retained), "retained", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(retained), "retained", StandardOpenOption.APPEND) store.seal(retained) - val stats = store.refreshStorageStats() + val stats = store.storageStatsStore.refresh() assertThat(stats.usedBytes).isEqualTo("activeretained".length.toLong()) } @@ -240,7 +240,7 @@ class TranscriptStoreTest { val link = segments.resolve("segment-000001.log") assumeTrue(runCatching { Files.createSymbolicLink(link, outside) }.isSuccess) - val stats = store.refreshStorageStats() + val stats = store.storageStatsStore.refresh() assertThat(stats.usedBytes).isEqualTo(2) } @@ -280,7 +280,7 @@ class TranscriptStoreTest { StandardOpenOption.WRITE, ) - val stats = store.refreshStorageStats() + val stats = store.storageStatsStore.refresh() assertThat(stats.usedBytes).isEqualTo(2) } @@ -293,7 +293,7 @@ class TranscriptStoreTest { val store = store(tmp, StoreOptions(telemetry = MicrometerAgentGatewayTelemetry(registry))) val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "bytes", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "bytes", StandardOpenOption.APPEND) store.seal(stable) @@ -317,10 +317,10 @@ class TranscriptStoreTest { val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 2) - store.appendContinuationDelimiter(stable, 2, AgentContinuation(reason = "restart", previousEpoch = 1)) - store.appendContinuationDelimiter(stable, 2, AgentContinuation(reason = "restart", previousEpoch = 1)) + store.continuationStore.appendDelimiter(stable, 2, AgentContinuation(reason = "restart", previousEpoch = 1)) + store.continuationStore.appendDelimiter(stable, 2, AgentContinuation(reason = "restart", previousEpoch = 1)) - val text = Files.readString(store.activeSegmentPath(stable)) + val text = Files.readString(store.segmentStore.activeSegmentPath(stable)) assertThat(Regex("continuation epoch=2").findAll(text).toList()).hasSize(1) assertThat(Regex("agent restarted").findAll(text).toList()).hasSize(1) assertThat(text).doesNotContain("updated setup") @@ -335,7 +335,7 @@ class TranscriptStoreTest { val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 2) - store.appendContinuationDelimiter( + store.continuationStore.appendDelimiter( stable, 2, AgentContinuation( @@ -344,7 +344,7 @@ class TranscriptStoreTest { ), ) - val text = Files.readString(store.activeSegmentPath(stable)) + val text = Files.readString(store.segmentStore.activeSegmentPath(stable)) assertThat(text).contains("agent restarted") assertThat(text).doesNotContain("setup transition") } @@ -357,7 +357,7 @@ class TranscriptStoreTest { val stable = "11111111-1111-1111-1111-111111111111" store.open(stable, 2) - store.appendContinuationDelimiter( + store.continuationStore.appendDelimiter( stable, 2, AgentContinuation( @@ -367,7 +367,7 @@ class TranscriptStoreTest { toSetupLabel = "GPU runner", ), ) - store.appendContinuationDelimiter( + store.continuationStore.appendDelimiter( stable, 2, AgentContinuation( @@ -378,7 +378,7 @@ class TranscriptStoreTest { ), ) - val text = Files.readString(store.activeSegmentPath(stable)) + val text = Files.readString(store.segmentStore.activeSegmentPath(stable)) assertThat(Regex("continuation epoch=2").findAll(text).toList()).hasSize(1) assertThat(text).contains("setup transition") assertThat(text).contains("fromSetup=\"Default runner token=[redacted]\"") diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptTailerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptTailerTest.kt index 3d2ae37..649ed2a 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptTailerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptTailerTest.kt @@ -19,7 +19,7 @@ class TranscriptTailerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "abcdef", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "abcdef", StandardOpenOption.APPEND) store.recoverMetadata(stable) val frames = mutableListOf() diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/ws/AgentAttachHandlerTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/ws/AgentAttachHandlerTest.kt index b819d6a..d7831a3 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/ws/AgentAttachHandlerTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/ws/AgentAttachHandlerTest.kt @@ -108,7 +108,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) store.recoverMetadata(stable) val telemetry = RecordingTelemetry() val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store, telemetry) @@ -151,7 +151,7 @@ class AgentAttachHandlerTest { store.open(stable, 1) val total = AgentAttachHandler.MAX_COLD_REPLAY_BYTES + 10 Files.writeString( - store.activeSegmentPath(stable), + store.segmentStore.activeSegmentPath(stable), "x".repeat(total.toInt()), StandardOpenOption.APPEND, ) @@ -181,7 +181,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "abcdef", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "abcdef", StandardOpenOption.APPEND) store.recoverMetadata(stable) val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store) val ws = wsSession("abc", "?mode=RESUME&epoch=1&offset=4&cursor=1&off=2") @@ -207,10 +207,10 @@ class AgentAttachHandlerTest { val stableForOff = "22222222-2222-2222-2222-222222222222" val store = transcriptStore(tmp) store.open(stableForCursor, 1) - Files.writeString(store.activeSegmentPath(stableForCursor), "abcdef", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stableForCursor), "abcdef", StandardOpenOption.APPEND) store.recoverMetadata(stableForCursor) store.open(stableForOff, 1) - Files.writeString(store.activeSegmentPath(stableForOff), "abcdef", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stableForOff), "abcdef", StandardOpenOption.APPEND) store.recoverMetadata(stableForOff) val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store) @@ -249,7 +249,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 2) - Files.writeString(store.activeSegmentPath(stable), "abcdef", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "abcdef", StandardOpenOption.APPEND) store.recoverMetadata(stable) val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store) val ws = wsSession("abc", "?mode=RESUME&epoch=1&offset=3") @@ -274,11 +274,11 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp, segmentBytes = 4, capBytes = 8) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) store.rotateIfNeeded(stable) - Files.writeString(store.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) store.rotateIfNeeded(stable) - Files.writeString(store.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) store.trimIfNeeded(stable) val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store) val ws = wsSession("abc", "?mode=RESUME&epoch=1&offset=0") @@ -306,11 +306,11 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp, segmentBytes = 4, capBytes = 8) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "1111", StandardOpenOption.APPEND) store.rotateIfNeeded(stable) - Files.writeString(store.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "2222", StandardOpenOption.APPEND) store.rotateIfNeeded(stable) - Files.writeString(store.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "3333", StandardOpenOption.APPEND) store.trimIfNeeded(stable) val telemetry = RecordingTelemetry() val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store, telemetry) @@ -338,7 +338,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) store.recoverMetadata(stable) val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store) val ws = wsSession("abc", "?mode=RESUME&epoch=1&offset=0") @@ -347,7 +347,7 @@ class AgentAttachHandlerTest { every { ws.sendMessage(capture(sent)) } returns Unit durableHandler.afterConnectionEstablished(ws) - Files.writeString(store.activeSegmentPath(stable), "-live", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "-live", StandardOpenOption.APPEND) await().atMost(Duration.ofSeconds(2)).until { sent.any { it.payload.contains("-live") } } val replayOutputIndex = sent.indexOfFirst { it.payload.contains("\"output\":\"hello\"") } @@ -367,7 +367,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "abc", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "abc", StandardOpenOption.APPEND) store.recoverMetadata(stable) val telemetry = RecordingTelemetry() val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store, telemetry) @@ -455,7 +455,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) store.recoverMetadata(stable) val telemetry = RecordingTelemetry() val durableHandler = attachHandler(props.copy(workspaceRoot = tmp.toString()), store, telemetry) @@ -490,7 +490,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) store.recoverMetadata(stable) val durableProps = props.copy( @@ -505,7 +505,7 @@ class AgentAttachHandlerTest { durableHandler.afterConnectionEstablished(ws) durableHandler.afterConnectionClosed(ws, CloseStatus.NORMAL) - Files.writeString(store.activeSegmentPath(stable), "-after-close", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "-after-close", StandardOpenOption.APPEND) Thread.sleep(100) assertThat(sent.any { it.payload.contains("\"output\":\"hello\"") }).isTrue @@ -519,7 +519,7 @@ class AgentAttachHandlerTest { val stable = "11111111-1111-1111-1111-111111111111" val store = transcriptStore(tmp) store.open(stable, 1) - Files.writeString(store.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "hello", StandardOpenOption.APPEND) store.recoverMetadata(stable) val durableProps = props.copy( @@ -534,7 +534,7 @@ class AgentAttachHandlerTest { durableHandler.afterConnectionEstablished(ws) durableHandler.shutdown() - Files.writeString(store.activeSegmentPath(stable), "-after-shutdown", StandardOpenOption.APPEND) + Files.writeString(store.segmentStore.activeSegmentPath(stable), "-after-shutdown", StandardOpenOption.APPEND) Thread.sleep(100) assertThat(sent.any { it.payload.contains("\"output\":\"hello\"") }).isTrue From fd89e7fbab647499c5167417e7fefbd4ec98d481 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 08:49:48 +0000 Subject: [PATCH 5/6] ci: trigger run for facade slimming From adda2afb2c6962c0a652404bac548e4b31319ce8 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Mon, 13 Jul 2026 09:42:32 +0000 Subject: [PATCH 6/6] fix: chained-call newlines in transcript store test --- .../personalstack/agentgateway/tmux/TranscriptStoreTest.kt | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt index 5fe1e58..d7e98e9 100644 --- a/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt +++ b/services/agent-gateway/src/test/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptStoreTest.kt @@ -88,7 +88,12 @@ class TranscriptStoreTest { val metadata = store.rotateIfNeeded(stable) assertThat(metadata.activeSegment).isEqualTo(1) - assertThat(store.segmentStore.activeSegmentPath(stable).fileName.toString()).isEqualTo("segment-000001.log") + assertThat( + store.segmentStore + .activeSegmentPath(stable) + .fileName + .toString(), + ).isEqualTo("segment-000001.log") } @Test