fix(kernel,tools,workflow): session-robustness QA sweep
Uncommitted work from the session-robustness-and-dox branch sweep
(docs/qa/QA-session-robustness-and-dox.md), verified alongside the
compression/context fixes:
- SandboxedToolExecutor: validate tool args centrally before dispatch. A
malformed/missing-arg call becomes a recoverable ERROR: (surfaced with the
tool's arg schema so the model can correct + retry) instead of stranding
the stage with no artifact. + validation test.
- PlanLinter: seed artifacts (analysis) produced by the planning phase count
as available producers, so a plan stage that `needs` them isn't flagged as
an unproduced-need; H1 unproduced-needs + trap-state checks. + tests.
- DefaultSessionOrchestrator: live-QA robustness fixes (event-tail /
per-stage budget + retry handling).
- workflow prompts/configs: DOX AGENTS.md alignment + freestyle/task/role
prompt tweaks.
- SessionOrchestratorIntegrationTest: coverage for the above.
- FreestylePlanningWorkflowTest: allow list_dir in analyst tools (follows the
list_dir wiring in 968cbfa).
- QA plan doc for the branch sweep.
This commit is contained in:
+36
-10
@@ -170,11 +170,13 @@ class DefaultSessionOrchestrator(
|
|||||||
is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) {
|
is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) {
|
||||||
completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount)
|
completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount)
|
||||||
} else {
|
} else {
|
||||||
failWorkflow(
|
// No outgoing edge matched: the stage finished but produced nothing that satisfies a
|
||||||
sessionId = enriched.sessionId,
|
// transition (e.g. the analyst never emitted a validated `analysis`). Treat it as a
|
||||||
stageId = enriched.currentStageId,
|
// retryable stage failure — a fresh attempt lets the model gather more before writing —
|
||||||
reason = "no transition condition matched from stage ${enriched.currentStageId.value}",
|
// rather than killing the run. Only terminal once the retry budget is spent.
|
||||||
retryExhausted = false,
|
retryStageOrFail(
|
||||||
|
enriched,
|
||||||
|
"no transition condition matched from stage ${enriched.currentStageId.value}",
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -185,15 +187,39 @@ class DefaultSessionOrchestrator(
|
|||||||
retryExhausted = false,
|
retryExhausted = false,
|
||||||
)
|
)
|
||||||
|
|
||||||
is TransitionDecision.NoMatch -> failWorkflow(
|
is TransitionDecision.NoMatch -> retryStageOrFail(
|
||||||
sessionId = enriched.sessionId,
|
enriched,
|
||||||
stageId = enriched.currentStageId,
|
"no matching transition from stage ${enriched.currentStageId.value}",
|
||||||
reason = "no matching transition from stage ${enriched.currentStageId.value}",
|
|
||||||
retryExhausted = false,
|
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A non-terminal stage whose outcome matched no outgoing transition (Stay/NoMatch) produced no
|
||||||
|
* usable artifact. Route it back through the stage's retry budget instead of failing the run:
|
||||||
|
* re-execute the same stage if attempts remain, else fail terminally with retryExhausted=true.
|
||||||
|
*/
|
||||||
|
private suspend fun retryStageOrFail(
|
||||||
|
ctx: EnrichedExecutionContext,
|
||||||
|
reason: String,
|
||||||
|
): WorkflowResult {
|
||||||
|
val attempt = orchestrationRepository.getState(ctx.sessionId).retryCount
|
||||||
|
val shouldRetry = retryCoordinator.shouldRetry(
|
||||||
|
sessionId = ctx.sessionId,
|
||||||
|
stageId = ctx.currentStageId,
|
||||||
|
currentAttempt = attempt,
|
||||||
|
policy = ctx.config.retryPolicy,
|
||||||
|
failureReason = reason,
|
||||||
|
)
|
||||||
|
if (!shouldRetry) {
|
||||||
|
return failWorkflow(ctx.sessionId, ctx.currentStageId, reason, retryExhausted = true)
|
||||||
|
}
|
||||||
|
return when (val r = enterStage(ctx, ctx.currentStageId)) {
|
||||||
|
is StepResult.Continue -> step(r.ctx)
|
||||||
|
is StepResult.Terminal -> r.result
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
/** Returns the cached validated artifact content for the given session + artifact id, or null if absent. */
|
/** Returns the cached validated artifact content for the given session + artifact id, or null if absent. */
|
||||||
fun validatedArtifactContent(sessionId: SessionId, artifactId: ArtifactId): String? =
|
fun validatedArtifactContent(sessionId: SessionId, artifactId: ArtifactId): String? =
|
||||||
|
|||||||
@@ -0,0 +1,75 @@
|
|||||||
|
# QA Plan: session-robustness-and-dox — master..feat/session-robustness-and-dox
|
||||||
|
|
||||||
|
**Status:** DRAFT
|
||||||
|
**Run date / operator:**
|
||||||
|
**BACKLOG item:** feat/session-robustness-and-dox full-branch sweep
|
||||||
|
|
||||||
|
Commits under test:
|
||||||
|
```
|
||||||
|
c8c2521 fix(kernel,inference): reliable artifact emission for local models
|
||||||
|
af9da3c fix(inference): raise llama HTTP request timeout 600s -> 1800s
|
||||||
|
5e0c7df feat(server): REST stage-approval route POST /sessions/{id}/approve
|
||||||
|
e72051a feat(kernel): emit_artifact tool + actionable validation feedback
|
||||||
|
0652d87 fix(inference): salvage truncated/malformed tool-call JSON from content
|
||||||
|
4737300 fix(server): anchor static-registry file_read to the workspace
|
||||||
|
82674e8 feat(inference): capability-aware routing for tool-heavy stages
|
||||||
|
238d353 feat(qa): remote NIM provider + headless-QA robustness
|
||||||
|
e0f623a feat(workflow): tasked execution loop replaces role_pipeline
|
||||||
|
a00bd4a feat(tools): propose_scope recalibration valve for the execution loop
|
||||||
|
3e44c6d feat(kernel): deterministic per-task claim stage + per-task write scope
|
||||||
|
c1e4c7b feat(transitions): tasks_ready predicate for execution loop
|
||||||
|
d26f20c docs(dox): build out the DOX AGENTS.md hierarchy
|
||||||
|
1cb7fec feat(kernel): read-only mode after read-before-write + mandatory task
|
||||||
|
201b599 fix(inference): recover tool calls emitted as JSON in message content
|
||||||
|
ccfa4b4 feat(tui): composer soft-wrap, alt+backspace, bracketed paste, borders
|
||||||
|
8abe7d9 fix(tui): collapse multiline tool summaries so action rows don't stripe
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Preconditions
|
||||||
|
|
||||||
|
- [x] **server build/branch:** `feat/session-robustness-and-dox`, `:apps:server:run` on :8080 — _rebuild only with the server stopped_
|
||||||
|
- [x] **llama-server + model:** router = gemma-4-12B-it-qat-UD-Q4_K_XL :10000 (tool-capable, ~48 tok/s), narrator = LFM2.5-1.2B :10001
|
||||||
|
- [ ] **external deps:** none required for the core loop; capability-aware routing (§C8) is only meaningfully exercised if **two** tool-capable providers are configured — otherwise it's a single-candidate no-op. OpenAI-compat/NIM path (238d353) needs a remote endpoint; **out of scope** unless one is wired.
|
||||||
|
- [ ] **config synced:** `~/.config/correx/workflows/` must contain the NEW `task_planning.toml` + `prompts/task_planner.md` and the REWRITTEN `role_pipeline.toml` (execution loop). Copy from `examples/workflows/` before running — stale role_pipeline = wrong graph.
|
||||||
|
- [ ] **fixtures/seed:** a real workspace with a `.correx/project.toml` and a small, genuinely-actionable request (so `task_decompose` produces >1 task and the loop iterates).
|
||||||
|
|
||||||
|
## Acceptance gate (one sentence)
|
||||||
|
|
||||||
|
> The branch is correct **iff** a real request drives the execution loop end-to-end —
|
||||||
|
> planner emits a task graph, each task is deterministically claimed, writes are scoped to
|
||||||
|
> the claimed task's `affected_paths` (and blocked outside it / in read-only mode), artifacts
|
||||||
|
> emit reliably on the local model, and a headless `POST /approve` clears an approval gate —
|
||||||
|
> every step evidenced by a recorded event.
|
||||||
|
|
||||||
|
## Checks
|
||||||
|
|
||||||
|
| # | Action | Expected observable evidence | Result |
|
||||||
|
|---|--------|------------------------------|--------|
|
||||||
|
| 1 | Start a `task_planning` session on a real request | `correx events <id>` shows the `planner` stage; a `task_decompose` tool invocation succeeds and task-created events appear; stage cannot complete without it (`require_task_decompose` gate) — try a run where the model skips decompose and confirm the stage is **blocked**, not completed | |
|
||||||
|
| 2 | Start a `role_pipeline` (execution loop) session | Stage chain analyst→architect→decomposer→implementer→reviewer in the log; `decomposer-to-implementer` fires on `tasks_ready` (not always_true); loop re-enters implementer via `review-2-more` while tasks remain, exits via `review-3-done` when drained | |
|
||||||
|
| 3 | Watch the per-task claim stage (`claimTask=true`) | On each implementer entry, a claim record for the next ready task + its context bundle injected as **L0** (`sourceType=claimedTask`); the loop advances by claim events, not model memory — verify the claimed task id changes across iterations | |
|
||||||
|
| 4 | Let the implementer write a file inside the claimed task's `affected_paths` | `FileWriteTool` invocation ASSESSED and allowed; file appears on disk under the workspace | |
|
||||||
|
| 5 | Force a write **outside** the claimed scope | `ToolCallAssessedEvent` blocks it (scope violation); the write does not land on disk; feedback surfaced to the stage | |
|
||||||
|
| 6 | Trigger read-only mode (a stage that reads before any task is claimed / after read-before-write) | `isReadOnlyMode` denies a subsequent write: `ToolCallAssessedEvent` with the read-only denial; no file mutation event | |
|
||||||
|
| 7 | Have the implementer call `propose_scope` with a `paths`/`reason` | Tool records a scope-widen proposal on the task; **operator approval required** (approval gate fires); on approve, paths added to `affected_paths` and the previously-blocked write now lands | |
|
||||||
|
| 8 | Confirm artifact emission on the local model (analysis/design/patch) | Each `*_validated` transition fires because the artifact actually emitted — `artifact_validated` events present; no stage stuck retrying because the artifact never materialized (the c8c2521 fix) | |
|
||||||
|
| 9 | Feed a truncated/malformed tool-call (or observe a real one from gemma) | Salvage path recovers it: the tool call executes instead of the turn failing; log shows the salvage of JSON-in-content (0652d87 / 201b599) | |
|
||||||
|
| 10 | With a session suspended on an approval, `POST /sessions/{id}/approve` `{"decision":"APPROVE"}` (no requestId) | HTTP 200; server resolves the pending requestId itself; the gate clears and the session advances — no WebSocket needed. Also test `REJECT` (session takes reject path) and a bad `decision` → 400, and no-pending → 404 | |
|
||||||
|
| 11 | `POST /approve` with `{"decision":"STEER","steeringNote":"..."}` | Treated as APPROVE with the note attached; note recorded on the approval response | |
|
||||||
|
| 12 | (If 2 tool-capable providers configured) run a tool-heavy stage | Routing selects the higher tool-capability provider, not just any healthy one; check `/metrics/tool-reliability` reflects observed per-model reliability | |
|
||||||
|
| 13 | TUI composer smoke: soft-wrap a long line, alt+backspace, bracketed paste | Composer wraps within borders; alt+backspace deletes a word; pasted multiline text lands as one block; multiline tool summaries render collapsed (no striped action rows) | |
|
||||||
|
| 14 | `correx replay <id>` on a completed loop session | Double-read digest matches — the claim/scope/read-only logic is derived from recorded events, replay-stable | |
|
||||||
|
|
||||||
|
## Out of scope (explicitly NOT covered this pass)
|
||||||
|
|
||||||
|
- OpenAI-compat / remote NIM provider (238d353) — no remote endpoint wired; unit-tested only.
|
||||||
|
- Capability-aware routing real differentiation (§C12) unless a second tool-capable provider is configured; single-provider makes it a no-op.
|
||||||
|
- The DOX AGENTS.md hierarchy (d26f20c) — docs only, nothing to exercise at runtime.
|
||||||
|
- Layer-2 LLM annotation of tool-intent; the AMD/ROCm perf envelope.
|
||||||
|
|
||||||
|
## Disposition
|
||||||
|
|
||||||
|
- **PASS** → move the BACKLOG entry into `RETRO.md` with run date + cited evidence. Set Status: PASSED.
|
||||||
|
- **FAIL** → file each miss as a numbered `BACKLOG.md` finding (action + repro + wrong/missing signal), fix, re-run only failed checks. Status: FAILED until green.
|
||||||
@@ -15,7 +15,7 @@ id = "execute_script"
|
|||||||
prompt = "prompts/healthcheck_execute.md"
|
prompt = "prompts/healthcheck_execute.md"
|
||||||
needs = ["healthcheck_script"]
|
needs = ["healthcheck_script"]
|
||||||
produces = [{ name = "healthcheck_result", kind = "process_result" }]
|
produces = [{ name = "healthcheck_result", kind = "process_result" }]
|
||||||
allowed_tools = ["ShellTool"]
|
allowed_tools = ["shell"]
|
||||||
token_budget = 2048
|
token_budget = 2048
|
||||||
|
|
||||||
[[transitions]]
|
[[transitions]]
|
||||||
|
|||||||
@@ -19,11 +19,13 @@ Then frame the work as a task (per the task policy):
|
|||||||
|
|
||||||
Either way later stages thread the named task through the plan; the rest wait to be claimed.
|
Either way later stages thread the named task through the plan; the rest wait to be claimed.
|
||||||
|
|
||||||
Emit the `analysis` artifact (JSON, schema provided):
|
Produce the `analysis` artifact by calling the **`emit_artifact`** tool with these fields:
|
||||||
- `summary`: the goal in your own words.
|
- `summary`: the goal in your own words.
|
||||||
- `requirements`: concrete, checkable requirements, one per line.
|
- `requirements`: concrete, checkable requirements, one per line.
|
||||||
- `affected_areas`: files/modules likely involved, one per line.
|
- `affected_areas`: files/modules likely involved, one per line.
|
||||||
|
|
||||||
|
Call `emit_artifact` once you have read enough — do not write the JSON as a plain message.
|
||||||
|
|
||||||
If — and only if — something genuinely blocks a plan (an ambiguous goal, a missing decision, a
|
If — and only if — something genuinely blocks a plan (an ambiguous goal, a missing decision, a
|
||||||
fork only the user can resolve), add a `questions` array. Each entry is an object:
|
fork only the user can resolve), add a `questions` array. Each entry is an object:
|
||||||
- `prompt` (required): the question, in full.
|
- `prompt` (required): the question, in full.
|
||||||
@@ -32,6 +34,11 @@ fork only the user can resolve), add a `questions` array. Each entry is an objec
|
|||||||
- `multiSelect` (optional, default false): true if more than one option may apply.
|
- `multiSelect` (optional, default false): true if more than one option may apply.
|
||||||
- `header` (optional): a 1–2 word label for the question (e.g. "Scope", "Stack").
|
- `header` (optional): a 1–2 word label for the question (e.g. "Scope", "Stack").
|
||||||
|
|
||||||
|
Greenfield or new-surface work (a new UI, app, or module) almost always hides such a fork even
|
||||||
|
when the high-level goal is clear: the build tool, styling approach, state/routing libraries, or
|
||||||
|
component library are choices only the user can pin, and a placeholder instruction does not
|
||||||
|
resolve them. Ask about stack/tooling in that case rather than guessing.
|
||||||
|
|
||||||
Example:
|
Example:
|
||||||
```json
|
```json
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ already covers the goal, load it with `task_context` and name it in your summary
|
|||||||
|
|
||||||
## Step 2 — Sweep the codebase (read-only)
|
## Step 2 — Sweep the codebase (read-only)
|
||||||
|
|
||||||
Use `file_read` (also lists a directory when given a directory path), `ShellTool` (grep, find, ls)
|
Use `file_read` (also lists a directory when given a directory path), `shell` (grep, find, ls)
|
||||||
to locate the files, modules, and patterns the request touches. Read broadly enough to understand:
|
to locate the files, modules, and patterns the request touches. Read broadly enough to understand:
|
||||||
- what already exists
|
- what already exists
|
||||||
- what the request changes or adds
|
- what the request changes or adds
|
||||||
|
|||||||
@@ -36,7 +36,7 @@ start = "analyst"
|
|||||||
id = "analyst"
|
id = "analyst"
|
||||||
prompt = "prompts/analyst.md"
|
prompt = "prompts/analyst.md"
|
||||||
produces = [{ name = "analysis", kind = "analysis" }]
|
produces = [{ name = "analysis", kind = "analysis" }]
|
||||||
allowed_tools = ["file_read", "ShellTool", "task_search", "task_context", "task_create"]
|
allowed_tools = ["file_read", "shell", "task_search", "task_context", "task_create"]
|
||||||
ground_references = true
|
ground_references = true
|
||||||
token_budget = 16384
|
token_budget = 16384
|
||||||
max_retries = 2
|
max_retries = 2
|
||||||
@@ -58,7 +58,7 @@ max_retries = 2
|
|||||||
id = "decomposer"
|
id = "decomposer"
|
||||||
prompt = "prompts/task_planner.md"
|
prompt = "prompts/task_planner.md"
|
||||||
needs = ["design"]
|
needs = ["design"]
|
||||||
allowed_tools = ["file_read", "ShellTool", "task_search", "task_context", "task_decompose", "task_create"]
|
allowed_tools = ["file_read", "shell", "task_search", "task_context", "task_decompose", "task_create"]
|
||||||
require_task_decompose = true
|
require_task_decompose = true
|
||||||
token_budget = 16384
|
token_budget = 16384
|
||||||
max_retries = 2
|
max_retries = 2
|
||||||
@@ -73,7 +73,7 @@ prompt = "prompts/implementer.md"
|
|||||||
produces = [{ name = "patch", kind = "file_written" }]
|
produces = [{ name = "patch", kind = "file_written" }]
|
||||||
claim_task = true
|
claim_task = true
|
||||||
allowed_tools = [
|
allowed_tools = [
|
||||||
"file_read", "file_write", "file_edit", "ShellTool", "propose_scope",
|
"file_read", "file_write", "file_edit", "shell", "propose_scope",
|
||||||
"task_create", "task_update", "task_context", "task_search",
|
"task_create", "task_update", "task_context", "task_search",
|
||||||
]
|
]
|
||||||
token_budget = 32768
|
token_budget = 32768
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ start = "planner"
|
|||||||
[[stages]]
|
[[stages]]
|
||||||
id = "planner"
|
id = "planner"
|
||||||
prompt = "prompts/task_planner.md"
|
prompt = "prompts/task_planner.md"
|
||||||
allowed_tools = ["file_read", "ShellTool", "task_search", "task_context", "task_decompose", "task_create"]
|
allowed_tools = ["file_read", "shell", "task_search", "task_context", "task_decompose", "task_create"]
|
||||||
require_task_decompose = true
|
require_task_decompose = true
|
||||||
token_budget = 16384
|
token_budget = 16384
|
||||||
max_retries = 2
|
max_retries = 2
|
||||||
|
|||||||
+14
@@ -18,6 +18,7 @@ import com.correx.core.tools.contract.FileAffectingTool
|
|||||||
import com.correx.core.tools.contract.Tool
|
import com.correx.core.tools.contract.Tool
|
||||||
import com.correx.core.tools.contract.ToolExecutor
|
import com.correx.core.tools.contract.ToolExecutor
|
||||||
import com.correx.core.tools.contract.ToolResult
|
import com.correx.core.tools.contract.ToolResult
|
||||||
|
import com.correx.core.tools.contract.ValidationResult
|
||||||
import com.correx.core.tools.registry.ToolRegistry
|
import com.correx.core.tools.registry.ToolRegistry
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
import kotlinx.coroutines.withContext
|
import kotlinx.coroutines.withContext
|
||||||
@@ -55,6 +56,19 @@ class SandboxedToolExecutor(
|
|||||||
// 2. emit started
|
// 2. emit started
|
||||||
emitStarted(sessionId, invocationId, toolName)
|
emitStarted(sessionId, invocationId, toolName)
|
||||||
|
|
||||||
|
// 3. validate args centrally. A malformed/missing-arg call is the model's mistake, not a
|
||||||
|
// fatal stage failure: return recoverable so the orchestrator surfaces it as ERROR: + the
|
||||||
|
// tool's arg schema and the model can correct + retry (bounded by MAX_TOOL_ROUNDS), instead
|
||||||
|
// of stranding the stage with no artifact ("no transition condition matched").
|
||||||
|
(tool.validateRequest(request) as? ValidationResult.Invalid)?.let { invalid ->
|
||||||
|
emitFailed(sessionId, invocationId, toolName, invalid.reason)
|
||||||
|
return@withContext ToolResult.Failure(
|
||||||
|
invocationId = invocationId,
|
||||||
|
reason = invalid.reason,
|
||||||
|
recoverable = true,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
// 4. create working dir
|
// 4. create working dir
|
||||||
val workingDir = workDir.resolve(sessionId.value).resolve(invocationId.value)
|
val workingDir = workDir.resolve(sessionId.value).resolve(invocationId.value)
|
||||||
Files.createDirectories(workingDir)
|
Files.createDirectories(workingDir)
|
||||||
|
|||||||
+87
@@ -0,0 +1,87 @@
|
|||||||
|
package com.correx.infrastructure.tools
|
||||||
|
|
||||||
|
import com.correx.core.approvals.Tier
|
||||||
|
import com.correx.core.events.EventDispatcher
|
||||||
|
import com.correx.core.events.events.EventPayload
|
||||||
|
import com.correx.core.events.events.NewEvent
|
||||||
|
import com.correx.core.events.events.StoredEvent
|
||||||
|
import com.correx.core.events.events.ToolExecutionFailedEvent
|
||||||
|
import com.correx.core.events.events.ToolRequest
|
||||||
|
import com.correx.core.events.stores.EventStore
|
||||||
|
import com.correx.core.events.types.SessionId
|
||||||
|
import com.correx.core.events.types.StageId
|
||||||
|
import com.correx.core.events.types.ToolInvocationId
|
||||||
|
import com.correx.core.tools.contract.ParamRole
|
||||||
|
import com.correx.core.tools.contract.Tool
|
||||||
|
import com.correx.core.tools.contract.ToolCapability
|
||||||
|
import com.correx.core.tools.contract.ToolExecutor
|
||||||
|
import com.correx.core.tools.contract.ToolResult
|
||||||
|
import com.correx.core.tools.contract.ValidationResult
|
||||||
|
import com.correx.core.tools.registry.ToolRegistry
|
||||||
|
import kotlinx.coroutines.flow.Flow
|
||||||
|
import kotlinx.coroutines.flow.emptyFlow
|
||||||
|
import kotlinx.coroutines.runBlocking
|
||||||
|
import kotlinx.serialization.json.JsonObject
|
||||||
|
import org.junit.jupiter.api.Assertions.assertEquals
|
||||||
|
import org.junit.jupiter.api.Assertions.assertTrue
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
import java.nio.file.Files
|
||||||
|
|
||||||
|
class SandboxedToolExecutorValidationTest {
|
||||||
|
|
||||||
|
private class CapturingEventStore : EventStore {
|
||||||
|
val payloads = mutableListOf<EventPayload>()
|
||||||
|
override suspend fun append(event: NewEvent): StoredEvent {
|
||||||
|
payloads += event.payload
|
||||||
|
val seq = payloads.size.toLong()
|
||||||
|
return StoredEvent(event.metadata, seq, seq, event.payload)
|
||||||
|
}
|
||||||
|
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = events.map { append(it) }
|
||||||
|
override fun read(sessionId: SessionId): List<StoredEvent> = emptyList()
|
||||||
|
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = emptyList()
|
||||||
|
override fun lastSequence(sessionId: SessionId): Long? = null
|
||||||
|
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
|
||||||
|
override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
|
||||||
|
override suspend fun lastGlobalSequence(): Long = payloads.size.toLong()
|
||||||
|
override fun allEvents(): Sequence<StoredEvent> = emptySequence()
|
||||||
|
override fun allSessionIds(): Set<SessionId> = emptySet()
|
||||||
|
}
|
||||||
|
|
||||||
|
private class SingleToolRegistry(private val tool: Tool) : ToolRegistry {
|
||||||
|
override fun resolve(name: String): Tool? = if (name == tool.name) tool else null
|
||||||
|
override fun all(): List<Tool> = listOf(tool)
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A tool that rejects every call at validation and would explode if execute() were reached. */
|
||||||
|
private class AlwaysInvalidTool : Tool, ToolExecutor {
|
||||||
|
override val name: String = "picky"
|
||||||
|
override val description: String = "fake"
|
||||||
|
override val tier: Tier = Tier.T2
|
||||||
|
override val requiredCapabilities: Set<ToolCapability> = emptySet()
|
||||||
|
override val paramRoles: Map<String, ParamRole> = emptyMap()
|
||||||
|
override val parametersSchema: JsonObject = JsonObject(emptyMap())
|
||||||
|
override fun validateRequest(request: ToolRequest): ValidationResult =
|
||||||
|
ValidationResult.Invalid("Missing or empty 'argv' parameter. Expected List<String>.")
|
||||||
|
override suspend fun execute(request: ToolRequest): ToolResult =
|
||||||
|
error("execute must not run when validation fails")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `bad args fail recoverably and never reach execute`() {
|
||||||
|
val tool = AlwaysInvalidTool()
|
||||||
|
val events = CapturingEventStore()
|
||||||
|
val exec = SandboxedToolExecutor(
|
||||||
|
delegate = tool,
|
||||||
|
registry = SingleToolRegistry(tool),
|
||||||
|
eventDispatcher = EventDispatcher(events),
|
||||||
|
workDir = Files.createTempDirectory("sbx-validate"),
|
||||||
|
)
|
||||||
|
val result = runBlocking {
|
||||||
|
exec.execute(ToolRequest(ToolInvocationId("inv"), SessionId("s"), StageId("st"), "picky", emptyMap()))
|
||||||
|
}
|
||||||
|
|
||||||
|
result as ToolResult.Failure
|
||||||
|
assertTrue(result.recoverable, "validation failures must be recoverable so the model can correct + retry")
|
||||||
|
assertEquals(1, events.payloads.filterIsInstance<ToolExecutionFailedEvent>().size)
|
||||||
|
}
|
||||||
|
}
|
||||||
+13
-6
@@ -20,7 +20,7 @@ data class PlanLintResult(
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Deterministic, pure-Kotlin lint over a compiled plan graph (plan-pipeline-spec §5). Zero inference,
|
* Deterministic, pure-Kotlin lint over a compiled plan graph (plan-pipeline-spec §5). Zero inference,
|
||||||
* zero I/O — a function of the graph alone, so it is replay-safe by construction. It complements
|
* zero I/O — a function of the graph and a fixed seed set, so it is replay-safe by construction. It complements
|
||||||
* [ExecutionPlanCompiler] (which already throws on unreachable-from-start / unknown-kind / bad-edge):
|
* [ExecutionPlanCompiler] (which already throws on unreachable-from-start / unknown-kind / bad-edge):
|
||||||
* the lint adds the checks the compiler does NOT make.
|
* the lint adds the checks the compiler does NOT make.
|
||||||
*
|
*
|
||||||
@@ -32,15 +32,22 @@ object PlanLinter {
|
|||||||
private const val STAGE_CEILING = 12
|
private const val STAGE_CEILING = 12
|
||||||
private const val FAN_OUT_THRESHOLD = 4
|
private const val FAN_OUT_THRESHOLD = 4
|
||||||
|
|
||||||
fun lint(graph: WorkflowGraph): PlanLintResult =
|
/**
|
||||||
|
* Artifacts that pre-exist in the session before the execution plan runs — produced by the
|
||||||
|
* planning phase, not by any plan stage. The architect prompt explicitly allows `needs` to
|
||||||
|
* reference these (notably `analysis`), so the linter must treat them as available producers.
|
||||||
|
*/
|
||||||
|
private val seedArtifacts = setOf("analysis")
|
||||||
|
|
||||||
|
fun lint(graph: WorkflowGraph, seeds: Set<String> = seedArtifacts): PlanLintResult =
|
||||||
PlanLintResult(
|
PlanLintResult(
|
||||||
hardFailures = unproducedNeeds(graph) + trapStates(graph),
|
hardFailures = unproducedNeeds(graph, seeds) + trapStates(graph),
|
||||||
softFindings = stageCount(graph) + fanOut(graph) + emptyBriefs(graph) + duplicateBriefs(graph),
|
softFindings = stageCount(graph) + fanOut(graph) + emptyBriefs(graph) + duplicateBriefs(graph),
|
||||||
)
|
)
|
||||||
|
|
||||||
/** H1: a stage `needs` an artifact that no stage `produces`. */
|
/** H1: a stage `needs` an artifact that neither a plan stage `produces` nor a seed provides. */
|
||||||
private fun unproducedNeeds(graph: WorkflowGraph): List<PlanLintFinding> {
|
private fun unproducedNeeds(graph: WorkflowGraph, seeds: Set<String>): List<PlanLintFinding> {
|
||||||
val produced = graph.stages.values.flatMap { it.produces }.map { it.name.value }.toSet()
|
val produced = graph.stages.values.flatMap { it.produces }.map { it.name.value }.toSet() + seeds
|
||||||
return graph.stages.entries.flatMap { (id, stage) ->
|
return graph.stages.entries.flatMap { (id, stage) ->
|
||||||
stage.needs.map { it.value }.filterNot { it in produced }.map { need ->
|
stage.needs.map { it.value }.filterNot { it in produced }.map { need ->
|
||||||
PlanLintFinding(
|
PlanLintFinding(
|
||||||
|
|||||||
+1
-1
@@ -73,7 +73,7 @@ class FreestylePlanningWorkflowTest {
|
|||||||
// analyst frames the work: search + open one task or decompose into a graph (the architect
|
// analyst frames the work: search + open one task or decompose into a graph (the architect
|
||||||
// threads the named task into the plan).
|
// threads the named task into the plan).
|
||||||
assertEquals(
|
assertEquals(
|
||||||
setOf("file_read", "ShellTool", "task_search", "task_context", "task_create", "task_decompose"),
|
setOf("file_read", "list_dir", "shell", "task_search", "task_context", "task_create", "task_decompose"),
|
||||||
graph.stages[StageId("analyst")]!!.allowedTools,
|
graph.stages[StageId("analyst")]!!.allowedTools,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|||||||
+14
@@ -78,6 +78,20 @@ class PlanLinterTest {
|
|||||||
assertTrue(h.detail.contains("z"))
|
assertTrue(h.detail.contains("z"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `a need satisfied by the analysis seed is not a failure`() {
|
||||||
|
val g = graph(
|
||||||
|
stages = mapOf(
|
||||||
|
"a" to stage(produces = listOf("x"), needs = listOf("analysis"), prompt = "research"),
|
||||||
|
"b" to stage(needs = listOf("x"), prompt = "build"),
|
||||||
|
),
|
||||||
|
edges = listOf("a" to "b", "b" to "done"),
|
||||||
|
start = "a",
|
||||||
|
)
|
||||||
|
val result = PlanLinter.lint(g)
|
||||||
|
assertTrue(result.passed, "analysis is a planning-phase seed, not an unproduced need")
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun `a cycle with no exit is a trap-state hard failure`() {
|
fun `a cycle with no exit is a trap-state hard failure`() {
|
||||||
val g = graph(
|
val g = graph(
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ import com.correx.core.events.events.OrchestrationPausedEvent
|
|||||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||||
import com.correx.core.events.events.ToolInvocationRequestedEvent
|
import com.correx.core.events.events.ToolInvocationRequestedEvent
|
||||||
import com.correx.core.events.events.WorkflowCompletedEvent
|
import com.correx.core.events.events.WorkflowCompletedEvent
|
||||||
|
import com.correx.core.events.events.RetryAttemptedEvent
|
||||||
import com.correx.core.events.events.WorkflowFailedEvent
|
import com.correx.core.events.events.WorkflowFailedEvent
|
||||||
import com.correx.core.events.events.WorkflowStartedEvent
|
import com.correx.core.events.events.WorkflowStartedEvent
|
||||||
import com.correx.core.events.events.ToolRequest
|
import com.correx.core.events.events.ToolRequest
|
||||||
@@ -70,6 +71,7 @@ import com.correx.core.sessions.projections.replay.EventReplayer
|
|||||||
import com.correx.core.tools.contract.ToolExecutor
|
import com.correx.core.tools.contract.ToolExecutor
|
||||||
import com.correx.core.tools.contract.ToolResult
|
import com.correx.core.tools.contract.ToolResult
|
||||||
import com.correx.core.transitions.evaluation.PromptResolver
|
import com.correx.core.transitions.evaluation.PromptResolver
|
||||||
|
import com.correx.core.transitions.resolution.DefaultTransitionResolver
|
||||||
import com.correx.core.transitions.graph.StageConfig
|
import com.correx.core.transitions.graph.StageConfig
|
||||||
import com.correx.core.transitions.graph.TransitionEdge
|
import com.correx.core.transitions.graph.TransitionEdge
|
||||||
import com.correx.core.transitions.graph.WorkflowGraph
|
import com.correx.core.transitions.graph.WorkflowGraph
|
||||||
@@ -214,6 +216,46 @@ class SessionOrchestratorIntegrationTest {
|
|||||||
assertTrue(failed.retryExhausted)
|
assertTrue(failed.retryExhausted)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `stage that matches no transition retries through the budget before failing`(): Unit = runBlocking {
|
||||||
|
val sessionId = SessionId("s-no-transition-retry")
|
||||||
|
// Stage A succeeds (no produces, no tools) but its only outgoing edge never matches, so the
|
||||||
|
// resolver returns Stay. That must consume the retry budget, not fail the run on first sight.
|
||||||
|
val graph = WorkflowGraph(
|
||||||
|
id = "no-transition-retry-test",
|
||||||
|
stages = mapOf(StageId("A") to StageConfig()),
|
||||||
|
transitions = setOf(
|
||||||
|
TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { false }),
|
||||||
|
),
|
||||||
|
start = StageId("A"),
|
||||||
|
)
|
||||||
|
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 2, backoffMs = 0))
|
||||||
|
// The default fixture evaluator returns true unconditionally; use one that honours the
|
||||||
|
// condition so the { false } edge actually produces a Stay.
|
||||||
|
val orchestrator = DefaultSessionOrchestrator(
|
||||||
|
repositories = repositories,
|
||||||
|
engines = engines.copy(
|
||||||
|
transitionResolver = DefaultTransitionResolver { condition, ctx -> condition.evaluate(ctx) },
|
||||||
|
),
|
||||||
|
retryCoordinator = retryCoordinator,
|
||||||
|
artifactStore = artifactStore,
|
||||||
|
decisionJournalRepository = decisionJournalRepository,
|
||||||
|
)
|
||||||
|
|
||||||
|
orchestrator.run(sessionId, graph, config)
|
||||||
|
|
||||||
|
val events = eventStore.read(sessionId)
|
||||||
|
val retries = events.count { it.payload is RetryAttemptedEvent }
|
||||||
|
assertEquals(2, retries, "expected the no-transition outcome to exhaust the retry budget")
|
||||||
|
val failed = events.find { it.payload is WorkflowFailedEvent }?.payload as? WorkflowFailedEvent
|
||||||
|
assertNotNull(failed)
|
||||||
|
assertTrue(failed!!.retryExhausted, "expected retryExhausted=true once the budget is spent")
|
||||||
|
assertTrue(
|
||||||
|
failed.reason.contains("no transition condition matched"),
|
||||||
|
"unexpected failure reason: ${failed.reason}",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun `workflow pauses for approval when required`(): Unit = runBlocking {
|
fun `workflow pauses for approval when required`(): Unit = runBlocking {
|
||||||
val sessionId = SessionId("s3")
|
val sessionId = SessionId("s3")
|
||||||
|
|||||||
Reference in New Issue
Block a user