feat(tasks): mirror scout run logs into posthog logs

Incorporates the scout run log mirror from PR #71094 (credit: Andrew
Maguire, @andrewm4894) into the agent-run telemetry branch, squashed from
the PR's final state so both delivery paths ship together.

Persisted task-run log entries are mirrored into the PostHog Logs product,
scoped to scout runs. There is no transport of its own: TaskRun.append_log
emits one structured stdout line per persisted entry (event=task_run_log)
and the per-cluster OTel collector that already tails container stdout
ships them into the region's internal project's Logs, parsing JSON keys
into queryable attributes and request_id (the run uuid) into a trace id so
one run groups as one trace.

- run_log_mirror.py translates each ACP JSONL entry: readable bodies for
  agent messages / tool calls / sandbox output / turn ends (8k char cap,
  bounded entries per call), severity mapping, and run-identity fields as
  log attributes
- TaskRun.append_log calls the mirror after the S3 write, guarded so any
  failure is logged and never breaks the run
- TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS setting gates which task origins
  mirror (default signals_scout; empty disables)

Complementary to the OTLP run-telemetry export on this branch: the mirror
carries full readable bodies for PostHog-authored scout runs into the
internal project, while the OTLP path stays metadata-only and covers
customer-driven cloud runs.

Generated-By: PostHog Code
Task-Id: 0c511836-2180-455a-9b58-45df2a0661ec

Author: tatoalo

docs/internal/sandboxes-setup-guide.md

@@ -252,6 +252,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:

posthog/settings/temporal.py

@@ -102,6 +102,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)
 

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

products/tasks/backend/models.py

@@ -1669,6 +1669,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(
@@ -1686,6 +1688,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 effective_rtk(self) -> bool | None:
         """rtk posture for analytics: the launch-persisted effective value, falling
         back to the user's explicit override for runs that never launched."""