diff --git a/src/ian/services/notifications.py b/src/ian/services/notifications.py
index 05de41a..faf634b 100644
--- a/src/ian/services/notifications.py
+++ b/src/ian/services/notifications.py
@@ -23,7 +23,7 @@
 from ian.config import DISCORD_BOT_TOKEN, DISCORD_LOG_CHANNEL_ID
 from ian.domain.reminders import get_valid_bound_members
 from ian.services import discord_api
-from ian.utils.console import eprint
+from ian.utils.logging import log_event
 
 
 LOG_CHANNEL_ID = DISCORD_LOG_CHANNEL_ID
@@ -33,14 +33,43 @@
 def send_discord_dm(user_id: str, text: str) -> bool:
     response = discord_api.create_dm_channel(user_id)
     if response.status_code != 200:
-        eprint(f"  [Discord] Failed to create DM channel for {user_id}: {response.text}")
+        log_event(
+            "discord_dm_delivery",
+            "notifications",
+            level="warning",
+            platform="Discord",
+            status="failure",
+            stage="create_channel",
+            user_id=user_id,
+            http_status=response.status_code,
+        )
         return False
 
     dm_channel_id = response.json()["id"]
     message_response = discord_api.send_channel_message(dm_channel_id, text)
     if message_response.status_code != 200:
-        eprint(f"  [Discord] Failed to send message to {user_id}: {message_response.text}")
+        log_event(
+            "discord_dm_delivery",
+            "notifications",
+            level="warning",
+            platform="Discord",
+            status="failure",
+            stage="send_message",
+            user_id=user_id,
+            channel_id=dm_channel_id,
+            http_status=message_response.status_code,
+        )
         return False
+    log_event(
+        "discord_dm_delivery",
+        "notifications",
+        platform="Discord",
+        status="success",
+        stage="send_message",
+        user_id=user_id,
+        channel_id=dm_channel_id,
+        http_status=message_response.status_code,
+    )
     return True
 
 
@@ -57,12 +86,35 @@ def send_discord_channel_message(channel_id: str, message: str) -> bool:
     try:
         response = discord_api.send_channel_message(channel_id, message)
         if response.status_code in (200, 201):
-            eprint(f"[notify_staff] 成功發送通知到 Discord channel {channel_id}")
+            log_event(
+                "discord_channel_message",
+                "notifications",
+                platform="Discord",
+                status="success",
+                channel_id=channel_id,
+                http_status=response.status_code,
+            )
             return True
-        eprint(f"[notify_staff] 發送失敗: {response.status_code} - {response.text}")
+        log_event(
+            "discord_channel_message",
+            "notifications",
+            level="warning",
+            platform="Discord",
+            status="failure",
+            channel_id=channel_id,
+            http_status=response.status_code,
+        )
         return False
     except Exception as e:
-        eprint(f"[notify_staff] 發送通知時發生錯誤: {e}")
+        log_event(
+            "discord_channel_message",
+            "notifications",
+            level="error",
+            platform="Discord",
+            status="error",
+            channel_id=channel_id,
+            error=e,
+        )
         return False
 
 
