fix(sec-core): expose skill ledger verdicts

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

@@ -1,6 +1,7 @@
 """Read-only daemon handlers for security and observability SQLite data."""
 
 import json
+from pathlib import PurePath
 from typing import Any
 
 from agent_sec_cli.daemon.errors import BadRequestError
@@ -17,7 +18,7 @@
 )
 from agent_sec_cli.observability.models import ObservabilityEventRecord
 from agent_sec_cli.observability.sqlite_reader import ObservabilityReader
-from agent_sec_cli.security_events.schema import SecurityEvent
+from agent_sec_cli.security_events.schema import SecurityEvent, extract_verdict
 from agent_sec_cli.security_events.sqlite_reader import SqliteEventReader
 from agent_sec_cli.utils.timestamp import (
     normalize_iso_to_utc_iso,
@@ -350,11 +351,53 @@ def _security_filters(params: dict[str, Any]) -> dict[str, str | None]:
 
 def _event_payload(event: SecurityEvent, *, include_details: bool) -> dict[str, Any]:
     payload = event.to_dict()
+    payload.update(_dashboard_event_fields(event))
     if not include_details:
         payload.pop("details", None)
     return payload
 
 
+def _dashboard_event_fields(event: SecurityEvent) -> dict[str, str]:
+    """Project optional fields needed by dashboard event rows."""
+    fields: dict[str, str] = {}
+    verdict = extract_verdict(event.details)
+    if verdict is not None:
+        fields["verdict"] = verdict
+
+    if event.category != "skill_ledger":
+        return fields
+
+    result = event.details.get("result")
+    result_data = result if isinstance(result, dict) else {}
+    request = event.details.get("request")
+    request_data = request if isinstance(request, dict) else {}
+
+    command = _first_non_empty_string(
+        result_data.get("command"), request_data.get("command")
+    )
+    if command is not None:
+        fields["command"] = command
+
+    if isinstance(result_data.get("results"), list):
+        return fields
+
+    skill_name = _first_non_empty_string(result_data.get("skill_name"))
+    if skill_name is None:
+        skill_dir = _first_non_empty_string(request_data.get("skill_dir"))
+        if skill_dir is not None:
+            skill_name = PurePath(skill_dir).name or None
+    if skill_name is not None:
+        fields["skill_name"] = skill_name
+    return fields
+
+
+def _first_non_empty_string(*values: Any) -> str | None:
+    for value in values:
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
 def _observability_item(row: ObservabilityEventRecord) -> dict[str, Any]:
     return {
         "kind": "observability",

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/security_middleware/backends/skill_ledger.py

@@ -47,6 +47,16 @@
 )
 _CAMEL_ACRONYM_BOUNDARY_RE = re.compile(r"(.)([A-Z][a-z]+)")
 _CAMEL_WORD_BOUNDARY_RE = re.compile(r"([a-z0-9])([A-Z])")
+_VERDICT_SEVERITY = {
+    "pass": 0,
+    "none": 1,
+    "warn": 2,
+    "unmanaged": 3,
+    "drifted": 4,
+    "deny": 5,
+    "tampered": 6,
+    "error": 7,
+}
 
 
 def _to_snake_case(value: str) -> str:
@@ -85,9 +95,60 @@ def _snake_case_event_keys(value: Any) -> Any:
     return normalized
 
 
+def _known_verdict(value: Any) -> str | None:
+    """Return a recognized Skill Ledger verdict value."""
+    if isinstance(value, str) and value in _VERDICT_SEVERITY:
+        return value
+    return None
+
+
+def _batch_verdict(results: Any) -> str | None:
+    """Return the most severe verdict from a batch command result."""
+    if not isinstance(results, list):
+        return None
+
+    verdicts: list[str] = []
+    for item in results:
+        if not isinstance(item, dict):
+            continue
+        for value in (item.get("status"), item.get("verdict")):
+            verdict = _known_verdict(value)
+            if verdict is not None:
+                verdicts.append(verdict)
+
+    if not verdicts:
+        return None
+    return max(verdicts, key=_VERDICT_SEVERITY.__getitem__)
+
+
+def _project_event_verdict(data: dict[str, Any]) -> str | None:
+    """Project command-specific output into the canonical event verdict."""
+    command = data.get("command")
+    if not isinstance(command, str):
+        return None
+
+    if command in {"init", "scan", "check"} and "results" in data:
+        return _batch_verdict(data.get("results"))
+    if command == "show":
+        return _known_verdict(data.get("latest_status"))
+    if command == "check":
+        return _known_verdict(data.get("status"))
+    if command in {"scan", "certify"}:
+        return _known_verdict(data.get("verdict"))
+    if command == "decide":
+        return _known_verdict(data.get("current_status")) or _known_verdict(
+            data.get("verdict")
+        )
+    return None
+
+
 def _normalize_event_result(data: dict[str, Any]) -> dict[str, Any]:
     """Build the Skill Ledger event-only result payload."""
-    return _snake_case_event_keys(data)
+    normalized = _snake_case_event_keys(data)
+    verdict = _project_event_verdict(normalized)
+    if verdict is not None:
+        normalized["verdict"] = verdict
+    return normalized
 
 
 class SkillLedgerBackend(BaseBackend):

src/agent-sec-core/tests/integration-test/skill-ledger/test_skill_ledger_integration.py

@@ -30,6 +30,13 @@
 import agent_sec_cli.security_events as security_events
 import pytest
 from agent_sec_cli.cli import app as cli_app
+from agent_sec_cli.daemon.handlers.security_query import (
+    security_events_list_handler,
+    security_summary_handler,
+)
+from agent_sec_cli.daemon.protocol import DaemonRequest
+from agent_sec_cli.daemon.runtime import DaemonRuntime
+from agent_sec_cli.security_events.sqlite_reader import SqliteEventReader
 from agent_sec_cli.skill_ledger import config as config_module
 from agent_sec_cli.skill_ledger.core import decision as decision_core
 from agent_sec_cli.skill_ledger.core import live_root as live_root_core
@@ -2402,6 +2409,94 @@ def test_show_reports_active_latest_decision_and_root_match(ws):
     assert out["warnings"] == [out["message"]]
 
 
+def test_show_event_flows_through_jsonl_sqlite_and_dashboard(ws, monkeypatch):
+    skill = make_skill(ws.skills_dir, "dashboard-show", {"data.txt": "safe"})
+    findings = write_findings_file(
+        ws.fixtures,
+        "dashboard-show-pass.json",
+        [{"rule": "ok", "level": "pass", "message": "pass"}],
+    )
+    event_data = ws.root / "events_dashboard_show"
+    event_data.mkdir()
+    env = ws.env({"AGENT_SEC_DATA_DIR": str(event_data)})
+
+    reset_security_event_writers()
+    try:
+        certified = run_skill_ledger(
+            ["certify", str(skill), "--findings", str(findings)],
+            env_extra=env,
+        )
+        shown = run_skill_ledger(["show", str(skill)], env_extra=env)
+    finally:
+        reset_security_event_writers()
+
+    assert certified.returncode == 0, certified.stderr
+    assert shown.returncode == 0, shown.stderr
+    stdout_result = parse_json_output(shown.stdout)
+    assert stdout_result["latestStatus"] == "pass"
+    assert "verdict" not in stdout_result
+
+    jsonl_events = read_security_events(event_data)
+    show_jsonl = next(
+        event
+        for event in jsonl_events
+        if event["details"]["result"].get("command") == "show"
+    )
+    assert show_jsonl["details"]["result"]["verdict"] == "pass"
+    assert show_jsonl["details"]["result"]["skill_name"] == skill.name
+
+    sqlite_reader = SqliteEventReader(path=event_data / "security-events.db")
+    try:
+        pass_events = sqlite_reader.query(
+            category="skill_ledger",
+            verdict="pass",
+        )
+        show_sqlite = next(
+            event
+            for event in pass_events
+            if event.details["result"].get("command") == "show"
+        )
+        assert show_sqlite.event_id == show_jsonl["event_id"]
+        assert sqlite_reader.count_by("verdict", category="skill_ledger") == {"pass": 2}
+    finally:
+        sqlite_reader.close()
+
+    monkeypatch.setenv("AGENT_SEC_DATA_DIR", str(event_data))
+    runtime = DaemonRuntime(socket_path=event_data / "daemon.sock")
+    list_result = security_events_list_handler(
+        DaemonRequest(
+            method="sec.events.list",
+            params={"category": "skill_ledger", "verdict": "pass"},
+        ),
+        runtime,
+    )
+    dashboard_show = next(
+        item
+        for item in list_result.data["items"]
+        if item["event_id"] == show_jsonl["event_id"]
+    )
+    assert dashboard_show["verdict"] == "pass"
+    assert dashboard_show["command"] == "show"
+    assert dashboard_show["skill_name"] == skill.name
+    assert "details" not in dashboard_show
+
+    summary_result = security_summary_handler(
+        DaemonRequest(
+            method="sec.summary",
+            params={"category": "skill_ledger", "latest_limit": 10},
+        ),
+        runtime,
+    )
+    summary_show = next(
+        item
+        for item in summary_result.data["latest_events"]
+        if item["event_id"] == show_jsonl["event_id"]
+    )
+    assert summary_show["verdict"] == "pass"
+    assert summary_show["command"] == "show"
+    assert summary_show["skill_name"] == skill.name
+
+
 def test_show_findings_summary_handles_missing_fields_and_empty_findings(ws):
     skill = make_skill(
         ws.skills_dir, "decision-show-missing-findings", {"data.txt": "v1"}

src/agent-sec-core/tests/unit-test/daemon/test_security_query_handler.py

@@ -179,6 +179,93 @@ def test_security_event_queries_read_sqlite_data(tmp_path: Path) -> None:
     assert summary_response.data["affected_runs"] == 2
 
 
+def test_skill_ledger_dashboard_fields_are_available_without_details(
+    tmp_path: Path,
+) -> None:
+    details = {
+        "request": {"command": "show", "skill_dir": "/opt/skills/demo"},
+        "result": {
+            "command": "show",
+            "verdict": "deny",
+            "skill_name": "demo",
+            "latest_status": "deny",
+        },
+    }
+    _write_security_event(
+        tmp_path,
+        event_id="skill-show",
+        event_type="skill_ledger",
+        category="skill_ledger",
+        details=details,
+    )
+
+    list_response = _call_daemon(tmp_path, "sec.events.list")
+    list_item = list_response.data["items"][0]
+    assert list_item["verdict"] == "deny"
+    assert list_item["command"] == "show"
+    assert list_item["skill_name"] == "demo"
+    assert "details" not in list_item
+
+    get_response = _call_daemon(
+        tmp_path,
+        "sec.events.get",
+        {"event_id": "skill-show"},
+    )
+    get_item = get_response.data["event"]
+    assert get_item["verdict"] == "deny"
+    assert get_item["command"] == "show"
+    assert get_item["skill_name"] == "demo"
+    assert get_item["details"] == details
+
+    summary_response = _call_daemon(tmp_path, "sec.summary")
+    latest_item = summary_response.data["latest_events"][0]
+    assert latest_item["verdict"] == "deny"
+    assert latest_item["command"] == "show"
+    assert latest_item["skill_name"] == "demo"
+    assert "details" not in latest_item
+
+
+def test_skill_ledger_dashboard_skill_name_fallback_skips_batch_events() -> None:
+    failed_event = SecurityEvent(
+        event_type="skill_ledger",
+        category="skill_ledger",
+        result="failed",
+        details={
+            "request": {"command": "check", "skill_dir": "/opt/skills/fallback"},
+            "result": {},
+        },
+    )
+    failed_payload = security_query._event_payload(
+        failed_event,
+        include_details=False,
+    )
+
+    assert failed_payload["command"] == "check"
+    assert failed_payload["skill_name"] == "fallback"
+    assert "verdict" not in failed_payload
+
+    batch_event = SecurityEvent(
+        event_type="skill_ledger",
+        category="skill_ledger",
+        details={
+            "request": {"command": "check", "skill_dir": "/opt/skills/not-a-skill"},
+            "result": {
+                "command": "check",
+                "verdict": "tampered",
+                "results": [{"skill_name": "a", "status": "tampered"}],
+            },
+        },
+    )
+    batch_payload = security_query._event_payload(
+        batch_event,
+        include_details=False,
+    )
+
+    assert batch_payload["command"] == "check"
+    assert batch_payload["verdict"] == "tampered"
+    assert "skill_name" not in batch_payload
+
+
 def test_security_summary_handler_uses_repository_summary(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -430,6 +517,7 @@ def test_security_events_list_filters_by_verdict(tmp_path: Path) -> None:
     assert len(response.data["items"]) == 2
     returned_ids = {item["event_id"] for item in response.data["items"]}
     assert returned_ids == {"deny-1", "deny-2"}
+    assert all(item["verdict"] == "deny" for item in response.data["items"])
 
 
 def test_security_events_list_normalizes_explicit_since_until(tmp_path: Path) -> None: