from ast import AST, Attribute, Call, ImportFrom, Name, parse, walk
from asyncio import gather
from asyncio import run as asyncio_run
from datetime import UTC, datetime
from json import dumps
from pathlib import Path
from types import SimpleNamespace
from unittest import IsolatedAsyncioTestCase, TestCase

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.testclient import TestClient
from httpx import ASGITransport, AsyncClient

from avalan.flow import (
    FlowDefinition,
    FlowDiagnostic,
    FlowDiagnosticCategory,
    FlowDiagnosticSeverity,
    FlowExecutor,
    FlowExecutorRunResult,
    FlowNodePlan,
    InMemoryFlowStateStore,
    compile_flow_source,
    default_flow_node_registry,
    inspect_flow_graph_source,
)
from avalan.server.routers import flow as flow_router
from avalan.server.routers.flow import router
from avalan.task import TaskClientUnsupportedOperationError, TaskRunState
from avalan.task.event import SanitizedTaskEvent, TaskEventCategory
from avalan.task.store import TaskStoreNotFoundError

_FORBIDDEN_ASYNCIO_IMPORTS = frozenset({"run", "to_thread"})
_FORBIDDEN_ASYNCIO_METHODS = frozenset({"run", "to_thread"})
_FORBIDDEN_LOOP_METHODS = frozenset({"run_until_complete"})
_FORBIDDEN_SYNC_FILE_METHODS = frozenset(
    {
        "read_bytes",
        "read_text",
        "write_bytes",
        "write_text",
    }
)


def _server_flow_async_boundary_violations(
    path: Path,
    source: str,
) -> tuple[str, ...]:
    tree = parse(source, filename=str(path))
    violations: list[str] = []
    for node in walk(tree):
        if isinstance(node, ImportFrom):
            violations.extend(_asyncio_import_violations(path, node))
            continue
        if isinstance(node, Call):
            violations.extend(_call_boundary_violations(path, node))
    return tuple(sorted(violations))


def _asyncio_import_violations(
    path: Path,
    node: ImportFrom,
) -> tuple[str, ...]:
    if node.module != "asyncio":
        return ()
    return tuple(
        f"{_node_location(path, node)} imports asyncio.{alias.name}"
        for alias in node.names
        if alias.name in _FORBIDDEN_ASYNCIO_IMPORTS
    )


def _call_boundary_violations(path: Path, node: Call) -> tuple[str, ...]:
    func = node.func
    if isinstance(func, Name) and func.id in _FORBIDDEN_ASYNCIO_METHODS:
        return (f"{_node_location(path, node)} calls {func.id}",)
    if not isinstance(func, Attribute):
        return ()
    violations: list[str] = []
    if func.attr in _FORBIDDEN_SYNC_FILE_METHODS:
        violations.append(f"{_node_location(path, node)} calls {func.attr}")
    if func.attr in _FORBIDDEN_LOOP_METHODS:
        violations.append(f"{_node_location(path, node)} calls {func.attr}")
    if (
        isinstance(func.value, Name)
        and func.value.id == "asyncio"
        and func.attr in _FORBIDDEN_ASYNCIO_METHODS
    ):
        violations.append(
            f"{_node_location(path, node)} calls asyncio.{func.attr}"
        )
    return tuple(violations)


def _node_location(path: Path, node: AST) -> str:
    return f"{path}:{getattr(node, 'lineno', 0)}"