diff --git a/src/ian/utils/logging.py b/src/ian/utils/logging.py
new file mode 100644
index 0000000..4e1a96d
--- /dev/null
+++ b/src/ian/utils/logging.py
@@ -0,0 +1,245 @@
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# Copyright (c) 2026 NTU AI Club
+#
+# This file is part of Ian, an open-source AI agent framework developed
+# and maintained by NTU AI Club.
+#
+# Ian is licensed under the GNU General Public License, either version 3
+# of the License, or (at your option) any later version.
+#
+# Ian is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+# See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ian. If not, see <https://www.gnu.org/licenses/>.
+#
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+import threading
+from collections.abc import Callable, Mapping
+from datetime import datetime
+from typing import Any, TextIO
+
+from ian.config import TZ_TPE
+
+
+LOG_LEVELS = frozenset({"debug", "info", "warning", "error", "critical"})
+REDACTED = "[REDACTED]"
+Identifier = str | int | None
+
+_IDENTIFIER_FIELDS = frozenset(
+    {
+        "account_id",
+        "channel_id",
+        "recipient_id",
+        "sender_id",
+        "session_id",
+        "user_id",
+    }
+)
+_CONTENT_FIELDS = frozenset(
+    {
+        "content",
+        "input",
+        "message",
+        "message_content",
+        "prompt",
+        "query",
+        "raw_content",
+        "request_body",
+        "response",
+        "text",
+        "user_message",
+    }
+)
+_RESERVED_FIELDS = frozenset(
+    {
+        "timestamp",
+        "level",
+        "event",
+        "component",
+        "platform",
+        "status",
+        "duration_ms",
+        "error_type",
+        "correlation_id",
+    }
+)
+
+
+def _stable_hash(value: Identifier) -> str:
+    if value is None:
+        return ""
+    text = str(value).strip()
+    if not text:
+        return ""
+    digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
+    return f"sha256:{digest}"
+
+
+def hash_account_id(account_id: Identifier) -> str:
+    """Return a stable pseudonym for a platform account identifier."""
+    return _stable_hash(account_id)
+
+
+def hash_email(email: str | None) -> str:
+    """Return a stable, case-insensitive pseudonym for an email address."""
+    if email is None:
+        return ""
+    return _stable_hash(str(email).strip().lower())
+
+
+def redact_token(token: str | None) -> str:
+    """Remove a token or secret while preserving whether a value was present."""
+    if token is None:
+        return ""
+    return REDACTED if str(token).strip() else ""
+
+
+def redact_user_content(content: str | None) -> str:
+    """Remove raw user content and retain only its character count."""
+    if content is None:
+        return ""
+    text = str(content)
+    return f"[REDACTED content_length={len(text)}]" if text else ""
+
+
+def _redaction_kind(field_name: str) -> str | None:
+    normalized = field_name.lower()
+    if (
+        "token" in normalized
+        or "secret" in normalized
+        or normalized.endswith("api_key")
+        or normalized == "authorization"
+    ):
+        return "token"
+    if normalized == "email" or normalized.endswith("_email"):
+        return "email"
+    if normalized in _IDENTIFIER_FIELDS or normalized.endswith("_account_id"):
+        return "identifier"
+    if normalized in _CONTENT_FIELDS or normalized.endswith("_content"):
+        return "content"
+    return None
+
+
+def _as_identifier(value: Any) -> Identifier:
+    if value is None or isinstance(value, str):
+        return value
+    if isinstance(value, int) and not isinstance(value, bool):
+        return value
+    return str(value)
+
+
+def _as_text(value: Any) -> str | None:
+    return None if value is None else str(value)
+
+
+def _sanitize_value(field_name: str, value: Any) -> Any:
+    kind = _redaction_kind(field_name)
+    if kind == "token":
+        return redact_token(_as_text(value))
+    if kind == "email":
+        return hash_email(_as_text(value))
+    if kind == "identifier":
+        return hash_account_id(_as_identifier(value))
+    if kind == "content":
+        return redact_user_content(_as_text(value))
+    if isinstance(value, Mapping):
+        return {str(key): _sanitize_value(str(key), item) for key, item in value.items()}
+    if isinstance(value, (list, tuple, set)):
+        return [_sanitize_value(field_name, item) for item in value]
+    return value
+
+
+def sanitize_log_fields(fields: Mapping[str, Any]) -> dict[str, Any]:
+    """Apply the default sensitive-field policy to structured log fields."""
+    return {str(key): _sanitize_value(str(key), value) for key, value in fields.items()}
+
+
+class StructuredLogger:
+    """Emit sanitized JSON Lines to stderr or another local text stream."""
+
+    def __init__(
+        self,
+        *,
+        stream: TextIO | None = None,
+        clock: Callable[[], datetime] | None = None,
+    ) -> None:
+        self._stream = stream
+        self._clock = clock or (lambda: datetime.now(TZ_TPE))
+        self._lock = threading.Lock()
+
+    def emit(
+        self,
+        event: str,
+        component: str,
+        *,
+        level: str = "info",
+        platform: str | None = None,
+        status: str | None = None,
+        duration_ms: float | int | None = None,
+        error: BaseException | None = None,
+        correlation_id: str | None = None,
+        **fields: Any,
+    ) -> dict[str, Any]:
+        if level not in LOG_LEVELS:
+            raise ValueError(f"Unsupported log level: {level}")
+        if not isinstance(event, str) or not event.strip():
+            raise ValueError("event and component are required")
+        if not isinstance(component, str) or not component.strip():
+            raise ValueError("event and component are required")
+
+        conflicting = _RESERVED_FIELDS.intersection(fields)
+        if conflicting:
+            names = ", ".join(sorted(conflicting))
+            raise ValueError(f"Reserved log fields must use named parameters: {names}")
+
+        timestamp = self._clock()
+        if timestamp.tzinfo is None:
+            timestamp = timestamp.replace(tzinfo=TZ_TPE)
+        timestamp = timestamp.astimezone(TZ_TPE)
+
+        entry: dict[str, Any] = {
+            "timestamp": timestamp.isoformat(),
+            "level": level,
+            "event": event,
+            "component": component,
+        }
+        optional = {
+            "platform": platform,
+            "status": status,
+            "duration_ms": duration_ms,
+            "error_type": type(error).__name__ if error is not None else None,
+            "correlation_id": correlation_id,
+        }
+        entry.update({key: value for key, value in optional.items() if value is not None})
+        entry.update(sanitize_log_fields(fields))
+
+        payload = json.dumps(
+            entry,
+            ensure_ascii=False,
+            separators=(",", ":"),
+            default=str,
+            allow_nan=False,
+        )
+        stream = self._stream or sys.stderr
+        with self._lock:
+            stream.write(payload + "\n")
+            stream.flush()
+        return entry
+
+
+_application_logger = StructuredLogger()
+
+
+def log_event(event: str, component: str, **kwargs: Any) -> dict[str, Any]:
+    """Emit an application event through the shared structured logger."""
+    return _application_logger.emit(event, component, **kwargs)
diff --git a/tests/services/test_discord_api.py b/tests/services/test_discord_api.py
index 91ad7bc..1edaa6e 100644
--- a/tests/services/test_discord_api.py
+++ b/tests/services/test_discord_api.py
@@ -18,6 +18,8 @@
 # along with Ian. If not, see <https://www.gnu.org/licenses/>.
 #
 
