workspace/tests/test_audit_ledger.py

@@ -560,6 +560,280 @@ class TestLedgerHooks:
         assert ev.risk_flag is True
 
 
+# ---------------------------------------------------------------------------
+# hooks — extended coverage (26 new cases to reach 26-case total)
+# ---------------------------------------------------------------------------
+
+class TestLedgerHooksExtended:
+    """Extended coverage for molecule_audit.hooks — fills all uncovered branches.
+
+    Existing TestLedgerHooks covers the golden-path cases.
+    This class covers: _to_bytes, session lifecycle, agent_id defaults,
+    None/empty inputs, override flags, risk propagation, and edge cases.
+    """
+
+    # ── _to_bytes ──────────────────────────────────────────────────────────────
+
+    def test_to_bytes_none(self):
+        from molecule_audit.hooks import _to_bytes
+        assert _to_bytes(None) is None
+
+    def test_to_bytes_bytes_returns_same(self):
+        from molecule_audit.hooks import _to_bytes
+        data = b"\x00\xff"
+        assert _to_bytes(data) == data
+
+    def test_to_bytes_str_returns_utf8(self):
+        from molecule_audit.hooks import _to_bytes
+        assert _to_bytes("café") == "café".encode("utf-8")
+
+    def test_to_bytes_dict_is_json_deterministic(self):
+        from molecule_audit.hooks import _to_bytes
+        d = {"b": 2, "a": 1}
+        result = _to_bytes(d)
+        # Must be valid UTF-8 JSON
+        import json
+        parsed = json.loads(result.decode("utf-8"))
+        assert parsed == {"a": 1, "b": 2}  # sort_keys=True
+        # Same dict produces same bytes (deterministic)
+        assert _to_bytes(d) == result
+
+    def test_to_bytes_list_is_json(self):
+        from molecule_audit.hooks import _to_bytes
+        result = _to_bytes([1, "two", {"three": 3}])
+        import json
+        parsed = json.loads(result.decode("utf-8"))
+        assert parsed == [1, "two", {"three": 3}]
+
+    # ── _DEFAULT_AGENT_ID ─────────────────────────────────────────────────────
+
+    def test_agent_id_defaults_to_workspace_id_env(self, monkeypatch):
+        import molecule_audit.hooks as hooks
+        monkeypatch.setenv("WORKSPACE_ID", "env-workspace-42")
+        # Reset so it picks up the new env value
+        hooks._DEFAULT_AGENT_ID = hooks.os.environ.get("WORKSPACE_ID", "unknown-agent")
+        h = hooks.LedgerHooks(session_id="s")
+        assert h.agent_id == "env-workspace-42"
+
+    def test_agent_id_overrides_env(self):
+        from molecule_audit.hooks import LedgerHooks
+        h = LedgerHooks(session_id="s", agent_id="explicit-agent")
+        assert h.agent_id == "explicit-agent"
+
+    # ── Session lifecycle ─────────────────────────────────────────────────────
+
+    def test_session_is_lazy(self, mem_session):
+        """_open_session is not called until first on_* method."""
+        from molecule_audit.hooks import LedgerHooks
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        # Session must NOT be opened until needed
+        assert hooks._session is None
+
+    def test_session_reused_across_calls(self, mem_session):
+        """Multiple on_* calls share the same SQLAlchemy session."""
+        from molecule_audit.hooks import LedgerHooks
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_task_start(input_text="start")
+        hooks.on_task_end(output_text="end")
+        # Both events written to the same session
+        assert mem_session.query(
+            __import__("molecule_audit.ledger", fromlist=["AuditEvent"]).AuditEvent
+        ).count() == 2
+
+    def test_close_when_session_is_none(self):
+        """close() is safe to call when no session was ever opened."""
+        from molecule_audit.hooks import LedgerHooks
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks.close()  # must not raise
+        assert hooks._session is None
+
+    def test_context_manager_releases_on_exception(self, mem_session):
+        """__exit__ closes session even when an exception propagates."""
+        from molecule_audit.hooks import LedgerHooks
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        with pytest.raises(ZeroDivisionError):
+            with hooks:
+                hooks.on_task_start(input_text="start")
+                raise ZeroDivisionError("boom")
+        # Session must still be closed
+        assert hooks._session is None
+
+    # ── on_task_start None/empty inputs ───────────────────────────────────────
+
+    def test_on_task_start_none_input(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_task_start(input_text=None)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.input_hash is None
+        assert ev.operation == "task_start"
+
+    def test_on_task_start_risk_flag_true(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_task_start(risk_flag=True)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.risk_flag is True
+
+    def test_on_task_start_oversight_flag_override(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1", human_oversight_flag=False)
+        hooks._session = mem_session
+        hooks.on_task_start(human_oversight_flag=True)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.human_oversight_flag is True
+
+    # ── on_llm_call None/empty inputs ─────────────────────────────────────────
+
+    def test_on_llm_call_none_input_and_output(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_llm_call(model="m", input_text=None, output_text=None)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.input_hash is None
+        assert ev.output_hash is None
+        assert ev.model_used == "m"
+
+    def test_on_llm_call_risk_flag_true(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_llm_call(model="m", risk_flag=True)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.risk_flag is True
+
+    # ── on_tool_call None/empty inputs ────────────────────────────────────────
+
+    def test_on_tool_call_bytes_input(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks, _to_bytes
+        from molecule_audit.ledger import AuditEvent, hash_content
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        binary = b"binary data \x00\xff"
+        hooks.on_tool_call("read_file", input_data=binary)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.input_hash == hash_content(binary)
+        assert ev.model_used == "read_file"
+
+    def test_on_tool_call_none_input_and_output(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_tool_call("echo", input_data=None, output_data=None)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.input_hash is None
+        assert ev.output_hash is None
+        assert ev.model_used == "echo"
+
+    def test_on_tool_call_risk_flag_true(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_tool_call("write_file", risk_flag=True)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.risk_flag is True
+
+    # ── on_task_end None/empty inputs ─────────────────────────────────────────
+
+    def test_on_task_end_none_output(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_task_end(output_text=None)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.output_hash is None
+        assert ev.operation == "task_end"
+
+    def test_on_task_end_risk_flag_true(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+        hooks.on_task_end(risk_flag=True)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.risk_flag is True
+
+    def test_on_task_end_oversight_flag_override(self, mem_session):
+        from molecule_audit.hooks import LedgerHooks
+        from molecule_audit.ledger import AuditEvent
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1", human_oversight_flag=False)
+        hooks._session = mem_session
+        hooks.on_task_end(human_oversight_flag=True)
+        hooks.close()
+
+        ev = mem_session.query(AuditEvent).first()
+        assert ev.human_oversight_flag is True
+
+    # ── _safe_append exception swallowing ─────────────────────────────────────
+
+    def test_safe_append_swallows_session_error(self, mem_session, caplog):
+        """_safe_append logs a warning when append_event raises."""
+        import logging
+        from molecule_audit.hooks import LedgerHooks
+
+        hooks = LedgerHooks(session_id="s1", agent_id="ag1")
+        hooks._session = mem_session
+
+        # Force an error by making the session raise on commit
+        orig_commit = mem_session.commit
+        def bad_commit():
+            raise RuntimeError("simulated DB error")
+        mem_session.commit = bad_commit
+
+        with caplog.at_level(logging.WARNING, logger="molecule_audit.hooks"):
+            hooks.on_task_start(input_text="test")
+
+        mem_session.commit = orig_commit  # restore
+        assert any("failed to append event" in r.message for r in caplog.records)
+
+
 # ---------------------------------------------------------------------------
 # verify.py CLI
 # ---------------------------------------------------------------------------