feat: undo session file mutations via server endpoint + CLI command
This commit is contained in:
@@ -5,6 +5,7 @@ import com.correx.apps.cli.commands.ProviderCommand
|
|||||||
import com.correx.apps.cli.commands.RunCommand
|
import com.correx.apps.cli.commands.RunCommand
|
||||||
import com.correx.apps.cli.commands.SessionCommand
|
import com.correx.apps.cli.commands.SessionCommand
|
||||||
import com.correx.apps.cli.commands.StatusCommand
|
import com.correx.apps.cli.commands.StatusCommand
|
||||||
|
import com.correx.apps.cli.commands.UndoCommand
|
||||||
import com.github.ajalt.clikt.core.CliktCommand
|
import com.github.ajalt.clikt.core.CliktCommand
|
||||||
import com.github.ajalt.clikt.core.subcommands
|
import com.github.ajalt.clikt.core.subcommands
|
||||||
import com.github.ajalt.clikt.parameters.options.flag
|
import com.github.ajalt.clikt.parameters.options.flag
|
||||||
@@ -23,4 +24,5 @@ fun buildCli(): CorrexCli = CorrexCli().subcommands(
|
|||||||
ApproveCommand(),
|
ApproveCommand(),
|
||||||
StatusCommand(),
|
StatusCommand(),
|
||||||
ProviderCommand(),
|
ProviderCommand(),
|
||||||
|
UndoCommand(),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -0,0 +1,29 @@
|
|||||||
|
package com.correx.apps.cli.commands
|
||||||
|
|
||||||
|
import com.correx.apps.cli.DEFAULT_PORT
|
||||||
|
import com.github.ajalt.clikt.core.CliktCommand
|
||||||
|
import com.github.ajalt.clikt.parameters.arguments.argument
|
||||||
|
import com.github.ajalt.clikt.parameters.options.default
|
||||||
|
import com.github.ajalt.clikt.parameters.options.option
|
||||||
|
import io.ktor.client.HttpClient
|
||||||
|
import io.ktor.client.engine.cio.CIO
|
||||||
|
import io.ktor.client.request.post
|
||||||
|
import io.ktor.client.statement.bodyAsText
|
||||||
|
import kotlinx.coroutines.runBlocking
|
||||||
|
|
||||||
|
class UndoCommand : CliktCommand(name = "undo") {
|
||||||
|
private val sessionId by argument("SESSION_ID")
|
||||||
|
private val host by option("--host").default("localhost")
|
||||||
|
private val port by option("--port").default("$DEFAULT_PORT")
|
||||||
|
|
||||||
|
override fun run(): Unit = runBlocking {
|
||||||
|
val portInt = port.toIntOrNull() ?: DEFAULT_PORT
|
||||||
|
val client = HttpClient(CIO)
|
||||||
|
try {
|
||||||
|
val response = client.post("http://$host:$portInt/sessions/$sessionId/undo")
|
||||||
|
echo(response.bodyAsText())
|
||||||
|
} finally {
|
||||||
|
client.close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -171,6 +171,13 @@ fun main() {
|
|||||||
embedder = embedder,
|
embedder = embedder,
|
||||||
l3MemoryStore = l3MemoryStore,
|
l3MemoryStore = l3MemoryStore,
|
||||||
)
|
)
|
||||||
|
val sessionUndoService = com.correx.apps.server.undo.SessionUndoService(
|
||||||
|
eventStore = eventStore,
|
||||||
|
reverser = com.correx.infrastructure.tools.filesystem.FileMutationReverser(
|
||||||
|
artifactStore = artifactStore,
|
||||||
|
allowedRoots = setOf(workspaceRoot, workingDir),
|
||||||
|
),
|
||||||
|
)
|
||||||
val module = ServerModule(
|
val module = ServerModule(
|
||||||
orchestrator = orchestrator,
|
orchestrator = orchestrator,
|
||||||
eventStore = eventStore,
|
eventStore = eventStore,
|
||||||
@@ -183,6 +190,7 @@ fun main() {
|
|||||||
orchestrationRepository = repositories.orchestrationRepository,
|
orchestrationRepository = repositories.orchestrationRepository,
|
||||||
approvalRepository = repositories.approvalRepository,
|
approvalRepository = repositories.approvalRepository,
|
||||||
toolRegistry = toolRegistry,
|
toolRegistry = toolRegistry,
|
||||||
|
sessionUndoService = sessionUndoService,
|
||||||
)
|
)
|
||||||
module.start()
|
module.start()
|
||||||
log.info("==============================")
|
log.info("==============================")
|
||||||
|
|||||||
@@ -48,6 +48,7 @@ class ServerModule(
|
|||||||
val orchestrationRepository: OrchestrationRepository,
|
val orchestrationRepository: OrchestrationRepository,
|
||||||
val approvalRepository: DefaultApprovalRepository,
|
val approvalRepository: DefaultApprovalRepository,
|
||||||
val toolRegistry: ToolRegistry,
|
val toolRegistry: ToolRegistry,
|
||||||
|
val sessionUndoService: com.correx.apps.server.undo.SessionUndoService,
|
||||||
// Long-lived scope owned by the module — backs the event-store subscription.
|
// Long-lived scope owned by the module — backs the event-store subscription.
|
||||||
// SupervisorJob so one failure doesn't kill the whole module;
|
// SupervisorJob so one failure doesn't kill the whole module;
|
||||||
// Dispatchers.Default since the work is non-blocking and CPU-light.
|
// Dispatchers.Default since the work is non-blocking and CPU-light.
|
||||||
|
|||||||
@@ -42,6 +42,7 @@ fun Route.sessionRoutes(module: ServerModule) {
|
|||||||
route("/{id}") {
|
route("/{id}") {
|
||||||
getSessionRoute(module)
|
getSessionRoute(module)
|
||||||
cancelSessionRoute(module)
|
cancelSessionRoute(module)
|
||||||
|
undoSessionRoute(module)
|
||||||
getEventsRoute(module)
|
getEventsRoute(module)
|
||||||
webSocket("/stream") {
|
webSocket("/stream") {
|
||||||
val id = call.parameters["id"] ?: return@webSocket
|
val id = call.parameters["id"] ?: return@webSocket
|
||||||
@@ -88,6 +89,15 @@ private fun Route.cancelSessionRoute(module: ServerModule) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun Route.undoSessionRoute(module: ServerModule) {
|
||||||
|
post("/undo") {
|
||||||
|
val id = call.parameters["id"]
|
||||||
|
?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id")
|
||||||
|
val summary = module.sessionUndoService.undo(TypeId(id))
|
||||||
|
call.respond(HttpStatusCode.OK, summary)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private fun Route.getEventsRoute(module: ServerModule) {
|
private fun Route.getEventsRoute(module: ServerModule) {
|
||||||
get("/events") {
|
get("/events") {
|
||||||
val id = call.parameters["id"]
|
val id = call.parameters["id"]
|
||||||
|
|||||||
@@ -0,0 +1,51 @@
|
|||||||
|
package com.correx.apps.server.undo
|
||||||
|
|
||||||
|
import com.correx.core.events.events.FileWrittenEvent
|
||||||
|
import com.correx.core.events.stores.EventStore
|
||||||
|
import com.correx.core.events.types.SessionId
|
||||||
|
import com.correx.infrastructure.tools.filesystem.FileMutationReverser
|
||||||
|
import com.correx.infrastructure.tools.filesystem.RevertResult
|
||||||
|
import kotlinx.serialization.Serializable
|
||||||
|
|
||||||
|
@Serializable
|
||||||
|
data class UndoSummary(
|
||||||
|
val sessionId: String,
|
||||||
|
val reverted: Int,
|
||||||
|
val deleted: Int,
|
||||||
|
val rejected: Int,
|
||||||
|
val failed: Int,
|
||||||
|
val details: List<String> = emptyList(),
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Reverses every file mutation a session performed, newest-first, using only the
|
||||||
|
* event log + CAS (the reverser). Newest-first so a path written multiple times in
|
||||||
|
* the session ends at its pre-session content.
|
||||||
|
*/
|
||||||
|
class SessionUndoService(
|
||||||
|
private val eventStore: EventStore,
|
||||||
|
private val reverser: FileMutationReverser,
|
||||||
|
) {
|
||||||
|
suspend fun undo(sessionId: SessionId): UndoSummary {
|
||||||
|
val mutations = eventStore.read(sessionId)
|
||||||
|
.mapNotNull { it.payload as? FileWrittenEvent }
|
||||||
|
.asReversed()
|
||||||
|
|
||||||
|
var reverted = 0
|
||||||
|
var deleted = 0
|
||||||
|
var rejected = 0
|
||||||
|
var failed = 0
|
||||||
|
val details = mutableListOf<String>()
|
||||||
|
|
||||||
|
for (event in mutations) {
|
||||||
|
when (val result = reverser.revert(event)) {
|
||||||
|
RevertResult.Restored -> { reverted++; details += "restored ${event.path}" }
|
||||||
|
RevertResult.Deleted -> { deleted++; details += "deleted ${event.path}" }
|
||||||
|
is RevertResult.Rejected -> { rejected++; details += "rejected ${event.path}: ${result.reason}" }
|
||||||
|
is RevertResult.Failed -> { failed++; details += "failed ${event.path}: ${result.reason}" }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return UndoSummary(sessionId.value, reverted, deleted, rejected, failed, details)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,127 @@
|
|||||||
|
package com.correx.apps.server.undo
|
||||||
|
|
||||||
|
import com.correx.core.artifactstore.ArtifactStore
|
||||||
|
import com.correx.core.events.events.EventMetadata
|
||||||
|
import com.correx.core.events.events.EventPayload
|
||||||
|
import com.correx.core.events.events.FileWrittenEvent
|
||||||
|
import com.correx.core.events.events.NewEvent
|
||||||
|
import com.correx.core.events.events.StoredEvent
|
||||||
|
import com.correx.core.events.stores.EventStore
|
||||||
|
import com.correx.core.events.types.ArtifactId
|
||||||
|
import com.correx.core.events.types.EventId
|
||||||
|
import com.correx.core.events.types.SessionId
|
||||||
|
import com.correx.core.events.types.ToolInvocationId
|
||||||
|
import com.correx.infrastructure.tools.filesystem.FileMutationReverser
|
||||||
|
import kotlinx.coroutines.flow.Flow
|
||||||
|
import kotlinx.coroutines.flow.emptyFlow
|
||||||
|
import kotlinx.coroutines.runBlocking
|
||||||
|
import kotlinx.datetime.Clock
|
||||||
|
import org.junit.jupiter.api.Assertions.assertEquals
|
||||||
|
import org.junit.jupiter.api.Assertions.assertFalse
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
import java.nio.file.Files
|
||||||
|
|
||||||
|
class SessionUndoServiceTest {
|
||||||
|
|
||||||
|
private class FakeArtifactStore : ArtifactStore {
|
||||||
|
val blobs = linkedMapOf<String, ByteArray>()
|
||||||
|
override suspend fun put(bytes: ByteArray): ArtifactId {
|
||||||
|
val id = ArtifactId(bytes.toString(Charsets.UTF_8).hashCode().toString())
|
||||||
|
blobs[id.value] = bytes
|
||||||
|
return id
|
||||||
|
}
|
||||||
|
override suspend fun get(id: ArtifactId): ByteArray? = blobs[id.value]
|
||||||
|
override suspend fun flushBefore(commit: suspend () -> Unit) = commit()
|
||||||
|
}
|
||||||
|
|
||||||
|
private class FakeEventStore(private val events: List<StoredEvent>) : EventStore {
|
||||||
|
override suspend fun append(event: NewEvent): StoredEvent = error("unused")
|
||||||
|
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused")
|
||||||
|
override fun read(sessionId: SessionId): List<StoredEvent> = events
|
||||||
|
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
|
||||||
|
events.filter { it.sequence >= fromSequence }
|
||||||
|
override fun lastSequence(sessionId: SessionId): Long? = events.lastOrNull()?.sequence
|
||||||
|
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
|
||||||
|
override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
|
||||||
|
override suspend fun lastGlobalSequence(): Long = 0L
|
||||||
|
override fun allEvents(): Sequence<StoredEvent> = emptySequence()
|
||||||
|
override fun allSessionIds(): Set<SessionId> = emptySet()
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun storedEvent(payload: EventPayload, sessionId: SessionId, seq: Long): StoredEvent =
|
||||||
|
StoredEvent(
|
||||||
|
metadata = EventMetadata(
|
||||||
|
eventId = EventId("e$seq"),
|
||||||
|
sessionId = sessionId,
|
||||||
|
timestamp = Clock.System.now(),
|
||||||
|
schemaVersion = 1,
|
||||||
|
causationId = null,
|
||||||
|
correlationId = null,
|
||||||
|
),
|
||||||
|
sequence = seq,
|
||||||
|
sessionSequence = seq,
|
||||||
|
payload = payload,
|
||||||
|
)
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `undo restores a file that was overwritten`(): Unit = runBlocking {
|
||||||
|
val dir = Files.createTempDirectory("undo-test").toRealPath()
|
||||||
|
val target = dir.resolve("file.txt")
|
||||||
|
Files.writeString(target, "NEW")
|
||||||
|
|
||||||
|
val fakeStore = FakeArtifactStore()
|
||||||
|
val preImageHash = fakeStore.put("OLD".toByteArray()).value
|
||||||
|
|
||||||
|
val sessionId = SessionId("session-overwrite")
|
||||||
|
val event = FileWrittenEvent(
|
||||||
|
invocationId = ToolInvocationId("inv-1"),
|
||||||
|
sessionId = sessionId,
|
||||||
|
path = target.toString(),
|
||||||
|
preImageHash = preImageHash,
|
||||||
|
postImageHash = null,
|
||||||
|
preExisted = true,
|
||||||
|
timestampMs = 0L,
|
||||||
|
)
|
||||||
|
val fakeEventStore = FakeEventStore(listOf(storedEvent(event, sessionId, 1L)))
|
||||||
|
val reverser = FileMutationReverser(fakeStore, setOf(dir))
|
||||||
|
val service = SessionUndoService(fakeEventStore, reverser)
|
||||||
|
|
||||||
|
val summary = service.undo(sessionId)
|
||||||
|
|
||||||
|
assertEquals(1, summary.reverted)
|
||||||
|
assertEquals(0, summary.deleted)
|
||||||
|
assertEquals(0, summary.rejected)
|
||||||
|
assertEquals(0, summary.failed)
|
||||||
|
assertEquals("OLD", Files.readString(target))
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `undo deletes a file that was newly created`(): Unit = runBlocking {
|
||||||
|
val dir = Files.createTempDirectory("undo-new").toRealPath()
|
||||||
|
val target = dir.resolve("new.txt")
|
||||||
|
Files.writeString(target, "CREATED")
|
||||||
|
|
||||||
|
val fakeStore = FakeArtifactStore()
|
||||||
|
val sessionId = SessionId("session-create")
|
||||||
|
val event = FileWrittenEvent(
|
||||||
|
invocationId = ToolInvocationId("inv-2"),
|
||||||
|
sessionId = sessionId,
|
||||||
|
path = target.toString(),
|
||||||
|
preImageHash = null,
|
||||||
|
postImageHash = null,
|
||||||
|
preExisted = false,
|
||||||
|
timestampMs = 0L,
|
||||||
|
)
|
||||||
|
val fakeEventStore = FakeEventStore(listOf(storedEvent(event, sessionId, 1L)))
|
||||||
|
val reverser = FileMutationReverser(fakeStore, setOf(dir))
|
||||||
|
val service = SessionUndoService(fakeEventStore, reverser)
|
||||||
|
|
||||||
|
val summary = service.undo(sessionId)
|
||||||
|
|
||||||
|
assertEquals(0, summary.reverted)
|
||||||
|
assertEquals(1, summary.deleted)
|
||||||
|
assertEquals(0, summary.rejected)
|
||||||
|
assertEquals(0, summary.failed)
|
||||||
|
assertFalse(Files.exists(target))
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user