From 4e085100450182fa83db48111376d05286ec0711 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 21 Jun 2026 08:07:33 +0000 Subject: [PATCH] =?UTF-8?q?feat(server):=20activate=20architect=20contradi?= =?UTF-8?q?ction-check=20hook=20(B=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the display-only ArchitectContradictionChecker (eae0a0c) live: ServerModule.start() subscribes to the architect stage's `design` ArtifactCreatedEvent, resolves the decision text (prefers the design schema's `approach` field; falls back to the capped artifact text), runs checker.check(...), and appends the non-null PossibleContradictionFlaggedEvent. Live-only and never halts a stage (runCatching). Checker built in Main.kt gated on project.enabled (same L3 "project:" namespace ProjectMemoryService.persist populates) and threaded into ServerModule as a nullable param. Replaces the standing TODO(wiring) in Main.kt. Co-Authored-By: Claude Opus 4.8 --- .../kotlin/com/correx/apps/server/Main.kt | 21 +- .../com/correx/apps/server/ServerModule.kt | 95 ++++++ .../server/ArchitectContradictionHookTest.kt | 276 ++++++++++++++++++ 3 files changed, 382 insertions(+), 10 deletions(-) create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/ArchitectContradictionHookTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 4cbc18e5..ea2cdad8 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -380,16 +380,16 @@ fun main() { privilegedLocations = privilegedLocations, allowedWorkspaceRoots = allowedWorkspaceRoots, ) - // TODO(wiring): ArchitectContradictionChecker (BACKLOG §B-§4) is complete but not yet live. - // `embedder` + `l3MemoryStore` are in scope here, but two preconditions are missing: - // (1) no server-side hook exposes the architect's decision *text* with an event-emit path - // (architect is a workflow-defined role/stage, not a known stage id; decisions become - // DecisionRecords via the journal reducer from generic events), and - // (2) decisions are only embedded into L3 at session end (ProjectMemoryService.persist), - // under turnId "project:", so an in-session architect run has nothing of its - // own to retrieve. Wiring this would mean inventing an architect-stage subscriber + - // emit path — deliberately left out per scope. To go live: subscribe to the architect - // decision, call ArchitectContradictionChecker.check(...), and emit the non-null result. + // Display-only architect contradiction surfacing (BACKLOG §B-§4): now live via the + // ServerModule.start() subscription on the architect's `design` ArtifactCreatedEvent (see + // ServerModule.handleArchitectArtifact). Gated on project.enabled because the checker queries the + // same L3 "project:" namespace that ProjectMemoryService.persist populates. + val architectContradictionChecker: com.correx.apps.server.memory.ArchitectContradictionChecker? = + if (correxConfig.project.enabled) { + com.correx.apps.server.memory.ArchitectContradictionChecker(embedder, l3MemoryStore) + } else { + null + } // Built from a config snapshot and reused by ConfigService's rebuild hook so toggling // project.enabled / personalization.* applies live to the next session. fun buildProjectMemory(cfg: CorrexConfig): com.correx.apps.server.memory.ProjectMemoryService? = @@ -477,6 +477,7 @@ fun main() { workspaceResolver = workspaceResolver, narrationMaxPerRun = correxConfig.router.narration.maxPerRun, projectMemory = projectMemory, + architectContradictionChecker = architectContradictionChecker, configHolder = configHolder, freestyleDriver = freestyleDriver, operatorProfile = operatorProfile, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index c5ea422f..269395a5 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -13,11 +13,15 @@ import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.config.OperatorProfile import com.correx.core.config.ProjectProfileLoader +import com.correx.apps.server.memory.ArchitectContradictionChecker import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.ArtifactContentStoredEvent +import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.NewEvent import com.correx.core.events.events.ExecutionPlanLockedEvent import com.correx.core.events.events.OperatorProfileBoundEvent +import com.correx.core.events.events.PossibleContradictionFlaggedEvent import com.correx.core.events.events.ProjectProfileBoundEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent @@ -26,6 +30,9 @@ import com.correx.core.events.stores.EventStore import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.jsonObject import com.correx.core.kernel.execution.WorkflowResult import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator import com.correx.core.kernel.orchestration.OrchestrationConfig @@ -87,6 +94,10 @@ class ServerModule( // Cross-session, repo-scoped memory. Null disables the hooks entirely (tests / static path). // Rebuilt by ConfigService when project.enabled is toggled live. projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = null, + // Display-only architect contradiction surfacing (BACKLOG §B-§4). Null disables the hook + // (tests / project.enabled=false). Unlike projectMemory there is no live config-rebuild hook: + // a server restart re-reads project.enabled, which is enough for this informational flag. + private val architectContradictionChecker: ArchitectContradictionChecker? = null, // Live, swappable config. Null only in tests that don't exercise config editing; defaults to a // holder seeded from defaults so callers always have a value to read. val configHolder: com.correx.core.config.ConfigHolder = @@ -187,6 +198,80 @@ class ServerModule( // Continuous, edge-triggered health watch (observability-spec §4). Live-only like narration: // probes read the environment and record degraded/restored events; replay reads those facts. healthMonitor?.start(moduleScope) + + // Display-only architect contradiction surfacing (BACKLOG §B-§4). Live-only like the hooks + // above: subscribeAll() replays nothing and ServerModule is never built under replay, so a + // restart never re-fires the flag (invariant #8). Never halts a stage — failures are logged + // and swallowed (runCatching) and the flag is emitted purely for the operator to eyeball. + architectContradictionChecker?.let { checker -> + eventStore.subscribeAll() + .filter { + it.payload is ArtifactCreatedEvent && + (it.payload as ArtifactCreatedEvent).stageId.value == ARCHITECT_STAGE_ID + } + .onEach { stored -> + runCatching { + handleArchitectArtifact(checker, stored.payload as ArtifactCreatedEvent) + }.onFailure { log.warn("architect contradiction check failed: {}", it.message) } + } + .launchIn(moduleScope) + } + } + + /** + * Runs the architect contradiction check for a freshly-created `design` artifact and, when the + * checker surfaces a non-null flag, appends the [PossibleContradictionFlaggedEvent]. Display-only: + * the flag is informational and never halts the stage. Factored out of the subscription so the + * extraction/dispatch path is unit-testable without a live event flow. + * + * @return the appended flag, or null when the artifact text could not be resolved or the checker + * found nothing near the architect's decision. + */ + internal suspend fun handleArchitectArtifact( + checker: ArchitectContradictionChecker, + event: ArtifactCreatedEvent, + ): PossibleContradictionFlaggedEvent? { + val decisionText = resolveArchitectDecisionText(event) ?: return null + val flag = checker.check(event.sessionId, event.stageId, decisionText) ?: return null + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(java.util.UUID.randomUUID().toString()), + sessionId = event.sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = flag, + ), + ) + return flag + } + + /** + * Resolves the architect's decision text for [event] from the CAS-stored `design` artifact. + * Finds the content hash by scanning the session for the [ArtifactContentStoredEvent] matching + * the artifact id (emitted at inference time, before [ArtifactCreatedEvent]), fetches the bytes, + * and prefers the design schema's `approach` field ("the chosen approach and why") as the + * decision summary; falls back to the whole artifact text, capped at [DECISION_TEXT_CAP] chars. + * + * @return the decision text, or null when no content hash, bytes, or text could be resolved. + */ + private suspend fun resolveArchitectDecisionText(event: ArtifactCreatedEvent): String? { + val contentHash = eventStore.read(event.sessionId) + .asSequence() + .map { it.payload } + .filterIsInstance() + .firstOrNull { it.artifactId == event.artifactId } + ?.contentHash ?: return null + val text = artifactStore.get(contentHash)?.toString(Charsets.UTF_8)?.takeIf { it.isNotBlank() } + ?: return null + val approach = runCatching { + (lenientJson.parseToJsonElement(text).jsonObject["approach"] as? JsonPrimitive) + ?.takeIf { it.isString }?.content + }.getOrNull()?.takeIf { it.isNotBlank() } + return (approach ?: text).take(DECISION_TEXT_CAP) } /** @@ -438,4 +523,14 @@ class ServerModule( .forEach { req -> approvalCoordinator.registerPendingRequest(req.id, sessionId) } } } + + companion object { + /** Workflow stage id that produces the `design` artifact (examples/workflows/role_pipeline.toml). */ + const val ARCHITECT_STAGE_ID = "architect" + + /** Cap on the architect decision text embedded when no `approach` field is present. */ + private const val DECISION_TEXT_CAP = 4000 + + private val lenientJson = Json { ignoreUnknownKeys = true; isLenient = true } + } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/ArchitectContradictionHookTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/ArchitectContradictionHookTest.kt new file mode 100644 index 00000000..a07b3289 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/ArchitectContradictionHookTest.kt @@ -0,0 +1,276 @@ +package com.correx.apps.server + +import com.correx.apps.server.memory.ArchitectContradictionChecker +import com.correx.apps.server.registry.ProviderRegistry +import com.correx.apps.server.registry.WorkflowRegistry +import com.correx.apps.server.registry.WorkflowSummary +import com.correx.apps.server.undo.SessionUndoService +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.context.builder.DefaultContextPackBuilder +import com.correx.core.context.compression.DefaultContextCompressor +import com.correx.core.events.EventDispatcher +import com.correx.core.events.events.ArtifactContentStoredEvent +import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.PossibleContradictionFlaggedEvent +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.ProviderId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.inference.Embedder +import com.correx.core.inference.InferenceProjector +import com.correx.core.inference.InferenceProvider +import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.ProviderHealth +import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer +import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.kernel.orchestration.OrchestrationConfig +import com.correx.core.kernel.orchestration.OrchestrationProjector +import com.correx.core.kernel.orchestration.OrchestrationRepository +import com.correx.core.kernel.orchestration.OrchestratorEngines +import com.correx.core.kernel.orchestration.OrchestratorRepositories +import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.risk.DefaultRiskAssessor +import com.correx.core.router.l3.L3Hit +import com.correx.core.router.l3.L3MemoryEntry +import com.correx.core.router.l3.L3MemoryStore +import com.correx.core.router.l3.L3Query +import com.correx.core.router.model.RouterConfig +import com.correx.core.sessions.DefaultSessionReducer +import com.correx.core.sessions.DefaultSessionRepository +import com.correx.core.sessions.SessionProjector +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.core.transitions.resolution.DefaultTransitionResolver +import com.correx.core.validation.pipeline.ValidationPipeline +import com.correx.infrastructure.InfrastructureModule +import com.correx.infrastructure.inference.commons.UnavailableProbe +import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.infrastructure.tools.FileEditConfig +import com.correx.infrastructure.tools.FileReadConfig +import com.correx.infrastructure.tools.FileWriteConfig +import com.correx.infrastructure.tools.ShellConfig +import com.correx.infrastructure.tools.ToolConfig +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.nio.file.Path +import java.util.UUID + +/** Always returns the same vector so the canned L3 store fully controls the scores. */ +private class OnesEmbedder(override val dimension: Int = 8) : Embedder { + override suspend fun embed(text: String): FloatArray = FloatArray(dimension) { 1f } +} + +/** Returns canned hits regardless of the query vector. */ +private class CannedL3MemoryStore(private val hits: List) : L3MemoryStore { + override suspend fun store(entry: L3MemoryEntry) = Unit + override suspend fun query(query: L3Query): List = hits.sortedByDescending { it.score }.take(query.k) + override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = hits.any { it.entry.turnId.startsWith(prefix) } + override suspend fun close() = Unit +} + +/** Minimal in-memory CAS keyed by content hash. */ +private class MapArtifactStore(private val bytesByHash: Map) : ArtifactStore { + override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("00".repeat(32)) + override suspend fun get(id: ArtifactId): ByteArray? = bytesByHash[id.value] + override suspend fun flushBefore(commit: suspend () -> Unit) { commit() } +} + +/** + * Exercises [ServerModule.handleArchitectArtifact]: the extraction/dispatch helper that backs the + * display-only architect contradiction subscription (BACKLOG §B-§4). A full subscription/integration + * test is not required — these assert the helper fetches the design artifact text and emits (or + * skips) the flag. The surrounding module is built with in-memory infra (no real model needed). + */ +class ArchitectContradictionHookTest { + + private val session = SessionId("new-session") + private val priorSession = SessionId("prior-session") + private val architectStage = StageId("architect") + private val designArtifact = ArtifactId("design") + private val contentHash = ArtifactId("cafebabe") + + private fun append(es: EventStore, payload: EventPayload, sid: SessionId = session) = runBlocking { + es.append( + NewEvent( + EventMetadata(EventId(UUID.randomUUID().toString()), sid, Clock.System.now(), 1, null, null), + payload, + ), + ) + } + + private fun priorDecisionHit(text: String, score: Float) = L3Hit( + entry = L3MemoryEntry( + id = UUID.randomUUID().toString(), + sessionId = priorSession, + turnId = "project:/repo", + text = text, + vector = FloatArray(8) { 1f }, + timestampMs = 0L, + ), + score = score, + ) + + private fun flagsIn(es: EventStore): List = + es.read(session).map { it.payload }.filterIsInstance() + + private fun buildModule( + eventStore: EventStore, + artifactStore: ArtifactStore, + checker: ArchitectContradictionChecker?, + tempDir: Path, + ): ServerModule { + val provider = InfrastructureModule.createLlamaCppProvider( + modelId = "test-model", + modelPath = "/dev/null", + baseUrl = "http://127.0.0.1:1", + ) + val providerRegistry = InfrastructureModule.createProviderRegistry(listOf(provider)) + val inferenceRouter = com.correx.core.inference.DefaultInferenceRouter( + providerRegistry, + com.correx.infrastructure.inference.FirstAvailableRoutingStrategy(), + ) + val toolConfig = ToolConfig( + shell = ShellConfig(enabled = false, allowedExecutables = emptySet(), workingDir = tempDir), + fileRead = FileReadConfig(enabled = false, allowedPaths = setOf(tempDir)), + fileWrite = FileWriteConfig(enabled = false, allowedPaths = setOf(tempDir), workingDir = tempDir), + fileEdit = FileEditConfig(enabled = false, allowedPaths = setOf(tempDir), workingDir = tempDir), + ) + val toolRegistry = InfrastructureModule.createToolRegistry(toolConfig) + val toolExecutor = InfrastructureModule.createToolExecutor( + registry = toolRegistry, + eventDispatcher = EventDispatcher(eventStore), + workDir = tempDir, + artifactStore = null, + ) + val engines = OrchestratorEngines( + transitionResolver = DefaultTransitionResolver { condition, ctx -> condition.evaluate(ctx) }, + contextPackBuilder = DefaultContextPackBuilder(DefaultContextCompressor()), + inferenceRouter = inferenceRouter, + validationPipeline = ValidationPipeline(validators = emptyList()), + approvalEngine = com.correx.core.approvals.domain.DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + toolRegistry = toolRegistry, + toolExecutor = toolExecutor, + ) + val repositories = OrchestratorRepositories( + eventStore = eventStore, + inferenceRepository = InferenceRepository(DefaultEventReplayer(eventStore, InferenceProjector())), + orchestrationRepository = OrchestrationRepository( + DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())), + ), + sessionRepository = DefaultSessionRepository( + DefaultEventReplayer(eventStore, SessionProjector(DefaultSessionReducer())), + ), + artifactRepository = InfrastructureModule.createArtifactRepository(eventStore), + approvalRepository = InfrastructureModule.createApprovalRepository(eventStore), + ) + val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = artifactStore, + tokenizer = provider.tokenizer, + decisionJournalRepository = InfrastructureModule.createDecisionJournalRepository(eventStore), + ) + val routerFacade = InfrastructureModule.createRouterFacade( + eventStore = eventStore, + inferenceRouter = inferenceRouter, + config = RouterConfig(), + tokenizer = provider.tokenizer, + ) + val noopProviderRegistry = object : ProviderRegistry { + override fun listAll() = emptyList() + override suspend fun healthCheckAll() = emptyMap() + } + val noopWorkflowRegistry = object : WorkflowRegistry { + override fun listAll(): List = emptyList() + override fun find(workflowId: String): WorkflowGraph? = null + } + return ServerModule( + orchestrator = orchestrator, + eventStore = eventStore, + artifactStore = artifactStore, + sessionRepository = repositories.sessionRepository, + workflowRegistry = noopWorkflowRegistry, + providerRegistry = noopProviderRegistry, + defaultOrchestrationConfig = OrchestrationConfig(sandboxRoot = tempDir), + routerFacade = routerFacade, + orchestrationRepository = repositories.orchestrationRepository, + approvalRepository = repositories.approvalRepository, + toolRegistry = toolRegistry, + sessionUndoService = SessionUndoService(eventStore, artifactStore, bootRoots = setOf(tempDir)), + resourceProbe = UnavailableProbe, + architectContradictionChecker = checker, + ) + } + + @Test + fun `emits the flag when the checker surfaces a near prior decision`(@TempDir tempDir: Path) = runBlocking { + val es = InMemoryEventStore() + val designJson = """{"approach":"Use Postgres for the event store.","components":["db.kt"]}""" + val artifacts = MapArtifactStore(mapOf(contentHash.value to designJson.toByteArray())) + // ArtifactContentStored precedes ArtifactCreated for the same artifactId (inference-time). + append(es, ArtifactContentStoredEvent(designArtifact, contentHash, session, architectStage)) + val created = ArtifactCreatedEvent(designArtifact, session, architectStage, schemaVersion = 1) + append(es, created) + val checker = ArchitectContradictionChecker( + OnesEmbedder(), + CannedL3MemoryStore(listOf(priorDecisionHit("Decided to use SQLite for the event store.", 0.9f))), + ) + + val flag = buildModule(es, artifacts, checker, tempDir).handleArchitectArtifact(checker, created) + + assertNotNull(flag, "expected a contradiction flag") + assertEquals(session, flag!!.sessionId) + assertEquals(architectStage, flag.stageId) + // The design schema's `approach` field is preferred as the decision text. + assertEquals("Use Postgres for the event store.", flag.decisionSummary) + assertEquals("Decided to use SQLite for the event store.", flag.related.single().summary) + // The flag is appended to the session stream (display-only, but recorded for the operator). + assertEquals(1, flagsIn(es).size) + } + + @Test + fun `returns null and appends nothing when the checker finds no related decision`( + @TempDir tempDir: Path, + ) = runBlocking { + val es = InMemoryEventStore() + val designJson = """{"approach":"Use Postgres for the event store.","components":["db.kt"]}""" + val artifacts = MapArtifactStore(mapOf(contentHash.value to designJson.toByteArray())) + append(es, ArtifactContentStoredEvent(designArtifact, contentHash, session, architectStage)) + val created = ArtifactCreatedEvent(designArtifact, session, architectStage, schemaVersion = 1) + append(es, created) + val checker = ArchitectContradictionChecker(OnesEmbedder(), CannedL3MemoryStore(emptyList())) + + val flag = buildModule(es, artifacts, checker, tempDir).handleArchitectArtifact(checker, created) + + assertNull(flag) + assertEquals(0, flagsIn(es).size) + } + + @Test + fun `returns null when no content hash was recorded for the artifact`(@TempDir tempDir: Path) = runBlocking { + val es = InMemoryEventStore() + val created = ArtifactCreatedEvent(designArtifact, session, architectStage, schemaVersion = 1) + append(es, created) + val checker = ArchitectContradictionChecker( + OnesEmbedder(), + CannedL3MemoryStore(listOf(priorDecisionHit("Decided to use SQLite.", 0.9f))), + ) + + val flag = buildModule(es, MapArtifactStore(emptyMap()), checker, tempDir) + .handleArchitectArtifact(checker, created) + + assertNull(flag) + } +}