diff --git a/docs/internal/sandboxes-setup-guide.md b/docs/internal/sandboxes-setup-guide.md index c515f5900842..602c5c7b2a42 100644 --- a/docs/internal/sandboxes-setup-guide.md +++ b/docs/internal/sandboxes-setup-guide.md @@ -177,6 +177,21 @@ SANDBOX_MCP_URL=https://.ngrok-free.app/mcp `SANDBOX_MCP_URL` overrides the `host.docker.internal` default (which only resolves from local Docker sandboxes, not Modal). Without it, sandbox agents can't reach the MCP server and lose access to the PostHog `execute-sql`, query, and tool-calling stack. +### Agent run telemetry (optional) + +To ship agent-server run metadata to PostHog Logs, set both of the first two; the third additionally produces one APM trace per run (root `task_run` span, a `turn` span per prompt, a `tool_call:` span per tool call) with trace/span ids stamped on the log records: + +```bash +SANDBOX_AGENT_OTEL_LOGS_URL=http://localhost:8000/i/v1/logs # or https://us.i.posthog.com/i/v1/logs +SANDBOX_AGENT_OTEL_LOGS_TOKEN= +SANDBOX_AGENT_OTEL_TRACES_URL=http://localhost:8000/i/v1/traces # optional, enables APM spans +``` + +In cloud, emission is additionally gated per run by the `tasks-agent-run-otel-telemetry` feature flag (org-targeted, stamped into run state at dispatch; it also gates the scout run-log mirror). `DEBUG` bypasses the flag, so locally these settings are the only switch. They're injected into the sandbox as `POSTHOG_AGENT_OTEL_LOGS_URL`/`_TOKEN`/`POSTHOG_AGENT_OTEL_TRACES_URL` (deliberately not standard `OTEL_*` names, so OTel SDKs in user code don't auto-export into the telemetry project). +The agent-server exports run/turn/tool lifecycle metadata (never message content or tool arguments), tagged with `run_id`/`task_id`/`team_id`/`user_id`/`distinct_id` resource attributes and `service.name=posthog-code-agent`. +Telemetry stays off when either of the first two vars is unset. +For local Docker sandboxes the localhost URLs are rewritten to `host.docker.internal` automatically; local ingestion requires the `capture-logs` service to be running. + ### MCP server `.env` `MODAL_DOCKER` (and the local Docker provider) both depend on the MCP server running at `localhost:8787`. The server reads its config from `services/mcp/.env` — without it, things like `POSTHOG_API_BASE_URL`, the UI-apps token, and analytics keys are missing and the server will either refuse to start or return broken responses to the sandbox. @@ -237,6 +252,39 @@ 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. + +In production 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. +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 +``` + +The mirror also has a **direct OTLP leg** that ships each batch straight to a logs ingest endpoint: + +```bash +TASK_RUN_LOGS_MIRROR_OTLP_URL=http://localhost:8000/i/v1/logs # prod: https://us.i.posthog.com/i/v1/logs +TASK_RUN_LOGS_MIRROR_OTLP_TOKEN= +``` + +The token pins the destination: scout runs execute for customer teams, +but their mirrored transcripts must only ever land in — and bill — PostHog's own internal logs project, never the customer's. +Records arrive under `service.name=task-run-log-mirror` with the run uuid as the trace id. + +Locally the direct leg is the only delivery path: `append_log` runs in the host Django process, +and the dev collector (`otel-collector-config.dev.yaml`) only tails docker-compose container stdout, +so without these settings the mirrored lines only show up in the Django phrocs pane. + +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 12f4eefb50f7..6fdaa8776611 100644 --- a/posthog/settings/temporal.py +++ b/posthog/settings/temporal.py @@ -51,6 +51,14 @@ SANDBOX_LLM_GATEWAY_URL: str | None = get_from_env("SANDBOX_LLM_GATEWAY_URL", None, optional=True) SANDBOX_MCP_URL: str | None = get_from_env("SANDBOX_MCP_URL", None, optional=True) +# OTLP destinations for agent-server run telemetry (PostHog Logs/APM). +# Full ingest URLs (e.g. https://us.i.posthog.com/i/v1/logs and .../i/v1/traces) +# plus the project API key of the telemetry project. Telemetry stays off unless +# URL + token are set; the traces URL additionally enables APM spans. +SANDBOX_AGENT_OTEL_LOGS_URL: str | None = get_from_env("SANDBOX_AGENT_OTEL_LOGS_URL", None, optional=True) +SANDBOX_AGENT_OTEL_LOGS_TOKEN: str | None = get_from_env("SANDBOX_AGENT_OTEL_LOGS_TOKEN", None, optional=True) +SANDBOX_AGENT_OTEL_TRACES_URL: str | None = get_from_env("SANDBOX_AGENT_OTEL_TRACES_URL", None, optional=True) + # client_id of the OAuthApplication used to mint the access token the PostHog setup wizard # uses when it runs inside a task sandbox (the "run the wizard in the cloud" onboarding path). # It must be the wizard's own app so the LLM gateway authorizes the token like a normal wizard @@ -94,6 +102,24 @@ "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") +) + +# Direct OTLP delivery for the mirror above. The token pins the destination: scout runs +# execute for customer teams, but their mirrored logs must only ever land in (and bill) +# PostHog's own internal logs project — so this is the internal project's API key, never +# derived from the run's team. Point locally at the dev logs ingest to see mirrored runs +# in /logs. Unset disables the direct leg (stdout emission for the collector remains). +TASK_RUN_LOGS_MIRROR_OTLP_URL: str | None = get_from_env("TASK_RUN_LOGS_MIRROR_OTLP_URL", None, optional=True) +TASK_RUN_LOGS_MIRROR_OTLP_TOKEN: str | None = get_from_env("TASK_RUN_LOGS_MIRROR_OTLP_TOKEN", None, optional=True) + 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/constants.py b/products/tasks/backend/constants.py index 8dc2f7548933..b329cc792fed 100644 --- a/products/tasks/backend/constants.py +++ b/products/tasks/backend/constants.py @@ -7,6 +7,10 @@ AGENT_PROXY_KEEP_STREAM_OPEN_FEATURE_FLAG = "tasks-agent-proxy-keep-stream-open" MODAL_VM_SANDBOX_FEATURE_FLAG = "tasks-modal-vm-sandbox" MODAL_NETWORK_ALLOWLIST_FEATURE_FLAG = "tasks-modal-network-allowlist" +AGENT_RUN_OTEL_TELEMETRY_FEATURE_FLAG = "tasks-agent-run-otel-telemetry" +# Run-state key the telemetry flag decision is stamped under at dispatch (temporal/client.py). +# Consumers read the stamp, so the decision stays stable for the run's whole lifetime. +AGENT_OTEL_TELEMETRY_STATE_KEY = "agent_otel_telemetry_enabled" def vm_sandbox_allowed_origin_products(payload: object) -> set[str]: @@ -365,6 +369,9 @@ def vm_sandbox_allowed_origins(*, distinct_id: str, organization_id: str) -> set "GH_TOKEN", "LLM_GATEWAY_URL", "POSTHOG_RESUME_RUN_ID", + "POSTHOG_AGENT_OTEL_LOGS_URL", + "POSTHOG_AGENT_OTEL_LOGS_TOKEN", + "POSTHOG_AGENT_OTEL_TRACES_URL", "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "DISABLE_TELEMETRY", "DISABLE_ERROR_REPORTING", diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index a427210a1cb2..4cf8974890c3 100644 --- a/products/tasks/backend/facade/api.py +++ b/products/tasks/backend/facade/api.py @@ -37,6 +37,7 @@ from posthog.models.integration import Integration from products.tasks.backend.constants import ( + AGENT_OTEL_TELEMETRY_STATE_KEY, MAX_CUSTOM_IMAGES_PER_TEAM, MAX_CUSTOM_IMAGES_PER_USER, RESERVED_SANDBOX_ENVIRONMENT_VARIABLE_KEYS, @@ -1687,6 +1688,12 @@ def _sync_automation_schedule(automation: TaskAutomation) -> None: "wizard_head_branch", "use_modal_directory_resume_snapshots", "use_modal_vm_sandbox", + # Rollout stamps written once at dispatch by _capture_run_feature_flags; a PATCHable + # value would let a task controller bypass the org feature flags (for telemetry, that + # means injecting the internal OTLP capture token into their sandbox and re-enabling + # the run-log mirror with the rollout off). + AGENT_OTEL_TELEMETRY_STATE_KEY, + "sandbox_event_ingest_enabled", "snapshot_external_id", "snapshot_kind", "snapshot_mount_path", diff --git a/products/tasks/backend/feature_flags.py b/products/tasks/backend/feature_flags.py index dc6f8ba68abb..894d8f0a3c0f 100644 --- a/products/tasks/backend/feature_flags.py +++ b/products/tasks/backend/feature_flags.py @@ -4,6 +4,8 @@ import posthoganalytics +from products.tasks.backend.constants import AGENT_OTEL_TELEMETRY_STATE_KEY, AGENT_RUN_OTEL_TELEMETRY_FEATURE_FLAG + logger = logging.getLogger(__name__) NATIVE_STEERING_SIGNALS_FEATURE_FLAG = "tasks-native-steering-signals" @@ -26,3 +28,32 @@ def is_native_steering_signals_enabled() -> bool: except Exception: logger.exception("native_steering_signals_feature_flag_check_failed") return False + + +def is_agent_otel_telemetry_enabled(*, distinct_id: str, organization_id: str) -> bool: + """Org-gated rollout of agent-run OTel telemetry; fail-closed when evaluation fails.""" + try: + return bool( + posthoganalytics.feature_enabled( + AGENT_RUN_OTEL_TELEMETRY_FEATURE_FLAG, + distinct_id=distinct_id, + groups={"organization": organization_id}, + group_properties={"organization": {"id": organization_id}}, + only_evaluate_locally=False, + send_feature_flag_events=False, + ) + ) + except Exception: + logger.exception("agent_otel_telemetry_flag_check_failed") + return False + + +def agent_otel_telemetry_enabled_for_state(state: dict | None) -> bool: + """Per-run telemetry decision, read from the flag value stamped into run state at dispatch. + + DEBUG bypasses the flag: the analytics SDK is disabled in local dev, where the + telemetry env settings / mirror settings are themselves the opt-in. + """ + if settings.DEBUG: + return True + return (state or {}).get(AGENT_OTEL_TELEMETRY_STATE_KEY) is True diff --git a/products/tasks/backend/logic/services/agentsh.py b/products/tasks/backend/logic/services/agentsh.py index 93b64cc67704..058ec7cb42a3 100644 --- a/products/tasks/backend/logic/services/agentsh.py +++ b/products/tasks/backend/logic/services/agentsh.py @@ -64,6 +64,8 @@ def _port_from_url(url: str | None) -> int | None: "SANDBOX_API_URL", "SANDBOX_LLM_GATEWAY_URL", "SANDBOX_MCP_URL", + "SANDBOX_AGENT_OTEL_LOGS_URL", + "SANDBOX_AGENT_OTEL_TRACES_URL", ) diff --git a/products/tasks/backend/logic/services/docker_sandbox.py b/products/tasks/backend/logic/services/docker_sandbox.py index 598ae9db63bd..30455c4f0296 100644 --- a/products/tasks/backend/logic/services/docker_sandbox.py +++ b/products/tasks/backend/logic/services/docker_sandbox.py @@ -84,6 +84,8 @@ { "POSTHOG_API_URL", "POSTHOG_SITE_URL", + "POSTHOG_AGENT_OTEL_LOGS_URL", + "POSTHOG_AGENT_OTEL_TRACES_URL", "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", "OTEL_EXPORTER_OTLP_ENDPOINT", } 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..167d8adbe78b --- /dev/null +++ b/products/tasks/backend/logic/services/run_log_mirror.py @@ -0,0 +1,281 @@ +"""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). So in production +dogfooding scout-run logs needs no transport of its own: emitting one structured stdout +line per persisted entry is enough. + +A direct OTLP leg (`TASK_RUN_LOGS_MIRROR_OTLP_URL` + `_TOKEN`) additionally ships each +batch straight to a logs ingest endpoint. The token pins the destination to PostHog's own +internal logs project: scout runs execute for customer teams, and their mirrored +transcripts must never land in — or bill — a customer's project. It is also the only +delivery path in local dev, where `append_log` runs in the host Django process whose +stdout the dev collector (docker containers only) never tails. + +Each mirrored line carries the run's uuid as `request_id` (and as the OTLP trace id on +the direct leg), 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 time +from datetime import datetime +from typing import Any + +from django.conf import settings + +import structlog + +from posthog.security.outbound_proxy import internal_requests + +from products.tasks.backend.metrics import RUN_LOG_MIRROR_ENTRIES_TOTAL, RUN_LOG_MIRROR_OTLP_BATCHES_TOTAL + +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"} + +_OTLP_SEVERITIES = {"info": ("INFO", 9), "warn": ("WARN", 13), "error": ("ERROR", 17)} + +_READABLE_SESSION_UPDATES = { + "agent_message", + "agent_message_chunk", + "agent_thought_chunk", + "user_message_chunk", +} + +_TOOL_SESSION_UPDATES = {"tool_call", "tool_call_update"} + +_LOG_EVENT_NAME = "task_run_log" + +_OTLP_SERVICE_NAME = "task-run-log-mirror" + +# (connect, read) — a scalar timeout applies to each stage independently, so a slow +# connect followed by a stalled read would block append_log for the sum. The mirror +# runs synchronously on that hot path; keep the worst case tightly bounded. +_OTLP_TIMEOUT = (1, 3) + +# Identifier-shaped fields (method names, timestamps) come from the same semi-trusted +# request as the entries; cap them so an oversized value can't push the stdout line +# past the collector's 100 KB drop threshold or bloat OTLP attributes. +MAX_IDENTIFIER_CHARS = 200 + +# ACP content blocks nest at most a few levels; an adversarially deep list must not +# RecursionError and abort the whole batch. +_MAX_CONTENT_DEPTH = 10 + + +def mirroring_enabled(origin_product: str) -> bool: + return origin_product in settings.TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS + + +def mirror_entries( + # list[Any], not list[dict]: entries arrive from a semi-trusted request payload, + # so the non-dict guard below is load-bearing. + entries: list[Any], + *, + 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] + records: list[tuple[str, dict[str, Any]]] = [] + 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, update, session_update), + } + method = notification.get("method") + if isinstance(method, str): + fields["acp_method"] = method[:MAX_IDENTIFIER_CHARS] + if session_update: + fields["acp_session_update"] = session_update[:MAX_IDENTIFIER_CHARS] + entry_timestamp = entry.get("timestamp") + if isinstance(entry_timestamp, str): + fields["entry_timestamp"] = entry_timestamp[:MAX_IDENTIFIER_CHARS] + + getattr(logger, _LOG_METHOD_NAMES[severity])(_LOG_EVENT_NAME, **fields) + records.append((severity, fields)) + + if records: + RUN_LOG_MIRROR_ENTRIES_TOTAL.labels(origin_product=origin_product).inc(len(records)) + _post_otlp(records, run_id=run_id) + + +def _post_otlp(records: list[tuple[str, dict[str, Any]]], *, run_id: str) -> None: + """Ship the batch straight to the configured logs OTLP endpoint. + + The token routes the records, so the destination is always the internal logs + project the settings point at — never the run's (customer) team. Best-effort: + a delivery failure is logged and never raises into the run. + """ + url = settings.TASK_RUN_LOGS_MIRROR_OTLP_URL + token = settings.TASK_RUN_LOGS_MIRROR_OTLP_TOKEN + if not url or not token or not records: + return + + # The run uuid without dashes is a valid 16-byte hex trace id, matching the + # request_id -> trace id mapping the production collector applies. + trace_id = run_id.replace("-", "") + trace_fields = {"traceId": trace_id} if len(trace_id) == 32 else {} + event_attribute = {"key": "event", "value": {"stringValue": _LOG_EVENT_NAME}} + log_records = [] + for severity, fields in records: + severity_text, severity_number = _OTLP_SEVERITIES[severity] + log_records.append( + { + "timeUnixNano": str(_time_unix_nano(fields.get("entry_timestamp"))), + "severityText": severity_text, + "severityNumber": severity_number, + "body": {"stringValue": fields["body"]}, + **trace_fields, + "attributes": [ + event_attribute, + *[ + {"key": key, "value": _otlp_attribute_value(value)} + for key, value in fields.items() + if key != "body" + ], + ], + } + ) + payload = { + "resourceLogs": [ + { + "resource": {"attributes": [{"key": "service.name", "value": {"stringValue": _OTLP_SERVICE_NAME}}]}, + "scopeLogs": [{"scope": {"name": __name__}, "logRecords": log_records}], + } + ] + } + try: + response = internal_requests.post( + url, + json=payload, + headers={"Authorization": f"Bearer {token}"}, + timeout=_OTLP_TIMEOUT, + ) + response.raise_for_status() + RUN_LOG_MIRROR_OTLP_BATCHES_TOTAL.labels(outcome="sent").inc() + except Exception as e: + RUN_LOG_MIRROR_OTLP_BATCHES_TOTAL.labels(outcome="failed").inc() + logger.warning("task_run_log_mirror_otlp_failed", task_run_id=run_id, error=str(e)) + + +def _otlp_attribute_value(value: Any) -> dict[str, Any]: + # OTLP JSON encodes 64-bit ints as strings. + if isinstance(value, bool): + return {"boolValue": value} + if isinstance(value, int): + return {"intValue": str(value)} + return {"stringValue": str(value)} + + +def _time_unix_nano(entry_timestamp: Any) -> int: + if isinstance(entry_timestamp, str): + try: + return int(datetime.fromisoformat(entry_timestamp).timestamp() * 1_000_000_000) + except (ValueError, OverflowError): + pass + return time.time_ns() + + +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, update: dict, session_update: str | None) -> str: + raw_params = notification.get("params") + params: dict = raw_params if isinstance(raw_params, dict) else {} + + body: str | None = None + if session_update in _READABLE_SESSION_UPDATES: + text = _extract_text(update.get("content")) + if text is not None: + body = f"[{session_update}] {text}" + else: + body = f"[{session_update}]" + elif session_update in _TOOL_SESSION_UPDATES: + title = update.get("title") or update.get("toolCallId") or "" + status = update.get("status") + body = f"[{session_update}] {title}" + (f" ({status})" if status else "") + elif session_update: + body = f"[{session_update}]" + 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: + method = notification.get("method") + body = f"[{method}]" if isinstance(method, str) else "[acp_message]" + return body[:MAX_BODY_CHARS] + + +def _extract_text(content: Any, depth: int = 0) -> str | None: + """Pull plain text out of an ACP content block (single block or list of blocks).""" + if depth > _MAX_CONTENT_DEPTH: + return None + 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, depth + 1) for block in content) if t] + return "\n".join(parts) if parts else None + return None diff --git a/products/tasks/backend/metrics.py b/products/tasks/backend/metrics.py index 3f6415c8701f..39a82319a3fe 100644 --- a/products/tasks/backend/metrics.py +++ b/products/tasks/backend/metrics.py @@ -59,6 +59,25 @@ ], ) +AGENT_OTEL_TELEMETRY_STAMPED_TOTAL = Counter( + "posthog_tasks_agent_otel_telemetry_stamped_total", + "Agent-run OTel telemetry rollout decisions stamped into run state at dispatch " + "(tasks-agent-run-otel-telemetry flag). First-time stamps only; resumes reuse the stamp.", + labelnames=["enabled"], +) + +RUN_LOG_MIRROR_ENTRIES_TOTAL = Counter( + "posthog_tasks_run_log_mirror_entries_total", + "Task-run log entries mirrored to stdout for the internal Logs project (run_log_mirror).", + labelnames=["origin_product"], +) + +RUN_LOG_MIRROR_OTLP_BATCHES_TOTAL = Counter( + "posthog_tasks_run_log_mirror_otlp_batches_total", + "Direct-OTLP mirror batch deliveries by outcome (the local-dev leg; unset in cloud).", + labelnames=["outcome"], +) + PREWARMED_ACTIVATED_TOTAL = Counter( "posthog_tasks_prewarmed_activated_total", "Pre-warmed Runs that received their first user message (the warm sandbox got used, not reaped)", diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 57f143634f2a..5b190c61f3dc 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -1675,6 +1675,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( @@ -1692,6 +1694,41 @@ 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.feature_flags import agent_otel_telemetry_enabled_for_state + from products.tasks.backend.logic.services.run_log_mirror import mirror_entries, mirroring_enabled + + if not settings.TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS: + return + + # Per-run rollout decision (tasks-agent-run-otel-telemetry), stamped into run + # state at dispatch; fail closed while the stamp is absent. + if not agent_otel_telemetry_enabled_for_state(self.state if isinstance(self.state, dict) else None): + 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.""" diff --git a/products/tasks/backend/presentation/serializers.py b/products/tasks/backend/presentation/serializers.py index e9b7e6be7a56..5a190e064945 100644 --- a/products/tasks/backend/presentation/serializers.py +++ b/products/tasks/backend/presentation/serializers.py @@ -600,6 +600,11 @@ def validate_origin_product(self, value): # would let them expose an arbitrary task to the whole team. The experiments # flow creates its tasks server-side through the facade, never through here. raise serializers.ValidationError("origin_product 'experiments' is reserved for the experiments flow") + if value == tasks_facade.TaskOriginProduct.SIGNALS_SCOUT: + # Scout tasks are created only by the signals scout harness. A forged scout origin + # would route the task's run logs into PostHog's internal Logs project + # (run_log_mirror) and inherit scout visibility semantics. + raise serializers.ValidationError("origin_product 'signals_scout' is reserved for signals scout runs") return value def validate_repository(self, value): diff --git a/products/tasks/backend/temporal/client.py b/products/tasks/backend/temporal/client.py index bc5c3bff73d5..4f5a80c7971d 100644 --- a/products/tasks/backend/temporal/client.py +++ b/products/tasks/backend/temporal/client.py @@ -15,10 +15,10 @@ from posthog.temporal.common.client import async_connect, sync_connect from posthog.temporal.oauth import PosthogMcpScopes -from products.tasks.backend.constants import SANDBOX_EVENT_INGEST_FEATURE_FLAG +from products.tasks.backend.constants import AGENT_OTEL_TELEMETRY_STATE_KEY, SANDBOX_EVENT_INGEST_FEATURE_FLAG from products.tasks.backend.error_telemetry import truncate_error_message -from products.tasks.backend.feature_flags import is_native_steering_signals_enabled -from products.tasks.backend.metrics import observe_task_run_workflow_start +from products.tasks.backend.feature_flags import is_agent_otel_telemetry_enabled, is_native_steering_signals_enabled +from products.tasks.backend.metrics import AGENT_OTEL_TELEMETRY_STAMPED_TOTAL, observe_task_run_workflow_start from products.tasks.backend.models import Task, TaskRun from products.tasks.backend.temporal.build_image.workflow import BuildSandboxImageInput from products.tasks.backend.temporal.constants import ( @@ -113,15 +113,21 @@ def _get_task_run_for_metrics(run_id: str) -> TaskRun | None: return None -def _capture_sandbox_event_ingest_flag(run_id: str) -> None: +def _capture_run_feature_flags(run_id: str) -> None: + """Evaluate per-run rollout flags once at dispatch and stamp them into run state. + + Idempotent per key, so retries and resumes keep the decision the run started with. + """ try: task_run = TaskRun.objects.select_related("task__created_by", "task__team").get(id=run_id) except Exception: - logger.exception("sandbox_event_ingest_capture_run_missing", extra={"run_id": run_id}) + logger.exception("run_feature_flag_capture_run_missing", extra={"run_id": run_id}) return state = task_run.state or {} - if isinstance(state.get("sandbox_event_ingest_enabled"), bool): + need_event_ingest = not isinstance(state.get("sandbox_event_ingest_enabled"), bool) + need_otel_telemetry = not isinstance(state.get(AGENT_OTEL_TELEMETRY_STATE_KEY), bool) + if not need_event_ingest and not need_otel_telemetry: return task = task_run.task @@ -130,33 +136,47 @@ def _capture_sandbox_event_ingest_flag(run_id: str) -> None: task.created_by.distinct_id if task.created_by and task.created_by.distinct_id else "process_task_workflow" ) - try: - enabled = bool( - posthoganalytics.feature_enabled( - SANDBOX_EVENT_INGEST_FEATURE_FLAG, - distinct_id=distinct_id, - groups={"organization": organization_id}, - group_properties={"organization": {"id": organization_id}}, - only_evaluate_locally=False, - send_feature_flag_events=False, + event_ingest_enabled = False + if need_event_ingest: + try: + event_ingest_enabled = bool( + posthoganalytics.feature_enabled( + SANDBOX_EVENT_INGEST_FEATURE_FLAG, + distinct_id=distinct_id, + groups={"organization": organization_id}, + group_properties={"organization": {"id": organization_id}}, + only_evaluate_locally=False, + send_feature_flag_events=False, + ) ) - ) - except Exception as e: - logger.warning( - "sandbox_event_ingest_capture_flag_failed", - extra={"run_id": run_id, "task_id": str(task.id), "error": str(e)}, - ) - enabled = False - - def _set_sandbox_event_ingest_flag(latest_state: dict[str, Any]) -> None: - if not isinstance(latest_state.get("sandbox_event_ingest_enabled"), bool): - latest_state["sandbox_event_ingest_enabled"] = enabled + except Exception as e: + logger.warning( + "sandbox_event_ingest_capture_flag_failed", + extra={"run_id": run_id, "task_id": str(task.id), "error": str(e)}, + ) + otel_telemetry_enabled = need_otel_telemetry and is_agent_otel_telemetry_enabled( + distinct_id=distinct_id, organization_id=organization_id + ) - captured_state = TaskRun.mutate_state_atomic(task_run.id, _set_sandbox_event_ingest_flag) - captured_enabled = captured_state.get("sandbox_event_ingest_enabled", enabled) + def _stamp_flags(latest_state: dict[str, Any]) -> None: + if need_event_ingest and not isinstance(latest_state.get("sandbox_event_ingest_enabled"), bool): + latest_state["sandbox_event_ingest_enabled"] = event_ingest_enabled + if need_otel_telemetry and not isinstance(latest_state.get(AGENT_OTEL_TELEMETRY_STATE_KEY), bool): + latest_state[AGENT_OTEL_TELEMETRY_STATE_KEY] = otel_telemetry_enabled + + captured_state = TaskRun.mutate_state_atomic(task_run.id, _stamp_flags) + if need_otel_telemetry: + AGENT_OTEL_TELEMETRY_STAMPED_TOTAL.labels( + enabled=str(bool(captured_state.get(AGENT_OTEL_TELEMETRY_STATE_KEY))).lower() + ).inc() logger.info( - "sandbox_event_ingest_captured", - extra={"run_id": run_id, "task_id": str(task.id), "sandbox_event_ingest_enabled": captured_enabled}, + "run_feature_flags_captured", + extra={ + "run_id": run_id, + "task_id": str(task.id), + "sandbox_event_ingest_enabled": captured_state.get("sandbox_event_ingest_enabled"), + "agent_otel_telemetry_enabled": captured_state.get(AGENT_OTEL_TELEMETRY_STATE_KEY), + }, ) @@ -196,7 +216,7 @@ async def execute_task_processing_workflow_async( task_run_for_metrics = await _aget_task_run_for_metrics(run_id) observe_task_run_workflow_start(task_run_for_metrics, outcome="attempted", reason="requested") await Team.objects.select_related("organization").aget(id=team_id) - await sync_to_async(_capture_sandbox_event_ingest_flag)(run_id) + await sync_to_async(_capture_run_feature_flags)(run_id) workflow_id = TaskRun.get_workflow_id(task_id, run_id, workflow_id_prefix) if workflow_id_prefix: @@ -279,7 +299,7 @@ def execute_task_processing_workflow( ) Team.objects.get(id=team_id) - _capture_sandbox_event_ingest_flag(run_id) + _capture_run_feature_flags(run_id) workflow_id = TaskRun.get_workflow_id(task_id, run_id, workflow_id_prefix) if workflow_id_prefix: @@ -431,7 +451,7 @@ def redispatch_orphaned_task_run(run_id: str) -> str: ) observe_task_run_workflow_start(task_run, outcome="attempted", reason="reconcile") - _capture_sandbox_event_ingest_flag(run_id) + _capture_run_feature_flags(run_id) try: client = sync_connect() asyncio.run( @@ -461,7 +481,7 @@ def redispatch_orphaned_task_run(run_id: str) -> str: def resume_task_in_cloud_workflow(run_id: str, workflow_id: str) -> None: - _capture_sandbox_event_ingest_flag(run_id) + _capture_run_feature_flags(run_id) client = sync_connect() asyncio.run( client.start_workflow( diff --git a/products/tasks/backend/temporal/process_task/activities/create_sandbox_from_snapshot.py b/products/tasks/backend/temporal/process_task/activities/create_sandbox_from_snapshot.py index 7a37cc291c89..2ee3266a2681 100644 --- a/products/tasks/backend/temporal/process_task/activities/create_sandbox_from_snapshot.py +++ b/products/tasks/backend/temporal/process_task/activities/create_sandbox_from_snapshot.py @@ -126,6 +126,7 @@ def create_sandbox_from_snapshot(input: CreateSandboxFromSnapshotInput) -> Creat access_token=access_token, team_id=ctx.team_id, sandbox_environment=sandbox_env, + otel_telemetry_enabled=ctx.agent_otel_telemetry_enabled, ) environment_variables.update(get_git_identity_env_vars(task, ctx.state)) diff --git a/products/tasks/backend/temporal/process_task/activities/get_task_processing_context.py b/products/tasks/backend/temporal/process_task/activities/get_task_processing_context.py index b38f4dd2111e..bde97d9b82c5 100644 --- a/products/tasks/backend/temporal/process_task/activities/get_task_processing_context.py +++ b/products/tasks/backend/temporal/process_task/activities/get_task_processing_context.py @@ -11,6 +11,7 @@ from posthog.temporal.common.utils import asyncify, close_db_connections from products.tasks.backend.constants import ( + AGENT_OTEL_TELEMETRY_STATE_KEY, AGENT_PROXY_KEEP_STREAM_OPEN_FEATURE_FLAG, CONTINUE_AS_NEW_FEATURE_FLAG, MODAL_DIRECTORY_RESUME_SNAPSHOTS_FEATURE_FLAG, @@ -23,6 +24,7 @@ vm_sandbox_default_base_origin_products, ) from products.tasks.backend.exceptions import TaskInvalidStateError, TaskRunNotReadyError +from products.tasks.backend.feature_flags import is_agent_otel_telemetry_enabled from products.tasks.backend.logic.services.sandbox_config import ( MAX_SANDBOX_CPU_CORES, MAX_SANDBOX_MEMORY_GB, @@ -82,6 +84,9 @@ class TaskProcessingContext: # Captured at workflow start so the sandbox event transport branch is # deterministic for the full run. sandbox_event_ingest_enabled: bool = False + # Captured at workflow start so telemetry env injection (and the run-log mirror, + # which reads the same state stamp) is deterministic for the full run. + agent_otel_telemetry_enabled: bool = False use_modal_vm_sandbox: bool = False use_modal_network_allowlist: bool = False # Burstable by default; the per-run state can opt out to pin a fixed-size box @@ -362,6 +367,33 @@ def _is_sandbox_event_ingest_enabled( return enabled +def _is_agent_otel_telemetry_enabled( + *, + distinct_id: str, + organization_id: str, + run_id: str, + state: dict | None = None, +) -> bool: + # DEBUG first: local dev disables the analytics SDK, so the dispatch capture always + # stamps False there - honoring the stamp would disable telemetry locally even with + # the SANDBOX_AGENT_OTEL_* settings (the local opt-in, which gate emission themselves) + # set. Matches agent_otel_telemetry_enabled_for_state, which the mirror reads. + if settings.DEBUG: + return True + + state_override = (state or {}).get(AGENT_OTEL_TELEMETRY_STATE_KEY) + if isinstance(state_override, bool): + return state_override + + enabled = is_agent_otel_telemetry_enabled(distinct_id=distinct_id, organization_id=organization_id) + log_with_activity_context( + "agent_otel_telemetry_flag_checked", + run_id=run_id, + agent_otel_telemetry_enabled=enabled, + ) + return enabled + + def _is_modal_vm_sandbox_enabled( *, distinct_id: str, @@ -760,6 +792,12 @@ def get_task_processing_context(input: GetTaskProcessingContextInput) -> TaskPro "debug", f"sandbox_event_ingest_enabled: {sandbox_event_ingest_enabled} for this task run", ) + agent_otel_telemetry_enabled = _is_agent_otel_telemetry_enabled( + distinct_id=distinct_id, + organization_id=organization_id, + run_id=run_id, + state=state, + ) use_modal_vm_sandbox = _is_modal_vm_sandbox_enabled( distinct_id=distinct_id, organization_id=organization_id, @@ -886,6 +924,7 @@ def get_task_processing_context(input: GetTaskProcessingContextInput) -> TaskPro use_modal_resume_snapshots=settings.TASKS_USE_MODAL_RESUME_SNAPSHOTS or use_modal_directory_resume_snapshots, use_modal_directory_resume_snapshots=use_modal_directory_resume_snapshots, sandbox_event_ingest_enabled=sandbox_event_ingest_enabled, + agent_otel_telemetry_enabled=agent_otel_telemetry_enabled, use_modal_vm_sandbox=use_modal_vm_sandbox, use_modal_network_allowlist=use_modal_network_allowlist, burstable_sandbox_resources_enabled=burstable_sandbox_resources_enabled, diff --git a/products/tasks/backend/temporal/process_task/activities/provision_sandbox.py b/products/tasks/backend/temporal/process_task/activities/provision_sandbox.py index a121568ae5dc..2c4ec913182f 100644 --- a/products/tasks/backend/temporal/process_task/activities/provision_sandbox.py +++ b/products/tasks/backend/temporal/process_task/activities/provision_sandbox.py @@ -45,6 +45,7 @@ get_sandbox_api_url, get_sandbox_github_token, get_sandbox_name_for_task, + get_sandbox_otel_env_vars, get_sandbox_snapshot_metadata, get_task_run_credential_user, parse_run_state, @@ -344,6 +345,9 @@ def _build_environment_variables( # the cache key) and cost attribution. Rely on Temporal retries instead. environment_variables["POSTHOG_DISABLE_MODEL_FALLBACK"] = "1" + if ctx.agent_otel_telemetry_enabled: + environment_variables.update(get_sandbox_otel_env_vars()) + if ctx.allowed_domains is not None: environment_variables.update(NETWORK_RESTRICTED_AGENT_ENV) diff --git a/products/tasks/backend/temporal/process_task/activities/tests/test_get_task_processing_context.py b/products/tasks/backend/temporal/process_task/activities/tests/test_get_task_processing_context.py index 6424664bb0ee..cab7cff82544 100644 --- a/products/tasks/backend/temporal/process_task/activities/tests/test_get_task_processing_context.py +++ b/products/tasks/backend/temporal/process_task/activities/tests/test_get_task_processing_context.py @@ -24,6 +24,7 @@ from products.tasks.backend.temporal.process_task.activities.get_task_processing_context import ( GetTaskProcessingContextInput, TaskProcessingContext, + _is_agent_otel_telemetry_enabled, _is_agent_proxy_keep_stream_open_enabled, _is_burstable_sandbox_resources_enabled, _is_continue_as_new_enabled, @@ -38,6 +39,26 @@ @pytest.mark.requires_secrets +class TestIsAgentOtelTelemetryEnabled: + @pytest.mark.parametrize( + "debug,state,expected", + [ + # DEBUG must win over the stamp: local dev always stamps False (SDK disabled), + # and the SANDBOX_AGENT_OTEL_* settings are the local opt-in. + (True, {"agent_otel_telemetry_enabled": False}, True), + (True, {}, True), + (False, {"agent_otel_telemetry_enabled": True}, True), + (False, {"agent_otel_telemetry_enabled": False}, False), + ], + ) + def test_debug_wins_then_stamp(self, debug, state, expected): + with override_settings(DEBUG=debug): + assert ( + _is_agent_otel_telemetry_enabled(distinct_id="d", organization_id="o", run_id="r", state=state) + is expected + ) + + class TestGetTaskProcessingContextActivity: def _create_task_with_repo(self, team, user, github_integration, repo_config): return Task.objects.create( diff --git a/products/tasks/backend/temporal/process_task/tests/test_provision_sandbox.py b/products/tasks/backend/temporal/process_task/tests/test_provision_sandbox.py index 20eebaba58b3..0806f9d08e1a 100644 --- a/products/tasks/backend/temporal/process_task/tests/test_provision_sandbox.py +++ b/products/tasks/backend/temporal/process_task/tests/test_provision_sandbox.py @@ -197,3 +197,58 @@ def test_build_environment_variables_disables_telemetry_when_restricted( assert all(env.get(k) == "1" for k in keys) else: assert not (keys & env.keys()) + + +@patch(f"{_PROVISION}.get_git_identity_env_vars", return_value={}) +@patch(f"{_PROVISION}.get_sandbox_jwt_public_key", return_value="pub") +@patch(f"{_PROVISION}.get_sandbox_api_url", return_value="https://api.example") +@pytest.mark.parametrize( + "url, token, traces_url, expected_keys", + [ + ( + "https://us.i.posthog.com/i/v1/logs", + "phc_telemetry", + "https://us.i.posthog.com/i/v1/traces", + {"POSTHOG_AGENT_OTEL_LOGS_URL", "POSTHOG_AGENT_OTEL_LOGS_TOKEN", "POSTHOG_AGENT_OTEL_TRACES_URL"}, + ), + ( + "https://us.i.posthog.com/i/v1/logs", + "phc_telemetry", + None, + {"POSTHOG_AGENT_OTEL_LOGS_URL", "POSTHOG_AGENT_OTEL_LOGS_TOKEN"}, + ), + ("https://us.i.posthog.com/i/v1/logs", None, None, set()), + (None, "phc_telemetry", None, set()), + # Traces alone are useless without the logs pair carrying the token. + (None, None, "https://us.i.posthog.com/i/v1/traces", set()), + ], +) +def test_build_environment_variables_injects_otel_env_only_when_fully_configured( + _api, _jwt, _git, url, token, traces_url, expected_keys +): + ctx = _context(agent_otel_telemetry_enabled=True) + + with override_settings( + SANDBOX_AGENT_OTEL_LOGS_URL=url, + SANDBOX_AGENT_OTEL_LOGS_TOKEN=token, + SANDBOX_AGENT_OTEL_TRACES_URL=traces_url, + ): + env = _build_environment_variables(ctx, MagicMock(), "", "access-token") + + assert {key for key in env if key.startswith("POSTHOG_AGENT_OTEL_")} == expected_keys + + +@patch(f"{_PROVISION}.get_git_identity_env_vars", return_value={}) +@patch(f"{_PROVISION}.get_sandbox_jwt_public_key", return_value="pub") +@patch(f"{_PROVISION}.get_sandbox_api_url", return_value="https://api.example") +def test_build_environment_variables_omits_otel_env_when_flag_disabled(_api, _jwt, _git): + ctx = _context() + + with override_settings( + SANDBOX_AGENT_OTEL_LOGS_URL="https://us.i.posthog.com/i/v1/logs", + SANDBOX_AGENT_OTEL_LOGS_TOKEN="phc_telemetry", + SANDBOX_AGENT_OTEL_TRACES_URL="https://us.i.posthog.com/i/v1/traces", + ): + env = _build_environment_variables(ctx, MagicMock(), "", "access-token") + + assert not any(key.startswith("POSTHOG_AGENT_OTEL_") for key in env) diff --git a/products/tasks/backend/temporal/process_task/tests/test_utils.py b/products/tasks/backend/temporal/process_task/tests/test_utils.py index 44769040e084..bfe800dbac81 100644 --- a/products/tasks/backend/temporal/process_task/tests/test_utils.py +++ b/products/tasks/backend/temporal/process_task/tests/test_utils.py @@ -1,6 +1,6 @@ from unittest.mock import MagicMock, patch -from django.test import TestCase +from django.test import SimpleTestCase, TestCase, override_settings from parameterized import parameterized @@ -16,6 +16,7 @@ McpServerConfig, RunState, build_imported_mcp_server_configs, + build_sandbox_environment_variables, get_git_identity_env_vars, get_github_credential_source, get_imported_mcp_server_configs, @@ -1066,3 +1067,61 @@ def test_collision_detection_is_case_insensitive(self): ) assert get_relayed_mcp_server_names(task_run, {"grafana"}) == ["Playwright", "internal-cli"] + + +class TestBuildSandboxEnvironmentVariables(SimpleTestCase): + @patch( + "products.tasks.backend.logic.services.connection_token.get_sandbox_jwt_public_key", + return_value="pub", + ) + @patch( + "products.tasks.backend.temporal.process_task.utils.get_sandbox_api_url", + return_value="https://api.example", + ) + def test_snapshot_resume_env_includes_otel_config_when_configured(self, _api, _jwt) -> None: + with override_settings( + SANDBOX_AGENT_OTEL_LOGS_URL="https://us.i.posthog.com/i/v1/logs", + SANDBOX_AGENT_OTEL_LOGS_TOKEN="phc_telemetry", + SANDBOX_AGENT_OTEL_TRACES_URL="https://us.i.posthog.com/i/v1/traces", + ): + env = build_sandbox_environment_variables(None, "access-token", 1, otel_telemetry_enabled=True) + + assert env["POSTHOG_AGENT_OTEL_LOGS_URL"] == "https://us.i.posthog.com/i/v1/logs" + assert env["POSTHOG_AGENT_OTEL_LOGS_TOKEN"] == "phc_telemetry" + assert env["POSTHOG_AGENT_OTEL_TRACES_URL"] == "https://us.i.posthog.com/i/v1/traces" + + @patch( + "products.tasks.backend.logic.services.connection_token.get_sandbox_jwt_public_key", + return_value="pub", + ) + @patch( + "products.tasks.backend.temporal.process_task.utils.get_sandbox_api_url", + return_value="https://api.example", + ) + def test_snapshot_resume_env_omits_otel_config_without_logs_pair(self, _api, _jwt) -> None: + with override_settings( + SANDBOX_AGENT_OTEL_LOGS_URL="https://us.i.posthog.com/i/v1/logs", + SANDBOX_AGENT_OTEL_LOGS_TOKEN=None, + SANDBOX_AGENT_OTEL_TRACES_URL="https://us.i.posthog.com/i/v1/traces", + ): + env = build_sandbox_environment_variables(None, "access-token", 1, otel_telemetry_enabled=True) + + assert not any(key.startswith("POSTHOG_AGENT_OTEL_") for key in env) + + @patch( + "products.tasks.backend.logic.services.connection_token.get_sandbox_jwt_public_key", + return_value="pub", + ) + @patch( + "products.tasks.backend.temporal.process_task.utils.get_sandbox_api_url", + return_value="https://api.example", + ) + def test_snapshot_resume_env_omits_otel_config_when_flag_disabled(self, _api, _jwt) -> None: + with override_settings( + SANDBOX_AGENT_OTEL_LOGS_URL="https://us.i.posthog.com/i/v1/logs", + SANDBOX_AGENT_OTEL_LOGS_TOKEN="phc_telemetry", + SANDBOX_AGENT_OTEL_TRACES_URL="https://us.i.posthog.com/i/v1/traces", + ): + env = build_sandbox_environment_variables(None, "access-token", 1) + + assert not any(key.startswith("POSTHOG_AGENT_OTEL_") for key in env) diff --git a/products/tasks/backend/temporal/process_task/utils.py b/products/tasks/backend/temporal/process_task/utils.py index 242f9e67a667..e7ebb32083d5 100644 --- a/products/tasks/backend/temporal/process_task/utils.py +++ b/products/tasks/backend/temporal/process_task/utils.py @@ -1118,6 +1118,7 @@ def build_sandbox_environment_variables( access_token: str, team_id: int, sandbox_environment: Optional[Any] = None, + otel_telemetry_enabled: bool = False, ) -> dict[str, str]: """Build the environment variables dict for a sandbox, merging user env vars from SandboxEnvironment. @@ -1148,6 +1149,27 @@ def build_sandbox_environment_variables( if settings.SANDBOX_LLM_GATEWAY_URL: env_vars["LLM_GATEWAY_URL"] = settings.SANDBOX_LLM_GATEWAY_URL + if otel_telemetry_enabled: + env_vars.update(get_sandbox_otel_env_vars()) + + return env_vars + + +def get_sandbox_otel_env_vars() -> dict[str, str]: + """OTLP config for agent-server run telemetry (PostHog Logs/APM). + + Deliberately POSTHOG_-prefixed rather than the standard OTEL_* names: the + sandbox env is inherited by user processes, and standard OTEL_* vars would + make any OTel SDK in user code auto-export into our telemetry project. + """ + if not (settings.SANDBOX_AGENT_OTEL_LOGS_URL and settings.SANDBOX_AGENT_OTEL_LOGS_TOKEN): + return {} + env_vars = { + "POSTHOG_AGENT_OTEL_LOGS_URL": settings.SANDBOX_AGENT_OTEL_LOGS_URL, + "POSTHOG_AGENT_OTEL_LOGS_TOKEN": settings.SANDBOX_AGENT_OTEL_LOGS_TOKEN, + } + if settings.SANDBOX_AGENT_OTEL_TRACES_URL: + env_vars["POSTHOG_AGENT_OTEL_TRACES_URL"] = settings.SANDBOX_AGENT_OTEL_TRACES_URL return env_vars diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py index 26f05f5b4888..2bb29f7e3bb4 100644 --- a/products/tasks/backend/tests/test_api.py +++ b/products/tasks/backend/tests/test_api.py @@ -4317,6 +4317,8 @@ def test_patch_cannot_mutate_protected_credential_state_keys(self, _mock_publish "inactivity_timeout_seconds": 600, "use_modal_directory_resume_snapshots": True, "use_modal_vm_sandbox": False, + "agent_otel_telemetry_enabled": False, + "sandbox_event_ingest_enabled": False, "snapshot_external_id": "im-real", "snapshot_kind": "directory", "snapshot_mount_path": "/tmp", @@ -4348,6 +4350,8 @@ def test_patch_cannot_mutate_protected_credential_state_keys(self, _mock_publish "wizard_config": {}, "use_modal_directory_resume_snapshots": False, "use_modal_vm_sandbox": True, + "agent_otel_telemetry_enabled": True, + "sandbox_event_ingest_enabled": True, "snapshot_external_id": "im-attacker", "snapshot_kind": "directory", "snapshot_mount_path": "/tmp/workspace", @@ -4378,6 +4382,8 @@ def test_patch_cannot_mutate_protected_credential_state_keys(self, _mock_publish assert "wizard_config" not in run.state # caller cannot mark a run as a wizard run assert run.state["use_modal_directory_resume_snapshots"] is True assert run.state["use_modal_vm_sandbox"] is False + assert run.state["agent_otel_telemetry_enabled"] is False + assert run.state["sandbox_event_ingest_enabled"] is False assert run.state["snapshot_external_id"] == "im-real" assert run.state["snapshot_kind"] == "directory" assert run.state["snapshot_mount_path"] == "/tmp" @@ -4394,6 +4400,7 @@ def test_patch_cannot_mutate_protected_credential_state_keys(self, _mock_publish "state": {}, "state_remove_keys": [ "github_credential_source", + "agent_otel_telemetry_enabled", "sandbox_id", "use_modal_directory_resume_snapshots", "use_modal_vm_sandbox", @@ -4412,6 +4419,7 @@ def test_patch_cannot_mutate_protected_credential_state_keys(self, _mock_publish self.assertEqual(response.status_code, status.HTTP_200_OK) run.refresh_from_db() assert run.state["github_credential_source"] == "caller_token" # protected key survives removal + assert run.state["agent_otel_telemetry_enabled"] is False # protected key survives removal assert run.state["sandbox_id"] == "sb-real" # protected key survives removal assert run.state["use_modal_directory_resume_snapshots"] is True # protected key survives removal assert run.state["use_modal_vm_sandbox"] is False # protected key survives removal diff --git a/products/tasks/backend/tests/test_presentation_serializers.py b/products/tasks/backend/tests/test_presentation_serializers.py index 8ba6c5f48cd8..3cc7d2446dc7 100644 --- a/products/tasks/backend/tests/test_presentation_serializers.py +++ b/products/tasks/backend/tests/test_presentation_serializers.py @@ -9,9 +9,24 @@ from products.tasks.backend.presentation.serializers import ( TaskRunCreateRequestSerializer, TaskRunLivingArtifactCreateRequestSerializer, + TaskWriteSerializer, ) +class TestTaskWriteSerializerOriginProduct(SimpleTestCase): + @parameterized.expand( + [ + ("image_builder", True), + ("signals_scout", True), + ("user_created", False), + ] + ) + def test_internal_only_origins_are_rejected(self, origin_product: str, expected_rejected: bool) -> None: + serializer = TaskWriteSerializer(data={"origin_product": origin_product}) + serializer.is_valid() + assert ("origin_product" in serializer.errors) is expected_rejected + + class TestTaskRunLivingArtifactCreateRequestSerializer(SimpleTestCase): @parameterized.expand( [ 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..7efdde6dc8e4 --- /dev/null +++ b/products/tasks/backend/tests/test_run_log_mirror.py @@ -0,0 +1,344 @@ +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.feature_flags import agent_otel_telemetry_enabled_for_state +from products.tasks.backend.logic.services.run_log_mirror import ( + MAX_BODY_CHARS, + MAX_ENTRIES_PER_CALL, + MAX_IDENTIFIER_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 _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) + + @parameterized.expand( + [ + ( + "protocol_request", + { + "method": "session/new", + "params": { + "mcpServers": [ + { + "headers": [ + {"name": "Authorization", "value": "Bearer protocol-secret"}, + ] + } + ] + }, + }, + "[session/new]", + ), + ( + "metadata_session_update", + _session_update_entry( + "available_commands_update", + availableCommands=[{"name": "example", "description": "metadata-secret"}], + )["notification"], + "[available_commands_update]", + ), + ] + ) + def test_protocol_metadata_omits_payload(self, _name, notification, expected_body): + mock_logger = _mirror([{"notification": notification}]) + body = mock_logger.info.call_args.kwargs["body"] + self.assertEqual(body, expected_body) + self.assertNotIn("secret", body) + + 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_identifier_fields_are_capped(self): + entry = {"timestamp": "9" * 100_000, "notification": {"method": "m" * 100_000}} + mock_logger = _mirror([entry]) + fields = mock_logger.info.call_args.kwargs + self.assertEqual(len(fields["acp_method"]), MAX_IDENTIFIER_CHARS) + self.assertEqual(len(fields["entry_timestamp"]), MAX_IDENTIFIER_CHARS) + + def test_deeply_nested_content_returns_type_only_body(self): + content: list = ["bottom"] + for _ in range(2_000): + content = [content] + mock_logger = _mirror([_session_update_entry("agent_message", content=content)]) + self.assertEqual(mock_logger.info.call_args.kwargs["body"], "[agent_message]") + + 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() + + +class TestMirrorOtlpDelivery(SimpleTestCase): + @override_settings( + TASK_RUN_LOGS_MIRROR_OTLP_URL="https://us.i.posthog.com/i/v1/logs", + TASK_RUN_LOGS_MIRROR_OTLP_TOKEN="phc_internal", + ) + def test_posts_one_otlp_batch_with_severity_trace_and_attributes(self): + entries = [ + _session_update_entry("agent_message", content={"type": "text", "text": "hello"}), + {"notification": {"method": "_posthog/error", "params": {"message": "boom"}}}, + ] + with patch("products.tasks.backend.logic.services.run_log_mirror.internal_requests") as mock_requests: + _mirror(entries) + + mock_requests.post.assert_called_once() + args, kwargs = mock_requests.post.call_args + assert args[0] == "https://us.i.posthog.com/i/v1/logs" + # The configured internal-project token routes the records — never a key + # derived from the run's (customer) team. + assert kwargs["headers"] == {"Authorization": "Bearer phc_internal"} + records = kwargs["json"]["resourceLogs"][0]["scopeLogs"][0]["logRecords"] + assert [r["severityText"] for r in records] == ["INFO", "ERROR"] + assert records[0]["body"]["stringValue"] == "[agent_message] hello" + assert records[1]["body"]["stringValue"] == "boom" + # The run uuid (dashes stripped) is the trace id, grouping the run. + assert records[0]["traceId"] == RUN_ID.replace("-", "") + attributes = {a["key"]: a["value"] for a in records[0]["attributes"]} + assert attributes["event"] == {"stringValue": "task_run_log"} + assert attributes["task_run_id"] == {"stringValue": RUN_ID} + # OTLP JSON encodes 64-bit ints as strings. + assert attributes["team_id"] == {"intValue": "2"} + + @override_settings( + TASK_RUN_LOGS_MIRROR_OTLP_URL="https://us.i.posthog.com/i/v1/logs", + TASK_RUN_LOGS_MIRROR_OTLP_TOKEN="phc_internal", + ) + def test_protocol_payloads_never_reach_the_otlp_payload(self): + entry = { + "notification": { + "method": "session/new", + "params": {"mcpServers": [{"headers": [{"name": "Authorization", "value": "Bearer protocol-secret"}]}]}, + } + } + with patch("products.tasks.backend.logic.services.run_log_mirror.internal_requests") as mock_requests: + _mirror([entry]) + + payload = mock_requests.post.call_args.kwargs["json"] + self.assertNotIn("protocol-secret", json.dumps(payload)) + records = payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"] + # The attribute set is closed: run identity plus entry type, never raw params. + self.assertEqual( + {a["key"] for a in records[0]["attributes"]}, + {"event", "request_id", "task_run_id", "task_id", "team_id", "origin_product", "acp_method"}, + ) + + @parameterized.expand( + [ + ("both_unset", None, None), + ("url_only", "https://us.i.posthog.com/i/v1/logs", None), + ("token_only", None, "phc_internal"), + ] + ) + def test_no_http_delivery_unless_fully_configured(self, _name, url, token): + entry = _session_update_entry("agent_message", content={"type": "text", "text": "hello"}) + with override_settings(TASK_RUN_LOGS_MIRROR_OTLP_URL=url, TASK_RUN_LOGS_MIRROR_OTLP_TOKEN=token): + with patch("products.tasks.backend.logic.services.run_log_mirror.internal_requests") as mock_requests: + _mirror([entry]) + + mock_requests.post.assert_not_called() + + +class TestAgentOtelTelemetryStateGate(SimpleTestCase): + @parameterized.expand( + [ + ("stamped_true", {"agent_otel_telemetry_enabled": True}, True), + ("stamped_false", {"agent_otel_telemetry_enabled": False}, False), + ("missing", {}, False), + ("none_state", None, False), + ] + ) + def test_fails_closed_outside_debug(self, _name, state, expected): + assert agent_otel_telemetry_enabled_for_state(state) is expected + + @override_settings(DEBUG=True) + def test_debug_bypasses_the_flag(self): + assert agent_otel_telemetry_enabled_for_state(None) is True + + +@override_settings(TASK_RUN_LOGS_MIRROR_ORIGIN_PRODUCTS=["signals_scout"]) +class TestAppendLogMirroring(TestCase): + organization: Organization + team: Team + + @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, telemetry_enabled: bool | None = True) -> TaskRun: + task = Task.objects.create( + team=self.team, + title="Test Task", + description="Test", + origin_product=origin_product, + ) + state = {"agent_otel_telemetry_enabled": telemetry_enabled} if telemetry_enabled is not None else {} + return TaskRun.objects.create(team=self.team, task=task, state=state) + + @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() + + @patch("products.tasks.backend.logic.services.run_log_mirror.logger") + @patch("products.tasks.backend.models.object_storage") + def test_no_mirroring_without_telemetry_flag_stamp(self, mock_storage, mock_logger): + mock_storage.read.return_value = None + run = self._create_run(Task.OriginProduct.SIGNALS_SCOUT, telemetry_enabled=None) + + 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() + + @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() diff --git a/products/tasks/backend/tests/test_temporal_client.py b/products/tasks/backend/tests/test_temporal_client.py index f57c38d75e93..3f3805d3afec 100644 --- a/products/tasks/backend/tests/test_temporal_client.py +++ b/products/tasks/backend/tests/test_temporal_client.py @@ -209,7 +209,7 @@ def test_does_not_overwrite_run_that_already_started(self, executor: str) -> Non self.assertIsNone(run.completed_at) @parameterized.expand([("sync",), ("async",)]) - def test_captures_sandbox_event_ingest_flag_before_starting_workflow(self, executor: str) -> None: + def test_captures_run_feature_flags_before_starting_workflow(self, executor: str) -> None: run = self._create_run() client = Mock() client.start_workflow = AsyncMock() @@ -229,14 +229,19 @@ def test_captures_sandbox_event_ingest_flag_before_starting_workflow(self, execu run.refresh_from_db() self.assertEqual(run.state["sandbox_event_ingest_enabled"], True) - flag.assert_called_once_with( - "tasks-cloud-runs-sandbox-event-ingest", - distinct_id="process_task_workflow", - groups={"organization": str(self.organization.id)}, - group_properties={"organization": {"id": str(self.organization.id)}}, - only_evaluate_locally=False, - send_feature_flag_events=False, - ) + self.assertEqual(run.state["agent_otel_telemetry_enabled"], True) + # Patching the shared posthoganalytics module attribute covers both evaluation + # sites (event ingest in client.py, telemetry in feature_flags.py). + self.assertEqual(flag.call_count, 2) + for flag_key in ("tasks-cloud-runs-sandbox-event-ingest", "tasks-agent-run-otel-telemetry"): + flag.assert_any_call( + flag_key, + distinct_id="process_task_workflow", + groups={"organization": str(self.organization.id)}, + group_properties={"organization": {"id": str(self.organization.id)}}, + only_evaluate_locally=False, + send_feature_flag_events=False, + ) def test_captures_sandbox_event_ingest_flag_before_resuming_workflow(self) -> None: run = self._create_run()