feat(sec-core): route prompt scan to daemon and add prompt model preload

src/agent-sec-core/Makefile

@@ -54,10 +54,8 @@ test-python: ## Run Python unit and integration tests
 	uv run --project agent-sec-cli python3 tests/e2e/skill-ledger/e2e_test.py
 
 .PHONY: test-prompt-scanner-e2e
-test-prompt-scanner-e2e: ## Run prompt-scanner e2e tests (downloads ML model on first run)
-	@echo "🔥 Warming up prompt-scanner (downloading ML model if not cached)..."
-	uv run --project agent-sec-cli agent-sec-cli scan-prompt warmup
-	@echo "🧪 Running prompt-scanner e2e tests..."
+test-prompt-scanner-e2e: ## Run daemon-backed prompt-scanner e2e tests
+	@echo "🧪 Running daemon-backed prompt-scanner e2e tests..."
 	uv run --project agent-sec-cli pytest tests/e2e/prompt-scanner/e2e_test.py
 
 .PHONY: test-e2e-rpm

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/correlation_context.py

@@ -117,6 +117,15 @@ def parse_trace_context(value: str | None) -> TraceContext | None:
     if not isinstance(payload, dict):
         raise ValueError("trace context must be a JSON object")
 
+    return parse_trace_context_payload(payload)
+
+
+def parse_trace_context_payload(
+    payload: Mapping[str, Any] | None,
+) -> TraceContext | None:
+    """Normalize a structured trace context payload into snake_case fields."""
+    if payload is None:
+        return None
     return TraceContext(**_normalized_fields(payload))
 
 

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/handlers/__init__.py

@@ -0,0 +1 @@
+"""Daemon method handlers."""

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/handlers/prompt_scan.py

@@ -0,0 +1,120 @@
+"""Daemon handler for the scan-prompt CLI-compatible method."""
+
+import asyncio
+from typing import Any
+
+from agent_sec_cli.daemon.errors import UnavailableError
+from agent_sec_cli.daemon.protocol import DaemonRequest
+from agent_sec_cli.daemon.registry import (
+    HandlerResult,
+    MethodRegistry,
+    MethodSpec,
+)
+from agent_sec_cli.daemon.runtime import DaemonRuntime
+
+
+def register_prompt_scan_methods(registry: MethodRegistry) -> None:
+    """Register prompt scanner daemon methods."""
+    registry.register(
+        MethodSpec(
+            method="scan-prompt",
+            handler=prompt_scan_handler,
+            lifecycle="security action",
+            queue="prompt-scan",
+            timeout_ms=30_000,
+            access_log=True,
+        )
+    )
+
+
+async def prompt_scan_handler(
+    request: DaemonRequest, runtime: DaemonRuntime
+) -> HandlerResult:
+    """Execute prompt scanning through security middleware."""
+    prompt_scan_state = runtime.prompt_scan_state
+    if prompt_scan_state.status != "ready" or not prompt_scan_state.loaded:
+        raise UnavailableError(_prompt_unavailable_message(runtime))
+
+    params = request.params
+    result = await asyncio.to_thread(
+        _invoke_prompt_scan,
+        trace_context=request.trace_context,
+        text=_string_param(params, "text"),
+        mode=_string_param(params, "mode", default="standard"),
+        source=_string_param(params, "source"),
+    )
+    return _action_result_to_handler_result(result)
+
+
+def _invoke_prompt_scan(
+    *,
+    trace_context: dict[str, Any],
+    text: str,
+    mode: str,
+    source: str,
+) -> Any:
+    from agent_sec_cli.security_middleware import (  # noqa: PLC0415 - lazy import: daemon handler execution only
+        invoke_with_context,
+    )
+
+    return invoke_with_context(
+        "prompt_scan",
+        caller="daemon",
+        trace_context=trace_context,
+        text=text,
+        mode=mode,
+        source=source,
+    )
+
+
+def _action_result_to_handler_result(result: Any) -> HandlerResult:
+    return HandlerResult(
+        data=result.data,
+        stdout=result.stdout,
+        stderr=result.error,
+        exit_code=result.exit_code,
+    )
+
+
+def _string_param(
+    params: dict[str, Any],
+    name: str,
+    default: str = "",
+) -> str:
+    value = params.get(name, default)
+    if value is None:
+        return default
+    return str(value)
+
+
+def _prompt_unavailable_message(runtime: DaemonRuntime) -> str:
+    prompt_scan_state = runtime.prompt_scan_state.to_dict()
+    status = prompt_scan_state.get("status", "unknown")
+    model = prompt_scan_state.get("model")
+    last_error = prompt_scan_state.get("last_error")
+
+    if status == "downloading":
+        parts = [
+            "prompt scanner is not ready: model download is still in progress",
+            "status=downloading",
+        ]
+    elif status == "loading":
+        parts = [
+            "prompt scanner is not ready: model download completed and the model is loading",
+            "status=loading",
+        ]
+    elif status == "degraded":
+        parts = [
+            "prompt scanner preload failed",
+            "retry with `agent-sec-cli scan-prompt warmup`",
+            "then restart the agent-sec daemon process",
+            "status=degraded",
+        ]
+    else:
+        parts = [f"prompt scanner is not ready: status={status}"]
+
+    if model:
+        parts.append(f"model={model}")
+    if last_error:
+        parts.append(f"last_error={last_error}")
+    return ", ".join(parts)

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/handlers/prompt_scan_protocol.md

