From df2e6523b03874ecde6f772c108c68c9323d1079 Mon Sep 17 00:00:00 2001 From: Andrew Maguire Date: Wed, 15 Jul 2026 15:23:14 +0100 Subject: [PATCH 1/3] feat(tasks): mirror scout run logs to posthog logs via otlp Scout (and, later, other task-origin) run logs live only as JSONL blobs in object storage, which makes sampling and eyeballing runs painful. Mirror entries persisted via TaskRun.append_log to a PostHog project's Logs product as OTLP/HTTP log records, gated by settings and scoped to signals_scout origin by default. Export runs on a Celery task so it never blocks or breaks the log write. --- docs/internal/sandboxes-setup-guide.md | 18 ++ posthog/settings/temporal.py | 12 ++ .../backend/logic/services/run_log_otlp.py | 167 +++++++++++++++ products/tasks/backend/models.py | 36 ++++ products/tasks/backend/tasks.py | 55 +++++ .../tasks/backend/tests/test_run_log_otlp.py | 202 ++++++++++++++++++ 6 files changed, 490 insertions(+) create mode 100644 products/tasks/backend/logic/services/run_log_otlp.py create mode 100644 products/tasks/backend/tasks.py create mode 100644 products/tasks/backend/tests/test_run_log_otlp.py diff --git a/docs/internal/sandboxes-setup-guide.md b/docs/internal/sandboxes-setup-guide.md index f304f9f95591..4f3b0544a5e0 100644 --- a/docs/internal/sandboxes-setup-guide.md +++ b/docs/internal/sandboxes-setup-guide.md @@ -237,6 +237,24 @@ repositories. > **Note:** This only works with `SANDBOX_PROVIDER=docker`. +### Optional: mirror run logs to PostHog Logs (dogfooding) + +Task-run log entries (the JSONL appended to object storage via `TaskRun.append_log`) can also be shipped to a PostHog project's Logs product as OTLP log records, +so runs can be browsed and sampled in the Logs UI instead of fetching S3 blobs. + +```bash +# The target project's OTLP logs endpoint and API token. +TASK_RUN_LOGS_OTLP_ENDPOINT=https://us.i.posthog.com/i/v1/logs +TASK_RUN_LOGS_OTLP_TOKEN=phc_... +# Which task origins to forward (comma-separated). Defaults to signals scouts only. +TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS=signals_scout +``` + +Forwarding is off unless both endpoint and token are set. +Records carry `service.name=`, a run-scoped trace id, and `task_run_id` / `task_id` / `team_id` attributes, +so one run can be pulled up with an attribute filter on `task_run_id`. +Export happens on a Celery task off the log-write path; failures are logged and never break the run. + ### 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..02d8b89ac277 100644 --- a/posthog/settings/temporal.py +++ b/posthog/settings/temporal.py @@ -76,6 +76,18 @@ "TASKS_CREDENTIAL_REFRESH_INITIAL_DELAY_SECONDS", 0, type_cast=int ) +# Mirror persisted task-run logs into a PostHog project's Logs product (dogfooding). +# Entries appended to a run's S3 JSONL log are also shipped as OTLP/HTTP log records to +# TASK_RUN_LOGS_OTLP_ENDPOINT (e.g. https://us.i.posthog.com/i/v1/logs), authenticated with +# the target project's API token. Disabled unless both endpoint and token are set. Only runs +# whose task origin_product is in TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS are forwarded — scoped +# to signals scouts for now; widen the list to cover more task origins. +TASK_RUN_LOGS_OTLP_ENDPOINT: str | None = get_from_env("TASK_RUN_LOGS_OTLP_ENDPOINT", None, optional=True) +TASK_RUN_LOGS_OTLP_TOKEN: str | None = get_from_env("TASK_RUN_LOGS_OTLP_TOKEN", None, optional=True) +TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS: list[str] = get_list( + os.getenv("TASK_RUN_LOGS_OTLP_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_otlp.py b/products/tasks/backend/logic/services/run_log_otlp.py new file mode 100644 index 000000000000..9fe63b1e8beb --- /dev/null +++ b/products/tasks/backend/logic/services/run_log_otlp.py @@ -0,0 +1,167 @@ +"""Convert persisted task-run log entries (ACP JSONL) into an OTLP/HTTP logs payload. + +Task-run logs are appended to object storage as one ACP notification envelope per line. +To dogfood the PostHog Logs product, entries can also be mirrored to a PostHog project's +OTLP logs endpoint (`/i/v1/logs`). This module is the pure translation layer: it maps each +entry to an OTLP log record with a run-scoped trace id so one run reads as one trace in the +Logs UI. Dispatch/transport live in `products.tasks.backend.tasks`. +""" + +import json +import uuid +from datetime import datetime +from typing import Any + +from django.conf import settings +from django.utils import timezone + +# Bodies larger than this are truncated — huge tool-call payloads would otherwise blow the +# ingestion endpoint's 2 MB request cap and are useless for eyeballing runs anyway. +MAX_BODY_CHARS = 32_000 + +_SEVERITY_NUMBERS = {"debug": 5, "info": 9, "warn": 13, "error": 17} + + +def otlp_forwarding_configured() -> bool: + return bool(settings.TASK_RUN_LOGS_OTLP_ENDPOINT and settings.TASK_RUN_LOGS_OTLP_TOKEN) + + +def otlp_forwarding_enabled(origin_product: str) -> bool: + return otlp_forwarding_configured() and origin_product in settings.TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS + + +def build_otlp_payload( + entries: list[dict], + *, + team_id: int, + task_id: str, + run_id: str, + origin_product: str, +) -> dict[str, Any] | None: + """Build an OTLP `ExportLogsServiceRequest` JSON body from persisted log entries.""" + records = [_log_record(entry, run_id=run_id) for entry in entries if isinstance(entry, dict)] + if not records: + return None + + resource_attributes = [ + _attribute("service.name", origin_product), + _attribute("team_id", str(team_id)), + _attribute("task_id", task_id), + _attribute("task_run_id", run_id), + ] + return { + "resourceLogs": [ + { + "resource": {"attributes": resource_attributes}, + "scopeLogs": [{"scope": {"name": "posthog.task_run"}, "logRecords": records}], + } + ] + } + + +def _log_record(entry: dict, *, run_id: str) -> dict[str, Any]: + raw_notification = entry.get("notification") + notification: dict = raw_notification if isinstance(raw_notification, dict) else {} + method = notification.get("method") + update = _session_update(notification) + session_update = update.get("sessionUpdate") if isinstance(update.get("sessionUpdate"), str) else None + + severity = _severity(notification, session_update) + # Record-level copies of the run identity make the Logs UI attribute filters usable + # without touching resource attributes. + attributes = [_attribute("task_run_id", run_id)] + if isinstance(method, str): + attributes.append(_attribute("acp.method", method)) + if session_update: + attributes.append(_attribute("acp.session_update", session_update)) + + return { + "timeUnixNano": str(_time_unix_nano(entry)), + "severityText": severity, + "severityNumber": _SEVERITY_NUMBERS[severity], + "body": {"stringValue": _body(notification, session_update)}, + "attributes": attributes, + # All records of a run share the run's uuid as trace id, so a run groups as one trace. + "traceId": uuid.UUID(run_id).hex, + } + + +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, session_update: str | None) -> 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 _SEVERITY_NUMBERS: + return level + if session_update == "agent_thought_chunk": + return "debug" + 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 + + +def _time_unix_nano(entry: dict) -> int: + timestamp = entry.get("timestamp") + parsed: datetime | None = None + if isinstance(timestamp, str): + try: + parsed = datetime.fromisoformat(timestamp) + except ValueError: + parsed = None + if parsed is None: + parsed = timezone.now() + return int(parsed.timestamp() * 1_000_000_000) + + +def _attribute(key: str, value: str) -> dict[str, Any]: + return {"key": key, "value": {"stringValue": value}} diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 0deb6b1a5fb1..ee8147e61d80 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._forward_logs_to_posthog_logs(entries) + if is_new_file and ttl_days is not None: try: object_storage.tag( @@ -1120,6 +1122,40 @@ def append_log(self, entries: list[dict], *, ttl_days: int | None = DEFAULT_LOG_ error=str(e), ) + def _forward_logs_to_posthog_logs(self, entries: list[dict]) -> None: + """Mirror persisted entries into a PostHog project's Logs product (dogfooding). + + Fire-and-forget: dispatch failures must never break the run's log write. + """ + from products.tasks.backend.logic.services.run_log_otlp import ( + otlp_forwarding_configured, + otlp_forwarding_enabled, + ) + + if not otlp_forwarding_configured(): + return + + try: + origin_product = self.task.origin_product + if not otlp_forwarding_enabled(origin_product): + return + + from products.tasks.backend.tasks import forward_task_run_logs_to_posthog_logs + + forward_task_run_logs_to_posthog_logs.delay( + 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.forward_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/tasks.py b/products/tasks/backend/tasks.py new file mode 100644 index 000000000000..242f9d20c4c7 --- /dev/null +++ b/products/tasks/backend/tasks.py @@ -0,0 +1,55 @@ +from django.conf import settings + +import requests +import structlog +from celery import shared_task + +from products.tasks.backend.logic.services.run_log_otlp import build_otlp_payload + +logger = structlog.get_logger(__name__) + +OTLP_EXPORT_TIMEOUT_SECONDS = 10 + + +@shared_task( + ignore_result=True, + autoretry_for=(requests.RequestException,), + retry_backoff=True, + max_retries=3, +) +def forward_task_run_logs_to_posthog_logs( + entries: list[dict], + team_id: int, + task_id: str, + run_id: str, + origin_product: str, +) -> None: + """Mirror persisted task-run log entries to a PostHog project's Logs product via OTLP/HTTP.""" + endpoint = settings.TASK_RUN_LOGS_OTLP_ENDPOINT + token = settings.TASK_RUN_LOGS_OTLP_TOKEN + if not endpoint or not token: + return + + payload = build_otlp_payload( + entries, + team_id=team_id, + task_id=task_id, + run_id=run_id, + origin_product=origin_product, + ) + if payload is None: + return + + response = requests.post( + endpoint, + json=payload, + headers={"Authorization": f"Bearer {token}"}, + timeout=OTLP_EXPORT_TIMEOUT_SECONDS, + ) + if response.status_code >= 400: + logger.warning( + "task_run.otlp_log_export_rejected", + run_id=run_id, + status_code=response.status_code, + body=response.text[:500], + ) diff --git a/products/tasks/backend/tests/test_run_log_otlp.py b/products/tasks/backend/tests/test_run_log_otlp.py new file mode 100644 index 000000000000..9d860d5dd379 --- /dev/null +++ b/products/tasks/backend/tests/test_run_log_otlp.py @@ -0,0 +1,202 @@ +import json +import uuid + +from unittest.mock import 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_otlp import MAX_BODY_CHARS, build_otlp_payload +from products.tasks.backend.models import Task, TaskRun + +RUN_ID = "0b166f65-9e52-4d1b-b3c4-1a9e3f6d3c21" +TASK_ID = "7d0e9a34-2f1c-4b8a-9c3d-5e6f7a8b9c0d" + + +def _build(entries: list[dict]) -> dict | None: + return build_otlp_payload(entries, team_id=2, task_id=TASK_ID, run_id=RUN_ID, origin_product="signals_scout") + + +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 TestBuildOtlpPayload(SimpleTestCase): + @parameterized.expand( + [ + ( + "agent_message", + _session_update_entry("agent_message", content={"type": "text", "text": "hello"}), + "info", + "[agent_message] hello", + ), + ( + "agent_thought_is_debug", + _session_update_entry("agent_thought_chunk", content={"type": "text", "text": "thinking"}), + "debug", + "[agent_thought_chunk] thinking", + ), + ( + "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"}}}, + "warn", + "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_severity, expected_body): + payload = _build([entry]) + assert payload is not None + record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] + self.assertEqual(record["severityText"], expected_severity) + self.assertEqual(record["body"]["stringValue"], expected_body) + + def test_unrecognized_entry_falls_back_to_json_body(self): + notification = {"method": "session/request_permission", "params": {"tool": "bash"}} + payload = _build([{"notification": notification}]) + assert payload is not None + record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] + self.assertEqual(json.loads(record["body"]["stringValue"]), notification) + + def test_payload_structure_carries_run_identity(self): + payload = _build([_session_update_entry("agent_message", content={"type": "text", "text": "hi"})]) + assert payload is not None + resource_attrs = { + attr["key"]: attr["value"]["stringValue"] for attr in payload["resourceLogs"][0]["resource"]["attributes"] + } + self.assertEqual( + resource_attrs, + {"service.name": "signals_scout", "team_id": "2", "task_id": TASK_ID, "task_run_id": RUN_ID}, + ) + record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] + self.assertEqual(record["traceId"], uuid.UUID(RUN_ID).hex) + self.assertEqual(record["timeUnixNano"], str(1784109600 * 1_000_000_000)) + record_attrs = {attr["key"]: attr["value"]["stringValue"] for attr in record["attributes"]} + self.assertEqual(record_attrs["acp.method"], "session/update") + self.assertEqual(record_attrs["acp.session_update"], "agent_message") + + def test_oversized_body_is_truncated(self): + entry = _session_update_entry("agent_message", content={"type": "text", "text": "x" * (MAX_BODY_CHARS * 2)}) + payload = _build([entry]) + assert payload is not None + record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] + self.assertEqual(len(record["body"]["stringValue"]), MAX_BODY_CHARS) + + @parameterized.expand([("empty", []), ("non_dict_entries", ["not-a-dict", 42])]) + def test_no_usable_entries_returns_none(self, _name, entries): + self.assertIsNone(_build(entries)) + + +@override_settings( + TASK_RUN_LOGS_OTLP_ENDPOINT="https://us.i.posthog.com/i/v1/logs", + TASK_RUN_LOGS_OTLP_TOKEN="phc_test", + TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS=["signals_scout"], +) +class TestAppendLogForwarding(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.tasks.forward_task_run_logs_to_posthog_logs.delay") + @patch("products.tasks.backend.models.object_storage") + def test_forwards_only_allowlisted_origin_products( + self, origin_product, expect_forwarded, mock_storage, mock_delay + ): + 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_forwarded: + mock_delay.assert_called_once_with( + entries=[message], + team_id=self.team.id, + task_id=str(run.task_id), + run_id=str(run.id), + origin_product=origin_product, + ) + else: + mock_delay.assert_not_called() + + @override_settings(TASK_RUN_LOGS_OTLP_ENDPOINT=None, TASK_RUN_LOGS_OTLP_TOKEN=None) + @patch("products.tasks.backend.tasks.forward_task_run_logs_to_posthog_logs.delay") + @patch("products.tasks.backend.models.object_storage") + def test_no_forwarding_when_unconfigured(self, mock_storage, mock_delay): + 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_delay.assert_not_called() + + @patch( + "products.tasks.backend.tasks.forward_task_run_logs_to_posthog_logs.delay", side_effect=RuntimeError("kaboom") + ) + @patch("products.tasks.backend.models.object_storage") + def test_dispatch_failure_does_not_break_log_write(self, mock_storage, mock_delay): + 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_delay.assert_called_once() From 8b1b456daea2da72d01edd99b382430c1c8e45b7 Mon Sep 17 00:00:00 2001 From: Andrew Maguire Date: Wed, 15 Jul 2026 15:46:47 +0100 Subject: [PATCH 2/3] feat(tasks): mirror scout run logs via stdout instead of otlp The per-cluster OTel collector already ships all container stdout into the region's internal PostHog project's Logs product, parsing JSON keys into attributes and request_id into a trace id. Emitting structured stdout lines at append_log gets scout run logs into PostHog Logs with no transport, credentials, or Celery hop of its own. --- docs/internal/sandboxes-setup-guide.md | 23 +-- posthog/settings/temporal.py | 18 +- .../backend/logic/services/run_log_mirror.py | 136 ++++++++++++++ .../backend/logic/services/run_log_otlp.py | 167 ------------------ products/tasks/backend/models.py | 25 ++- products/tasks/backend/tasks.py | 55 ------ ...run_log_otlp.py => test_run_log_mirror.py} | 125 ++++++------- 7 files changed, 219 insertions(+), 330 deletions(-) create mode 100644 products/tasks/backend/logic/services/run_log_mirror.py delete mode 100644 products/tasks/backend/logic/services/run_log_otlp.py delete mode 100644 products/tasks/backend/tasks.py rename products/tasks/backend/tests/{test_run_log_otlp.py => test_run_log_mirror.py} (54%) diff --git a/docs/internal/sandboxes-setup-guide.md b/docs/internal/sandboxes-setup-guide.md index 4f3b0544a5e0..e92bb5f90093 100644 --- a/docs/internal/sandboxes-setup-guide.md +++ b/docs/internal/sandboxes-setup-guide.md @@ -237,23 +237,24 @@ repositories. > **Note:** This only works with `SANDBOX_PROVIDER=docker`. -### Optional: mirror run logs to PostHog Logs (dogfooding) +### Task-run log mirroring to PostHog Logs (dogfooding) -Task-run log entries (the JSONL appended to object storage via `TaskRun.append_log`) can also be shipped to a PostHog project's Logs product as OTLP log records, +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 -# The target project's OTLP logs endpoint and API token. -TASK_RUN_LOGS_OTLP_ENDPOINT=https://us.i.posthog.com/i/v1/logs -TASK_RUN_LOGS_OTLP_TOKEN=phc_... -# Which task origins to forward (comma-separated). Defaults to signals scouts only. -TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS=signals_scout +# Which task origins to mirror (comma-separated). Defaults to signals scouts only. +# Set empty to disable. +TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS=signals_scout ``` -Forwarding is off unless both endpoint and token are set. -Records carry `service.name=`, a run-scoped trace id, and `task_run_id` / `task_id` / `team_id` attributes, -so one run can be pulled up with an attribute filter on `task_run_id`. -Export happens on a Celery task off the log-write path; failures are logged and never break the run. +Mirroring failures are logged and never break the run's log write. ### How `MODAL_DOCKER` works diff --git a/posthog/settings/temporal.py b/posthog/settings/temporal.py index 02d8b89ac277..d8759d212567 100644 --- a/posthog/settings/temporal.py +++ b/posthog/settings/temporal.py @@ -76,16 +76,14 @@ "TASKS_CREDENTIAL_REFRESH_INITIAL_DELAY_SECONDS", 0, type_cast=int ) -# Mirror persisted task-run logs into a PostHog project's Logs product (dogfooding). -# Entries appended to a run's S3 JSONL log are also shipped as OTLP/HTTP log records to -# TASK_RUN_LOGS_OTLP_ENDPOINT (e.g. https://us.i.posthog.com/i/v1/logs), authenticated with -# the target project's API token. Disabled unless both endpoint and token are set. Only runs -# whose task origin_product is in TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS are forwarded — scoped -# to signals scouts for now; widen the list to cover more task origins. -TASK_RUN_LOGS_OTLP_ENDPOINT: str | None = get_from_env("TASK_RUN_LOGS_OTLP_ENDPOINT", None, optional=True) -TASK_RUN_LOGS_OTLP_TOKEN: str | None = get_from_env("TASK_RUN_LOGS_OTLP_TOKEN", None, optional=True) -TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS: list[str] = get_list( - os.getenv("TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS", "signals_scout") +# 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") 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..0a3d851ea027 --- /dev/null +++ b/products/tasks/backend/logic/services/run_log_mirror.py @@ -0,0 +1,136 @@ +"""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 + +_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.""" + 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/logic/services/run_log_otlp.py b/products/tasks/backend/logic/services/run_log_otlp.py deleted file mode 100644 index 9fe63b1e8beb..000000000000 --- a/products/tasks/backend/logic/services/run_log_otlp.py +++ /dev/null @@ -1,167 +0,0 @@ -"""Convert persisted task-run log entries (ACP JSONL) into an OTLP/HTTP logs payload. - -Task-run logs are appended to object storage as one ACP notification envelope per line. -To dogfood the PostHog Logs product, entries can also be mirrored to a PostHog project's -OTLP logs endpoint (`/i/v1/logs`). This module is the pure translation layer: it maps each -entry to an OTLP log record with a run-scoped trace id so one run reads as one trace in the -Logs UI. Dispatch/transport live in `products.tasks.backend.tasks`. -""" - -import json -import uuid -from datetime import datetime -from typing import Any - -from django.conf import settings -from django.utils import timezone - -# Bodies larger than this are truncated — huge tool-call payloads would otherwise blow the -# ingestion endpoint's 2 MB request cap and are useless for eyeballing runs anyway. -MAX_BODY_CHARS = 32_000 - -_SEVERITY_NUMBERS = {"debug": 5, "info": 9, "warn": 13, "error": 17} - - -def otlp_forwarding_configured() -> bool: - return bool(settings.TASK_RUN_LOGS_OTLP_ENDPOINT and settings.TASK_RUN_LOGS_OTLP_TOKEN) - - -def otlp_forwarding_enabled(origin_product: str) -> bool: - return otlp_forwarding_configured() and origin_product in settings.TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS - - -def build_otlp_payload( - entries: list[dict], - *, - team_id: int, - task_id: str, - run_id: str, - origin_product: str, -) -> dict[str, Any] | None: - """Build an OTLP `ExportLogsServiceRequest` JSON body from persisted log entries.""" - records = [_log_record(entry, run_id=run_id) for entry in entries if isinstance(entry, dict)] - if not records: - return None - - resource_attributes = [ - _attribute("service.name", origin_product), - _attribute("team_id", str(team_id)), - _attribute("task_id", task_id), - _attribute("task_run_id", run_id), - ] - return { - "resourceLogs": [ - { - "resource": {"attributes": resource_attributes}, - "scopeLogs": [{"scope": {"name": "posthog.task_run"}, "logRecords": records}], - } - ] - } - - -def _log_record(entry: dict, *, run_id: str) -> dict[str, Any]: - raw_notification = entry.get("notification") - notification: dict = raw_notification if isinstance(raw_notification, dict) else {} - method = notification.get("method") - update = _session_update(notification) - session_update = update.get("sessionUpdate") if isinstance(update.get("sessionUpdate"), str) else None - - severity = _severity(notification, session_update) - # Record-level copies of the run identity make the Logs UI attribute filters usable - # without touching resource attributes. - attributes = [_attribute("task_run_id", run_id)] - if isinstance(method, str): - attributes.append(_attribute("acp.method", method)) - if session_update: - attributes.append(_attribute("acp.session_update", session_update)) - - return { - "timeUnixNano": str(_time_unix_nano(entry)), - "severityText": severity, - "severityNumber": _SEVERITY_NUMBERS[severity], - "body": {"stringValue": _body(notification, session_update)}, - "attributes": attributes, - # All records of a run share the run's uuid as trace id, so a run groups as one trace. - "traceId": uuid.UUID(run_id).hex, - } - - -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, session_update: str | None) -> 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 _SEVERITY_NUMBERS: - return level - if session_update == "agent_thought_chunk": - return "debug" - 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 - - -def _time_unix_nano(entry: dict) -> int: - timestamp = entry.get("timestamp") - parsed: datetime | None = None - if isinstance(timestamp, str): - try: - parsed = datetime.fromisoformat(timestamp) - except ValueError: - parsed = None - if parsed is None: - parsed = timezone.now() - return int(parsed.timestamp() * 1_000_000_000) - - -def _attribute(key: str, value: str) -> dict[str, Any]: - return {"key": key, "value": {"stringValue": value}} diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index ee8147e61d80..1d8bb1078895 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -1103,7 +1103,7 @@ def append_log(self, entries: list[dict], *, ttl_days: int | None = DEFAULT_LOG_ object_storage.write(self.log_url, content) - self._forward_logs_to_posthog_logs(entries) + self._mirror_logs_to_posthog_logs(entries) if is_new_file and ttl_days is not None: try: @@ -1122,28 +1122,23 @@ def append_log(self, entries: list[dict], *, ttl_days: int | None = DEFAULT_LOG_ error=str(e), ) - def _forward_logs_to_posthog_logs(self, entries: list[dict]) -> None: - """Mirror persisted entries into a PostHog project's Logs product (dogfooding). + 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: dispatch failures must never break the run's log write. + Fire-and-forget: mirroring failures must never break the run's log write. """ - from products.tasks.backend.logic.services.run_log_otlp import ( - otlp_forwarding_configured, - otlp_forwarding_enabled, - ) + from products.tasks.backend.logic.services.run_log_mirror import mirror_entries, mirroring_enabled - if not otlp_forwarding_configured(): + if not settings.TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS: return try: origin_product = self.task.origin_product - if not otlp_forwarding_enabled(origin_product): + if not mirroring_enabled(origin_product): return - from products.tasks.backend.tasks import forward_task_run_logs_to_posthog_logs - - forward_task_run_logs_to_posthog_logs.delay( - entries=entries, + mirror_entries( + entries, team_id=self.team_id, task_id=str(self.task_id), run_id=str(self.id), @@ -1151,7 +1146,7 @@ def _forward_logs_to_posthog_logs(self, entries: list[dict]) -> None: ) except Exception as e: logger.warning( - "task_run.forward_logs_to_posthog_logs_failed", + "task_run.mirror_logs_to_posthog_logs_failed", task_run_id=str(self.id), error=str(e), ) diff --git a/products/tasks/backend/tasks.py b/products/tasks/backend/tasks.py deleted file mode 100644 index 242f9d20c4c7..000000000000 --- a/products/tasks/backend/tasks.py +++ /dev/null @@ -1,55 +0,0 @@ -from django.conf import settings - -import requests -import structlog -from celery import shared_task - -from products.tasks.backend.logic.services.run_log_otlp import build_otlp_payload - -logger = structlog.get_logger(__name__) - -OTLP_EXPORT_TIMEOUT_SECONDS = 10 - - -@shared_task( - ignore_result=True, - autoretry_for=(requests.RequestException,), - retry_backoff=True, - max_retries=3, -) -def forward_task_run_logs_to_posthog_logs( - entries: list[dict], - team_id: int, - task_id: str, - run_id: str, - origin_product: str, -) -> None: - """Mirror persisted task-run log entries to a PostHog project's Logs product via OTLP/HTTP.""" - endpoint = settings.TASK_RUN_LOGS_OTLP_ENDPOINT - token = settings.TASK_RUN_LOGS_OTLP_TOKEN - if not endpoint or not token: - return - - payload = build_otlp_payload( - entries, - team_id=team_id, - task_id=task_id, - run_id=run_id, - origin_product=origin_product, - ) - if payload is None: - return - - response = requests.post( - endpoint, - json=payload, - headers={"Authorization": f"Bearer {token}"}, - timeout=OTLP_EXPORT_TIMEOUT_SECONDS, - ) - if response.status_code >= 400: - logger.warning( - "task_run.otlp_log_export_rejected", - run_id=run_id, - status_code=response.status_code, - body=response.text[:500], - ) diff --git a/products/tasks/backend/tests/test_run_log_otlp.py b/products/tasks/backend/tests/test_run_log_mirror.py similarity index 54% rename from products/tasks/backend/tests/test_run_log_otlp.py rename to products/tasks/backend/tests/test_run_log_mirror.py index 9d860d5dd379..a41fece9e7be 100644 --- a/products/tasks/backend/tests/test_run_log_otlp.py +++ b/products/tasks/backend/tests/test_run_log_mirror.py @@ -1,7 +1,6 @@ import json -import uuid -from unittest.mock import patch +from unittest.mock import MagicMock, patch from django.test import SimpleTestCase, TestCase, override_settings @@ -9,15 +8,17 @@ from posthog.models import Organization, Team -from products.tasks.backend.logic.services.run_log_otlp import MAX_BODY_CHARS, build_otlp_payload +from products.tasks.backend.logic.services.run_log_mirror import MAX_BODY_CHARS, 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 _build(entries: list[dict]) -> dict | None: - return build_otlp_payload(entries, team_id=2, task_id=TASK_ID, run_id=RUN_ID, origin_product="signals_scout") +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: @@ -31,7 +32,7 @@ def _session_update_entry(session_update: str, **update_fields) -> dict: } -class TestBuildOtlpPayload(SimpleTestCase): +class TestMirrorEntries(SimpleTestCase): @parameterized.expand( [ ( @@ -40,12 +41,6 @@ class TestBuildOtlpPayload(SimpleTestCase): "info", "[agent_message] hello", ), - ( - "agent_thought_is_debug", - _session_update_entry("agent_thought_chunk", content={"type": "text", "text": "thinking"}), - "debug", - "[agent_thought_chunk] thinking", - ), ( "tool_call_without_content", _session_update_entry("tool_call", title="grep", status="in_progress"), @@ -61,7 +56,7 @@ class TestBuildOtlpPayload(SimpleTestCase): ( "console_level_passthrough", {"notification": {"method": "_posthog/console", "params": {"level": "warn", "message": "careful"}}}, - "warn", + "warning", "careful", ), ( @@ -83,55 +78,45 @@ class TestBuildOtlpPayload(SimpleTestCase): ), ] ) - def test_severity_and_body_mapping(self, _name, entry, expected_severity, expected_body): - payload = _build([entry]) - assert payload is not None - record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] - self.assertEqual(record["severityText"], expected_severity) - self.assertEqual(record["body"]["stringValue"], expected_body) + 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"}} - payload = _build([{"notification": notification}]) - assert payload is not None - record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] - self.assertEqual(json.loads(record["body"]["stringValue"]), notification) - - def test_payload_structure_carries_run_identity(self): - payload = _build([_session_update_entry("agent_message", content={"type": "text", "text": "hi"})]) - assert payload is not None - resource_attrs = { - attr["key"]: attr["value"]["stringValue"] for attr in payload["resourceLogs"][0]["resource"]["attributes"] - } - self.assertEqual( - resource_attrs, - {"service.name": "signals_scout", "team_id": "2", "task_id": TASK_ID, "task_run_id": RUN_ID}, - ) - record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] - self.assertEqual(record["traceId"], uuid.UUID(RUN_ID).hex) - self.assertEqual(record["timeUnixNano"], str(1784109600 * 1_000_000_000)) - record_attrs = {attr["key"]: attr["value"]["stringValue"] for attr in record["attributes"]} - self.assertEqual(record_attrs["acp.method"], "session/update") - self.assertEqual(record_attrs["acp.session_update"], "agent_message") + 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)}) - payload = _build([entry]) - assert payload is not None - record = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0] - self.assertEqual(len(record["body"]["stringValue"]), MAX_BODY_CHARS) + mock_logger = _mirror([entry]) + self.assertEqual(len(mock_logger.info.call_args.kwargs["body"]), MAX_BODY_CHARS) @parameterized.expand([("empty", []), ("non_dict_entries", ["not-a-dict", 42])]) - def test_no_usable_entries_returns_none(self, _name, entries): - self.assertIsNone(_build(entries)) + 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_OTLP_ENDPOINT="https://us.i.posthog.com/i/v1/logs", - TASK_RUN_LOGS_OTLP_TOKEN="phc_test", - TASK_RUN_LOGS_OTLP_ORIGIN_PRODUCTS=["signals_scout"], -) -class TestAppendLogForwarding(TestCase): +@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") @@ -152,11 +137,9 @@ def _create_run(self, origin_product: str) -> TaskRun: (Task.OriginProduct.USER_CREATED, False), ] ) - @patch("products.tasks.backend.tasks.forward_task_run_logs_to_posthog_logs.delay") + @patch("products.tasks.backend.logic.services.run_log_mirror.logger") @patch("products.tasks.backend.models.object_storage") - def test_forwards_only_allowlisted_origin_products( - self, origin_product, expect_forwarded, mock_storage, mock_delay - ): + 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"}) @@ -165,38 +148,36 @@ def test_forwards_only_allowlisted_origin_products( run.append_log([message, chunk]) mock_storage.write.assert_called_once() - if expect_forwarded: - mock_delay.assert_called_once_with( - entries=[message], - team_id=self.team.id, - task_id=str(run.task_id), - run_id=str(run.id), - origin_product=origin_product, - ) + 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_delay.assert_not_called() + mock_logger.info.assert_not_called() - @override_settings(TASK_RUN_LOGS_OTLP_ENDPOINT=None, TASK_RUN_LOGS_OTLP_TOKEN=None) - @patch("products.tasks.backend.tasks.forward_task_run_logs_to_posthog_logs.delay") + @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_forwarding_when_unconfigured(self, mock_storage, mock_delay): + 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_delay.assert_not_called() + mock_logger.info.assert_not_called() @patch( - "products.tasks.backend.tasks.forward_task_run_logs_to_posthog_logs.delay", side_effect=RuntimeError("kaboom") + "products.tasks.backend.logic.services.run_log_mirror.mirror_entries", + side_effect=RuntimeError("kaboom"), ) @patch("products.tasks.backend.models.object_storage") - def test_dispatch_failure_does_not_break_log_write(self, mock_storage, mock_delay): + 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_delay.assert_called_once() + mock_mirror.assert_called_once() From 3139655be7a128abd7267050c8864e7af7309d4f Mon Sep 17 00:00:00 2001 From: Andrew Maguire Date: Wed, 15 Jul 2026 15:51:56 +0100 Subject: [PATCH 3/3] fix(tasks): bound mirrored entries per append_log call origin_product is user-settable on task creation, so a hostile append must not be able to flood stdout and the log collector with an unbounded entry list. --- .../tasks/backend/logic/services/run_log_mirror.py | 12 ++++++++++++ products/tasks/backend/tests/test_run_log_mirror.py | 12 +++++++++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/products/tasks/backend/logic/services/run_log_mirror.py b/products/tasks/backend/logic/services/run_log_mirror.py index 0a3d851ea027..5869d10b7189 100644 --- a/products/tasks/backend/logic/services/run_log_mirror.py +++ b/products/tasks/backend/logic/services/run_log_mirror.py @@ -25,6 +25,11 @@ # 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"} @@ -41,6 +46,13 @@ def mirror_entries( 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 diff --git a/products/tasks/backend/tests/test_run_log_mirror.py b/products/tasks/backend/tests/test_run_log_mirror.py index a41fece9e7be..c65c9d76ef0e 100644 --- a/products/tasks/backend/tests/test_run_log_mirror.py +++ b/products/tasks/backend/tests/test_run_log_mirror.py @@ -8,7 +8,7 @@ from posthog.models import Organization, Team -from products.tasks.backend.logic.services.run_log_mirror import MAX_BODY_CHARS, mirror_entries +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" @@ -107,6 +107,16 @@ def test_oversized_body_is_truncated(self): 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)