diff --git a/frontend/src/scenes/inbox/components/config/scouts/ScoutBadges.tsx b/frontend/src/scenes/inbox/components/config/scouts/ScoutBadges.tsx index eb949b719419..0df818441cdc 100644 --- a/frontend/src/scenes/inbox/components/config/scouts/ScoutBadges.tsx +++ b/frontend/src/scenes/inbox/components/config/scouts/ScoutBadges.tsx @@ -2,6 +2,28 @@ import { LemonTag, Tooltip } from '@posthog/lemon-ui' import type { ScoutOriginEnumApi } from 'products/signals/frontend/generated/api.schemas' +/** + * Shown when a scout paused itself after failing repeatedly (`status=paused_by_system`, + * `pause_reason=repeated_failures`). Without it the pause is only visible in the API + * response, and a scout that has silently stopped looks the same as one a person turned off. + */ +export function ScoutPausedBadge(): JSX.Element { + return ( + + This scout paused itself because its last few runs all failed. It retries about once a day and + resumes its normal schedule on the first successful run. Turn it on to resume it right away. + + } + > + + Paused + + + ) +} + /** Canonical (PostHog-maintained) vs Custom (team-authored) scout badge. */ export function ScoutOriginBadge({ origin }: { origin: ScoutOriginEnumApi }): JSX.Element { return ( diff --git a/frontend/src/scenes/inbox/components/config/scouts/ScoutConfigControls.test.tsx b/frontend/src/scenes/inbox/components/config/scouts/ScoutConfigControls.test.tsx index 0dbf9126a357..1b79a92754b4 100644 --- a/frontend/src/scenes/inbox/components/config/scouts/ScoutConfigControls.test.tsx +++ b/frontend/src/scenes/inbox/components/config/scouts/ScoutConfigControls.test.tsx @@ -20,6 +20,7 @@ const config: SignalScoutConfigApi = { run_cron_schedule: '0 9 * * *', output_destinations: {}, last_run_at: null, + consecutive_failure_count: 0, created_at: '2026-07-21T12:00:00Z', } diff --git a/frontend/src/scenes/inbox/components/config/scouts/ScoutRowCard.tsx b/frontend/src/scenes/inbox/components/config/scouts/ScoutRowCard.tsx index b30a160c461f..d5d867d4f29c 100644 --- a/frontend/src/scenes/inbox/components/config/scouts/ScoutRowCard.tsx +++ b/frontend/src/scenes/inbox/components/config/scouts/ScoutRowCard.tsx @@ -21,7 +21,7 @@ import { ScoutRollup, } from '../../../utils/scoutRunsWindow' import { agentSetupModalLogic } from '../../shell/agentSetupModalLogic' -import { ScoutOriginBadge } from './ScoutBadges' +import { ScoutOriginBadge, ScoutPausedBadge } from './ScoutBadges' import { ScoutConfigForm, ScoutEnabledSwitch } from './ScoutConfigControls' import { ScoutRunBoxes } from './ScoutRunBoxes' @@ -129,6 +129,9 @@ export function ScoutRowCard({ + {config.status === 'paused_by_system' && config.pause_reason === 'repeated_failures' ? ( + + ) : null}
diff --git a/frontend/src/scenes/inbox/logics/scoutCreateModalLogic.test.ts b/frontend/src/scenes/inbox/logics/scoutCreateModalLogic.test.ts index 7537bab63d72..1b7c4c03ace9 100644 --- a/frontend/src/scenes/inbox/logics/scoutCreateModalLogic.test.ts +++ b/frontend/src/scenes/inbox/logics/scoutCreateModalLogic.test.ts @@ -40,6 +40,7 @@ const CREATED_SCOUT: SignalScoutCreateResponseApi = { run_cron_schedule: null, output_destinations: {}, last_run_at: null, + consecutive_failure_count: 0, created_at: '2026-07-24T00:00:00Z', }, } diff --git a/frontend/src/scenes/inbox/logics/scoutFleetLogic.test.ts b/frontend/src/scenes/inbox/logics/scoutFleetLogic.test.ts index 8d35bc5fb29f..fb0d0aed3aa3 100644 --- a/frontend/src/scenes/inbox/logics/scoutFleetLogic.test.ts +++ b/frontend/src/scenes/inbox/logics/scoutFleetLogic.test.ts @@ -45,6 +45,7 @@ const BASE_CONFIG: SignalScoutConfigApi = { run_cron_schedule: null, output_destinations: {}, last_run_at: null, + consecutive_failure_count: 0, created_at: '2026-07-22T00:00:00Z', } diff --git a/posthog/models/activity_logging/activity_log.py b/posthog/models/activity_logging/activity_log.py index 6c3a1f4906f6..749fe93609bc 100644 --- a/posthog/models/activity_logging/activity_log.py +++ b/posthog/models/activity_logging/activity_log.py @@ -396,11 +396,13 @@ class Meta: "Subscription": [ "next_delivery_date", ], - # `last_run_at` is written by the scout coordinator on every tick (~every 15 min per scout). - # When that is the only change, suppress the activity signal entirely so run bookkeeping - # never spams the audit log. + # `last_run_at` is written by the scout coordinator on every tick (~every 15 min per scout), + # and the failure streak by the runner on every run outcome. When those are the only + # change, suppress the activity signal entirely so run bookkeeping never spams the audit + # log. A breaker trip is NOT suppressed: it moves `status`, which logs like any pause. "SignalScoutConfig": [ "last_run_at", + "consecutive_failure_count", ], } @@ -760,6 +762,7 @@ class Meta: # Run bookkeeping, not user intent — keep it out of change detection even when it # rides along with a real change (belt-and-suspenders with signal_exclusions above). "last_run_at", + "consecutive_failure_count", # Companion bookkeeping that rides along with every logged `status` change; the # activity log entry itself already carries who and when. "status_changed_at", diff --git a/products/signals/backend/admin.py b/products/signals/backend/admin.py index 8300a4d69073..7960b8565c43 100644 --- a/products/signals/backend/admin.py +++ b/products/signals/backend/admin.py @@ -116,6 +116,7 @@ class SignalScoutConfigAdmin(admin.ModelAdmin): "pause_reason", "status_changed_at", "status_changed_by", + "consecutive_failure_count", ) list_select_related = ("team", "team__organization") show_full_result_count = False diff --git a/products/signals/backend/migrations/0079_signalscoutconfig_consecutive_failure_count.py b/products/signals/backend/migrations/0079_signalscoutconfig_consecutive_failure_count.py new file mode 100644 index 000000000000..2ddf6fd8e8ea --- /dev/null +++ b/products/signals/backend/migrations/0079_signalscoutconfig_consecutive_failure_count.py @@ -0,0 +1,13 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [("signals", "0078_backfill_scout_status")] + + operations = [ + migrations.AddField( + model_name="signalscoutconfig", + name="consecutive_failure_count", + field=models.PositiveIntegerField(db_default=0, default=0), + ), + ] diff --git a/products/signals/backend/migrations/0080_reconcile_scout_status_enabled.py b/products/signals/backend/migrations/0080_reconcile_scout_status_enabled.py new file mode 100644 index 000000000000..914b71863a55 --- /dev/null +++ b/products/signals/backend/migrations/0080_reconcile_scout_status_enabled.py @@ -0,0 +1,33 @@ +from django.db import migrations + + +def reconcile_enabled_wins(apps, schema_editor): + # Re-runs 0078's enabled-wins reconciliation over rows old instances drifted during the + # 0077 rollout window (enabled-only writes from pods that predated the dual-write land + # after the backfill). `enabled` carries the operator's intent for any mismatch: no + # system writer existed before the constraints in 0081, so a disagreeing row can only be + # a human toggle that missed the `status` side. Small table, plain UPDATEs. + SignalScoutConfig = apps.get_model("signals", "SignalScoutConfig") + manager = SignalScoutConfig._default_manager + manager.filter(enabled=False, status__in=["active", "pending_pause"]).update( + status="paused_by_user", pause_reason=None + ) + manager.filter(enabled=True, status__in=["paused_by_system", "paused_by_user"]).update( + status="active", pause_reason=None + ) + # Reason coherence for 0081's second constraint: a reason on a runnable/user-paused row is + # stray; a reasonless warning is meaningless (runnable anyway); a reasonless system pause + # can only predate system writers, so it reads as a human's off-switch. + manager.filter(status__in=["active", "paused_by_user"], pause_reason__isnull=False).update(pause_reason=None) + manager.filter(status="pending_pause", pause_reason__isnull=True).update(status="active") + manager.filter(status="paused_by_system", pause_reason__isnull=True).update(status="paused_by_user") + + +class Migration(migrations.Migration): + dependencies = [ + ("signals", "0079_signalscoutconfig_consecutive_failure_count"), + ] + + operations = [ + migrations.RunPython(reconcile_enabled_wins, reverse_code=migrations.RunPython.noop), + ] diff --git a/products/signals/backend/migrations/0081_scout_status_constraints.py b/products/signals/backend/migrations/0081_scout_status_constraints.py new file mode 100644 index 000000000000..45f864867c9a --- /dev/null +++ b/products/signals/backend/migrations/0081_scout_status_constraints.py @@ -0,0 +1,28 @@ +from django.db import migrations, models + +from posthog.migration_helpers import AddConstraintNotValid + + +class Migration(migrations.Migration): + dependencies = [ + ("signals", "0080_reconcile_scout_status_enabled"), + ] + + operations = [ + AddConstraintNotValid( + model_name="signalscoutconfig", + constraint=models.CheckConstraint( + name="scout_config_enabled_matches_status", + condition=models.Q(enabled=True, status__in=["active", "pending_pause"]) + | models.Q(enabled=False, status__in=["paused_by_system", "paused_by_user"]), + ), + ), + AddConstraintNotValid( + model_name="signalscoutconfig", + constraint=models.CheckConstraint( + name="scout_config_pause_reason_matches_status", + condition=models.Q(status__in=["pending_pause", "paused_by_system"], pause_reason__isnull=False) + | models.Q(status__in=["active", "paused_by_user"], pause_reason__isnull=True), + ), + ), + ] diff --git a/products/signals/backend/migrations/0082_validate_scout_status_constraints.py b/products/signals/backend/migrations/0082_validate_scout_status_constraints.py new file mode 100644 index 000000000000..993741ac9324 --- /dev/null +++ b/products/signals/backend/migrations/0082_validate_scout_status_constraints.py @@ -0,0 +1,14 @@ +from django.db import migrations + +from posthog.migration_helpers import ValidateConstraint + + +class Migration(migrations.Migration): + dependencies = [ + ("signals", "0081_scout_status_constraints"), + ] + + operations = [ + ValidateConstraint(model_name="signalscoutconfig", name="scout_config_enabled_matches_status"), + ValidateConstraint(model_name="signalscoutconfig", name="scout_config_pause_reason_matches_status"), + ] diff --git a/products/signals/backend/migrations/max_migration.txt b/products/signals/backend/migrations/max_migration.txt index 0d601943f305..8440e2ffd447 100644 --- a/products/signals/backend/migrations/max_migration.txt +++ b/products/signals/backend/migrations/max_migration.txt @@ -1 +1 @@ -0078_backfill_scout_status +0082_validate_scout_status_constraints diff --git a/products/signals/backend/models.py b/products/signals/backend/models.py index 15b50904f47a..b7ce00749f42 100644 --- a/products/signals/backend/models.py +++ b/products/signals/backend/models.py @@ -1237,6 +1237,14 @@ class PauseReason(models.TextChoices): # Stamped by the coordinator after each dispatch; drives the due-check. Written every # run, so it is excluded from activity logging (see field_exclusions below). last_run_at = models.DateTimeField(null=True, blank=True) + # Failure-streak circuit breaker over this lane's run outcomes, maintained by the runner: + # bumped on a failed run, zeroed on a successful one. At `FAILURE_STREAK_PAUSE_THRESHOLD` + # the runner pauses the lane (`transition_status_by_system`, `repeated_failures`) and the + # coordinator holds it to one probe per `AUTO_PAUSE_PROBE_INTERVAL_S`. Without it a lane + # that can never succeed re-dispatches forever, taking a full-length sandbox lease per + # interval to produce nothing. Written on every run, so it is excluded from activity + # logging like `last_run_at`; the pause itself logs through `status` like any other. + consecutive_failure_count = models.PositiveIntegerField(default=0, db_default=0) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) created_by = models.ForeignKey( @@ -1261,13 +1269,20 @@ class Meta: default_manager_name = "all_teams" constraints = [ models.UniqueConstraint(fields=["team", "skill_name"], name="unique_scout_config_per_team_skill"), - # A CHECK constraint tying `enabled` to `status` is deliberately deferred to a - # follow-up migration: enforcing it in the same deploy that introduces the - # dual-write breaks rolling deploys, because not-yet-replaced instances still - # write `enabled` alone and a NOT VALID constraint already checks new writes. - # That follow-up must first re-run the enabled-wins reconciliation over drifted - # rows (enabled-only writes from old instances during the rollout window land - # after the 0075 backfill), then add and validate the constraint. + # Backstop for the dual-write in `save`: added NOT VALID + validated (0080–0082) + # only after the 0077 deploy fully rolled, because enforcing it in the same deploy + # that introduced the dual-write would break not-yet-replaced instances that still + # write `enabled` alone. + models.CheckConstraint( + name="scout_config_enabled_matches_status", + condition=models.Q(enabled=True, status__in=["active", "pending_pause"]) + | models.Q(enabled=False, status__in=["paused_by_system", "paused_by_user"]), + ), + models.CheckConstraint( + name="scout_config_pause_reason_matches_status", + condition=models.Q(status__in=["pending_pause", "paused_by_system"], pause_reason__isnull=False) + | models.Q(status__in=["active", "paused_by_user"], pause_reason__isnull=True), + ), ] def save(self, *args: Any, **kwargs: Any) -> None: @@ -1329,7 +1344,18 @@ def transition_status_by_system( if new_status == self.Status.PAUSED_BY_USER: raise ValueError("Only a user write may set paused_by_user.") with transaction.atomic(): - locked = type(self).all_teams.select_for_update().get(pk=self.pk) + # One ordered query locks the whole team's rows, not just ours: the cap check below + # counts sibling rows, so two concurrent resumes locking only their own rows would + # each read the other as still paused and both slip under the cap. A single ordered + # lock set also keeps concurrent transitions on one team deadlock-free. Team config + # counts are small (capped) and transitions rare, so the wider lock is cheap. + team_rows = { + row.pk: row + for row in type(self).all_teams.select_for_update().filter(team_id=self.team_id).order_by("pk") + } + locked = team_rows.get(self.pk) + if locked is None: + return False if locked.status == self.Status.PAUSED_BY_USER: return False if locked.status != self.Status.ACTIVE and locked.pause_reason != pause_reason: @@ -1351,7 +1377,7 @@ def transition_status_by_system( ) if new_status in self.RUNNABLE_STATUSES and locked.status not in self.RUNNABLE_STATUSES: - peers = type(self).all_teams.filter(team_id=locked.team_id, enabled=True).exclude(pk=locked.pk).count() + peers = sum(1 for row in team_rows.values() if row.enabled and row.pk != locked.pk) if peers >= MAX_ENABLED_SCOUTS_PER_TEAM: return False recorded_reason = None if new_status == self.Status.ACTIVE else pause_reason @@ -1362,17 +1388,28 @@ def transition_status_by_system( locked.status_changed_at = timezone.now() locked.status_changed_by = None locked.enabled = new_status in self.RUNNABLE_STATUSES - locked.save( - update_fields=[ - "status", - "pause_reason", - "status_changed_at", - "status_changed_by", - "enabled", - "updated_at", - ] - ) - for field in ("status", "pause_reason", "status_changed_at", "status_changed_by", "enabled"): + update_fields = [ + "status", + "pause_reason", + "status_changed_at", + "status_changed_by", + "enabled", + "updated_at", + ] + if new_status == self.Status.ACTIVE: + # A resume always starts with a clean failure streak — otherwise the very next + # failed run would re-trip the breaker off stale evidence. + locked.consecutive_failure_count = 0 + update_fields.append("consecutive_failure_count") + locked.save(update_fields=update_fields) + for field in ( + "status", + "pause_reason", + "status_changed_at", + "status_changed_by", + "enabled", + "consecutive_failure_count", + ): setattr(self, field, getattr(locked, field)) return True diff --git a/products/signals/backend/scout_harness/AGENTS.md b/products/signals/backend/scout_harness/AGENTS.md index bc8e6e46d1d2..4a788cc7e292 100644 --- a/products/signals/backend/scout_harness/AGENTS.md +++ b/products/signals/backend/scout_harness/AGENTS.md @@ -38,6 +38,7 @@ In production it is driven by `SignalsScoutCoordinatorWorkflow` (periodic tick e See the module docstring at `profile/builders.py` for the authoritative section list — when adding or renaming a section, bump `INVENTORY_SOURCE_VERSION` so the cache invalidates cleanly. - `limits.py` Runtime ceilings as module constants: `DEFAULT_MAX_RUNTIME_S` (per-run budget), `ACTIVITY_SLACK_S`, and `WORKFLOW_HARD_CEILING_S` (`= DEFAULT_MAX_RUNTIME_S + ACTIVITY_SLACK_S`, the activity-level ceiling that gates the workflow's `start_to_close_timeout`). + Also the failure-streak circuit breaker's two knobs: `FAILURE_STREAK_PAUSE_THRESHOLD` (consecutive failed runs before the breaker pauses the lane) and `AUTO_PAUSE_PROBE_INTERVAL_S` (how long a paused lane holds before the coordinator dispatches one probe). - `team_limits.py` Single source of truth for a team's effective scout caps + metadata, resolved from the `signals-scout` flag payload in one read. The same three-layer cap resolution (`team_configs[team]` → `default_team_config` → code constant) the coordinator enforces at dispatch, plus enrollment (`_parse_enrollment` → `guaranteed_team_ids` / `skip_team_ids`, with a cloud/dev-gated fallback) and the editorial banner string. `guaranteed_team_ids` may contain the `"*"` wildcard (`ENROLL_ALL_TOKEN`): with it, enrollment inverts from an explicit allowlist to "every team that has enabled scout configs" — the self-serve gate, where the product-autonomy- gated UI creates the configs; explicit ids alongside `"*"` are still force-provisioned and `skip_team_ids` still hard-excludes. The global per-tick ceiling is flag-tunable too (`_resolve_global_max_runs_per_tick` ← `max_runs_per_tick_global`, default `MAX_RUNS_PER_TICK`). Kept free of the temporalio stack so it stays cheap to import on the API path; both the coordinator and the metadata endpoint import from here so the reported caps never drift from what dispatch allows. `resolve_team_metadata()` backs the metadata viewset; @@ -64,6 +65,11 @@ In production it is driven by `SignalsScoutCoordinatorWorkflow` (periodic tick e - The harness inserts the bridge row at the start of a run (inside `_spawn_and_run`). `SignalScoutRun` is now a thin bridge to a Tasks `TaskRun` — run status / timing / error live on `task_run`, not on the bridge row. Single-flight is a best-effort app-layer guard: `_has_running_run` skips dispatch when a prior run for the same `(team, skill_name)` has `task_run.status = IN_PROGRESS`. The old `WHERE status='running'` partial unique index was dropped in the bridge simplification; `_self_heal_stale_runs` now reaps the orphan case at the app layer (failing any `QUEUED`/`IN_PROGRESS` run older than `STALE_RUN_CUTOFF_S` before the guard), so a worker crash no longer wedges a lane permanently. A `task_run.status`-based DB constraint is still a possible follow-up for stronger single-flight guarantees. +- A `(team, skill)` lane that never succeeds is its own failure mode, distinct from the orphan case above: every dispatch is well-formed, takes a sandbox lease for the full runtime cap, produces nothing, and books a `failed` run — indefinitely, since the lane stays due every tick and nothing reconciles "this has never worked". + The runner maintains a failure-streak circuit breaker for it: `consecutive_failure_count` on the config row is bumped on a failed run and zeroed on a successful one, and at `FAILURE_STREAK_PAUSE_THRESHOLD` the runner pauses the lane through `SignalScoutConfig.transition_status_by_system` (`status=paused_by_system`, `pause_reason=repeated_failures`, which also syncs `enabled=False` like any pause). + The breaker is half-open on purpose — a pause is not a tombstone, so the coordinator dispatches one probe per `AUTO_PAUSE_PROBE_INTERVAL_S` to lanes paused with `repeated_failures` (`_collect_probe_runs`, cooldown anchored on `last_run_at` so a failed probe restarts it with no extra bookkeeping); a probe that succeeds resumes the lane through the same helper and normal cadence returns with nobody intervening. + The helper's reason-scoped ownership rule keeps it honest: the breaker may only resume its own `repeated_failures` pauses and can never touch `paused_by_user`, so an operator switching a scout off always wins. + The streak is read-only on the config serializers (`scout-configs-list`); any config edit resets it, a resume (human `enabled=true` or a successful probe) always starts it clean, and the trip emits `signals_scout_config_auto_paused` once. - The sandbox is opened with the team's MCP token plus the harness-internal tools. The skill body is loaded into the system prompt; each scout has its own `SignalScoutConfig` row (keyed on `(team, skill_name)`) whose `enabled` flag, `run_interval_minutes`, and optional project-local cron `run_cron_schedule` the coordinator's per-scout due-check honors. - Scout sandbox GitHub credentials are **always read-only**: the runner requests `github_read_access` on every scout run, so provisioning mints an ephemeral downscoped installation token (`contents`/`metadata`/`pull_requests` read, team-level installs only, never persisted — `get_readonly_github_token` in the tasks product) instead of the write-capable token that task creation would otherwise attach, or injects nothing when the mint isn't possible. The token backs the preinstalled `gh` CLI via `GH_TOKEN`/`GITHUB_TOKEN`. @@ -84,6 +90,8 @@ In production it is driven by `SignalsScoutCoordinatorWorkflow` (periodic tick e - Each run emits scout-owned lifecycle analytics events (best-effort, keyed on the team): `signals_scout_run_started` (the run cleared the guards and a TaskRun exists), `signals_scout_run_finished` (terminal: `completed`/`failed`/`cancelled` + runtime + emit count), and `signals_scout_run_reaped` (a stranded orphan was reaped by `_self_heal_stale_runs`). They join on `run_id`/`task_run_id` and are the event-derived (no-warehouse-lag) basis for throughput, stall, and worker-death alerting — a `started` with no `finished` is a run that died before finalize; a reaped run emits no `finished`. When the `scouts-model-selection` gate (or a runtime pin) routes the run, `started` and `finished` also carry `model` / `runtime_adapter`, so run outcomes are sliceable by model without joining through `$ai_generation`; absence means the agent-server default served it. + A `finished` run that died at the per-turn poll wall also carries the turn-log diagnostics from `TurnPollTimeout.diagnostics()` (`poll_timeout_stage` + elapsed/stale/line counts), because every wall failure raises one error string and the fleet's timeout rate is otherwise a single undifferentiated bucket: `no_turn_output` (the agent never emitted a turn-relevant line — it never got going), `stalled_after_output` (worked, then went quiet past the salvage window), `active_at_budget` (still streaming when the budget expired — the budget is the constraint, not the agent). + `signals_scout_config_auto_paused` fires once when a `(team, skill)` lane's failure-streak breaker trips (see the run-lifecycle bullet above) — the alertable signal for a wedged lane, which is otherwise invisible in a stream of individually-unremarkable `failed` runs. The report channel adds `signals_scout_report_emitted` / `signals_scout_report_edited` (plus customer-facing `$scout_report_*` copies), stamped with derived classification properties (`report_kind` = `finding`/`self_improvement`, `is_self_improvement_report`) via `_report_classification_props` in `tools/report.py` — classified server-side off the prompt's mandated title prefix (`prompt.SELF_IMPROVEMENT_REPORT_TITLE_PREFIX`), so self-improvement reports are separable without downstream title heuristics. That helper is the single extension point for future derived telemetry dimensions on these events — add new flags there (both events pick them up), not as model columns. - Emit happens via the harness's `emit_signal_*` tools, which call `emit_signal()` with `source_product="signals_scout"` and `source_type="cross_source_issue"`. From there the signal flows through the same emitter → buffer → grouping v2 path as any other source. @@ -98,6 +106,7 @@ In production it is driven by `SignalsScoutCoordinatorWorkflow` (periodic tick e - **Coordinator** — `temporal/agentic/scout_coordinator.py` and `scout_scheduler.py`. Polls every `COORDINATOR_INTERVAL_MINUTES = 30`; dispatches each scout whose per-scout schedule (`run_interval_minutes`, default every 24 hours, or an optional project-local cron `run_cron_schedule` that takes precedence) is due, most-overdue first, hard cap `MAX_RUNS_PER_TICK = 50` per tick, `ScheduleOverlapPolicy.SKIP` to drop ticks rather than queue them. + A lane paused by the failure-streak breaker (`status=paused_by_system`, `pause_reason=repeated_failures`) is excluded by the `enabled=True` dispatch filter like any other pause; the coordinator additionally dispatches one probe per `AUTO_PAUSE_PROBE_INTERVAL_S` to such lanes (`_collect_probe_runs`). - **Models** — `SignalScoutConfig`, `SignalScoutRun`, `SignalScratchpad`, `SignalScoutNote`, `SignalProjectProfile` in `../models.py`. - **Source variant** — `SignalSourceConfig.SourceProduct.SIGNALS_SCOUT` paired with `SourceType.CROSS_SOURCE_ISSUE`. - **Scout fleet** — the `signals-scout-*` skills live at diff --git a/products/signals/backend/scout_harness/limits.py b/products/signals/backend/scout_harness/limits.py index 548e6f5cebbb..f32ccbb340c9 100644 --- a/products/signals/backend/scout_harness/limits.py +++ b/products/signals/backend/scout_harness/limits.py @@ -30,6 +30,26 @@ # ticks. STALE_RUN_CUTOFF_S = 2 * WORKFLOW_HARD_CEILING_S +# Consecutive failed runs after which a scout config trips its circuit breaker and is +# auto-paused (`SignalScoutConfig.auto_paused_at`). Nothing else in the harness notices a +# scout that has never once succeeded: every dispatch takes a fresh sandbox lease for the +# full runtime cap, produces nothing, and books a `failed` run — so a permanently broken +# (team, skill) lane costs a lease per interval forever. Five in a row is well past any +# plausible run of bad luck (a flaky sandbox spawn, one upstream provider error) while still +# tripping within hours on the tight cadences, not days. +FAILURE_STREAK_PAUSE_THRESHOLD = 5 + +# Cooldown a paused lane holds before the coordinator dispatches one probe — the half-open +# state. A pause is not a tombstone: whatever wedged the lane (a broken sandbox env, a skill +# the model can't work through) usually gets fixed without anyone thinking to un-pause a +# scout, so the breaker has to re-test itself. A successful probe resumes the lane; a failed +# one restarts the cooldown through its own `last_run_at` stamp. Set to a day so a wedged +# lane costs one lease per day instead of one per interval. Deliberately independent of the +# scout's own schedule: for a scout on a slower-than-daily cadence the probe IS more frequent +# than its healthy schedule, which trades at most one lease per day for recovery within a day +# rather than up to a full (possibly 30-day) interval after the underlying cause is fixed. +AUTO_PAUSE_PROBE_INTERVAL_S = 24 * 60 * 60 + # Per-team ceiling on ENABLED scout configs — the per-team cost cap. Each enabled scout # is a recurring LLM sandbox run, so this bounds what one team can switch on. Set high so # teams can freely author scouts with minimal friction; it's a backstop against runaway diff --git a/products/signals/backend/scout_harness/runner.py b/products/signals/backend/scout_harness/runner.py index dbd8a3abdc86..50436b78d09e 100644 --- a/products/signals/backend/scout_harness/runner.py +++ b/products/signals/backend/scout_harness/runner.py @@ -7,6 +7,7 @@ from datetime import timedelta from typing import TYPE_CHECKING, Any +from django.db.models import F from django.utils import timezone import posthoganalytics @@ -20,7 +21,11 @@ from products.signals.backend.models import SignalScoutConfig, SignalScoutRun from products.signals.backend.scout_harness.derived_metadata import stamp_derived_metadata from products.signals.backend.scout_harness.lazy_seed import canonical_skill_names, sync_canonical_skills -from products.signals.backend.scout_harness.limits import DEFAULT_MAX_RUNTIME_S, STALE_RUN_CUTOFF_S +from products.signals.backend.scout_harness.limits import ( + DEFAULT_MAX_RUNTIME_S, + FAILURE_STREAK_PAUSE_THRESHOLD, + STALE_RUN_CUTOFF_S, +) from products.signals.backend.scout_harness.model_selection import resolve_scout_model from products.signals.backend.scout_harness.prompt import ( HARNESS_PROMPT_VERSION, @@ -41,7 +46,7 @@ resolve_acting_user_id_for_team, ) from products.tasks.backend.facade import api as tasks_facade -from products.tasks.backend.facade.agents import CustomPromptSandboxContext, MultiTurnSession +from products.tasks.backend.facade.agents import CustomPromptSandboxContext, MultiTurnSession, TurnPollTimeout if TYPE_CHECKING: from products.tasks.backend.models import TaskRun @@ -295,6 +300,10 @@ async def arun_signals_scout( emitted_count, _ = await database_sync_to_async(_read_run_metrics, thread_sensitive=False)( run_id, team.parent_team_id or team.id ) + # A run that got all the way through closes the breaker: the lane works, so any streak + # it had accumulated is stale and a standing auto-pause is lifted (this is also how the + # half-open probe recovers a paused lane once its underlying cause is fixed). + await database_sync_to_async(_clear_failure_streak, thread_sensitive=False)(config.pk) _capture_run_finished( team=team, config=config, @@ -346,6 +355,9 @@ async def arun_signals_scout( if row_persisted else (0, None) ) + # Advance the breaker before the event so the failure that trips it is the one whose + # `error_message` explains the pause. + streak = await database_sync_to_async(_record_failure_streak, thread_sensitive=False)(config.pk) _capture_run_finished( team=team, config=config, @@ -360,7 +372,17 @@ async def arun_signals_scout( runtime_adapter=runtime_adapter, error_type=type(exc).__name__, error_message=str(exc)[:300], + extra_properties=_poll_timeout_properties(exc), ) + if streak is not None and streak.tripped: + _capture_config_auto_paused( + team=team, + config=config, + skill_name=skill.name, + run_id=run_id, + failure_count=streak.count, + reason=str(exc)[:300], + ) return RunResult( run_id=str(run_id) if row_persisted else None, task_run_id=None, @@ -390,7 +412,10 @@ async def arun_signals_scout( }, ) # Synchronous, no DB read — the loop is collapsing, so don't await anything here; - # `emitted_count` is left unknown rather than risk a query during cancellation. + # `emitted_count` is left unknown rather than risk a query during cancellation. The + # failure-streak breaker is deliberately untouched too: a cancelled run says nothing + # about whether this lane can succeed, and counting worker shutdowns toward the streak + # would pause healthy scouts after a few deploys. _capture_run_finished( team=team, config=config, @@ -740,6 +765,95 @@ def _create_run_row( ) +@dataclass(frozen=True) +class _FailureStreak: + """Breaker state after a failed run. `tripped` is the *transition* into paused, not the + paused state itself — a re-failed probe leaves the status where it is without tripping + again, so the alerting event fires once per wedge rather than once per doomed run.""" + + count: int + tripped: bool + + +def _clear_failure_streak(config_id: Any) -> None: + """Zero the breaker after a successful run and lift its pause. Best-effort: the run + succeeded, so a bookkeeping failure here must not turn it into a failure. The streak reset + is filtered so the common case (a healthy lane) does no write at all; the resume goes + through the transition helper, whose reason scoping means only the breaker's own + `repeated_failures` pause can be lifted here — never a human's, never another writer's.""" + try: + SignalScoutConfig.all_teams.filter(pk=config_id).exclude(consecutive_failure_count=0).update( + consecutive_failure_count=0 + ) + config = SignalScoutConfig.all_teams.filter(pk=config_id).first() + if ( + config is not None + and config.status == SignalScoutConfig.Status.PAUSED_BY_SYSTEM + and config.pause_reason == SignalScoutConfig.PauseReason.REPEATED_FAILURES + ): + resumed = config.transition_status_by_system( + SignalScoutConfig.Status.ACTIVE, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + ) + if not resumed: + # Refused resumes are legitimate (team back at its enabled-scout cap, or the + # pause changed hands since the read) — the lane stays paused, worth a trace. + logger.info( + "signals_scout: probe succeeded but resume was refused", + extra={"scout_config_id": str(config_id)}, + ) + except Exception: + logger.exception("signals_scout: failed to clear failure streak", extra={"scout_config_id": str(config_id)}) + + +def _record_failure_streak(config_id: Any) -> _FailureStreak | None: + """Bump the failure streak and pause the lane at the threshold. Returns None when the row + is gone or the write failed — the caller only uses the result to decide whether to emit the + auto-paused event, and a failure here must never mask the run's own error. + + The bump is an atomic `F()` increment, not read-then-write: the runner's single-flight guard + means one run per (team, skill) at a time, but a config edit's streak reset can land + concurrently, and a stale absolute write would resurrect the streak the edit just cleared. + The pause goes through the transition helper: `tripped` is True only when the helper actually + moved the status, so a re-failed probe (already paused, transition is a no-op) re-arms the + cooldown via its own `last_run_at` stamp without firing the trip event again. The error text + rides on the events, not the row — the row records only the reason taxonomy + (`repeated_failures`). + """ + try: + updated = SignalScoutConfig.all_teams.filter(pk=config_id).update( + consecutive_failure_count=F("consecutive_failure_count") + 1 + ) + if not updated: + return None + config = SignalScoutConfig.all_teams.filter(pk=config_id).first() + if config is None: + return None + count = config.consecutive_failure_count + tripped = False + if count >= FAILURE_STREAK_PAUSE_THRESHOLD: + tripped = config.transition_status_by_system( + SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + ) + return _FailureStreak(count=count, tripped=tripped) + except Exception: + logger.exception("signals_scout: failed to record failure streak", extra={"scout_config_id": str(config_id)}) + return None + + +def _poll_timeout_properties(exc: BaseException) -> dict[str, Any] | None: + """Turn-log diagnostics for a run that died at the per-turn poll wall, or None for any other + failure. Every wall failure raises the same error string, which is why the fleet's timeout + rate reads as one cause; these properties split it into the populations that need different + fixes — an agent that never emitted a single turn-relevant line (never started), one that + worked and then went silent, and one still streaming when the budget ran out (the budget, + not the agent, is the constraint).""" + if not isinstance(exc, TurnPollTimeout): + return None + return exc.diagnostics() + + def _run_row_exists(run_id: Any, team_id: int) -> bool: return SignalScoutRun.objects.unscoped().filter(team_id=team_id, id=run_id).exists() @@ -845,6 +959,44 @@ def _capture_run_reaped( ) +def _capture_config_auto_paused( + *, + team: Team, + config: SignalScoutConfig, + skill_name: str, + run_id: Any, + failure_count: int, + reason: str, +) -> None: + """Emit a scout-owned event when a lane's failure-streak breaker trips. + + The state is also readable on the config row (and its API surface), but a wedge needs to be + *noticed*, not looked up: a lane that has never once succeeded is otherwise indistinguishable + from healthy traffic in the `signals_scout_run_finished` stream, which is why one tenant could + fail every run for days with an empty inbox and nobody see it. Fires only on the transition, so + a count here is a count of wedges. Best-effort: a capture failure must never affect the run. + """ + try: + posthoganalytics.capture( + event="signals_scout_config_auto_paused", + distinct_id=str(team.uuid), + properties={ + "skill_name": skill_name, + "scout_config_id": str(config.id), + "run_id": str(run_id), + "consecutive_failure_count": failure_count, + "failure_streak_threshold": FAILURE_STREAK_PAUSE_THRESHOLD, + "auto_pause_reason": reason, + }, + groups=groups(team.organization, team), + ) + except Exception: + logger.warning( + "signals_scout: failed to capture config auto-paused analytics event", + extra={"team_id": team.id, "skill_name": skill_name}, + ) + + def _attach_run_shape_props( properties: dict[str, Any], *, @@ -888,6 +1040,7 @@ def _capture_run_finished( runtime_adapter: str | None = None, error_type: str | None = None, error_message: str | None = None, + extra_properties: dict[str, Any] | None = None, ) -> None: """Emit the scout-owned per-run analytics event. @@ -902,6 +1055,9 @@ def _capture_run_finished( are attached so the failure rate is breakable down by cause without digging into worker logs — the bulk of scout failures fail in this layer before the `process-task` workflow's own `task_run_failed` event ever fires, so this is the only event that carries their reason. + `extra_properties` carries cause-specific detail the error string can't (today: the turn-log + diagnostics behind a per-turn poll timeout, which is a single string covering several + distinct failures). """ properties: dict[str, Any] = { "skill_name": skill.name, @@ -921,6 +1077,8 @@ def _capture_run_finished( if error_type is not None: properties["error_type"] = error_type properties["error_message"] = error_message + if extra_properties: + properties.update(extra_properties) try: posthoganalytics.capture( event="signals_scout_run_finished", diff --git a/products/signals/backend/scout_harness/serializers.py b/products/signals/backend/scout_harness/serializers.py index f68b45b1d051..f114e236c7c6 100644 --- a/products/signals/backend/scout_harness/serializers.py +++ b/products/signals/backend/scout_harness/serializers.py @@ -1949,6 +1949,15 @@ class SignalScoutConfigSerializer(serializers.ModelSerializer): allow_null=True, help_text="When the coordinator last dispatched this scout. Null if it has never run.", ) + consecutive_failure_count = serializers.IntegerField( + read_only=True, + help_text=( + "How many of this scout's runs have failed in a row. Back to 0 after a successful " + "run or any config edit. At the failure limit the scout pauses itself (`status` " + "becomes `paused_by_system` with `pause_reason` `repeated_failures`) and retries " + "about once a day; a successful retry resumes it, and so does setting `enabled=true`." + ), + ) @extend_schema_field(OpenApiTypes.STR) def get_description(self, obj: SignalScoutConfig) -> str: @@ -1979,6 +1988,7 @@ class Meta: "run_cron_schedule", "output_destinations", "last_run_at", + "consecutive_failure_count", "created_at", ] read_only_fields = ["id", "created_at"] @@ -2070,6 +2080,14 @@ def update(self, instance: SignalScoutConfig, validated_data: dict) -> SignalSco # never resume. Any other non-empty edit still clears a pending pause, since a human # tending the config is exactly the signal the warning exists to detect; an empty # PATCH is not an edit and must not count as human contact. + # A human tending the config resets the breaker's evidence — the failure streak is stale + # the moment someone acts on the lane. Lives here rather than in the viewsets so every + # human write path (PATCH, and both POST upserts through `_upsert_scout_config`) gets it; + # an empty write is not an edit and must not count. The pause itself (if the breaker + # tripped) is a status and lifts only through `enabled=true` below or a successful probe, + # both of which re-check the enabled-scout cap — an unrelated edit must not sidestep that. + if validated_data and instance.consecutive_failure_count: + validated_data["consecutive_failure_count"] = 0 if "enabled" in validated_data and validated_data["enabled"] != instance.enabled: target = ( SignalScoutConfig.Status.ACTIVE @@ -2086,6 +2104,10 @@ def update(self, instance: SignalScoutConfig, validated_data: dict) -> SignalSco validated_data["pause_reason"] = None validated_data["status_changed_at"] = timezone.now() validated_data["status_changed_by"] = getattr(request, "user", None) + if target == SignalScoutConfig.Status.ACTIVE: + # Same rule as `transition_status_by_system`: a resume starts with a clean + # failure streak, or the next failed run re-trips the breaker off stale evidence. + validated_data["consecutive_failure_count"] = 0 return super().update(instance, validated_data) class Meta: diff --git a/products/signals/backend/scout_harness/views.py b/products/signals/backend/scout_harness/views.py index 2481c21f8d3f..0bce5777b4c5 100644 --- a/products/signals/backend/scout_harness/views.py +++ b/products/signals/backend/scout_harness/views.py @@ -23,7 +23,7 @@ import dataclasses from dataclasses import dataclass from datetime import timedelta -from typing import cast +from typing import Any, cast from django.db import transaction from django.utils import timezone @@ -1915,7 +1915,7 @@ def partial_update(self, request: Request, *args, **kwargs) -> Response: if enabling: _reject_if_enabled_cap_reached(team_id, config.skill_name) # Fold `enabled_by` into the same save so enabling logs one activity entry, not two. - save_kwargs = {} + save_kwargs: dict[str, Any] = {} if enabling: save_kwargs["enabled_by"] = request.user instance = serializer.save(**save_kwargs) diff --git a/products/signals/backend/temporal/agentic/scout_coordinator.py b/products/signals/backend/temporal/agentic/scout_coordinator.py index f9280fa770e3..ad8e8710a633 100644 --- a/products/signals/backend/temporal/agentic/scout_coordinator.py +++ b/products/signals/backend/temporal/agentic/scout_coordinator.py @@ -21,6 +21,7 @@ from products.signals.backend.models import SignalScoutConfig from products.signals.backend.scout_harness.config_registry import live_scout_skill_names, register_missing_configs from products.signals.backend.scout_harness.lazy_seed import sync_canonical_skills +from products.signals.backend.scout_harness.limits import AUTO_PAUSE_PROBE_INTERVAL_S # Per-team cap resolution + the flag-payload read live in the temporalio-free `team_limits` module # so the HTTP metadata surface can share them. Imported by name so the planning code below calls @@ -181,6 +182,7 @@ def _collect_planned_runs( team_configs = _canonicalize_team_config_keys(team_configs or {}) default_team_config = default_team_config or {} due: list[_DueRun] = [] + paused_by_team = _breaker_paused_configs_by_team() for team, needs_seed in _participating_teams(enrollment): # Scouts held back from this team via the `withheld_skills` denylist (resolved most- # specific-first from this team's `team_configs` entry, then the fleet `default_team_config`): @@ -229,6 +231,7 @@ def _collect_planned_runs( if overdue_s is None: continue due.append(_DueRun(overdue_s, str(config.pk), team.id, config.skill_name)) + due.extend(_collect_probe_runs(paused_by_team.get(team.id, []), live_skills, now)) if not due: return [] @@ -378,8 +381,19 @@ def _participating_teams(enrollment: Enrollment) -> list[tuple[Team, bool]]: wildcard_ids: set[int] = set() if enrollment.wildcard: # Config rows persist under the canonical parent team, so these ids are already canonical. + # Breaker-paused configs keep the team enrolled even when nothing else is enabled: a + # wildcard team whose only scout tripped the breaker would otherwise drop out of + # participation entirely, and the recovery probe it was promised could never dispatch. wildcard_ids = set( - SignalScoutConfig.all_teams.filter(enabled=True).values_list("team_id", flat=True).distinct() + SignalScoutConfig.all_teams.filter( + Q(enabled=True) + | Q( + status=SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + ) + ) + .values_list("team_id", flat=True) + .distinct() ) wildcard_ids -= skip_canonical wildcard_ids -= explicit # explicit wins the tag — it gets the seed pass below @@ -391,6 +405,63 @@ def _participating_teams(enrollment: Enrollment) -> list[tuple[Team, bool]]: return [(teams[team_id], team_id in explicit) for team_id in sorted(all_ids) if team_id in teams] +def _breaker_paused_configs_by_team() -> dict[int, list[SignalScoutConfig]]: + """One cross-team fetch of every lane the failure-streak breaker has paused, keyed by team. + + Deliberately a single fleet-wide query hoisted out of `_collect_planned_runs`'s per-team + loop: breaker trips are rare, so a per-team lookup would be one mostly-empty round-trip per + participating team per tick. The filter is reason-scoped — only the breaker's own pauses — + so a human's pause or another writer's is never fetched, let alone probed. Genuinely + cross-team, which is what `all_teams` is for; per-team scoping happens at the call site. + """ + paused_by_team: dict[int, list[SignalScoutConfig]] = {} + paused = SignalScoutConfig.all_teams.filter( + status=SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + ) + for config in paused: + paused_by_team.setdefault(config.team_id, []).append(config) + return paused_by_team + + +def _collect_probe_runs(paused_configs: list[SignalScoutConfig], live_skills: set[str], now: datetime) -> list[_DueRun]: + """The half-open side of the failure-streak breaker: one probe per cooldown for paused lanes. + + A `(team, skill)` lane that fails every run still looks due every tick, and each dispatch + takes a sandbox lease for the full runtime cap to produce nothing — so an unrecoverable lane + costs a lease per interval indefinitely, with the only trace in the failure event stream. + Once the runner trips the breaker (`runner._record_failure_streak` → + `transition_status_by_system`), the pause syncs `enabled=False`, so the main dispatch query + stops seeing the lane. This is the reason-scoped exception that keeps the breaker half-open: + lanes paused with `repeated_failures` — never a human's pause, never another writer's, + guaranteed by `_breaker_paused_configs_by_team`'s filter — get one probe per + `AUTO_PAUSE_PROBE_INTERVAL_S`. A probe that succeeds resumes the lane + (`runner._clear_failure_streak`); a failed probe restarts the cooldown through its own + `last_run_at` stamp, with no extra bookkeeping. + """ + probes: list[_DueRun] = [] + for config in paused_configs: + if config.skill_name not in live_skills: + continue + # `last_run_at` is stamped on every dispatch, including the failed run that tripped the + # breaker, so the first probe lands one full cooldown after the trip. A null (possible + # only through manual row surgery) probes immediately rather than never. + cooldown_elapsed_s = ( + (now - config.last_run_at).total_seconds() if config.last_run_at else float(AUTO_PAUSE_PROBE_INTERVAL_S) + ) + overdue_s = cooldown_elapsed_s - AUTO_PAUSE_PROBE_INTERVAL_S + if overdue_s < 0: + continue + logger.info( + "signals_scout coordinator: dispatching probe for auto-paused scout", + team_id=config.team_id, + skill_name=config.skill_name, + consecutive_failure_count=config.consecutive_failure_count, + ) + probes.append(_DueRun(overdue_s, str(config.pk), config.team_id, config.skill_name)) + return probes + + def _overdue_seconds(config: SignalScoutConfig, now: datetime, project_timezone: tzinfo) -> float | None: """Seconds past due, or None if not yet due. Never-run rolling schedules are maximally overdue.""" if config.run_cron_schedule: diff --git a/products/signals/backend/test/test_scout_coordinator.py b/products/signals/backend/test/test_scout_coordinator.py index 80c54da3dd51..646ececf9caa 100644 --- a/products/signals/backend/test/test_scout_coordinator.py +++ b/products/signals/backend/test/test_scout_coordinator.py @@ -24,6 +24,7 @@ from products.signals.backend.models import SignalScoutConfig from products.signals.backend.scout_harness.config_registry import register_missing_configs from products.signals.backend.scout_harness.lazy_seed import HARNESS_SEEDED_BY, sync_canonical_skills +from products.signals.backend.scout_harness.limits import AUTO_PAUSE_PROBE_INTERVAL_S # The flag-payload read + per-team cap resolution live in `scout_harness/team_limits.py`; helpers # defined there are imported and patched there (see `_PAYLOAD_PATH` / `_IS_CLOUD_PATH`). @@ -51,6 +52,8 @@ SignalsScoutCoordinatorWorkflow, StampDispatchedRunsInput, _allocate_tick_budget, + _breaker_paused_configs_by_team, + _collect_probe_runs, _DueRun, _overdue_seconds, fetch_enabled_signals_scout_runs_activity, @@ -281,6 +284,30 @@ async def test_wildcard_dispatches_team_with_enabled_config(ateam): assert any(p.team_id == ateam.id and p.skill_name == "signals-scout-errors" for p in planned) +@pytest.mark.asyncio +@pytest.mark.django_db +@pytest.mark.flag_off +async def test_wildcard_keeps_a_fully_breaker_paused_team_enrolled_for_probes(ateam): + # A wildcard team whose ONLY scout the breaker paused has no enabled config left, so an + # enabled-only wildcard scan would drop the team from participation before probe collection + # ever ran — the promised recovery probe could never dispatch and the lane would stay paused + # until a human noticed. + await database_sync_to_async(_create_skill)(ateam, "signals-scout-errors") + await database_sync_to_async(_create_config)( + ateam, + "signals-scout-errors", + status=SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + consecutive_failure_count=5, + last_run_at=timezone.now() - timedelta(seconds=AUTO_PAUSE_PROBE_INTERVAL_S + 60), + ) + + with patch(_PAYLOAD_PATH, return_value={"guaranteed_team_ids": ["*"]}): + planned = await _run_activity() + + assert any(p.team_id == ateam.id and p.skill_name == "signals-scout-errors" for p in planned) + + @pytest.mark.asyncio @pytest.mark.django_db @pytest.mark.flag_off @@ -475,6 +502,92 @@ async def test_config_whose_skill_is_gone_is_skipped(ateam): # ── Schedule: deterministic due-check, no sampling ────────────────────────────── +@pytest.mark.django_db +class TestFailureStreakProbeCollection: + # The half-open side of the breaker. The pause itself removes the lane from the normal + # `enabled=True` dispatch query, so if probe collection regresses — wrong reason scope, + # wrong cooldown arithmetic, a human pause probed — a wedged lane either never recovers + # or a deliberately-paused scout keeps burning sandbox leases. + @parameterized.expand( + [ + ( + "cooldown_elapsed_probes", + SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + SignalScoutConfig.PauseReason.REPEATED_FAILURES, + AUTO_PAUSE_PROBE_INTERVAL_S + 60, + True, + ), + ( + "inside_cooldown_holds", + SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + SignalScoutConfig.PauseReason.REPEATED_FAILURES, + AUTO_PAUSE_PROBE_INTERVAL_S - 60, + False, + ), + ( + "user_pause_is_never_probed", + SignalScoutConfig.Status.PAUSED_BY_USER, + None, + AUTO_PAUSE_PROBE_INTERVAL_S + 60, + False, + ), + ( + "another_writers_pause_is_never_probed", + SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + SignalScoutConfig.PauseReason.NO_OUTPUT, + AUTO_PAUSE_PROBE_INTERVAL_S + 60, + False, + ), + ] + ) + def test_only_the_breakers_own_cooled_down_pauses_are_probed( + self, + _name: str, + status: SignalScoutConfig.Status, + pause_reason: SignalScoutConfig.PauseReason | None, + last_run_seconds_ago: int, + expect_probe: bool, + ) -> None: + now = timezone.now() + team = Team.objects.create(organization=Organization.objects.create(name="probe-org"), name="probe-team") + with team_scope(team.id, canonical=True): + _create_config( + team, + "signals-scout-general", + status=status, + pause_reason=pause_reason, + consecutive_failure_count=5, + last_run_at=now - timedelta(seconds=last_run_seconds_ago), + ) + + paused_by_team = _breaker_paused_configs_by_team() + probes = _collect_probe_runs(paused_by_team.get(team.id, []), {"signals-scout-general"}, now) + + assert [p.skill_name for p in probes] == (["signals-scout-general"] if expect_probe else []) + + +@pytest.mark.asyncio +@pytest.mark.django_db +async def test_planning_dispatches_a_probe_for_a_breaker_paused_lane(ateam): + # Wiring guard for the whole half-open path: a lane the breaker paused is invisible to the + # `enabled=True` dispatch query, so only `_collect_probe_runs` inside `_collect_planned_runs` + # can bring it back — if that call is dropped, a wedged lane whose cause was fixed stays + # paused forever with no human in the loop. + await database_sync_to_async(_create_skill)(ateam, "signals-scout-broken") + await database_sync_to_async(_create_config)( + ateam, + "signals-scout-broken", + status=SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + consecutive_failure_count=5, + last_run_at=timezone.now() - timedelta(seconds=AUTO_PAUSE_PROBE_INTERVAL_S + 60), + ) + + planned = await _run_activity() + + assert [p.skill_name for p in planned] == ["signals-scout-broken"] + + class TestCronScheduleDueCheck: @parameterized.expand( [ diff --git a/products/signals/backend/test/test_scout_harness.py b/products/signals/backend/test/test_scout_harness.py index fdeb1466616b..8aa7dafdc1b5 100644 --- a/products/signals/backend/test/test_scout_harness.py +++ b/products/signals/backend/test/test_scout_harness.py @@ -4,6 +4,7 @@ import json import random import asyncio +from contextlib import contextmanager from dataclasses import replace from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING @@ -31,7 +32,7 @@ from products.signals.backend.report_charts import ReportChart from products.signals.backend.scout_harness.derived_metadata import DERIVED_METADATA_KEY from products.signals.backend.scout_harness.lazy_seed import HARNESS_SEEDED_BY, _compute_row_hash -from products.signals.backend.scout_harness.limits import STALE_RUN_CUTOFF_S +from products.signals.backend.scout_harness.limits import FAILURE_STREAK_PAUSE_THRESHOLD, STALE_RUN_CUTOFF_S from products.signals.backend.scout_harness.model_selection import ScoutModel from products.signals.backend.scout_harness.prompt import _REPORT_CHARTS, HARNESS_PROMPT_VERSION, build_run_prompt from products.signals.backend.scout_harness.runner import RunResult, _ai_stage, _create_run_row, arun_signals_scout @@ -1134,6 +1135,90 @@ async def test_failed_run_captures_run_finished_event(ateam, aerrors_skill): assert props["error_message"] == "sandbox refused to start" +@contextmanager +def _stubbed_spawn_dependencies(): + with ( + patch( + "products.signals.backend.scout_harness.runner.get_or_create_signals_sandbox_env", + return_value="env-id", + ), + patch( + "products.signals.backend.scout_harness.runner.resolve_acting_user_id_for_team", + return_value=42, + ), + ): + yield + + +@pytest.mark.asyncio +@pytest.mark.django_db +async def test_failure_streak_pauses_scout_once_and_a_success_resumes_it(ateam, aerrors_skill): + TaskRun = apps.get_model("tasks", "TaskRun") + # The wedge this exists for: a (team, skill) lane that can never succeed stays due every + # tick, so it re-dispatches forever and takes a full-length sandbox lease each time to + # produce nothing. Nothing else in the harness notices, so the breaker has to. + session, result = await database_sync_to_async(_make_fake_session, thread_sensitive=False)(ateam, "close-out") + + async def _run_once(*, failing: bool, capture): + start = ( + AsyncMock(side_effect=RuntimeError("poll_for_turn: timed out after 900s")) + if failing + else _fake_start_invoking_hook(session, result) + ) + with ( + patch("products.signals.backend.scout_harness.runner.MultiTurnSession.start", new=start), + patch("products.signals.backend.scout_harness.runner.posthoganalytics.capture", new=capture), + # Canonical-skill sync is disk + DB work unrelated to the breaker, and this test + # calls the entrypoint once per run in the streak. + patch("products.signals.backend.scout_harness.runner.sync_canonical_skills"), + _stubbed_spawn_dependencies(), + ): + return await arun_signals_scout(team_id=ateam.id, skill_name="signals-scout-errors") + + def _paused_events(capture) -> list: + return [c for c in capture.call_args_list if c.kwargs["event"] == "signals_scout_config_auto_paused"] + + async def _reload() -> SignalScoutConfig: + return await database_sync_to_async(SignalScoutConfig.objects.get)( + team=ateam, skill_name="signals-scout-errors" + ) + + capture = MagicMock() + for _ in range(FAILURE_STREAK_PAUSE_THRESHOLD - 1): + await _run_once(failing=True, capture=capture) + config = await _reload() + assert config.consecutive_failure_count == FAILURE_STREAK_PAUSE_THRESHOLD - 1 + assert config.status == SignalScoutConfig.Status.ACTIVE + assert _paused_events(capture) == [] + + await _run_once(failing=True, capture=capture) + config = await _reload() + assert config.consecutive_failure_count == FAILURE_STREAK_PAUSE_THRESHOLD + assert config.status == SignalScoutConfig.Status.PAUSED_BY_SYSTEM + assert config.pause_reason == SignalScoutConfig.PauseReason.REPEATED_FAILURES + assert config.enabled is False + trip = _paused_events(capture) + assert len(trip) == 1 + assert trip[0].kwargs["properties"]["consecutive_failure_count"] == FAILURE_STREAK_PAUSE_THRESHOLD + assert "timed out after 900s" in trip[0].kwargs["properties"]["auto_pause_reason"] + + # A failed probe leaves the lane paused but must not re-alert — otherwise the event stops + # being a count of wedges and becomes a count of doomed runs again. + await _run_once(failing=True, capture=capture) + config = await _reload() + assert config.status == SignalScoutConfig.Status.PAUSED_BY_SYSTEM + assert len(_paused_events(capture)) == 1 + + # A probe that gets through resumes the lane with a clean streak, so a lane whose cause + # was fixed recovers without anyone having to un-pause it by hand. + assert (await _run_once(failing=False, capture=capture)).status == TaskRun.Status.COMPLETED.value + config = await _reload() + assert config.consecutive_failure_count == 0 + assert config.status == SignalScoutConfig.Status.ACTIVE + assert config.pause_reason is None + assert config.enabled is True + + @pytest.mark.asyncio @pytest.mark.django_db async def test_run_skipped_when_no_acting_user(ateam, aerrors_skill): diff --git a/products/signals/backend/test/test_scout_harness_api.py b/products/signals/backend/test/test_scout_harness_api.py index 57b2dedb96cf..bf062201be36 100644 --- a/products/signals/backend/test/test_scout_harness_api.py +++ b/products/signals/backend/test/test_scout_harness_api.py @@ -1401,6 +1401,46 @@ def test_partial_update_without_enabled_clears_a_pending_pause(self) -> None: assert config.status == SignalScoutConfig.Status.ACTIVE assert config.pause_reason is None + def test_partial_update_resets_the_failure_streak_but_not_a_breaker_pause(self) -> None: + # An unrelated edit resets the breaker's evidence, but must not resume the pause — + # resuming through an edit would sidestep the enabled-scout cap that `enabled=true` + # and the probe's resume both re-check. + config = SignalScoutConfig.objects.create( + team=self.team, + skill_name="signals-scout-foo", + enabled=False, + status=SignalScoutConfig.Status.PAUSED_BY_SYSTEM, + pause_reason=SignalScoutConfig.PauseReason.REPEATED_FAILURES, + ) + SignalScoutConfig.objects.filter(pk=config.pk).update(consecutive_failure_count=5) + + response = self.client.patch(self._detail_url(str(config.id)), data={"run_interval_minutes": 60}, format="json") + + assert response.status_code == status.HTTP_200_OK + config.refresh_from_db() + assert config.consecutive_failure_count == 0 + assert config.status == SignalScoutConfig.Status.PAUSED_BY_SYSTEM + assert config.pause_reason == SignalScoutConfig.PauseReason.REPEATED_FAILURES + + def test_create_upsert_on_existing_config_resets_the_failure_streak(self) -> None: + # The POST path re-registers an existing config through `_upsert_scout_config`, which is + # a human edit like PATCH — the pre-trip streak must reset there too, or an edit through + # this path leaves the scout one failure from pausing despite the intervening human touch. + self._make_skill("signals-scout-foo") + config = SignalScoutConfig.objects.create(team=self.team, skill_name="signals-scout-foo") + SignalScoutConfig.objects.filter(pk=config.pk).update(consecutive_failure_count=4) + + response = self.client.post( + self._list_url(), + data={"skill_name": "signals-scout-foo", "run_interval_minutes": 120}, + format="json", + ) + + assert response.status_code == status.HTTP_200_OK + config.refresh_from_db() + assert config.consecutive_failure_count == 0 + assert config.run_interval_minutes == 120 + def test_resending_enabled_false_does_not_escalate_a_system_pause(self) -> None: # Clients resend whole config objects; an unchanged `enabled=false` must not convert # a system pause into a user pause the system may never resume. diff --git a/products/signals/backend/test/test_scout_status.py b/products/signals/backend/test/test_scout_status.py index 8f3d55e04055..7bc379367ee3 100644 --- a/products/signals/backend/test/test_scout_status.py +++ b/products/signals/backend/test/test_scout_status.py @@ -158,6 +158,31 @@ def test_system_resume_refused_when_team_is_at_the_enabled_cap(self) -> None: paused.refresh_from_db() assert paused.enabled is False + def test_system_resume_starts_with_a_clean_failure_streak(self) -> None: + # Without the reset, the first failed run after a resume would re-trip the breaker + # off the stale pre-pause streak instead of five fresh failures. + config = self._config(status=Status.PAUSED_BY_SYSTEM, pause_reason=Reason.REPEATED_FAILURES) + SignalScoutConfig.objects.filter(pk=config.pk).update(consecutive_failure_count=5) + + applied = config.transition_status_by_system(Status.ACTIVE, pause_reason=Reason.REPEATED_FAILURES) + + assert applied is True + config.refresh_from_db() + assert config.consecutive_failure_count == 0 + + def test_human_re_enable_starts_with_a_clean_failure_streak(self) -> None: + config = self._config(status=Status.PAUSED_BY_SYSTEM, pause_reason=Reason.REPEATED_FAILURES) + SignalScoutConfig.objects.filter(pk=config.pk).update(consecutive_failure_count=5) + config.refresh_from_db() + + serializer = SignalScoutConfigUpdateSerializer(config, data={"enabled": True}, partial=True, context={}) + assert serializer.is_valid() + config = serializer.save() + + config.refresh_from_db() + assert config.status == Status.ACTIVE + assert config.consecutive_failure_count == 0 + def test_system_transition_clears_the_human_attribution_stamp(self) -> None: # Without the clear, a system pause would keep pointing at whichever human made the # previous transition and read as their action. diff --git a/products/signals/frontend/generated/api.schemas.ts b/products/signals/frontend/generated/api.schemas.ts index f3795c5e46fe..5281b0011bd7 100644 --- a/products/signals/frontend/generated/api.schemas.ts +++ b/products/signals/frontend/generated/api.schemas.ts @@ -1950,6 +1950,8 @@ export interface SignalScoutConfigApi { * @nullable */ readonly last_run_at: string | null + /** How many of this scout's runs have failed in a row. Back to 0 after a successful run or any config edit. At the failure limit the scout pauses itself (`status` becomes `paused_by_system` with `pause_reason` `repeated_failures`) and retries about once a day; a successful retry resumes it, and so does setting `enabled=true`. */ + readonly consecutive_failure_count: number readonly created_at: string } diff --git a/products/tasks/backend/facade/agents.py b/products/tasks/backend/facade/agents.py index c08c7de0dfeb..2fa03cf136e3 100644 --- a/products/tasks/backend/facade/agents.py +++ b/products/tasks/backend/facade/agents.py @@ -12,6 +12,7 @@ CustomPromptSandboxContext, EmptyAgentTurnError, OutputFn, + TurnPollTimeout, create_task_and_trigger, extract_json_from_text, poll_for_turn, @@ -27,6 +28,7 @@ "LocalSkillsCache", "MultiTurnSession", "OutputFn", + "TurnPollTimeout", "create_task_and_trigger", "extract_json_from_text", "poll_for_turn", diff --git a/products/tasks/backend/logic/services/custom_prompt_internals.py b/products/tasks/backend/logic/services/custom_prompt_internals.py index 11d9fd5f2198..8ad4f034aad0 100644 --- a/products/tasks/backend/logic/services/custom_prompt_internals.py +++ b/products/tasks/backend/logic/services/custom_prompt_internals.py @@ -72,6 +72,11 @@ # earlier finalization fingerprint and report a bogus success, so a failed progress line stays decisive. FAILED_PROGRESS_STATUS = "failed" +# The relay echoes the user's own prompt into the turn log as these `session/update` kinds. They are +# input, not agent output, so the turn-relevant growth counting discounts them like the transient +# side-channels above. +_PROMPT_ECHO_UPDATES = frozenset({"user_message", "user_message_chunk"}) + @dataclass(frozen=True) class AgentError: @@ -124,6 +129,63 @@ class CustomPromptSandboxContext: mint fails, the sandbox starts without a token.""" +class TurnPollTimeout(RuntimeError): + """The per-turn poll budget ran out with no `end_turn` and nothing salvageable. + + Carries what the turn log looked like at the wall, because the message alone cannot tell + apart failures that need opposite fixes: a turn that never produced a single turn-relevant + line (the sandbox agent never got going — waiting longer would change nothing), one that + worked and then went silent past the salvage window (a dropped stream mid-run), and one + still emitting when the budget expired (the budget is the binding constraint, not the + agent). Callers surface `diagnostics()` as analytics properties so the fleet's timeout rate + is splittable by cause instead of collapsing into one signature. + """ + + def __init__( + self, + message: str, + *, + stage: str, + elapsed: int, + stale_seconds: int, + total_lines: int, + turn_relevant_lines: int, + ): + super().__init__(message) + self.stage = stage + self.elapsed = elapsed + self.stale_seconds = stale_seconds + self.total_lines = total_lines + self.turn_relevant_lines = turn_relevant_lines + + def diagnostics(self) -> dict[str, Any]: + return { + "poll_timeout_stage": self.stage, + "poll_elapsed_seconds": self.elapsed, + "poll_stale_seconds": self.stale_seconds, + "turn_log_total_lines": self.total_lines, + "turn_log_relevant_lines": self.turn_relevant_lines, + } + + +# `TurnPollTimeout.stage` values. Low-cardinality on purpose: the numbers ride in the other +# diagnostic properties, so this stays usable as a breakdown key. +POLL_TIMEOUT_NO_TURN_OUTPUT = "no_turn_output" +POLL_TIMEOUT_STALLED_AFTER_OUTPUT = "stalled_after_output" +POLL_TIMEOUT_ACTIVE_AT_BUDGET = "active_at_budget" + + +def _classify_poll_timeout(*, turn_relevant_lines: int, stale_seconds: int) -> str: + if turn_relevant_lines == 0: + return POLL_TIMEOUT_NO_TURN_OUTPUT + # Past the salvage floor the stream is provably not live (the floor is one SSE read window), + # and salvage has already declined, so the turn died mid-flight. Below it, the turn was still + # producing turn-relevant output when the budget ran out. + if stale_seconds >= STALE_TURN_SALVAGE_SECONDS: + return POLL_TIMEOUT_STALLED_AFTER_OUTPUT + return POLL_TIMEOUT_ACTIVE_AT_BUDGET + + class EmptyAgentTurnError(RuntimeError): """Raised when the agent emitted end_turn but no agent_message (Claude Agent SDK short-circuits).""" @@ -236,6 +298,10 @@ async def poll_for_turn( consecutive_storage_errors = 0 # Elapsed time when we last saw new log lines last_new_lines_at = 0 + # Running tally of turn-relevant lines this turn produced. Zero at the wall means the agent + # never got going at all, which `last_new_lines_at` alone can't distinguish from "went quiet + # on the very first poll" — see `_classify_poll_timeout`. + turn_relevant_lines = 0 # Remember assistant text, as the agent message and end_message could arrive in different poll slices latest_assistant_text: str | None = None # Cursor at start of this turn — passed to _drain_final_log so the terminal-status drain can @@ -279,7 +345,9 @@ async def poll_for_turn( # _ended_on_pending_finalization. if total_lines > skip_lines: new_lines = (full_log or "").strip().split("\n")[skip_lines:] - if (total_lines - skip_lines) - _transient_growth(new_lines) > 0: + relevant_growth = (total_lines - skip_lines) - _transient_growth(new_lines) + if relevant_growth > 0: + turn_relevant_lines += relevant_growth last_new_lines_at = elapsed stale_seconds = elapsed - last_new_lines_at # Warn once per minute of silence (not every poll) so stalls surface without flooding logs. @@ -382,7 +450,27 @@ async def poll_for_turn( ) if salvaged is not None: return salvaged - raise RuntimeError(f"custom_prompt - poll_for_turn: timed out after {elapsed}s") + stage = _classify_poll_timeout(turn_relevant_lines=turn_relevant_lines, stale_seconds=stale_seconds) + logger.warning( + "custom_prompt - poll_for_turn: timed out after %ds, run=%s, stage=%s, stale_for=%ds, " + "total_lines=%d, turn_relevant_lines=%d", + elapsed, + task_run.id, + stage, + stale_seconds, + skip_lines, + turn_relevant_lines, + ) + # `stage` is in the message as well as the diagnostics so the cause survives everywhere the + # error string is the only thing that gets persisted (TaskRun.error_message, Temporal). + raise TurnPollTimeout( + f"custom_prompt - poll_for_turn: timed out after {elapsed}s (stage={stage})", + stage=stage, + elapsed=elapsed, + stale_seconds=stale_seconds, + total_lines=skip_lines, + turn_relevant_lines=turn_relevant_lines, + ) async def _read_turn_log_with_retry( @@ -694,9 +782,12 @@ def _is_failed_progress(notification: dict) -> bool: def _transient_growth(lines: list[str]) -> int: - """Count how many of `lines` are transient relay side-channel notifications (network audits, - credential refreshes, sandbox stdout, informational progress). A failed progress line is not - transient — it must count as real activity so the growth check can't discount a failure away.""" + """Count how many of `lines` carry no agent turn-state: transient relay side-channels (network + audits, credential refreshes, sandbox stdout, informational progress) and echoed prompts. The + relay echoes the user's own message into the turn log, so without discounting it every turn + counts at least one "turn-relevant" line and the `no_turn_output` timeout classification — + the agent never got going at all — could never fire. A failed progress line is not transient — + it must count as real activity so the growth check can't discount a failure away.""" count = 0 for line in lines: line = line.strip() @@ -706,7 +797,15 @@ def _transient_growth(lines: list[str]) -> int: notification = json.loads(line).get("notification") except json.JSONDecodeError: continue - if not isinstance(notification, dict) or notification.get("method") not in TRANSIENT_SIDE_CHANNEL_METHODS: + if not isinstance(notification, dict): + continue + method = notification.get("method") + if method == "session/update": + update = (notification.get("params") or {}).get("update") + if isinstance(update, dict) and update.get("sessionUpdate") in _PROMPT_ECHO_UPDATES: + count += 1 + continue + if method not in TRANSIENT_SIDE_CHANNEL_METHODS: continue if _is_failed_progress(notification): continue diff --git a/products/tasks/backend/tests/test_multi_turn_session.py b/products/tasks/backend/tests/test_multi_turn_session.py index 9ec669632fd4..9b546071fd12 100644 --- a/products/tasks/backend/tests/test_multi_turn_session.py +++ b/products/tasks/backend/tests/test_multi_turn_session.py @@ -19,6 +19,7 @@ AgentError, CustomPromptSandboxContext, EmptyAgentTurnError, + TurnPollTimeout, _extract_agent_error, create_task_and_trigger, poll_for_turn, @@ -759,6 +760,71 @@ async def test_informational_progress_after_fingerprint_still_salvages(self, sta assert last_message == "close-out summary" +class TestPollForTurnTimeoutDiagnosis: + """A wall failure must say which failure it was. + + Every exhausted poll budget raises the same string, so a fleet-wide timeout rate reads as + one cause when it is really three that need opposite fixes: an agent that never produced a + turn-relevant line (never got going), one that worked and then went quiet, and one still + streaming when the budget ran out (the budget is the constraint, not the agent). + """ + + @parameterized.expand( + [ + ("empty_log", lambda: "", "no_turn_output"), + # A run whose only output is relay side-channel noise never started a turn either — + # counting those lines as output would misfile it as a mid-run stall. + ("side_channel_noise_only", lambda: _console_line("agentsh network events"), "no_turn_output"), + # The relay echoes the user's prompt into every turn log, so counting it as output + # would make no_turn_output unreachable — every never-started turn would misfile as a + # mid-run stall. + ("prompt_echo_only", lambda: _user_message_line("prompt"), "no_turn_output"), + ("worked_then_silent", lambda: _agent_message_line("still working"), "stalled_after_output"), + ] + ) + @pytest.mark.asyncio + async def test_static_log_shape_is_classified(self, _name: str, build_log, expected_stage: str): + fake = FakeTaskRun() + with ( + patch("posthog.storage.object_storage.read", return_value=build_log()), + patch("asyncio.sleep", new=AsyncMock()), + patch("products.tasks.backend.logic.services.custom_prompt_internals.POLL_INTERVAL_SECONDS", 10), + patch("products.tasks.backend.logic.services.custom_prompt_internals.MAX_POLL_SECONDS", 30), + patch("products.tasks.backend.logic.services.custom_prompt_internals.STALE_TURN_SALVAGE_SECONDS", 15), + patch("products.tasks.backend.models.TaskRun.objects.get", return_value=fake), + ): + with pytest.raises(TurnPollTimeout) as exc_info: + await poll_for_turn(fake, skip_lines=0) + + assert exc_info.value.stage == expected_stage + assert exc_info.value.diagnostics()["poll_timeout_stage"] == expected_stage + # The stage rides in the message too — TaskRun.error_message and Temporal only keep the string. + assert f"stage={expected_stage}" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_log_still_growing_at_the_wall_is_not_reported_as_a_stall(self): + # The population that would justify a bigger budget: turn-relevant output right up to the + # wall. Lumping it in with the dead turns is what makes "is 15 minutes too tight?" + # unanswerable from the failure data. + logs = ["\n".join(_agent_message_line(f"step {i}") for i in range(n + 1)) for n in range(4)] + poll_iter = iter(logs) + + fake = FakeTaskRun() + with ( + patch("posthog.storage.object_storage.read", side_effect=lambda *a, **k: next(poll_iter, logs[-1])), + patch("asyncio.sleep", new=AsyncMock()), + patch("products.tasks.backend.logic.services.custom_prompt_internals.POLL_INTERVAL_SECONDS", 10), + patch("products.tasks.backend.logic.services.custom_prompt_internals.MAX_POLL_SECONDS", 30), + patch("products.tasks.backend.logic.services.custom_prompt_internals.STALE_TURN_SALVAGE_SECONDS", 15), + patch("products.tasks.backend.models.TaskRun.objects.get", return_value=fake), + ): + with pytest.raises(TurnPollTimeout) as exc_info: + await poll_for_turn(fake, skip_lines=0) + + assert exc_info.value.stage == "active_at_budget" + assert exc_info.value.turn_relevant_lines == 3 + + class TestPollForTurnTerminalDrain: """Terminal-status drain must recover an agent_message from *this* turn only. diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 702eec10246d..e34df0b25b8e 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -63239,6 +63239,8 @@ export namespace Schemas { * @nullable */ readonly last_run_at: string | null; + /** How many of this scout's runs have failed in a row. Back to 0 after a successful run or any config edit. At the failure limit the scout pauses itself (`status` becomes `paused_by_system` with `pause_reason` `repeated_failures`) and retries about once a day; a successful retry resumes it, and so does setting `enabled=true`. */ + readonly consecutive_failure_count: number; readonly created_at: string; }