@@ -0,0 +1,169 @@
+# scan-prompt daemon protocol
+
+This document defines the response contract for the daemon `scan-prompt` method.
+It is the method-level contract for callers such as the CLI, Cosh hooks, and
+other subprocess clients.
+
+## Response layers
+
+`scan-prompt` responses have three layers. Callers must handle them in this
+order.
+
+### 1. Transport failure: no `DaemonResponse`
+
+The client did not receive a valid daemon response.
+
+Examples:
+
+- daemon socket does not exist
+- daemon connection or request timeout
+- daemon process exits before writing a response
+- daemon response is not valid protocol data
+- daemon response exceeds the configured size limit
+- daemon runtime path cannot be resolved, for example when `XDG_RUNTIME_DIR` is
+  not set and no explicit daemon socket path is provided
+
+Caller behavior:
+
+```python
+try:
+    response = client.call("scan-prompt", params=params)
+except (DaemonClientError, DaemonRuntimePathError):
+    # Daemon is unreachable, the protocol is broken, or the local runtime
+    # socket path cannot be resolved.
+    exit(1)
+```
+
+### 2. Daemon failure: `ok=false`
+
+The daemon received the request, but the method could not be dispatched or
+executed at the daemon/method boundary.
+
+`ok=false` responses are not scan results. Callers must not parse `data` or
+`stdout` as an action result.
+
+Expected shape:
+
+```json
+{
+  "id": "req-1",
+  "ok": false,
+  "data": {},
+  "stdout": "",
+  "stderr": "prompt scanner is not ready: status=loading",
+  "exit_code": 1,
+  "error": {
+    "code": "unavailable",
+    "message": "prompt scanner is not ready: status=loading"
+  }
+}
+```
+
+`scan-prompt` daemon failures include:
+
+- unknown daemon method
+- malformed daemon request
+- prompt scanner runtime unavailable, including preload states such as
+  `pending`, `downloading`, `loading`, or `degraded`
+- daemon method timeout
+- unexpected handler crash
+
+Unexpected handler crashes should be logged by the daemon and returned as
+`internal_error` without exposing arbitrary exception details to callers.
+
+Caller behavior:
+
+```python
+if not response.ok:
+    echo_error(response.stderr or response.error["message"])
+    exit(response.exit_code or 1)
+```
+
+### 3. Action result: `ok=true`
+
+The daemon successfully dispatched `scan-prompt`, and the handler returned a
+scan action result.
+
+For `ok=true`, `exit_code` is the action/CLI semantic exit code. It may be
+non-zero even though daemon dispatch succeeded.
+
+Expected successful scan shape:
+
+```json
+{
+  "id": "req-1",
+  "ok": true,
+  "data": {
+    "ok": true,
+    "verdict": "pass"
+  },
+  "stdout": "{...same scan result as JSON...}",
+  "stderr": "",
+  "exit_code": 0
+}
+```
+
+Expected scanner error result shape:
+
+```json
+{
+  "id": "req-1",
+  "ok": true,
+  "data": {
+    "ok": false,
+    "verdict": "error",
+    "summary": "Scanner error: model exploded"
+  },
+  "stdout": "{...same error verdict as JSON...}",
+  "stderr": "Scanner error: model exploded",
+  "exit_code": 1
+}
+```
+
+`scan-prompt` action results include:
+
+- `PASS`, `WARN`, and `DENY` scan verdicts: `ok=true`, `exit_code=0`
+- backend validation failures, such as missing/empty `text` or invalid `mode`:
+  `ok=true`, `exit_code=1`, with `stderr` describing the validation error
+- scanner-produced `ERROR` verdicts: `ok=true`, `exit_code=1`, with structured
+  error verdict data
+- scanner domain exceptions that can be converted to an error verdict:
+  `ok=true`, `exit_code=1`, with structured error verdict data
+
+Caller behavior:
+
+```python
+if response.ok:
+    rendered = render_action_output_if_present(response)
+    if response.exit_code != 0:
+        if not rendered:
+            echo_error(response.stderr or "scan-prompt failed")
+        exit(response.exit_code)
+    exit(0)
+```
+
+Callers should render structured action output before exiting with a non-zero
+action `exit_code`, so JSON consumers can still parse the error verdict. If an
+action failure has no structured output, callers should display `stderr`.
+
+## Request parameters
+
+`scan-prompt` request params:
+
+```json
+{
+  "text": "prompt text to scan",
+  "mode": "fast|standard|strict",
+  "source": "optional input source label"
+}
+```
+
+Rules:
+
+- `text` is required and must contain non-whitespace content.
+- `mode` is optional and defaults to `standard`.
+- `mode` must be one of `fast`, `standard`, or `strict`.
+- `source` is optional and defaults to an empty string.
+
+Invalid `text` or `mode` is handled by the prompt scan backend and returned as
+an action failure: `ok=true`, `exit_code=1`.

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/health.py

