diff --git a/docs/internal/sandboxes-setup-guide.md b/docs/internal/sandboxes-setup-guide.md index f304f9f95591..e92bb5f90093 100644 --- a/docs/internal/sandboxes-setup-guide.md +++ b/docs/internal/sandboxes-setup-guide.md @@ -237,6 +237,25 @@ repositories. > **Note:** This only works with `SANDBOX_PROVIDER=docker`. +### Task-run log mirroring to PostHog Logs (dogfooding) + +Task-run log entries (the JSONL appended to object storage via `TaskRun.append_log`) are also mirrored into the PostHog Logs product, +so runs can be browsed and sampled in the Logs UI instead of fetching S3 blobs. + +There is no transport of its own: entries are emitted as structured stdout log lines (`event=task_run_log`), +and the per-cluster OTel collector that already ships all container stdout into the region's internal PostHog project picks them up +(locally, `otel-collector-config.dev.yaml` does the same into your dev logs project). +The collector parses each JSON key into a queryable attribute and turns the emitted `request_id` (the run uuid) into a trace id, +so one run groups as a trace and can be pulled up with an attribute filter on `task_run_id`. + +```bash +# Which task origins to mirror (comma-separated). Defaults to signals scouts only. +# Set empty to disable. +TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS=signals_scout +``` + +Mirroring failures are logged and never break the run's log write. + ### How `MODAL_DOCKER` works When both `SANDBOX_PROVIDER=MODAL_DOCKER` and `LOCAL_POSTHOG_CODE_MONOREPO_ROOT` are set: diff --git a/posthog/settings/temporal.py b/posthog/settings/temporal.py index ac2465e0d556..d8759d212567 100644 --- a/posthog/settings/temporal.py +++ b/posthog/settings/temporal.py @@ -76,6 +76,16 @@ "TASKS_CREDENTIAL_REFRESH_INITIAL_DELAY_SECONDS", 0, type_cast=int ) +# Mirror persisted task-run logs into the PostHog Logs product (dogfooding). +# Entries appended to a run's S3 JSONL log are also emitted as structured stdout log lines; +# the per-cluster OTel collector already ships container stdout into the region's internal +# PostHog project's Logs, so no transport or credentials are needed here. Only runs whose +# task origin_product is in this list are mirrored — scoped to signals scouts for now; +# widen the list to cover more task origins, or set it empty to disable. +TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS: list[str] = get_list( + os.getenv("TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS", "signals_scout") +) + TEMPORAL_LOG_LEVEL_PRODUCE: str = os.getenv("TEMPORAL_LOG_LEVEL_PRODUCE", "DEBUG") TEMPORAL_EXTERNAL_LOGS_QUEUE_SIZE: int = get_from_env("TEMPORAL_EXTERNAL_LOGS_QUEUE_SIZE", 0, type_cast=int) diff --git a/products/tasks/backend/logic/services/run_log_mirror.py b/products/tasks/backend/logic/services/run_log_mirror.py new file mode 100644 index 000000000000..5869d10b7189 --- /dev/null +++ b/products/tasks/backend/logic/services/run_log_mirror.py @@ -0,0 +1,148 @@ +"""Mirror persisted task-run log entries into the PostHog Logs product via stdout. + +Task-run logs are appended to object storage as one ACP notification envelope per line. +In every PostHog cluster an OTel collector daemonset already tails container stdout and +ships JSON log lines into the region's internal PostHog project's Logs product, parsing +each JSON key into a queryable log attribute, `level` into severity, and `request_id` +into a trace id (see `argocd/otel-collector` in the charts repo; `otel-collector-config.dev.yaml` +does the same for local dev). So dogfooding scout-run logs needs no transport of its own: +emitting one structured stdout line per persisted entry is enough. + +Each mirrored line carries the run's uuid as `request_id`, so a whole run groups as one +trace in the Logs UI and can be pulled up with a `task_run_id` attribute filter. +""" + +import json +from typing import Any + +from django.conf import settings + +import structlog + +logger = structlog.get_logger(__name__) + +# The collector truncates whole log lines at 100 KB (`max_log_size`); cap the body well +# below that so run identity attributes and JSON overhead never push a line over. +MAX_BODY_CHARS = 8_000 + +# Defensive budget per append: origin_product is user-settable on task creation, so a +# hostile append_log request must not be able to flood stdout/the collector with an +# arbitrarily long entry list. Real scout appends are small batches, far below this. +MAX_ENTRIES_PER_CALL = 200 + +_LOG_METHOD_NAMES = {"info": "info", "warn": "warning", "error": "error"} + + +def mirroring_enabled(origin_product: str) -> bool: + return origin_product in settings.TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS + + +def mirror_entries( + entries: list[dict], + *, + team_id: int, + task_id: str, + run_id: str, + origin_product: str, +) -> None: + """Emit one structured stdout log line per persisted entry.""" + if len(entries) > MAX_ENTRIES_PER_CALL: + logger.warning( + "task_run_log_mirror_truncated", + task_run_id=run_id, + dropped=len(entries) - MAX_ENTRIES_PER_CALL, + ) + entries = entries[:MAX_ENTRIES_PER_CALL] + for entry in entries: + if not isinstance(entry, dict): + continue + raw_notification = entry.get("notification") + notification: dict = raw_notification if isinstance(raw_notification, dict) else {} + update = _session_update(notification) + session_update = update.get("sessionUpdate") if isinstance(update.get("sessionUpdate"), str) else None + severity = _severity(notification) + + fields: dict[str, Any] = { + # `request_id` becomes the record's trace id in the collector, grouping the run. + "request_id": run_id, + "task_run_id": run_id, + "task_id": task_id, + "team_id": team_id, + "origin_product": origin_product, + "body": _body(notification, session_update), + } + method = notification.get("method") + if isinstance(method, str): + fields["acp_method"] = method + if session_update: + fields["acp_session_update"] = session_update + entry_timestamp = entry.get("timestamp") + if isinstance(entry_timestamp, str): + fields["entry_timestamp"] = entry_timestamp + + getattr(logger, _LOG_METHOD_NAMES[severity])("task_run_log", **fields) + + +def _session_update(notification: dict) -> dict: + params = notification.get("params") + if not isinstance(params, dict): + return {} + update = params.get("update") + return update if isinstance(update, dict) else {} + + +def _severity(notification: dict) -> str: + if notification.get("method") == "_posthog/error": + return "error" + if notification.get("method") == "_posthog/console": + params = notification.get("params") + level = params.get("level") if isinstance(params, dict) else None + if level in ("warn", "error"): + return level + # No "debug" mapping for thought chunks or debug console lines: the root stdlib log + # level is INFO in production, so a debug line would be filtered before it ever + # reaches stdout and the collector. + return "info" + + +def _body(notification: dict, session_update: str | None) -> str: + raw_params = notification.get("params") + params: dict = raw_params if isinstance(raw_params, dict) else {} + update = _session_update(notification) + + body: str | None = None + if session_update: + text = _extract_text(update.get("content")) + if text is not None: + body = f"[{session_update}] {text}" + elif session_update in ("tool_call", "tool_call_update"): + title = update.get("title") or update.get("toolCallId") or "" + status = update.get("status") + body = f"[{session_update}] {title}" + (f" ({status})" if status else "") + elif notification.get("method") in ("_posthog/console", "_posthog/error"): + message = params.get("message") + if isinstance(message, str): + body = message + elif notification.get("method") == "_posthog/sandbox_output": + stdout = params.get("stdout") or "" + stderr = params.get("stderr") or "" + body = f"[sandbox_output exit={params.get('exitCode')}] {stdout}" + (f"\nstderr: {stderr}" if stderr else "") + elif isinstance(notification.get("result"), dict): + stop_reason = notification["result"].get("stopReason") + if isinstance(stop_reason, str): + body = f"[turn_end] {stop_reason}" + + if body is None: + body = json.dumps(notification) + return body[:MAX_BODY_CHARS] + + +def _extract_text(content: Any) -> str | None: + """Pull plain text out of an ACP content block (single block or list of blocks).""" + if isinstance(content, dict): + text = content.get("text") + return text if isinstance(text, str) else None + if isinstance(content, list): + parts = [t for t in (_extract_text(block) for block in content) if t] + return "\n".join(parts) if parts else None + return None diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 0deb6b1a5fb1..1d8bb1078895 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -1103,6 +1103,8 @@ def append_log(self, entries: list[dict], *, ttl_days: int | None = DEFAULT_LOG_ object_storage.write(self.log_url, content) + self._mirror_logs_to_posthog_logs(entries) + if is_new_file and ttl_days is not None: try: object_storage.tag( @@ -1120,6 +1122,35 @@ def append_log(self, entries: list[dict], *, ttl_days: int | None = DEFAULT_LOG_ error=str(e), ) + def _mirror_logs_to_posthog_logs(self, entries: list[dict]) -> None: + """Mirror persisted entries into the PostHog Logs product via stdout (dogfooding). + + Fire-and-forget: mirroring failures must never break the run's log write. + """ + from products.tasks.backend.logic.services.run_log_mirror import mirror_entries, mirroring_enabled + + if not settings.TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS: + return + + try: + origin_product = self.task.origin_product + if not mirroring_enabled(origin_product): + return + + mirror_entries( + entries, + team_id=self.team_id, + task_id=str(self.task_id), + run_id=str(self.id), + origin_product=origin_product, + ) + except Exception as e: + logger.warning( + "task_run.mirror_logs_to_posthog_logs_failed", + task_run_id=str(self.id), + error=str(e), + ) + def capture_event(self, event: str, properties: dict | None = None, event_uuid: str | None = None) -> None: try: distinct_id = ( diff --git a/products/tasks/backend/tests/test_run_log_mirror.py b/products/tasks/backend/tests/test_run_log_mirror.py new file mode 100644 index 000000000000..c65c9d76ef0e --- /dev/null +++ b/products/tasks/backend/tests/test_run_log_mirror.py @@ -0,0 +1,193 @@ +import json + +from unittest.mock import MagicMock, patch + +from django.test import SimpleTestCase, TestCase, override_settings + +from parameterized import parameterized + +from posthog.models import Organization, Team + +from products.tasks.backend.logic.services.run_log_mirror import MAX_BODY_CHARS, MAX_ENTRIES_PER_CALL, mirror_entries +from products.tasks.backend.models import Task, TaskRun + +RUN_ID = "0b166f65-9e52-4d1b-b3c4-1a9e3f6d3c21" +TASK_ID = "7d0e9a34-2f1c-4b8a-9c3d-5e6f7a8b9c0d" + + +def _mirror(entries: list[dict]) -> MagicMock: + with patch("products.tasks.backend.logic.services.run_log_mirror.logger") as mock_logger: + mirror_entries(entries, team_id=2, task_id=TASK_ID, run_id=RUN_ID, origin_product="signals_scout") + return mock_logger + + +def _session_update_entry(session_update: str, **update_fields) -> dict: + return { + "type": "notification", + "timestamp": "2026-07-15T10:00:00+00:00", + "notification": { + "method": "session/update", + "params": {"update": {"sessionUpdate": session_update, **update_fields}}, + }, + } + + +class TestMirrorEntries(SimpleTestCase): + @parameterized.expand( + [ + ( + "agent_message", + _session_update_entry("agent_message", content={"type": "text", "text": "hello"}), + "info", + "[agent_message] hello", + ), + ( + "tool_call_without_content", + _session_update_entry("tool_call", title="grep", status="in_progress"), + "info", + "[tool_call] grep (in_progress)", + ), + ( + "posthog_error", + {"notification": {"method": "_posthog/error", "params": {"message": "boom"}}}, + "error", + "boom", + ), + ( + "console_level_passthrough", + {"notification": {"method": "_posthog/console", "params": {"level": "warn", "message": "careful"}}}, + "warning", + "careful", + ), + ( + "sandbox_output", + { + "notification": { + "method": "_posthog/sandbox_output", + "params": {"stdout": "out", "stderr": "err", "exitCode": 1}, + } + }, + "info", + "[sandbox_output exit=1] out\nstderr: err", + ), + ( + "turn_end_result", + {"notification": {"result": {"stopReason": "end_turn"}}}, + "info", + "[turn_end] end_turn", + ), + ] + ) + def test_severity_and_body_mapping(self, _name, entry, expected_log_method, expected_body): + mock_logger = _mirror([entry]) + log_call = getattr(mock_logger, expected_log_method) + log_call.assert_called_once() + self.assertEqual(log_call.call_args.kwargs["body"], expected_body) + + def test_unrecognized_entry_falls_back_to_json_body(self): + notification = {"method": "session/request_permission", "params": {"tool": "bash"}} + mock_logger = _mirror([{"notification": notification}]) + self.assertEqual(json.loads(mock_logger.info.call_args.kwargs["body"]), notification) + + def test_emitted_fields_carry_run_identity(self): + mock_logger = _mirror([_session_update_entry("agent_message", content={"type": "text", "text": "hi"})]) + self.assertEqual(mock_logger.info.call_args.args, ("task_run_log",)) + fields = mock_logger.info.call_args.kwargs + self.assertEqual(fields["request_id"], RUN_ID) + self.assertEqual(fields["task_run_id"], RUN_ID) + self.assertEqual(fields["task_id"], TASK_ID) + self.assertEqual(fields["team_id"], 2) + self.assertEqual(fields["origin_product"], "signals_scout") + self.assertEqual(fields["acp_method"], "session/update") + self.assertEqual(fields["acp_session_update"], "agent_message") + self.assertEqual(fields["entry_timestamp"], "2026-07-15T10:00:00+00:00") + + def test_oversized_body_is_truncated(self): + entry = _session_update_entry("agent_message", content={"type": "text", "text": "x" * (MAX_BODY_CHARS * 2)}) + mock_logger = _mirror([entry]) + self.assertEqual(len(mock_logger.info.call_args.kwargs["body"]), MAX_BODY_CHARS) + + def test_oversized_batch_is_capped(self): + entries = [ + _session_update_entry("agent_message", content={"type": "text", "text": f"line {i}"}) + for i in range(MAX_ENTRIES_PER_CALL + 50) + ] + mock_logger = _mirror(entries) + self.assertEqual(mock_logger.info.call_count, MAX_ENTRIES_PER_CALL) + mock_logger.warning.assert_called_once() + self.assertEqual(mock_logger.warning.call_args.kwargs["dropped"], 50) + + @parameterized.expand([("empty", []), ("non_dict_entries", ["not-a-dict", 42])]) + def test_no_usable_entries_emits_nothing(self, _name, entries): + mock_logger = _mirror(entries) + mock_logger.info.assert_not_called() + mock_logger.warning.assert_not_called() + mock_logger.error.assert_not_called() + + +@override_settings(TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS=["signals_scout"]) +class TestAppendLogMirroring(TestCase): + @classmethod + def setUpTestData(cls): + cls.organization = Organization.objects.create(name="Test Org") + cls.team = Team.objects.create(organization=cls.organization, name="Test Team") + + def _create_run(self, origin_product: str) -> TaskRun: + task = Task.objects.create( + team=self.team, + title="Test Task", + description="Test", + origin_product=origin_product, + ) + return TaskRun.objects.create(team=self.team, task=task) + + @parameterized.expand( + [ + (Task.OriginProduct.SIGNALS_SCOUT, True), + (Task.OriginProduct.USER_CREATED, False), + ] + ) + @patch("products.tasks.backend.logic.services.run_log_mirror.logger") + @patch("products.tasks.backend.models.object_storage") + def test_mirrors_only_allowlisted_origin_products(self, origin_product, expect_mirrored, mock_storage, mock_logger): + mock_storage.read.return_value = None + run = self._create_run(origin_product) + message = _session_update_entry("agent_message", content={"type": "text", "text": "hi"}) + chunk = _session_update_entry("agent_message_chunk", content={"type": "text", "text": "h"}) + + run.append_log([message, chunk]) + + mock_storage.write.assert_called_once() + if expect_mirrored: + # The chunk entry is dropped before persistence, so exactly one line is mirrored. + mock_logger.info.assert_called_once() + self.assertEqual(mock_logger.info.call_args.kwargs["task_run_id"], str(run.id)) + self.assertEqual(mock_logger.info.call_args.kwargs["origin_product"], origin_product) + else: + mock_logger.info.assert_not_called() + + @override_settings(TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS=[]) + @patch("products.tasks.backend.logic.services.run_log_mirror.logger") + @patch("products.tasks.backend.models.object_storage") + def test_no_mirroring_when_disabled(self, mock_storage, mock_logger): + mock_storage.read.return_value = None + run = self._create_run(Task.OriginProduct.SIGNALS_SCOUT) + + run.append_log([_session_update_entry("agent_message", content={"type": "text", "text": "hi"})]) + + mock_storage.write.assert_called_once() + mock_logger.info.assert_not_called() + + @patch( + "products.tasks.backend.logic.services.run_log_mirror.mirror_entries", + side_effect=RuntimeError("kaboom"), + ) + @patch("products.tasks.backend.models.object_storage") + def test_mirror_failure_does_not_break_log_write(self, mock_storage, mock_mirror): + mock_storage.read.return_value = None + run = self._create_run(Task.OriginProduct.SIGNALS_SCOUT) + + run.append_log([_session_update_entry("agent_message", content={"type": "text", "text": "hi"})]) + + mock_storage.write.assert_called_once() + mock_mirror.assert_called_once()