test(workspace): add 26-case coverage for molecule_audit.hooks

Covers LedgerHooks class (EU AI Act Art. 12 pipeline hooks):
- _to_bytes helper: 6 cases (None, bytes passthrough, str, dict, list, sort)
- LedgerHooks init: 6 cases (session_id, agent_id, db_url, oversight flag)
- Context manager: 3 cases (enter/exit, close, no-session noop)
- Session management: 3 cases (opens lazily, reuses session, close resets)
- Hook methods: 6 cases (on_task_start, on_llm_call, on_tool_call,
  on_task_end, oversight override, agent_id propagation)
- Error swallowing: 2 cases (append error logged, multiple errors swallowed)

Net new: 26 tests for previously zero-covered hooks module.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

workspace/tests/test_molecule_audit_hooks.py

@@ -0,0 +1,227 @@
+"""Tests for molecule_audit.hooks — EU AI Act Art. 12 pipeline hooks.
+
+Covers:
+- LedgerHooks context manager (session lifecycle)
+- on_task_start / on_llm_call / on_tool_call / on_task_end hook methods
+- _safe_append error swallowing
+- _to_bytes helper
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Ensure workspace root is on path
+_ws_root = __file__.rsplit("/tests/", 1)[0]
+if _ws_root not in sys.path:
+    sys.path.insert(0, _ws_root)
+
+from molecule_audit.hooks import LedgerHooks, _to_bytes
+
+
+class TestToBytes:
+    """Unit tests for the _to_bytes helper."""
+
+    def test_none_returns_none(self):
+        assert _to_bytes(None) is None
+
+    def test_bytes_passthrough(self):
+        data = b"hello"
+        assert _to_bytes(data) == data
+
+    def test_str_encoded_utf8(self):
+        assert _to_bytes("hello") == b"hello"
+        assert _to_bytes("こんにちは") == "こんにちは".encode("utf-8")
+
+    def test_dict_json_serialized(self):
+        result = _to_bytes({"key": "value", "num": 42})
+        assert b'"key"' in result and b'"value"' in result and b'"num"' in result
+
+    def test_list_json_serialized(self):
+        result = _to_bytes([1, 2, "three"])
+        # JSON encodes "three" as a string, so it has quotes
+        assert b'"three"' in result or b'three' in result
+        assert b"1" in result
+
+    def test_dict_sorted_keys(self):
+        """Dicts are JSON-serialized with sorted keys for deterministic output."""
+        a = _to_bytes({"b": 1, "a": 2})
+        b = _to_bytes({"a": 2, "b": 1})
+        assert a == b
+
+
+class TestLedgerHooksInit:
+    """LedgerHooks constructor and defaults."""
+
+    def test_session_id_required(self):
+        hooks = LedgerHooks(session_id="task-123")
+        assert hooks.session_id == "task-123"
+
+    def test_agent_id_from_env(self):
+        import os
+        env_id = os.environ.get("WORKSPACE_ID", "unknown-agent")
+        hooks = LedgerHooks(session_id="s1")
+        assert hooks.agent_id == env_id
+
+    def test_agent_id_override(self):
+        hooks = LedgerHooks(session_id="s1", agent_id="explicit-agent")
+        assert hooks.agent_id == "explicit-agent"
+
+    def test_db_url_stored(self):
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        assert hooks._db_url == "sqlite:///:memory:"
+
+    def test_human_oversight_default(self):
+        hooks = LedgerHooks(session_id="s1")
+        assert hooks._default_human_oversight is False
+
+    def test_human_oversight_true(self):
+        hooks = LedgerHooks(session_id="s1", human_oversight_flag=True)
+        assert hooks._default_human_oversight is True
+
+
+class TestLedgerHooksContextManager:
+    """LedgerHooks context manager lifecycle."""
+
+    def test_enter_returns_self(self):
+        hooks = LedgerHooks(session_id="s1")
+        with hooks as entered:
+            assert entered is hooks
+
+    def test_exit_closes_session(self):
+        mock_session = MagicMock()
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        # Pre-open the session via a mock
+        hooks._session = mock_session
+        hooks.__exit__(None, None, None)
+        mock_session.close.assert_called_once()
+
+    def test_exit_no_session_noop(self):
+        hooks = LedgerHooks(session_id="s1")
+        # No session opened — should not raise
+        hooks.__exit__(None, None, None)
+
+
+class TestLedgerHooksOpenSession:
+    """Lazy session opening."""
+
+    def test_opens_session_on_first_call(self):
+        mock_factory = MagicMock()
+        mock_session = MagicMock()
+        mock_factory.return_value = mock_session
+
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        with patch("molecule_audit.hooks.get_session_factory", return_value=mock_factory):
+            session = hooks._open_session()
+        assert session is mock_session
+        mock_factory.assert_called_once()
+
+    def test_reuses_same_session(self):
+        mock_factory = MagicMock()
+        mock_session = MagicMock()
+        mock_factory.return_value = mock_session
+
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        with patch("molecule_audit.hooks.get_session_factory", return_value=mock_factory):
+            s1 = hooks._open_session()
+            s2 = hooks._open_session()
+        assert s1 is s2
+        # Factory called only once (lazy)
+        assert mock_factory.call_count == 1
+
+    def test_close_resets_session(self):
+        mock_session = MagicMock()
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        hooks._session = mock_session
+        hooks.close()
+        assert hooks._session is None
+        mock_session.close.assert_called_once()
+
+
+class TestLedgerHooksHookMethods:
+    """Hook methods call _safe_append with correct kwargs."""
+
+    def _mock_hooks(self):
+        """Return a LedgerHooks with all ledger functions mocked."""
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        mock_session = MagicMock()
+        hooks._session = mock_session
+        return hooks, mock_session
+
+    def test_on_task_start_calls_append(self):
+        hooks, mock_session = self._mock_hooks()
+        with patch("molecule_audit.hooks.append_event") as mock_append:
+            hooks.on_task_start(input_text="user prompt", risk_flag=True)
+        mock_append.assert_called_once()
+        call_kwargs = mock_append.call_args.kwargs
+        assert call_kwargs["operation"] == "task_start"
+        assert call_kwargs["human_oversight_flag"] is False
+        assert call_kwargs["risk_flag"] is True
+
+    def test_on_task_start_respects_human_oversight_override(self):
+        hooks, _ = self._mock_hooks()
+        with patch("molecule_audit.hooks.append_event") as mock_append:
+            hooks.on_task_start(human_oversight_flag=True)
+        assert mock_append.call_args.kwargs["human_oversight_flag"] is True
+
+    def test_on_llm_call_includes_model(self):
+        hooks, _ = self._mock_hooks()
+        with patch("molecule_audit.hooks.append_event") as mock_append:
+            hooks.on_llm_call(model="hermes-4-405b", input_text="prompt", output_text="response")
+        call_kwargs = mock_append.call_args.kwargs
+        assert call_kwargs["operation"] == "llm_call"
+        assert call_kwargs["model_used"] == "hermes-4-405b"
+
+    def test_on_tool_call_includes_tool_name(self):
+        hooks, _ = self._mock_hooks()
+        with patch("molecule_audit.hooks.append_event") as mock_append:
+            hooks.on_tool_call(tool_name="search", input_data={"q": "test"}, output_data=["result"])
+        call_kwargs = mock_append.call_args.kwargs
+        assert call_kwargs["operation"] == "tool_call"
+        assert call_kwargs["model_used"] == "search"
+
+    def test_on_task_end_records_output(self):
+        hooks, _ = self._mock_hooks()
+        with patch("molecule_audit.hooks.append_event") as mock_append:
+            hooks.on_task_end(output_text="final result")
+        call_kwargs = mock_append.call_args.kwargs
+        assert call_kwargs["operation"] == "task_end"
+
+    def test_hooks_use_instance_agent_id(self):
+        hooks = LedgerHooks(session_id="s1", agent_id="my-workspace", db_url="sqlite:///:memory:")
+        mock_session = MagicMock()
+        hooks._session = mock_session
+        with patch("molecule_audit.hooks.append_event") as mock_append:
+            hooks.on_task_start()
+        assert mock_append.call_args.kwargs["agent_id"] == "my-workspace"
+
+
+class TestLedgerHooksSafeAppend:
+    """_safe_append swallows all exceptions without re-raising."""
+
+    def test_append_error_swallowed_and_logged(self, caplog):
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        mock_session = MagicMock()
+        hooks._session = mock_session
+
+        error = RuntimeError("DB write failed")
+        with patch("molecule_audit.hooks.append_event", side_effect=error):
+            with caplog.at_level(logging.WARNING):
+                # Should not raise
+                hooks.on_task_start(input_text="test")
+        assert any("DB write failed" in r.message for r in caplog.records)
+
+    def test_multiple_exceptions_swallowed(self):
+        hooks = LedgerHooks(session_id="s1", db_url="sqlite:///:memory:")
+        mock_session = MagicMock()
+        hooks._session = mock_session
+
+        with patch("molecule_audit.hooks.append_event", side_effect=RuntimeError("err")):
+            # All three calls should succeed (no exception)
+            hooks.on_task_start(input_text="a")
+            hooks.on_llm_call(model="m", input_text="b")
+            hooks.on_tool_call(tool_name="t", input_data={})