@@ -19,7 +19,7 @@ def build_health_snapshot(runtime: DaemonRuntime) -> dict[str, Any]:
         "pid": os.getpid(),
         "uptime_seconds": runtime.uptime_seconds(),
         "socket": str(runtime.socket_path),
-        "prompt_scan": runtime.prompt_scan.to_dict(),
+        "prompt_scan": runtime.prompt_scan_state.to_dict(),
         "jobs": runtime.jobs.status(),
         "queues": runtime.queues.to_dict(),
     }
@@ -30,17 +30,15 @@ def health_handler(_request: DaemonRequest, runtime: DaemonRuntime) -> HandlerRe
     return HandlerResult(data=build_health_snapshot(runtime))
 
 
-def create_default_registry() -> MethodRegistry:
-    """Create the C02 daemon registry with only daemon.health registered."""
-    registry = MethodRegistry()
+def register_health_methods(registry: MethodRegistry) -> None:
+    """Register daemon health methods."""
     registry.register(
         MethodSpec(
             method="daemon.health",
             handler=health_handler,
             lifecycle="admin",
             queue="admin",
             timeout_ms=1000,
-            access_log=True,
+            access_log=False,
         )
     )
-    return registry

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/jobs/__init__.py

@@ -0,0 +1,15 @@
+"""Daemon background job package."""
+
+from agent_sec_cli.daemon.jobs.base import (
+    BackgroundJob,
+    JobManager,
+    JobStatus,
+    PeriodicBackgroundJob,
+)
+
+__all__ = [
+    "BackgroundJob",
+    "JobManager",
+    "JobStatus",
+    "PeriodicBackgroundJob",
+]

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/jobs.py → src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/daemon/jobs/base.py

@@ -197,22 +197,6 @@ def started(self) -> bool:
         return self._started
 
 
-def register_default_jobs(job_manager: JobManager) -> None:
-    """Register daemon jobs that should start with every daemon instance.
-
-    Jobs registered here must subclass ``BackgroundJob`` and provide:
-    ``name`` for the stable health identifier, ``start()`` for async startup,
-    ``stop()`` for graceful async shutdown, and ``status()`` for the
-    JSON-serializable health snapshot. Periodic jobs should subclass
-    ``PeriodicBackgroundJob`` instead, pass ``interval_seconds`` to its
-    constructor, and implement ``run_once()`` for one scheduled iteration.
-    """
-    # C02 has no default background jobs. Future daemon jobs should be added
-    # here with job_manager.register(MyJob()) so every daemon startup path uses
-    # the same registration order and lifecycle semantics.
-    pass
-
-
 def next_cycle_start(
     started_monotonic: float,
     finished_monotonic: float,