/pentestgpt_agent/tests/test_loop.py

import json
from collections.abc import AsyncIterator
from pathlib import Path

import pytest
from unified_agent import (
    AgentEvent,
    CommandRun,
    RunOptions,
    SandboxPolicy,
    ToolCall,
    ToolResult,
    TurnCompleted,
    UnifiedAgent,
)

from pentestgpt_agent.agents import (
    EXECUTOR_INSTRUCTIONS,
    SUPERVISOR_INSTRUCTIONS,
    AgentContractError,
    Executor,
    Supervisor,
)
from pentestgpt_agent.execution import ExecutionOutcome, ValidExecution
from pentestgpt_agent.loop import PentestLoop
from pentestgpt_agent.memory import MemoryKernel, RunSpec, RunStatus
from pentestgpt_agent.plan import (
    SupervisorDecision,
    TaskKind,
    TaskProposal,
    TaskStatus,
    compile_plan,
)
from pentestgpt_agent.trace import EpisodeRunner, TraceStore


def _latest_observation_basis(prompt: str) -> list[str]:
    state = json.loads(prompt.split("\n\n", 1)[1])
    return [state["observations"][-1]["id"]]


class TwoDecisionSupervisorBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        if self.calls == 0:
            output = {
                "base_revision": 0,
                "new_tasks": [
                    {
                        "id": "discover-http",
                        "kind": "discover",
                        "target": "http://127.0.0.1:8080",
                        "objective": "Inspect the HTTP service.",
                        "done_when": "The HTTP response is recorded.",
                        "basis_ids": [],
                        "depends_on": [],
                    }
                ],
                "next_task_id": "discover-http",
                "finish": False,
                "finish_basis_ids": [],
                "summary": "Begin with HTTP discovery.",
            }
        else:
            output = {
                "base_revision": 2,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": _latest_observation_basis(prompt),
                "summary": "The bounded assessment is complete.",
            }
        self.calls += 1
        yield TurnCompleted(success=True, structured_output=output)


class ThreeDecisionSupervisorBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        if self.calls == 0:
            output = {
                "base_revision": 0,
                "new_tasks": [
                    {
                        "id": "discover-http",
                        "kind": "discover",
                        "target": "http://127.0.0.1:8080",
                        "objective": "Inspect the HTTP service.",
                        "done_when": "The HTTP response is recorded.",
                        "basis_ids": [],
                        "depends_on": [],
                    }
                ],
                "next_task_id": "discover-http",
                "finish": False,
                "finish_basis_ids": [],
                "summary": "Begin with HTTP discovery.",
            }
        elif self.calls == 1:
            output = {
                "base_revision": 2,
                "new_tasks": [],
                "next_task_id": "discover-http",
                "finish": False,
                "finish_basis_ids": [],
                "summary": "Retry the ready task with its grounded fallback receipt.",
            }
        else:
            output = {
                "base_revision": 4,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": _latest_observation_basis(prompt),
                "summary": "The retried task completed.",
            }
        self.calls += 1
        yield TurnCompleted(success=True, structured_output=output)


class GroundedExecutorBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        assert opts.output_schema is not None
        envelope = json.loads(prompt.split("\n\n", 1)[1])
        assert envelope["goal"] == "Assess the authorized target."
        assert opts.output_schema["properties"]["outcome"]["enum"] == [
            "done",
            "progress",
            "blocked",
            "failed",
        ]
        outcome_contract = opts.output_schema["properties"]["outcome"]["description"]
        assert "done_when is met" in outcome_contract
        assert "external prerequisite" in outcome_contract
        assert "premise is falsified" in outcome_contract
        assert "observation" not in opts.output_schema["properties"]
        evidence_contract = opts.output_schema["properties"]["evidence_excerpt"]["description"]
        assert "One verbatim contiguous substring" in evidence_contract
        assert "Do not add labels" in evidence_contract
        yield CommandRun(command="curl http://127.0.0.1:8080", exit_code=0, output="200 OK")
        yield TurnCompleted(
            success=True,
            structured_output={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": "The HTTP service responded successfully.",
                "evidence_excerpt": "200 OK",
            },
        )


class ActionThenMaxTurnsBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield CommandRun(
            command="find / -iname '*flag*'",
            exit_code=0,
            output="partial target output",
        )
        yield TurnCompleted(
            success=False,
            stop_reason="error_max_turns",
            error="Reached maximum number of turns (2)",
        )
        raise RuntimeError("Command failed with exit code 1")


class RetrySupervisorBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        if self.calls == 0:
            output = {
                "base_revision": 0,
                "new_tasks": [
                    {
                        "id": "discover-http",
                        "kind": "discover",
                        "target": "http://127.0.0.1:8080",
                        "objective": "Inspect the HTTP service.",
                        "done_when": "The HTTP response is recorded.",
                        "basis_ids": [],
                        "depends_on": [],
                    }
                ],
                "next_task_id": "discover-http",
                "finish": False,
                "finish_basis_ids": [],
                "summary": "Begin with HTTP discovery.",
            }
        else:
            output = {
                "base_revision": 3,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": _latest_observation_basis(prompt),
                "summary": "The retried task completed.",
            }
        self.calls += 1
        yield TurnCompleted(success=True, structured_output=output)


class CrashThenSucceedBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        self.calls += 1
        if self.calls == 1:
            raise RuntimeError("provider unavailable")
        yield CommandRun(command="curl http://127.0.0.1:8080", exit_code=0, output="200 OK")
        yield TurnCompleted(
            success=True,
            structured_output={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": "The HTTP service responded successfully.",
                "evidence_excerpt": "200 OK",
            },
        )


class ActionThenMalformedResultBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield CommandRun(command="curl http://127.0.0.1:8080", exit_code=0, output="200 OK")
        yield TurnCompleted(success=True, structured_output={"unexpected": "shape"})


class ActionThenUnsupportedEvidenceBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        self.calls += 1
        if self.calls == 1:
            yield CommandRun(
                command="curl http://127.0.0.1:8080",
                exit_code=0,
                output="200 OK",
            )
        yield TurnCompleted(
            success=True,
            structured_output={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": (
                    "Claimed an uncaptured credential."
                    if self.calls == 1
                    else "Corrected the result using the grounded receipt."
                ),
                "evidence_excerpt": "admin_password=secret" if self.calls == 1 else "200 OK",
            },
        )


class MustNotRunBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        raise AssertionError("provider must not run for a terminal persisted run")
        yield  # pragma: no cover


class FinishRevisionTwoBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield TurnCompleted(
            success=True,
            structured_output={
                "base_revision": 2,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": _latest_observation_basis(prompt),
                "summary": "The recovered evidence completed the run.",
            },
        )


class AlwaysCrashBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        raise RuntimeError("provider unavailable during interrupted run")
        yield  # pragma: no cover


class FinishRevisionThreeBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield TurnCompleted(
            success=True,
            structured_output={
                "base_revision": 3,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": _latest_observation_basis(prompt),
                "summary": "The recovered retry completed the run.",
            },
        )


class FinishAfterRecoveredSupervisorBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        state = json.loads(prompt.split("\n\n", 1)[1])
        assert state["revision"] == 2
        self.calls += 1
        yield TurnCompleted(
            success=True,
            structured_output={
                "base_revision": 2,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": [state["observations"][-1]["id"]],
                "summary": "The recovered Supervisor plan completed successfully.",
            },
        )


class InvalidThenValidSupervisorBackend:
    name = "scripted"

    def __init__(self) -> None:
        self.calls = 0

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        if self.calls == 0:
            output = {
                "base_revision": 0,
                "new_tasks": [
                    {
                        "id": "outside-scope",
                        "kind": "discover",
                        "target": "http://192.0.2.1",
                        "objective": "Inspect an invalid target.",
                        "done_when": "A response is recorded.",
                        "basis_ids": [],
                        "depends_on": [],
                    }
                ],
                "next_task_id": "outside-scope",
                "finish": False,
                "finish_basis_ids": [],
                "summary": "This proposal must be rejected.",
            }
        elif self.calls == 1:
            output = {
                "base_revision": 0,
                "new_tasks": [
                    {
                        "id": "discover-http",
                        "kind": "discover",
                        "target": "http://127.0.0.1:8080",
                        "objective": "Inspect the HTTP service.",
                        "done_when": "The HTTP response is recorded.",
                        "basis_ids": [],
                        "depends_on": [],
                    }
                ],
                "next_task_id": "discover-http",
                "finish": False,
                "finish_basis_ids": [],
                "summary": "Begin with scoped discovery.",
            }
        else:
            output = {
                "base_revision": 2,
                "new_tasks": [],
                "next_task_id": None,
                "finish": True,
                "finish_basis_ids": _latest_observation_basis(prompt),
                "summary": "The bounded assessment is complete.",
            }
        self.calls += 1
        yield TurnCompleted(success=True, structured_output=output)


class AlwaysInvalidSupervisorBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield TurnCompleted(success=True, structured_output={"invalid": "decision"})


class MalformedStructuredTransportBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield CommandRun(
            command="read-authorized-target",
            exit_code=0,
            output="200 OK FLAG{transport-recovery}",
        )
        yield ToolCall(
            name="StructuredOutput",
            input={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": (
                    "Captured the exact target result.</summary>\n"
                    '<parameter name="evidence_excerpt">FLAG{transport-recovery}'
                ),
            },
            call_id="structured-1",
        )
        yield ToolResult(
            call_id="structured-1",
            output=(
                "Output does not match required schema: root: must have required property "
                "'evidence_excerpt'"
            ),
            is_error=True,
        )
        yield TurnCompleted(
            success=False,
            stop_reason="error_max_structured_output_retries",
            error="Failed to provide valid structured output after 5 attempts",
        )


class MaxTurnsStructuredTransportBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield CommandRun(
            command="read-authorized-target",
            exit_code=0,
            output="200 OK FLAG{max-turn-recovery}",
        )
        yield ToolCall(
            name="StructuredOutput",
            input={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": "Captured the exact target result before the provider turn limit.",
                "evidence_excerpt": "FLAG{max-turn-recovery}",
            },
            call_id="structured-max-turn",
        )
        yield ToolResult(
            call_id="structured-max-turn",
            output="Reached maximum number of turns",
            is_error=True,
        )
        yield TurnCompleted(
            success=False,
            stop_reason="error_max_turns",
            error="Reached maximum number of turns (6)",
        )


class MaxTurnsMultipleStructuredCandidatesBackend:
    name = "scripted"

    async def stream(self, prompt: str, opts: RunOptions) -> AsyncIterator[AgentEvent]:
        yield CommandRun(
            command="fingerprint-authorized-target",
            exit_code=0,
            output="service=Roundcube 1.6.16",
        )
        yield ToolCall(
            name="StructuredOutput",
            input={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": (
                    "Captured the exact service version.</summary>\n"
                    '<parameter name="evidence_excerpt">service=Roundcube 1.6.16'
                ),
            },
            call_id="structured-grounded",
        )
        yield ToolResult(
            call_id="structured-grounded",
            output=(
                "Output does not match required schema: root: must have required property "
                "'evidence_excerpt'"
            ),
            is_error=True,
        )
        yield ToolCall(
            name="StructuredOutput",
            input={
                "task_id": "discover-http",
                "outcome": "done",
                "summary": (
                    "Retried with an approximate version.</summary>\n"
                    '<parameter name="evidence_excerpt">service=Roundcube 1.6.x'
                ),
            },
            call_id="structured-ungrounded",
        )
        yield ToolResult(
            call_id="structured-ungrounded",
            output=(
                "Output does not match required schema: root: must have required property "
                "'evidence_excerpt'"
            ),
            is_error=True,
        )
        yield TurnCompleted(
            success=False,
            stop_reason="error_max_turns",
            error="Reached maximum number of turns (7)",
        )


@pytest.mark.asyncio
async def test_one_grounded_task_can_drive_a_run_to_completion(tmp_path: Path) -> None:
    assert "fewest actions" in EXECUTOR_INSTRUCTIONS
    assert "one minimal goal-bearing substitution" in EXECUTOR_INSTRUCTIONS
    assert "Task kind is a hard boundary" in EXECUTOR_INSTRUCTIONS
    assert "DISCOVER maps the surface without vulnerability payloads" in EXECUTOR_INSTRUCTIONS
    assert "Never recursively search /" in EXECUTOR_INSTRUCTIONS
    assert "smallest complete exact quote" in EXECUTOR_INSTRUCTIONS
    assert "actionable endpoint, parameter, or result" in EXECUTOR_INSTRUCTIONS
    assert "If that quote does not directly demonstrate done_when, do not return done" in (
        EXECUTOR_INSTRUCTIONS
    )
    assert "single-token redirection or IFS-style payload" in EXECUTOR_INSTRUCTIONS
    traces = TraceStore(tmp_path / "runs")
    supervisor = Supervisor(
        EpisodeRunner(
            UnifiedAgent(
                TwoDecisionSupervisorBackend(),
                workspace=tmp_path / "supervisor",
                sandbox=SandboxPolicy.READ_ONLY,
                instructions=SUPERVISOR_INSTRUCTIONS,
            ),
            traces,
        )
    )
    executor = Executor(
        EpisodeRunner(
            UnifiedAgent(
                GroundedExecutorBackend(),
                workspace=tmp_path / "executor",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        )
    )
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    loop = PentestLoop(memory=memory, supervisor=supervisor, executor=executor, traces=traces)

    result = await loop.run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert result.revision == 3
    assert len(result.tasks) == 1
    assert result.tasks[0].status is TaskStatus.DONE
    assert len(result.observations) == 1
    assert result.observations[0].statement == "200 OK"
    assert result.observations[0].evidence_sequences == (1,)

    next_supervisor_trace = traces.load("run-1", "supervisor-r2")
    next_supervisor_prompt = next_supervisor_trace.input["rendered_prompt"]
    assert "200 OK" in next_supervisor_prompt
    assert '"recent_diagnostics"' in next_supervisor_prompt
    assert "The HTTP service responded successfully." not in next_supervisor_prompt
    assert "curl http://127.0.0.1:8080" not in next_supervisor_prompt


@pytest.mark.asyncio
async def test_max_turn_failure_after_an_action_is_atomically_terminal(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    supervisor = Supervisor(
        EpisodeRunner(
            UnifiedAgent(
                TwoDecisionSupervisorBackend(),
                workspace=tmp_path / "supervisor",
                sandbox=SandboxPolicy.READ_ONLY,
                instructions=SUPERVISOR_INSTRUCTIONS,
            ),
            traces,
        )
    )
    executor = Executor(
        EpisodeRunner(
            UnifiedAgent(
                ActionThenMaxTurnsBackend(),
                workspace=tmp_path / "executor",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        ),
        max_turns=2,
    )
    memory = MemoryKernel(tmp_path / "state.sqlite3")

    result = await PentestLoop(
        memory=memory,
        supervisor=supervisor,
        executor=executor,
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.FAILED
    assert result.revision == 2
    assert result.tasks[0].status is TaskStatus.FAILED
    assert len(result.attempts) == 1
    assert result.attempts[0].status == "error"
    assert result.attempts[0].failure_kind == "max_turns"
    assert result.attempts[0].failure_message == "Reached maximum number of turns (2)"
    assert result.attempts[0].finished_revision == 2
    assert result.observations == ()
    assert [(transition.revision, transition.kind) for transition in result.transitions] == [
        (0, "run_created"),
        (1, "plan_committed"),
        (2, "attempt_failed"),
    ]
    assert result.transitions[-1].detail == {
        "failure_kind": "max_turns",
        "failure_message": "Reached maximum number of turns (2)",
        "retry_attempt_id": None,
        "terminal": True,
    }


@pytest.mark.asyncio
async def test_provider_failure_before_actions_retries_once_in_the_same_run(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    supervisor_backend = RetrySupervisorBackend()
    executor_backend = CrashThenSucceedBackend()
    supervisor = Supervisor(
        EpisodeRunner(
            UnifiedAgent(
                supervisor_backend,
                workspace=tmp_path / "supervisor",
                sandbox=SandboxPolicy.READ_ONLY,
                instructions=SUPERVISOR_INSTRUCTIONS,
            ),
            traces,
        )
    )
    executor = Executor(
        EpisodeRunner(
            UnifiedAgent(
                executor_backend,
                workspace=tmp_path / "executor",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        )
    )

    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=supervisor,
        executor=executor,
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert result.revision == 4
    assert result.tasks[0].status is TaskStatus.DONE
    assert [attempt.status for attempt in result.attempts] == ["error", "done"]
    assert result.attempts[0].failure_kind == "provider"
    assert result.attempts[0].failure_message == "RuntimeError: provider unavailable"
    assert result.attempts[1].failure_kind is None
    assert len(result.observations) == 1
    assert executor_backend.calls == 2
    assert supervisor_backend.calls == 2


@pytest.mark.asyncio
async def test_malformed_result_after_an_action_is_a_terminal_validation_failure(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    supervisor = Supervisor(
        EpisodeRunner(
            UnifiedAgent(
                TwoDecisionSupervisorBackend(),
                workspace=tmp_path / "supervisor",
                sandbox=SandboxPolicy.READ_ONLY,
                instructions=SUPERVISOR_INSTRUCTIONS,
            ),
            traces,
        )
    )
    executor = Executor(
        EpisodeRunner(
            UnifiedAgent(
                ActionThenMalformedResultBackend(),
                workspace=tmp_path / "executor",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        )
    )

    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=supervisor,
        executor=executor,
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.FAILED
    assert result.tasks[0].status is TaskStatus.FAILED
    assert result.attempts[0].status == "error"
    assert result.attempts[0].failure_kind == "validation"
    assert result.attempts[0].failure_message == (
        "Executor result has unexpected or missing fields"
    )
    assert result.observations == ()


@pytest.mark.asyncio
async def test_uncaptured_done_evidence_degrades_to_grounded_progress_and_retries(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    supervisor_backend = ThreeDecisionSupervisorBackend()
    supervisor = Supervisor(
        EpisodeRunner(
            UnifiedAgent(
                supervisor_backend,
                workspace=tmp_path / "supervisor",
                sandbox=SandboxPolicy.READ_ONLY,
                instructions=SUPERVISOR_INSTRUCTIONS,
            ),
            traces,
        )
    )
    executor_backend = ActionThenUnsupportedEvidenceBackend()
    executor = Executor(
        EpisodeRunner(
            UnifiedAgent(
                executor_backend,
                workspace=tmp_path / "executor",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        )
    )

    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=supervisor,
        executor=executor,
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert result.tasks[0].status is TaskStatus.DONE
    assert [attempt.status for attempt in result.attempts] == ["progress", "done"]
    assert [observation.statement for observation in result.observations] == ["200 OK"]
    assert result.transitions[2].detail["evidence_fallback"] is True
    reused = [
        transition
        for transition in result.transitions
        if transition.detail.get("reused_observation_id")
    ]
    assert [transition.detail["reused_observation_id"] for transition in reused] == [
        result.observations[0].id
    ]
    assert executor_backend.calls == 2
    assert supervisor_backend.calls == 3


@pytest.mark.asyncio
async def test_reopening_a_completed_run_does_not_invoke_a_provider(tmp_path: Path) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    first_result = await PentestLoop(
        memory=memory,
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    TwoDecisionSupervisorBackend(),
                    workspace=tmp_path / "supervisor-first",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    GroundedExecutorBackend(),
                    workspace=tmp_path / "executor-first",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)
    must_not_run = MustNotRunBackend()

    reopened = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    must_not_run,
                    workspace=tmp_path / "supervisor-reopen",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    must_not_run,
                    workspace=tmp_path / "executor-reopen",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert reopened == first_result


@pytest.mark.asyncio
async def test_restart_commits_a_completed_executor_trace_without_replaying_it(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    plan_commit = memory.commit_plan(
        compile_plan(
            SupervisorDecision(
                base_revision=0,
                new_tasks=(
                    TaskProposal(
                        "discover-http",
                        TaskKind.DISCOVER,
                        "http://127.0.0.1:8080",
                        "Inspect the HTTP service.",
                        "The HTTP response is recorded.",
                    ),
                ),
                next_task_id="discover-http",
                finish=False,
                summary="Begin discovery.",
            ),
            initial,
        )
    )
    assert plan_commit.lease is not None
    active = memory.snapshot("run-1")
    episode_id = f"executor-{plan_commit.lease.attempt_id}"
    await Executor(
        EpisodeRunner(
            UnifiedAgent(
                GroundedExecutorBackend(),
                workspace=tmp_path / "executor-before-crash",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        )
    ).execute(active, plan_commit.lease, episode_id=episode_id)
    assert memory.snapshot("run-1").attempts[0].status == "active"

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    FinishRevisionTwoBackend(),
                    workspace=tmp_path / "supervisor-after-crash",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "executor-after-crash",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert recovered.status is RunStatus.COMPLETED
    assert recovered.revision == 3
    assert recovered.tasks[0].status is TaskStatus.DONE
    assert [attempt.status for attempt in recovered.attempts] == ["done"]
    assert len(recovered.observations) == 1


@pytest.mark.asyncio
async def test_restart_retries_a_failed_executor_trace_only_when_it_has_no_actions(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    plan_commit = memory.commit_plan(
        compile_plan(
            SupervisorDecision(
                base_revision=0,
                new_tasks=(
                    TaskProposal(
                        "discover-http",
                        TaskKind.DISCOVER,
                        "http://127.0.0.1:8080",
                        "Inspect the HTTP service.",
                        "The HTTP response is recorded.",
                    ),
                ),
                next_task_id="discover-http",
                finish=False,
                summary="Begin discovery.",
            ),
            initial,
        )
    )
    assert plan_commit.lease is not None
    with pytest.raises(AgentContractError, match="provider unavailable"):
        await Executor(
            EpisodeRunner(
                UnifiedAgent(
                    AlwaysCrashBackend(),
                    workspace=tmp_path / "executor-before-crash",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ).execute(
            memory.snapshot("run-1"),
            plan_commit.lease,
            episode_id=f"executor-{plan_commit.lease.attempt_id}",
        )

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    FinishRevisionThreeBackend(),
                    workspace=tmp_path / "supervisor-after-crash",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    GroundedExecutorBackend(),
                    workspace=tmp_path / "executor-after-crash",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert recovered.status is RunStatus.COMPLETED
    assert recovered.revision == 4
    assert [attempt.status for attempt in recovered.attempts] == ["error", "done"]
    assert recovered.attempts[0].failure_kind == "provider"
    assert recovered.attempts[0].failure_message == (
        "RuntimeError: provider unavailable during interrupted run"
    )


@pytest.mark.asyncio
async def test_restart_recovers_a_terminal_executor_event_when_output_write_was_interrupted(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    plan_commit = memory.commit_plan(
        compile_plan(
            SupervisorDecision(
                base_revision=0,
                new_tasks=(
                    TaskProposal(
                        "discover-http",
                        TaskKind.DISCOVER,
                        "http://127.0.0.1:8080",
                        "Inspect the HTTP service.",
                        "The HTTP response is recorded.",
                    ),
                ),
                next_task_id="discover-http",
                finish=False,
                summary="Begin discovery.",
            ),
            initial,
        )
    )
    assert plan_commit.lease is not None
    episode_id = f"executor-{plan_commit.lease.attempt_id}"
    await Executor(
        EpisodeRunner(
            UnifiedAgent(
                GroundedExecutorBackend(),
                workspace=tmp_path / "executor-before-crash",
                sandbox=SandboxPolicy.WORKSPACE_WRITE,
                instructions=EXECUTOR_INSTRUCTIONS,
            ),
            traces,
        )
    ).execute(
        memory.snapshot("run-1"),
        plan_commit.lease,
        episode_id=episode_id,
    )
    output_path = tmp_path / "runs" / "run-1" / "traces" / episode_id / "output.json"
    output_path.unlink()

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    FinishRevisionTwoBackend(),
                    workspace=tmp_path / "supervisor-after-crash",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "executor-after-crash",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert recovered.status is RunStatus.COMPLETED
    assert recovered.tasks[0].status is TaskStatus.DONE
    assert recovered.attempts[0].status == "done"
    assert recovered.observations[0].statement == "200 OK"


@pytest.mark.asyncio
async def test_restart_retries_when_crash_created_only_the_executor_episode_directory(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    plan_commit = memory.commit_plan(
        compile_plan(
            SupervisorDecision(
                base_revision=0,
                new_tasks=(
                    TaskProposal(
                        "discover-http",
                        TaskKind.DISCOVER,
                        "http://127.0.0.1:8080",
                        "Inspect the HTTP service.",
                        "The HTTP response is recorded.",
                    ),
                ),
                next_task_id="discover-http",
                finish=False,
                summary="Begin discovery.",
            ),
            initial,
        )
    )
    assert plan_commit.lease is not None
    episode_id = f"executor-{plan_commit.lease.attempt_id}"
    (tmp_path / "runs" / "run-1" / "traces" / episode_id).mkdir(parents=True)

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    FinishRevisionThreeBackend(),
                    workspace=tmp_path / "supervisor-after-crash",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    GroundedExecutorBackend(),
                    workspace=tmp_path / "executor-after-crash",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert recovered.status is RunStatus.COMPLETED
    assert [attempt.status for attempt in recovered.attempts] == ["error", "done"]
    assert recovered.attempts[0].failure_kind == "interrupted"


@pytest.mark.asyncio
async def test_restart_commits_a_completed_supervisor_trace_without_replaying_it(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    uncommitted = await Supervisor(
        EpisodeRunner(
            UnifiedAgent(
                TwoDecisionSupervisorBackend(),
                workspace=tmp_path / "supervisor-before-crash",
                sandbox=SandboxPolicy.READ_ONLY,
                instructions=SUPERVISOR_INSTRUCTIONS,
            ),
            traces,
        )
    ).decide(initial, episode_id="supervisor-r0")
    assert uncommitted.base_revision == 0
    assert memory.snapshot("run-1").revision == 0
    supervisor_after_crash = FinishAfterRecoveredSupervisorBackend()

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    supervisor_after_crash,
                    workspace=tmp_path / "supervisor-after-crash",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    GroundedExecutorBackend(),
                    workspace=tmp_path / "executor-after-crash",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert recovered.status is RunStatus.COMPLETED
    assert recovered.revision == 3
    assert supervisor_after_crash.calls == 1
    assert recovered.tasks[0].status is TaskStatus.DONE


@pytest.mark.asyncio
async def test_invalid_supervisor_plan_retries_in_a_new_logged_episode(tmp_path: Path) -> None:
    traces = TraceStore(tmp_path / "runs")
    supervisor_backend = InvalidThenValidSupervisorBackend()
    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    supervisor_backend,
                    workspace=tmp_path / "supervisor",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    GroundedExecutorBackend(),
                    workspace=tmp_path / "executor",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert supervisor_backend.calls == 3
    assert traces.exists("run-1", "supervisor-r0")
    assert traces.exists("run-1", "supervisor-r0-a2")
    assert traces.exists("run-1", "supervisor-r2")


@pytest.mark.asyncio
async def test_exhausted_supervisor_contract_failures_settle_the_run(tmp_path: Path) -> None:
    traces = TraceStore(tmp_path / "runs")
    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    AlwaysInvalidSupervisorBackend(),
                    workspace=tmp_path / "supervisor",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "executor",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.FAILED
    assert result.revision == 1
    assert result.tasks == ()
    assert result.attempts == ()
    assert [(transition.revision, transition.kind) for transition in result.transitions] == [
        (0, "run_created"),
        (1, "supervisor_failed"),
    ]
    assert result.transitions[-1].detail["failure_kind"] == "supervisor_contract"
    assert (
        "Supervisor result has unexpected or missing fields"
        in result.transitions[-1].detail["failure_message"]
    )


@pytest.mark.asyncio
async def test_decision_limit_atomically_settles_the_run(tmp_path: Path) -> None:
    traces = TraceStore(tmp_path / "runs")
    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    TwoDecisionSupervisorBackend(),
                    workspace=tmp_path / "supervisor",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    GroundedExecutorBackend(),
                    workspace=tmp_path / "executor",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
        max_decisions=1,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.FAILED
    assert result.revision == 3
    assert result.tasks[0].status is TaskStatus.DONE
    assert all(attempt.status != "active" for attempt in result.attempts)
    assert result.transitions[-1].kind == "decision_limit_reached"
    assert result.transitions[-1].detail == {
        "failure_kind": "decision_limit",
        "failure_message": "run reached the limit of 1 Supervisor decisions",
    }


@pytest.mark.asyncio
async def test_known_structured_transport_failure_is_recovered_without_replaying_actions(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    TwoDecisionSupervisorBackend(),
                    workspace=tmp_path / "supervisor",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MalformedStructuredTransportBackend(),
                    workspace=tmp_path / "executor",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert [attempt.status for attempt in result.attempts] == ["done"]
    assert result.observations[0].statement == "FLAG{transport-recovery}"
    assert result.transitions[2].detail["recovered_transport_failure"] is True
    trace = traces.load("run-1", result.observations[0].trace_episode_id)
    assert trace.output is not None and trace.output["success"] is False


@pytest.mark.asyncio
async def test_exact_result_at_max_turns_is_recovered_without_replaying_actions(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    TwoDecisionSupervisorBackend(),
                    workspace=tmp_path / "supervisor",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MaxTurnsStructuredTransportBackend(),
                    workspace=tmp_path / "executor",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert [attempt.status for attempt in result.attempts] == ["done"]
    assert result.observations[0].statement == "FLAG{max-turn-recovery}"
    assert result.transitions[2].detail["recovered_transport_failure"] is True
    trace = traces.load("run-1", result.observations[0].trace_episode_id)
    assert trace.output is not None and trace.output["success"] is False


@pytest.mark.asyncio
async def test_earlier_grounded_result_is_recovered_after_ungrounded_retry(
    tmp_path: Path,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    result = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    TwoDecisionSupervisorBackend(),
                    workspace=tmp_path / "supervisor",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MaxTurnsMultipleStructuredCandidatesBackend(),
                    workspace=tmp_path / "executor",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(
        RunSpec(
            run_id="run-1",
            goal="Assess the authorized target.",
            allowed_targets=("http://127.0.0.1:8080",),
        )
    )

    assert result.status is RunStatus.COMPLETED
    assert [attempt.status for attempt in result.attempts] == ["done"]
    assert result.observations[0].statement == "service=Roundcube 1.6.16"
    assert result.transitions[2].detail["recovered_transport_failure"] is True


@pytest.mark.asyncio
async def test_decision_budget_is_persistent_across_process_restart(tmp_path: Path) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    commit = memory.commit_plan(
        compile_plan(
            SupervisorDecision(
                base_revision=0,
                new_tasks=(
                    TaskProposal(
                        "discover-http",
                        TaskKind.DISCOVER,
                        "http://127.0.0.1:8080",
                        "Inspect the HTTP service.",
                        "The HTTP response is recorded.",
                    ),
                ),
                next_task_id="discover-http",
                finish=False,
                summary="Begin discovery.",
            ),
            initial,
        )
    )
    assert commit.lease is not None
    after_attempt = memory.commit_execution(
        ValidExecution(
            run_id="run-1",
            task_id="discover-http",
            attempt_id=commit.lease.attempt_id,
            lease_revision=commit.lease.revision,
            trace_episode_id="executor-before-restart",
            outcome=ExecutionOutcome.DONE,
            summary="Recorded the response.",
            observation="200 OK",
            evidence_sequences=(1,),
        )
    )
    assert after_attempt.revision == 2

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "supervisor-after-restart",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "executor-after-restart",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
        max_decisions=1,
    ).run(spec)

    assert recovered.status is RunStatus.FAILED
    assert recovered.transitions[-1].kind == "decision_limit_reached"
    assert not traces.exists("run-1", "supervisor-r2")


@pytest.mark.parametrize("input_content", ["{not-json", "[]"])
@pytest.mark.asyncio
async def test_restart_settles_a_malformed_trace_without_leaving_an_active_lease(
    tmp_path: Path,
    input_content: str,
) -> None:
    traces = TraceStore(tmp_path / "runs")
    memory = MemoryKernel(tmp_path / "state.sqlite3")
    spec = RunSpec(
        run_id="run-1",
        goal="Assess the authorized target.",
        allowed_targets=("http://127.0.0.1:8080",),
    )
    initial = memory.open_run(spec)
    commit = memory.commit_plan(
        compile_plan(
            SupervisorDecision(
                base_revision=0,
                new_tasks=(
                    TaskProposal(
                        "discover-http",
                        TaskKind.DISCOVER,
                        "http://127.0.0.1:8080",
                        "Inspect the HTTP service.",
                        "The HTTP response is recorded.",
                    ),
                ),
                next_task_id="discover-http",
                finish=False,
                summary="Begin discovery.",
            ),
            initial,
        )
    )
    assert commit.lease is not None
    episode_id = commit.lease.trace_episode_id
    assert episode_id is not None
    episode_dir = tmp_path / "runs" / "run-1" / "traces" / episode_id
    episode_dir.mkdir(parents=True)
    (episode_dir / "input.json").write_text(input_content, encoding="utf-8")
    (episode_dir / "events.jsonl").write_text("", encoding="utf-8")

    recovered = await PentestLoop(
        memory=MemoryKernel(tmp_path / "state.sqlite3"),
        supervisor=Supervisor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "supervisor-after-restart",
                    sandbox=SandboxPolicy.READ_ONLY,
                    instructions=SUPERVISOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        executor=Executor(
            EpisodeRunner(
                UnifiedAgent(
                    MustNotRunBackend(),
                    workspace=tmp_path / "executor-after-restart",
                    sandbox=SandboxPolicy.WORKSPACE_WRITE,
                    instructions=EXECUTOR_INSTRUCTIONS,
                ),
                traces,
            )
        ),
        traces=traces,
    ).run(spec)

    assert recovered.status is RunStatus.FAILED
    assert recovered.tasks[0].status is TaskStatus.FAILED
    assert recovered.attempts[0].status.value == "error"
    assert recovered.attempts[0].failure_kind == "trace_corrupt"
    assert all(attempt.status.value != "active" for attempt in recovered.attempts)