+import json
+
 from ian.services import notifications
 from ian.services.agent import logging as agent_logging
 
@@ -88,7 +90,7 @@ def fake_post(url, headers, json, timeout):
     ]
 
 
-def test_send_discord_dm_uses_shared_client_and_preserves_failure_message(
+def test_send_discord_dm_uses_shared_client_and_redacts_failure_details(
     monkeypatch, capsys
 ):
     from ian.services import discord_api
@@ -101,8 +103,13 @@ def fake_create_dm_channel(user_id):
 
     assert notifications.send_discord_dm("user-1", "hello") is False
 
-    captured = capsys.readouterr()
-    assert "Failed to create DM channel for user-1: cannot create" in captured.err
+    log_entry = json.loads(capsys.readouterr().err)
+    assert log_entry["event"] == "discord_dm_delivery"
+    assert log_entry["status"] == "failure"
+    assert log_entry["stage"] == "create_channel"
+    assert log_entry["user_id"].startswith("sha256:")
+    assert "user-1" not in json.dumps(log_entry)
+    assert "cannot create" not in json.dumps(log_entry)
 
 
 def test_send_discord_channel_message_uses_shared_client_for_success(monkeypatch, capsys):
@@ -120,7 +127,11 @@ def fake_send_channel_message(channel_id, message):
     assert calls == [("channel-1", "hello")]
 
     captured = capsys.readouterr()
-    assert "成功發送通知到 Discord channel channel-1" in captured.err
+    log_entry = json.loads(captured.err)
+    assert log_entry["event"] == "discord_channel_message"
+    assert log_entry["status"] == "success"
+    assert log_entry["channel_id"].startswith("sha256:")
+    assert "channel-1" not in captured.err
 
 
 def test_agent_logging_uses_shared_client_for_failure(monkeypatch, capsys):
diff --git a/tests/services/test_notifications.py b/tests/services/test_notifications.py
index 19880e7..61b7753 100644
--- a/tests/services/test_notifications.py
+++ b/tests/services/test_notifications.py
@@ -18,6 +18,7 @@
 # along with Ian. If not, see <https://www.gnu.org/licenses/>.
 #
 
+import json
 from types import SimpleNamespace
 
 import pytest
@@ -99,6 +100,59 @@ def test_send_notification_to_members_aggregates_delivery_results(
     assert sleep_calls == [0.5] * len(delivery_results)
 
 
+@pytest.mark.parametrize(
+    ("create_status", "send_status", "expected", "expected_stage"),
+    [
+        pytest.param(500, None, False, "create_channel", id="create-channel-failure"),
+        pytest.param(200, 500, False, "send_message", id="send-message-failure"),
+        pytest.param(200, 200, True, "send_message", id="success"),
+    ],
+)
+def test_send_discord_dm_emits_redacted_delivery_result(
+    monkeypatch,
+    capsys,
+    create_status,
+    send_status,
+    expected,
+    expected_stage,
+):
+    send_calls = []
+    monkeypatch.setattr(
+        notifications.discord_api,
+        "create_dm_channel",
+        lambda _user_id: SimpleNamespace(
+            status_code=create_status,
+            text="private create response",
+            json=lambda: {"id": "dm-channel-1"},
+        ),
+    )
+    monkeypatch.setattr(
+        notifications.discord_api,
+        "send_channel_message",
+        lambda channel_id, message: send_calls.append((channel_id, message))
+        or SimpleNamespace(status_code=send_status, text="private send response"),
+    )
+
+    result = notifications.send_discord_dm("user-1", "private message")
+
+    assert result is expected
+    assert send_calls == (
+        [] if send_status is None else [("dm-channel-1", "private message")]
+    )
+    log_entry = json.loads(capsys.readouterr().err)
+    assert log_entry["status"] == ("success" if expected else "failure")
+    assert log_entry["stage"] == expected_stage
+    serialized = json.dumps(log_entry)
+    for sensitive in (
+        "user-1",
+        "dm-channel-1",
+        "private message",
+        "private create response",
+        "private send response",
+    ):
+        assert sensitive not in serialized
+
+
 @pytest.mark.parametrize(
     ("status_code", "expected"),
     [
@@ -109,7 +163,7 @@ def test_send_notification_to_members_aggregates_delivery_results(
     ],
 )
 def test_send_discord_channel_message_handles_response_statuses(
-    monkeypatch, status_code, expected
+    monkeypatch, capsys, status_code, expected
 ):
     calls = []
     monkeypatch.setattr(
@@ -121,15 +175,25 @@ def test_send_discord_channel_message_handles_response_statuses(
 
     assert notifications.send_discord_channel_message("channel-1", "Notice") is expected
     assert calls == [("channel-1", "Notice")]
+    log_entry = json.loads(capsys.readouterr().err)
+    assert log_entry["status"] == ("success" if expected else "failure")
+    assert log_entry["http_status"] == status_code
+    assert "channel-1" not in json.dumps(log_entry)
+    assert "response body" not in json.dumps(log_entry)
 
 
-def test_send_discord_channel_message_handles_api_exception(monkeypatch):
+def test_send_discord_channel_message_handles_api_exception(monkeypatch, capsys):
     def fail(*_args):
         raise RuntimeError("Discord unavailable")
 
     monkeypatch.setattr(notifications.discord_api, "send_channel_message", fail)
 
     assert notifications.send_discord_channel_message("channel-1", "Notice") is False
+    log_entry = json.loads(capsys.readouterr().err)
+    assert log_entry["level"] == "error"
+    assert log_entry["status"] == "error"
+    assert log_entry["error_type"] == "RuntimeError"
+    assert "Discord unavailable" not in json.dumps(log_entry)
 
 
 @pytest.mark.parametrize(
diff --git a/tests/test_logging.py b/tests/test_logging.py
new file mode 100644
index 0000000..e857527
--- /dev/null
+++ b/tests/test_logging.py
@@ -0,0 +1,321 @@
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# Copyright (c) 2026 NTU AI Club
+#
+# This file is part of Ian, an open-source AI agent framework developed
+# and maintained by NTU AI Club.
+#
+# Ian is licensed under the GNU General Public License, either version 3
+# of the License, or (at your option) any later version.
+#
+# Ian is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+# See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ian. If not, see <https://www.gnu.org/licenses/>.
+#
+
+import io
+import json
+from datetime import UTC, datetime
+
+import pytest
+
+from ian.config import TZ_TPE
+from ian.utils.logging import (
+    REDACTED,
+    StructuredLogger,
+    hash_account_id,
+    hash_email,
+    log_event,
+    redact_token,
+    redact_user_content,
+    sanitize_log_fields,
+)
+
+
+def test_structured_logger_emits_json_line_with_standard_fields():
+    stream = io.StringIO()
+    logger = StructuredLogger(
+        stream=stream,
+        clock=lambda: datetime(2026, 7, 13, 9, 2, 3, tzinfo=TZ_TPE),
+    )
+
+    entry = logger.emit(
+        "request_completed",
+        "webhook",
+        platform="LINE",
+        status="success",
+        duration_ms=12.5,
+        correlation_id="request-1",
+        result_count=2,
+    )
+
+    assert json.loads(stream.getvalue()) == entry == {
+        "timestamp": "2026-07-13T09:02:03+08:00",
+        "level": "info",
+        "event": "request_completed",
+        "component": "webhook",
+        "platform": "LINE",
+        "status": "success",
+        "duration_ms": 12.5,
+        "correlation_id": "request-1",
+        "result_count": 2,
+    }
+    assert stream.getvalue().endswith("\n")
+
+
+def test_shared_logger_preserves_stderr_console_workflow(capsys):
+    log_event("service_started", "reminder", status="success")
+
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert json.loads(captured.err)["event"] == "service_started"
+
+
+def test_structured_logger_interprets_naive_clock_in_project_timezone():
+    stream = io.StringIO()
+    logger = StructuredLogger(
+        stream=stream,
+        clock=lambda: datetime(2026, 7, 13, 9, 2, 3),
+    )
+
+    logger.emit("service_started", "reminder")
+
+    assert json.loads(stream.getvalue())["timestamp"] == "2026-07-13T09:02:03+08:00"
+
+
+def test_structured_logger_converts_aware_clock_to_project_timezone():
+    stream = io.StringIO()
+    logger = StructuredLogger(
+        stream=stream,
+        clock=lambda: datetime(2026, 7, 13, 1, 2, 3, tzinfo=UTC),
+    )
+
+    logger.emit("service_started", "reminder")
+
+    assert json.loads(stream.getvalue())["timestamp"] == "2026-07-13T09:02:03+08:00"
+
+
+def test_structured_logger_emits_one_json_object_per_line():
+    stream = io.StringIO()
+    logger = StructuredLogger(stream=stream)
+
+    logger.emit("service_started", "reminder")
+    logger.emit("service_stopped", "reminder")
+
+    entries = [json.loads(line) for line in stream.getvalue().splitlines()]
+    assert [entry["event"] for entry in entries] == [
+        "service_started",
+        "service_stopped",
+    ]
+
+
+@pytest.mark.parametrize(
+    ("helper", "first", "second"),
+    [
+        pytest.param(hash_account_id, "account-123", "account-456", id="account-id"),
+        pytest.param(hash_email, "Member@Example.test", "other@example.test", id="email"),
+    ],
+)
+def test_hash_helpers_are_stable_without_exposing_values(helper, first, second):
+    hashed = helper(first)
+
+    assert hashed == helper(first)
+    assert hashed != helper(second)
+    assert first.lower() not in hashed
+    assert hashed.startswith("sha256:")
+
+
+def test_email_hashing_is_case_insensitive():
+    assert hash_email("Member@Example.test") == hash_email("member@example.test")
+
+
+def test_account_id_hashing_accepts_numeric_and_missing_identifiers():
+    assert hash_account_id(12345).startswith("sha256:")
+    assert hash_account_id(None) == ""
+
+
+@pytest.mark.parametrize(
+    ("helper", "value"),
+    [
+        pytest.param(hash_account_id, "", id="empty-account-id"),
+        pytest.param(hash_account_id, "   ", id="blank-account-id"),
+        pytest.param(hash_email, None, id="missing-email"),
+        pytest.param(hash_email, "", id="empty-email"),
+        pytest.param(hash_email, "   ", id="blank-email"),
+    ],
+)
+def test_hash_helpers_treat_empty_or_blank_values_as_missing(helper, value):
+    assert helper(value) == ""
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        pytest.param("secret-token", REDACTED, id="token"),
+        pytest.param("", "", id="empty-token"),
+        pytest.param(None, "", id="missing-token"),
+    ],
+)
+def test_redact_token_removes_secret(value, expected):
+    assert redact_token(value) == expected
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        pytest.param(None, "", id="missing"),
+        pytest.param("", "", id="empty"),
+        pytest.param("private user question", 21, id="ascii"),
+        pytest.param("台灣 AI", 5, id="unicode"),
+    ],
+)
+def test_redact_user_content_preserves_only_length(raw, expected):
+    redacted = redact_user_content(raw)
+
+    if isinstance(expected, int):
+        assert redacted == f"[REDACTED content_length={expected}]"
+        assert raw not in redacted
+    else:
+        assert redacted == expected
+
+
+def test_sanitize_log_fields_redacts_nested_sensitive_values():
+    fields = sanitize_log_fields(
+        {
+            "account_id": "account-123",
+            "email": "member@example.test",
+            "access_token": "token-123",
+            "user_message": "private question",
+            "query": "private search",
+            "context": {
+                "sender_id": "sender-123",
+                "api_key": "key-123",
+            },
+            "safe_count": 3,
+        }
+    )
+
+    serialized = json.dumps(fields)
+    for sensitive in (
+        "account-123",
+        "member@example.test",
+        "token-123",
+        "private question",
+        "private search",
+        "sender-123",
+        "key-123",
+    ):
+        assert sensitive not in serialized
+    assert fields["access_token"] == REDACTED
+    assert fields["context"]["api_key"] == REDACTED
+    assert fields["safe_count"] == 3
+
+
+@pytest.mark.parametrize(
+    ("field_name", "value"),
+    [
+        pytest.param("backup_account_id", "account-123", id="account-id-suffix"),
+        pytest.param("contact_email", "member@example.test", id="email-suffix"),
+        pytest.param("oauth_token_value", "token-123", id="token-substring"),
+        pytest.param("generated_content", "private answer", id="content-suffix"),
+        pytest.param("authorization", "Bearer secret", id="authorization"),
+    ],
+)
+def test_sanitize_log_fields_applies_sensitive_field_naming_rules(field_name, value):
+    sanitized = sanitize_log_fields({field_name: value})
+
+    assert value not in json.dumps(sanitized)
+
+
+def test_sanitize_log_fields_handles_nested_sequences_and_non_string_values():
+    fields = sanitize_log_fields(
+        {
+            "items": [
+                {"account_id": 12345},
+                {"email": None},
+                {"user_message": "private question"},
+            ],
+            "flags": (True, False),
+        }
+    )
+
+    serialized = json.dumps(fields)
+    assert fields["items"][0]["account_id"].startswith("sha256:")
+    assert fields["items"][1]["email"] == ""
+    assert fields["flags"] == [True, False]
+    assert "12345" not in serialized
+    assert "private question" not in serialized
+
+
+def test_sanitize_log_fields_coerces_unexpected_identifier_types_before_hashing():
+    fields = sanitize_log_fields({"account_id": True})
+
+    assert fields["account_id"] == hash_account_id("True")
+
+
+def test_structured_logger_records_error_type_without_error_message():
+    stream = io.StringIO()
+    logger = StructuredLogger(stream=stream)
+
+    logger.emit(
+        "operation_failed",
+        "member_store",
+        level="error",
+        status="error",
+        error=RuntimeError("token-123 leaked"),
+    )
+
+    entry = json.loads(stream.getvalue())
+    assert entry["error_type"] == "RuntimeError"
+    assert "token-123" not in stream.getvalue()
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "message"),
+    [
+        pytest.param({"level": "verbose"}, "Unsupported log level", id="level"),
+        pytest.param({"event": ""}, "event and component are required", id="event"),
+        pytest.param({"event": "   "}, "event and component are required", id="blank-event"),
+        pytest.param(
+            {"component": "   "},
+            "event and component are required",
+            id="blank-component",
+        ),
+        pytest.param(
+            {"fields": {"timestamp": "override"}},
+            "Reserved log fields",
+            id="reserved-field",
+        ),
+    ],
+)
+def test_structured_logger_rejects_invalid_schema(kwargs, message):
+    logger = StructuredLogger(stream=io.StringIO())
+    event = kwargs.pop("event", "event")
+    component = kwargs.pop("component", "component")
+    fields = kwargs.pop("fields", {})
+
+    with pytest.raises(ValueError, match=message):
+        logger.emit(event, component, **kwargs, **fields)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        pytest.param(float("nan"), id="nan"),
+        pytest.param(float("inf"), id="positive-infinity"),
+        pytest.param(float("-inf"), id="negative-infinity"),
+    ],
+)
+def test_structured_logger_rejects_non_finite_numbers_that_are_invalid_json(value):
+    stream = io.StringIO()
+    logger = StructuredLogger(stream=stream)
+
+    with pytest.raises(ValueError, match="JSON compliant"):
+        logger.emit("measurement_recorded", "metrics", measurement=value)
+
+    assert stream.getvalue() == ""