refactor: carve TranscriptStore into lease, metadata, segment and stats stores

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)
         }

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)

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+""")
+    }
+}

services/agent-gateway/src/main/kotlin/com/jorisjonkers/personalstack/agentgateway/tmux/TranscriptLeaseStore.kt

@@ -0,0 +1,93 @@
+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
+    }
+}

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<java.nio.file.Path>()).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<java.nio.file.Path>,
+    ): 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<Long> =
+        properties
+            .getProperty("delimiterEpochs", "")
+            .split(',')
+            .mapNotNull { it.trim().takeIf(String::isNotBlank)?.toLong() }
+            .toSet()
+}