Add automation inventory surface

tests/test_automation_status.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import asyncio
+import json
 from datetime import datetime
 from pathlib import Path
 from zoneinfo import ZoneInfo
@@ -182,3 +184,106 @@ def test_working_memory_frontmatter_evidence(tmp_path: Path) -> None:
     assert source["status"] == "ok"
     assert evidence[0]["run_id"] == "run-1"
     assert evidence[0]["output_validated"] is False
+
+
+def _scheduled_definition(enabled: bool = False):
+    return {
+        "id": "00000000-0000-0000-0000-000000000456",
+        "name": "One Shot",
+        "enabled": enabled,
+        "trigger_type": "scheduled",
+        "trigger_config": {
+            "trigger_type": "scheduled",
+            "at": "2026-06-26T09:00:00+02:00",
+            "timezone": "Europe/Berlin",
+        },
+        "source": "db",
+    }
+
+
+def test_inventory_report_uses_db_definition_rows(monkeypatch) -> None:
+    async def fake_load_definitions(args, warnings):
+        return [dict(_definition(), source="db"), _scheduled_definition()], {"status": "ok", "source": "db"}
+
+    async def fake_temporal(host, namespace, definitions, *, timeout_seconds):
+        return {
+            ACTIVITY_ID: {
+                "schedule_id": f"activity-schedule-{ACTIVITY_ID}",
+                "available": True,
+                "paused": False,
+                "missed_catchup_window": 0,
+                "last_fired_at": None,
+            },
+        }, {"status": "ok", "count": 1}
+
+    monkeypatch.setattr(status, "load_definitions", fake_load_definitions)
+    monkeypatch.setattr(status, "load_temporal_visibility", fake_temporal)
+    args = status.parse_inventory_args(["--format", "json"])
+
+    report, exit_code = asyncio.run(status.build_inventory_report(args))
+
+    assert exit_code == 0
+    assert report["sources"]["definitions"] == {"status": "ok", "source": "db"}
+    assert report["summary"]["automation_count"] == 2
+    assert report["automations"][0]["definition_source"] == "db"
+    assert report["automations"][0]["temporal"]["status"] == "active"
+    assert report["automations"][1]["schedule_id"].endswith("-once")
+
+
+def test_inventory_file_fallback_when_db_url_missing(monkeypatch) -> None:
+    monkeypatch.setattr(status, "file_definitions", lambda: [dict(_definition(), source="files")])
+    args = status.parse_inventory_args(["--db-url", "", "--temporal-host", ""])
+
+    report, exit_code = asyncio.run(status.build_inventory_report(args))
+
+    assert exit_code == 0
+    assert report["sources"]["definitions"]["status"] == "degraded"
+    assert report["automations"][0]["definition_source"] == "files"
+    assert "ACTCORE_DB_URL is not set" in report["warnings"][0]
+
+
+def test_inventory_filters_disabled_definitions() -> None:
+    definitions = [_definition(enabled=True), _scheduled_definition(enabled=False)]
+
+    filtered = status.filter_inventory_definitions(
+        definitions,
+        ids=[],
+        names=[],
+        enabled=False,
+        trigger_types=set(),
+    )
+
+    assert [item["name"] for item in filtered] == ["One Shot"]
+
+
+def test_inventory_temporal_unavailable_is_warning_not_failure(monkeypatch) -> None:
+    async def fake_load_definitions(args, warnings):
+        return [_definition()], {"status": "ok", "source": "db"}
+
+    async def fake_temporal(host, namespace, definitions, *, timeout_seconds):
+        return {}, {"status": "unavailable", "warning": "Temporal unavailable: nope"}
+
+    monkeypatch.setattr(status, "load_definitions", fake_load_definitions)
+    monkeypatch.setattr(status, "load_temporal_visibility", fake_temporal)
+    args = status.parse_inventory_args([])
+
+    report, exit_code = asyncio.run(status.build_inventory_report(args))
+
+    assert exit_code == 0
+    assert report["automations"][0]["temporal"]["status"] == "not_checked"
+    assert report["warnings"] == ["Temporal unavailable: nope"]
+
+
+def test_inventory_cli_emits_json(monkeypatch, capsys) -> None:
+    monkeypatch.setattr(status, "file_definitions", lambda: [dict(_definition(), source="files")])
+
+    exit_code = asyncio.run(status.async_inventory_main([
+        "--db-url", "",
+        "--temporal-host", "",
+        "--format", "json",
+    ]))
+
+    payload = json.loads(capsys.readouterr().out)
+    assert exit_code == 0
+    assert payload["mode"] == "automation-inventory"
+    assert payload["automations"][0]["name"] == "Daily Check"