class FlowRouterTestCase(TestCase):
    def setUp(self) -> None:
        self.app = FastAPI()
        self.app.include_router(router, prefix="/flows")
        self.client = TestClient(self.app)

    def test_flow_router_module_keeps_async_boundaries(self) -> None:
        module_path = Path(flow_router.__file__)

        violations = _server_flow_async_boundary_violations(
            Path("server/routers/flow.py"),
            module_path.read_text(encoding="utf-8"),
        )

        self.assertEqual(violations, ())

    def test_flow_router_boundary_audit_rejects_sync_bridges(self) -> None:
        violations = _server_flow_async_boundary_violations(
            Path("server/routers/flow.py"),
            """from asyncio import run, to_thread

async def load(path, loop):
    path.read_text()
    asyncio.run(load_async())
    loop.run_until_complete(load_async())
    return to_thread(path.write_text, "private")
""",
        )

        self.assertEqual(
            violations,
            (
                "server/routers/flow.py:1 imports asyncio.run",
                "server/routers/flow.py:1 imports asyncio.to_thread",
                "server/routers/flow.py:4 calls read_text",
                "server/routers/flow.py:5 calls asyncio.run",
                "server/routers/flow.py:6 calls run_until_complete",
                "server/routers/flow.py:7 calls to_thread",
            ),
        )

    def test_validate_success_and_malformed_source_are_safe(self) -> None:
        success = self.client.post(
            "/flows/validate",
            json={"source": _strict_flow_source()},
        )
        failure = self.client.post(
            "/flows/validate",
            json={"source": "[flow\nsecret = 'private customer prompt'"},
        )

        self.assertEqual(success.status_code, 200)
        self.assertTrue(success.json()["ok"])
        self.assertEqual(success.json()["definition"]["name"], "server-flow")
        self.assertEqual(failure.status_code, 200)
        self.assertFalse(failure.json()["ok"])
        self.assertEqual(
            failure.json()["diagnostics"][0]["code"],
            "flow.malformed_toml",
        )
        self.assertNotIn("private customer prompt", failure.text)

    def test_compile_and_graph_inspect_match_sdk_public_shapes(
        self,
    ) -> None:
        compile_sdk = asyncio_run(compile_flow_source(_graph_flow_source()))
        inspect_sdk = asyncio_run(
            inspect_flow_graph_source(_graph_flow_source())
        )

        compiled = self.client.post(
            "/flows/compile",
            json={"source": _graph_flow_source()},
        )
        inspected = self.client.post(
            "/flows/graph/inspect",
            json={"source": _graph_flow_source()},
        )

        self.assertEqual(compiled.status_code, 200)
        self.assertEqual(inspected.status_code, 200)
        self.assertEqual(
            compiled.json(),
            flow_router._public_value(compile_sdk.as_public_dict()),
        )
        self.assertEqual(
            inspected.json(),
            flow_router._public_value(inspect_sdk.as_public_dict()),
        )
        self.assertTrue(compiled.json()["ok"])
        self.assertTrue(inspected.json()["ok"])
        self.assertEqual(
            compiled.json()["canonical_source"],
            {"format": "toml", "strict": True},
        )
        self.assertEqual(
            inspected.json()["inspection"]["schema_version"],
            "flow.graph.inspection.v1",
        )
        self.assertNotIn("Private graph label", compiled.text)
        self.assertNotIn("Private graph label", inspected.text)

    def test_compile_and_graph_inspect_failures_are_safe(self) -> None:
        invalid_compile_sdk = asyncio_run(
            compile_flow_source(_invalid_graph_flow_source())
        )
        invalid_inspect_sdk = asyncio_run(
            inspect_flow_graph_source(_invalid_graph_flow_source())
        )
        file_compile_sdk = asyncio_run(
            compile_flow_source(_file_graph_flow_source())
        )
        file_inspect_sdk = asyncio_run(
            inspect_flow_graph_source(_file_graph_flow_source())
        )

        invalid_compile = self.client.post(
            "/flows/compile",
            json={"source": _invalid_graph_flow_source()},
        )
        invalid_inspect = self.client.post(
            "/flows/graph/inspect",
            json={"source": _invalid_graph_flow_source()},
        )
        file_compile = self.client.post(
            "/flows/compile",
            json={"source": _file_graph_flow_source()},
        )
        file_inspect = self.client.post(
            "/flows/graph/inspect",
            json={"source": _file_graph_flow_source()},
        )

        self.assertEqual(
            invalid_compile.json(),
            flow_router._public_value(invalid_compile_sdk.as_public_dict()),
        )
        self.assertEqual(
            invalid_inspect.json(),
            flow_router._public_value(invalid_inspect_sdk.as_public_dict()),
        )
        self.assertEqual(
            file_compile.json(),
            flow_router._public_value(file_compile_sdk.as_public_dict()),
        )
        self.assertEqual(
            file_inspect.json(),
            flow_router._public_value(file_inspect_sdk.as_public_dict()),
        )
        self.assertEqual(
            invalid_compile.json()["diagnostics"][0]["code"],
            "flow.graph.unsupported_executable_edge",
        )
        self.assertEqual(
            invalid_inspect.json()["diagnostics"][0]["code"],
            "flow.graph.unsupported_executable_edge",
        )
        self.assertEqual(
            file_compile.json()["diagnostics"][0]["code"],
            "flow.graph.read_failure",
        )
        self.assertEqual(
            file_inspect.json()["diagnostics"][0]["code"],
            "flow.graph.read_failure",
        )
        for response in (
            invalid_compile,
            invalid_inspect,
            file_compile,
            file_inspect,
        ):
            self.assertEqual(response.status_code, 200)
            self.assertFalse(response.json()["ok"])
            self.assertNotIn("Private graph label", response.text)
            self.assertNotIn("customer-token", response.text)

    def test_request_validation_errors_are_sanitized(self) -> None:
        invalid_source = self.client.post(
            "/flows/validate",
            json={"source": {"secret": "private customer prompt"}},
        )
        invalid_resume = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _strict_flow_source(),
                "decisions": {"review": "private resume token"},
            },
        )
        invalid_query = self.client.get(
            "/flows/runs/run-1/events",
            params={"after_sequence": -1},
        )
        invalid_mode = self.client.post(
            "/flows/mermaid/parse",
            json={
                "source": "graph TD\nA --> B",
                "mode": "private-mode-token",
            },
        )

        self.assertEqual(invalid_source.status_code, 422)
        self.assertEqual(invalid_resume.status_code, 422)
        self.assertEqual(invalid_query.status_code, 422)
        self.assertEqual(invalid_mode.status_code, 422)
        for response, path in (
            (invalid_source, "body.source"),
            (invalid_resume, "body.decisions.review"),
            (invalid_query, "query.after_sequence"),
            (invalid_mode, "body.mode"),
        ):
            payload = response.json()
            self.assertFalse(payload["ok"])
            self.assertEqual(
                payload["diagnostics"][0]["code"],
                "flow.definition.request_invalid",
            )
            self.assertEqual(payload["diagnostics"][0]["path"], path)
            self.assertNotIn("private", response.text)
            self.assertNotIn("customer prompt", response.text)
            self.assertNotIn("resume token", response.text)
            self.assertNotIn("private-mode-token", response.text)

    def test_request_validation_response_fallbacks_are_sanitized(
        self,
    ) -> None:
        empty = flow_router._request_validation_response(
            RequestValidationError([])
        )
        missing_location = flow_router._request_validation_response(
            RequestValidationError(
                [{"msg": "private validator detail", "type": "value_error"}]
            )
        )

        self.assertEqual(empty.status_code, 422)
        self.assertEqual(missing_location.status_code, 422)
        self.assertIn(b'"path":"request"', empty.body)
        self.assertIn(b'"path":"request"', missing_location.body)
        self.assertNotIn(b"private validator detail", missing_location.body)

    def test_mermaid_parse_render_and_compare(self) -> None:
        parsed = self.client.post(
            "/flows/mermaid/parse",
            json={"source": "graph TD\nA[Start] --> B[Done]"},
        )
        rendered = self.client.post(
            "/flows/mermaid/render",
            json={"source": "graph TD\nA[Start] --> B[Done]"},
        )
        compared = self.client.post(
            "/flows/mermaid/compare",
            json={
                "diagram_source": "graph TD\nA --> B",
                "definition_source": _strict_flow_source(),
            },
        )
        rejected = self.client.post(
            "/flows/mermaid/parse",
            json={
                "source": "graph TD\nA & B --> C\n%% private prompt",
                "mode": "executable",
            },
        )

        self.assertTrue(parsed.json()["ok"])
        self.assertEqual(parsed.json()["view"]["nodes"][0]["label"], "Start")
        self.assertTrue(rendered.json()["ok"])
        self.assertIn('A["Start"]', rendered.json()["source"])
        self.assertFalse(compared.json()["ok"])
        self.assertIn("flow.view.binding", compared.text)
        self.assertFalse(rejected.json()["ok"])
        self.assertNotIn("private prompt", rejected.text)

    def test_mermaid_render_and_compare_negative_inputs_are_safe(self) -> None:
        rendered = self.client.post(
            "/flows/mermaid/render",
            json={
                "source": "graph TD\nA & B --> C\n%% private prompt",
                "mode": "executable",
            },
        )
        compared = self.client.post(
            "/flows/mermaid/compare",
            json={
                "diagram_source": "graph TD\nA & B --> C",
                "definition_source": "[flow",
                "mode": "executable",
            },
        )

        self.assertEqual(rendered.status_code, 200)
        self.assertFalse(rendered.json()["ok"])
        self.assertNotIn("private prompt", rendered.text)
        self.assertEqual(compared.status_code, 200)
        self.assertFalse(compared.json()["ok"])
        self.assertIn("flow.malformed_toml", compared.text)

    def test_run_persists_inspect_trace_and_rejects_duplicate_run_id(
        self,
    ) -> None:
        self.app.state.flow_state_store = InMemoryFlowStateStore()

        run = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "private input"},
                "run_id": "run-1",
            },
        )
        inspect = self.client.get("/flows/runs/run-1/inspect")
        trace = self.client.get("/flows/runs/run-1/trace")
        duplicate = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "private input"},
                "run_id": "run-1",
            },
        )

        self.assertEqual(run.status_code, 200)
        self.assertTrue(run.json()["ok"], run.text)
        self.assertEqual(run.json()["outputs"], {"answer": "private input"})
        self.assertNotIn(
            "private input",
            dumps(run.json()["inspection"], sort_keys=True),
        )
        self.assertNotIn(
            "private input",
            dumps(run.json()["trace"], sort_keys=True),
        )
        self.assertEqual(run.json()["record_revision"], 1)
        self.assertEqual(inspect.status_code, 200)
        self.assertEqual(
            inspect.json()["inspection"]["state"],
            "succeeded",
        )
        self.assertNotIn(
            "private input",
            dumps(inspect.json()["inspection"], sort_keys=True),
        )
        self.assertEqual(trace.status_code, 200)
        self.assertNotIn("private input", trace.text)
        self.assertEqual(duplicate.status_code, 409)
        self.assertEqual(
            duplicate.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.run_conflict",
        )

    def test_run_rejects_invalid_source_and_supports_configured_runner(
        self,
    ) -> None:
        calls: list[str] = []

        async def runner(
            node: FlowNodePlan,
            inputs: dict[str, object],
        ) -> object:
            calls.append(node.name)
            return dict(inputs)

        invalid = self.client.post(
            "/flows/run",
            json={"source": "[flow", "inputs": {"payload": "safe"}},
        )
        no_run_id = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
            },
        )
        self.app.state.flow_node_runner = runner
        custom_runner = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
            },
        )
        self.app.state.flow_node_registry = default_flow_node_registry()
        with_registry = self.client.post(
            "/flows/validate",
            json={"source": _strict_flow_source()},
        )

        self.assertEqual(invalid.status_code, 200)
        self.assertFalse(invalid.json()["ok"])
        self.assertEqual(no_run_id.status_code, 200)
        self.assertNotIn("run_id", no_run_id.json())
        self.assertEqual(custom_runner.status_code, 200)
        self.assertEqual(calls, ["echo"])
        self.assertTrue(with_registry.json()["ok"])

    def test_run_with_executor_diagnostics_does_not_persist(self) -> None:
        self.app.state.flow_state_store = InMemoryFlowStateStore()
        self.app.state.flow_executor = _DiagnosticOnlyExecutor()

        response = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
                "run_id": "run-1",
            },
        )
        missing = self.client.get("/flows/runs/run-1/inspect")

        self.assertEqual(response.status_code, 200)
        self.assertFalse(response.json()["ok"])
        self.assertEqual(
            response.json()["diagnostics"][0]["code"],
            "flow.execution.test_diagnostic",
        )
        self.assertEqual(missing.status_code, 404)

    def test_run_without_store_for_run_id_and_missing_run_fail_safely(
        self,
    ) -> None:
        run = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
                "run_id": "run-1",
            },
        )
        self.app.state.flow_state_store = InMemoryFlowStateStore()
        missing = self.client.get("/flows/runs/missing/inspect")

        self.assertEqual(run.status_code, 503)
        self.assertEqual(
            run.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.store_unavailable",
        )
        self.assertEqual(missing.status_code, 404)
        self.assertEqual(
            missing.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.run_not_found",
        )

    def test_inspect_without_store_fails_safely(self) -> None:
        response = self.client.get("/flows/runs/run-1/inspect")

        self.assertEqual(response.status_code, 503)
        self.assertEqual(
            response.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.store_unavailable",
        )

    def test_resume_reports_conflicts_and_can_update_record(self) -> None:
        self.app.state.flow_state_store = InMemoryFlowStateStore()
        run = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
                "run_id": "run-1",
            },
        )
        resumed = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _strict_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
                "expected_revision": 1,
            },
        )
        invalid_source = self.client.post(
            "/flows/runs/run-1/resume",
            json={"source": "[flow", "decisions": {"review": {}}},
        )
        conflict = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _strict_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
                "expected_revision": 1,
            },
        )

        self.assertTrue(run.json()["ok"], run.text)
        self.assertEqual(resumed.status_code, 200)
        self.assertFalse(resumed.json()["ok"], resumed.text)
        self.assertEqual(
            resumed.json()["diagnostics"][0]["code"],
            "flow.execution.unknown_resume_node",
        )
        self.assertEqual(resumed.json()["record_revision"], 2)
        self.assertEqual(invalid_source.status_code, 200)
        self.assertFalse(invalid_source.json()["ok"])
        self.assertEqual(conflict.status_code, 409)
        self.assertEqual(
            conflict.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.revision_conflict",
        )

    def test_resume_without_store_and_missing_run_fail_safely(self) -> None:
        no_store = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _strict_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
            },
        )
        self.app.state.flow_state_store = InMemoryFlowStateStore()
        missing = self.client.post(
            "/flows/runs/missing/resume",
            json={
                "source": _strict_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
            },
        )

        self.assertEqual(no_store.status_code, 503)
        self.assertEqual(
            no_store.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.store_unavailable",
        )
        self.assertEqual(missing.status_code, 404)
        self.assertEqual(
            missing.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.run_not_found",
        )

    def test_resume_compiles_graph_source_before_executor(self) -> None:
        store = InMemoryFlowStateStore()
        self.app.state.flow_state_store = store
        run = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
                "run_id": "run-1",
            },
        )
        executor = _CapturingResumeExecutor()
        self.app.state.flow_executor = executor

        response = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _graph_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
            },
        )
        record = asyncio_run(store.get_flow_execution("run-1"))

        self.assertTrue(run.json()["ok"], run.text)
        self.assertEqual(response.status_code, 200)
        self.assertFalse(response.json()["ok"])
        self.assertEqual(
            response.json()["diagnostics"][0]["code"],
            "flow.execution.test_diagnostic",
        )
        self.assertEqual(response.json()["record_revision"], 1)
        self.assertEqual(record.revision, 1)
        self.assertIsInstance(executor.seen_flow, FlowDefinition)
        assert isinstance(executor.seen_flow, FlowDefinition)
        self.assertEqual(executor.seen_flow.name, "server-graph-flow")
        self.assertEqual(
            [(edge.source, edge.target) for edge in executor.seen_flow.edges],
            [("echo", "finish")],
        )
        self.assertNotIn("Private graph label", str(executor.seen_flow))
        self.assertFalse(hasattr(executor.seen_flow, "graph"))
        self.assertNotIn("Private graph label", response.text)

    def test_resume_rejects_graph_source_before_executor(self) -> None:
        store = InMemoryFlowStateStore()
        self.app.state.flow_state_store = store
        run = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
                "run_id": "run-1",
            },
        )
        executor = _CapturingResumeExecutor()
        self.app.state.flow_executor = executor

        invalid = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _invalid_graph_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
            },
        )
        file_backed = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _file_graph_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
            },
        )
        record = asyncio_run(store.get_flow_execution("run-1"))

        self.assertTrue(run.json()["ok"], run.text)
        self.assertEqual(invalid.status_code, 200)
        self.assertFalse(invalid.json()["ok"])
        self.assertEqual(file_backed.status_code, 200)
        self.assertFalse(file_backed.json()["ok"])
        self.assertEqual(
            invalid.json()["diagnostics"][0]["code"],
            "flow.graph.unsupported_executable_edge",
        )
        self.assertEqual(
            file_backed.json()["diagnostics"][0]["code"],
            "flow.graph.read_failure",
        )
        self.assertEqual(record.revision, 1)
        self.assertIsNone(executor.seen_flow)
        for response in (invalid, file_backed):
            self.assertNotIn("Private graph label", response.text)
            self.assertNotIn("customer-token", response.text)

    def test_validate_and_run_compile_graph_source_before_executor(
        self,
    ) -> None:
        executor = _CapturingDiagnosticExecutor()
        self.app.state.flow_executor = executor

        validated = self.client.post(
            "/flows/validate",
            json={"source": _graph_flow_source()},
        )
        run = self.client.post(
            "/flows/run",
            json={
                "source": _graph_flow_source(),
                "inputs": {"payload": "safe"},
            },
        )

        self.assertEqual(validated.status_code, 200)
        self.assertTrue(validated.json()["ok"], validated.text)
        self.assertEqual(
            validated.json()["definition"]["name"],
            "server-graph-flow",
        )
        self.assertNotIn("Private graph label", validated.text)
        self.assertEqual(run.status_code, 200)
        self.assertFalse(run.json()["ok"])
        self.assertIsInstance(executor.seen_flow, FlowDefinition)
        assert isinstance(executor.seen_flow, FlowDefinition)
        self.assertEqual(
            [(edge.source, edge.target) for edge in executor.seen_flow.edges],
            [("echo", "finish")],
        )
        self.assertNotIn("Private graph label", str(executor.seen_flow))
        self.assertFalse(hasattr(executor.seen_flow, "graph"))

    def test_run_rejects_invalid_graph_source_before_executor(self) -> None:
        executor = _CapturingDiagnosticExecutor()
        self.app.state.flow_executor = executor

        run = self.client.post(
            "/flows/run",
            json={"source": _invalid_graph_flow_source()},
        )

        self.assertEqual(run.status_code, 200)
        self.assertFalse(run.json()["ok"])
        self.assertEqual(
            run.json()["diagnostics"][0]["code"],
            "flow.graph.unsupported_executable_edge",
        )
        self.assertIsNone(executor.seen_flow)
        self.assertNotIn("Private graph label", run.text)

    def test_raw_source_rejects_file_graph_reference_before_executor(
        self,
    ) -> None:
        executor = _CapturingDiagnosticExecutor()
        self.app.state.flow_executor = executor

        validated = self.client.post(
            "/flows/validate",
            json={"source": _file_graph_flow_source()},
        )
        run = self.client.post(
            "/flows/run",
            json={"source": _file_graph_flow_source()},
        )

        self.assertEqual(validated.status_code, 200)
        self.assertFalse(validated.json()["ok"])
        self.assertEqual(run.status_code, 200)
        self.assertFalse(run.json()["ok"])
        for response in (validated, run):
            self.assertEqual(
                response.json()["diagnostics"][0]["code"],
                "flow.graph.read_failure",
            )
            self.assertEqual(
                response.json()["diagnostics"][0]["path"],
                "graph.source",
            )
            self.assertNotIn("customer-token", response.text)
            self.assertNotIn("private", response.text)
        self.assertIsNone(executor.seen_flow)

    def test_resume_with_executor_diagnostics_skips_state_update(self) -> None:
        self.app.state.flow_state_store = InMemoryFlowStateStore()
        run = self.client.post(
            "/flows/run",
            json={
                "source": _strict_flow_source(),
                "inputs": {"payload": "safe"},
                "run_id": "run-1",
            },
        )
        self.app.state.flow_executor = _DiagnosticOnlyExecutor()

        response = self.client.post(
            "/flows/runs/run-1/resume",
            json={
                "source": _strict_flow_source(),
                "decisions": {"review": {"decision": "approved"}},
            },
        )

        self.assertTrue(run.json()["ok"], run.text)
        self.assertEqual(response.status_code, 200)
        self.assertFalse(response.json()["ok"])
        self.assertEqual(response.json()["record_revision"], 1)

    def test_events_and_cancel_delegate_to_task_client(self) -> None:
        task_client = _FakeFlowTaskClient()
        self.app.state.flow_task_client = task_client

        events = self.client.get(
            "/flows/runs/run-1/events",
            params={"after_sequence": 1},
        )
        with self.client.stream(
            "GET",
            "/flows/runs/run-1/events",
            params={"stream": "true"},
        ) as streamed:
            stream_lines = list(streamed.iter_lines())
        cancelled = self.client.post("/flows/runs/run-1/cancel")
        cancel_conflict = self.client.post("/flows/runs/conflict/cancel")

        self.assertEqual(events.status_code, 200)
        self.assertEqual(events.json()["events"][0]["event_type"], "flow_end")
        self.assertEqual(task_client.after_sequences[0], 1)
        self.assertEqual(task_client.after_sequences[1], None)
        self.assertEqual(streamed.status_code, 200)
        self.assertIn("event: flow.event", stream_lines)
        self.assertIn("event: flow.events.completed", stream_lines)
        self.assertNotIn("private", "\n".join(stream_lines))
        self.assertEqual(cancelled.status_code, 200)
        self.assertEqual(cancelled.json()["run"]["state"], "cancel_requested")
        self.assertEqual(cancel_conflict.status_code, 409)
        self.assertEqual(
            cancel_conflict.json()["detail"]["diagnostics"][0]["code"],
            "task.cancel_unavailable",
        )

    def test_events_and_cancel_missing_runs_are_safe(self) -> None:
        self.app.state.flow_task_client = _MissingFlowTaskClient()

        events = self.client.get("/flows/runs/missing/events")
        cancel = self.client.post("/flows/runs/missing/cancel")

        self.assertEqual(events.status_code, 404)
        self.assertEqual(
            events.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.events_not_found",
        )
        self.assertEqual(cancel.status_code, 404)
        self.assertEqual(
            cancel.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.run_not_found",
        )

    def test_events_and_cancel_require_task_client(self) -> None:
        events = self.client.get("/flows/runs/run-1/events")
        cancel = self.client.post("/flows/runs/run-1/cancel")

        self.assertEqual(events.status_code, 503)
        self.assertEqual(
            events.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.events_unavailable",
        )
        self.assertEqual(cancel.status_code, 503)
        self.assertEqual(
            cancel.json()["detail"]["diagnostics"][0]["code"],
            "flow.task.cancel_unavailable",
        )


class FlowRouterAsyncTestCase(IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        self.app = FastAPI()
        self.app.include_router(router, prefix="/flows")
        self.client = AsyncClient(
            transport=ASGITransport(app=self.app),
            base_url="http://testserver",
        )

    async def asyncTearDown(self) -> None:
        await self.client.aclose()

    async def test_validate_and_run_graph_sources_concurrently(self) -> None:
        valid_validate, invalid_validate, valid_run, file_run = await gather(
            self.client.post(
                "/flows/validate",
                json={"source": _graph_flow_source()},
            ),
            self.client.post(
                "/flows/validate",
                json={"source": _invalid_graph_flow_source()},
            ),
            self.client.post(
                "/flows/run",
                json={
                    "source": _graph_flow_source(),
                    "inputs": {"payload": "safe"},
                },
            ),
            self.client.post(
                "/flows/run",
                json={"source": _file_graph_flow_source()},
            ),
        )

        self.assertEqual(valid_validate.status_code, 200)
        self.assertTrue(valid_validate.json()["ok"], valid_validate.text)
        self.assertEqual(invalid_validate.status_code, 200)
        self.assertFalse(invalid_validate.json()["ok"])
        self.assertEqual(
            invalid_validate.json()["diagnostics"][0]["code"],
            "flow.graph.unsupported_executable_edge",
        )
        self.assertEqual(valid_run.status_code, 200)
        self.assertTrue(valid_run.json()["ok"], valid_run.text)
        self.assertEqual(valid_run.json()["outputs"], {"answer": "safe"})
        self.assertEqual(file_run.status_code, 200)
        self.assertFalse(file_run.json()["ok"])
        self.assertEqual(
            file_run.json()["diagnostics"][0]["code"],
            "flow.graph.read_failure",
        )
        for response in (
            valid_validate,
            invalid_validate,
            valid_run,
            file_run,
        ):
            self.assertNotIn("Private graph label", response.text)
            self.assertNotIn("customer-token", response.text)


class _FakeFlowTaskClient:
    def __init__(self) -> None:
        self.after_sequences: list[int | None] = []

    async def events(
        self,
        run_id: str,
        *,
        after_sequence: int | None = None,
    ) -> tuple[SanitizedTaskEvent, ...]:
        self.after_sequences.append(after_sequence)
        return (
            SanitizedTaskEvent(
                event_id="event-1",
                run_id=run_id,
                sequence=2,
                event_type="flow_end",
                category=TaskEventCategory.ENGINE,
                created_at=datetime(2026, 6, 8, tzinfo=UTC),
                payload={"status": "succeeded"},
            ),
        )

    async def cancel(self, run_id: str) -> object:
        if run_id == "conflict":
            raise TaskClientUnsupportedOperationError(
                code="task.cancel_unavailable",
                operation="cancel",
                message="terminal",
            )
        return SimpleNamespace(
            run_id=run_id,
            state=TaskRunState.CANCEL_REQUESTED,
        )


class _MissingFlowTaskClient:
    async def events(
        self,
        run_id: str,
        *,
        after_sequence: int | None = None,
    ) -> tuple[SanitizedTaskEvent, ...]:
        _ = run_id, after_sequence
        raise TaskStoreNotFoundError("missing")

    async def cancel(self, run_id: str) -> object:
        _ = run_id
        raise TaskStoreNotFoundError("missing")


class _DiagnosticOnlyExecutor(FlowExecutor):
    async def run(
        self, *args: object, **kwargs: object
    ) -> FlowExecutorRunResult:
        _ = args, kwargs
        return FlowExecutorRunResult(diagnostics=(_diagnostic(),))

    async def resume(
        self,
        *args: object,
        **kwargs: object,
    ) -> FlowExecutorRunResult:
        _ = args, kwargs
        return FlowExecutorRunResult(diagnostics=(_diagnostic(),))


class _CapturingDiagnosticExecutor(FlowExecutor):
    def __init__(self) -> None:
        super().__init__()
        self.seen_flow: object | None = None

    async def run(
        self, *args: object, **kwargs: object
    ) -> FlowExecutorRunResult:
        _ = kwargs
        self.seen_flow = args[0] if args else None
        return FlowExecutorRunResult(diagnostics=(_diagnostic(),))


class _CapturingResumeExecutor(FlowExecutor):
    def __init__(self) -> None:
        super().__init__()
        self.seen_flow: object | None = None

    async def resume(
        self,
        *args: object,
        **kwargs: object,
    ) -> FlowExecutorRunResult:
        _ = kwargs
        self.seen_flow = args[0] if args else None
        return FlowExecutorRunResult(diagnostics=(_diagnostic(),))


def _diagnostic() -> FlowDiagnostic:
    return FlowDiagnostic(
        code="flow.execution.test_diagnostic",
        path="flow",
        category=FlowDiagnosticCategory.EXECUTION,
        severity=FlowDiagnosticSeverity.ERROR,
        message="Flow execution returned a diagnostic.",
        hint="Inspect the flow run.",
    )


def _strict_flow_source() -> str:
    return """
[flow]
name = "server-flow"
version = "1"

[[inputs]]
name = "payload"
type = "string"

[[outputs]]
name = "answer"
type = "text"

[entry]
type = "node"
node = "echo"

[output_behavior]
type = "map"

[output_behavior.outputs]
answer = "echo.value"

[nodes.echo]
type = "pass-through"

[nodes.echo.mapping]
value = "input.payload"
"""


def _graph_flow_source() -> str:
    return """
[flow]
name = "server-graph-flow"
version = "1"

[[inputs]]
name = "payload"
type = "string"

[[outputs]]
name = "answer"
type = "text"

[entry]
type = "node"
node = "echo"

[output_behavior]
type = "map"

[output_behavior.outputs]
answer = "finish.value"

[graph]
format = "mermaid"
source = "inline"
mode = "executable"
diagram = '''
flowchart LR
echo route_1@-->|Private graph label| finish
'''

[nodes.echo]
type = "pass-through"

[nodes.echo.mapping]
value = "input.payload"

[nodes.finish]
type = "pass-through"

[nodes.finish.mapping]
value = "echo.value"
"""


def _invalid_graph_flow_source() -> str:
    return """
[flow]
name = "server-graph-flow"
version = "1"

[[inputs]]
name = "payload"
type = "string"

[[outputs]]
name = "answer"
type = "text"

[entry]
type = "node"
node = "echo"

[output_behavior]
type = "map"

[output_behavior.outputs]
answer = "finish.value"

[graph]
format = "mermaid"
source = "inline"
mode = "executable"
diagram = '''
flowchart LR
echo -->|Private graph label| finish
'''

[nodes.echo]
type = "pass-through"

[nodes.finish]
type = "pass-through"
"""


def _file_graph_flow_source() -> str:
    return """
[flow]
name = "server-file-graph-flow"
version = "1"

[[inputs]]
name = "payload"
type = "string"

[[outputs]]
name = "answer"
type = "text"

[entry]
type = "node"
node = "echo"

[output_behavior]
type = "map"

[output_behavior.outputs]
answer = "finish.value"

[graph]
format = "mermaid"
source = "file"
mode = "executable"
path = "private/customer-token.mmd"

[nodes.echo]
type = "pass-through"

[nodes.finish]
type = "pass-through"
"""
