From 14069b7f2f3dbd4f46bb8473b881dcaf86fbde9b Mon Sep 17 00:00:00 2001 From: helloiamvu Date: Tue, 14 Jul 2026 10:26:08 +0200 Subject: [PATCH 1/4] infra(35-06): earnings-archive ingest-side bucket + poller job + daily scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add infra/earnings_archive.tf: the storage + scheduling for the continuous earnings-call audio archive, on the INGEST side of the Phase 28 audio firewall. Declares (mirroring the existing Phase 28 infra shapes): - google_storage_bucket.earnings_archive — PRIVATE archive in the ingest project (google_project.ingest), public_access_prevention = "enforced", uniform access, force_destroy = false, NO serving-project IAM binding (audio unreachable from serving). No lifecycle-delete: the 35-01 gate was signed `approve` with NO retention cap (D-27.9 amend, no conditions). - google_service_account.earnings_archive_fleet + _rw — least-priv harvest + transcription fleet SA with objectAdmin on the archive bucket ONLY (copied from batch.tf backfill_progress_rw). - google_cloud_run_v2_job.earnings_archive_poller — the 35-07 poller container (ingest project, image at the operator-seeded AR path via var.image_earnings_archive_poller; Phase 28 operator-seeded-image / image_tag=latest first-apply convention, documented in-file). - google_service_account.sched_earnings_archive + _invoker + google_cloud_scheduler_job.earnings_archive_daily — dedicated invoker SA with run.invoker on the poller Job only, daily cron :run'ing that job (mirrors scheduler.tf sched_incremental -> weather_incremental; both the :run URI and invoker binding reference the declared job, no dangling ref). Wires new variables (image_earnings_archive_poller, earnings_archive_bucket_name, earnings_archive_poll_schedule default "0 7 * * *") in variables.tf and outputs (earnings_archive_bucket, earnings_archive_fleet_sa_email, earnings_archive_poller_job) in outputs.tf. Verified: `tofu init -backend=false && tofu validate` -> "Success! The configuration is valid." (terraform binary absent in this env; OpenTofu v1.12.3 is the tool the existing infra root was authored with). tofu fmt clean on all three files. Requirements: ARCHIVE-INFRA-01 Co-Authored-By: Claude Fable 5 --- infra/earnings_archive.tf | 186 ++++++++++++++++++++++++++++++++++++++ infra/outputs.tf | 20 ++++ infra/variables.tf | 23 +++++ 3 files changed, 229 insertions(+) create mode 100644 infra/earnings_archive.tf diff --git a/infra/earnings_archive.tf b/infra/earnings_archive.tf new file mode 100644 index 00000000..dd798111 --- /dev/null +++ b/infra/earnings_archive.tf @@ -0,0 +1,186 @@ +# Phase 35 (35-06) — continuous earnings-call AUDIO ARCHIVE storage + scheduling. +# +# The LOCKED storage decision (35-CONTEXT) puts the growing internal earnings-call +# audio archive on the Phase 28 GCE platform, on the INGEST side of the audio +# firewall — audio must NEVER be reachable from the internet-facing serving +# project (projects.tf: ingest = "transient-audio island"; serving = "audio +# toolchain absent"). This file declares the four standing resources that make +# that real: +# +# 1. A PRIVATE GCS archive bucket in the ingest project (public access +# prevention enforced, uniform access, NO serving-project IAM binding). +# 2. A least-privilege harvest + transcription FLEET SA with objectAdmin on +# that bucket ONLY (mirrors batch.tf backfill_progress_rw least-priv shape). +# 3. The daily archive POLLER as a real Cloud Run Job (image at the +# operator-seeded Artifact Registry path, Phase 28 precedent) — declared +# FIRST so the scheduler + invoker binding reference a resource that exists. +# 4. A daily Cloud Scheduler job + its own dedicated invoker SA that :run's the +# poller Job (mirrors scheduler.tf sched_incremental → weather_incremental). +# +# D-27.9 amendment (35-01 GATE, signed `approve` 2026-07-14, NO conditions): audio +# may be RETAINED as an internal archive artifact. The operator attached NO +# retention lifecycle cap and NO source exclusions — so this bucket carries NO +# lifecycle-delete rule (a retained archive), unlike the transient +# earnings-audio-handoff bucket (cloud_run.tf, 1-day reaper) or the +# backfill-progress bucket (batch.tf, 90-day reaper). If the gate is ever +# re-signed `amend-with-conditions` with an age cap, add a lifecycle_rule here. + +locals { + earnings_archive_image = { + poller = "${local.ar_image_base}/${var.image_earnings_archive_poller}:${var.image_tag}" + } +} + +# ===================================================================== +# Ingest-side PRIVATE archive bucket (35-06 Task 1) — audio firewall +# ===================================================================== +# Retained earnings-call audio + our own SDK transcripts land here. Placed in the +# INGEST project (google_project.ingest) — never the serving project — and +# public_access_prevention = "enforced" + uniform access + NO serving-project IAM +# member is the structural firewall (T-35-07): the internet-facing serving +# surface has no path to a stored audio object. Co-located with the ingest audio +# toolchain (upper(var.weather_region) = US-CENTRAL1), matching the +# backfill_progress bucket's location convention. force_destroy = false: this is +# a retained archive, not a scratch bucket. +resource "google_storage_bucket" "earnings_archive" { + project = google_project.ingest.project_id + name = var.earnings_archive_bucket_name != "" ? var.earnings_archive_bucket_name : "mostlyright-earnings-archive-${google_project.ingest.number}" + location = upper(var.weather_region) + uniform_bucket_level_access = true + force_destroy = false + public_access_prevention = "enforced" + + # NO lifecycle_rule: the 35-01 gate was signed `approve` with NO retention cap + # (D-27.9 amendment, no conditions) — the archive is retained indefinitely. + # Contrast the transient earnings-audio-handoff bucket (cloud_run.tf) which + # reaps at age = 1 day. + + labels = { + phase = "35" + role = "earnings-archive" + } + + depends_on = [google_project_service.enabled] +} + +# ===================================================================== +# Least-privilege harvest + transcription FLEET SA (35-06 Task 1) +# ===================================================================== +# The single identity the harvest downloaders + the batch transcription fleet run +# as. It is bound to the archive bucket ONLY (objectAdmin, below) — nothing else. +# CRITICAL FIREWALL: no serving-project SA or principal is ever granted access to +# the archive bucket; audio must be unreachable from serving (T-35-07/T-35-08). +resource "google_service_account" "earnings_archive_fleet" { + project = google_project.ingest.project_id + account_id = "earnings-archive-fleet" + display_name = "Earnings archive harvest + transcription fleet runtime SA" + description = "Runs the Phase 35 harvest downloaders + batch transcription fleet + the daily archive poller. objectAdmin on the ingest-side earnings-archive bucket ONLY; no serving grant; audio never leaves ingest." + + depends_on = [google_project_service.enabled] +} + +# The fleet SA reads (dedup) + writes (upload) the archive bucket. objectAdmin on +# the ONE bucket only — least privilege, copied from batch.tf backfill_progress_rw. +resource "google_storage_bucket_iam_member" "earnings_archive_rw" { + bucket = google_storage_bucket.earnings_archive.name + role = "roles/storage.objectAdmin" + member = "serviceAccount:${google_service_account.earnings_archive_fleet.email}" +} + +# ===================================================================== +# Daily archive POLLER — Cloud Run Job, ingest project (35-06 Task 2) +# ===================================================================== +# The 35-07 poller.py container: checks the source feeds (Castify RSS / +# AlphaStreet channel) for new calls, dedups against what is already in the +# archive bucket, and enqueues new harvest work. Declared as a REAL Cloud Run Job +# so the scheduler's :run target + invoker binding below reference a resource +# that actually exists (no dangling reference — terraform validate proves it). +# Mirrors the SHAPE of google_cloud_run_v2_job.capture (cloud_run.tf): ingest +# project, var.serving_region, deletion_protection = false, single container. +# +# OPERATOR-SEEDED IMAGE (Phase 28 precedent): exactly like the existing +# capture/stt/rolefact/weather-incremental jobs, NO Dockerfile for this job is +# committed in the SDK monorepo. The poller Dockerfile lives in the SEPARATE +# archive repo (added in 35-07 Task 2); the OPERATOR builds + pushes it to the +# reused Artifact Registry path (local.earnings_archive_image.poller) BEFORE the +# first `terraform apply`. The var.image_tag default "latest" matches the Phase +# 28 first-apply convention; CI/operator pins an immutable digest/tag per deploy. +# Reuses the fleet SA (Task 1): the poller reads the archive bucket to dedup. +resource "google_cloud_run_v2_job" "earnings_archive_poller" { + project = google_project.ingest.project_id + name = "earnings-archive-poller" + location = var.serving_region + + # 35-full deploy: allow TF to replace a tainted/failed initial revision. + deletion_protection = false + + template { + template { + service_account = google_service_account.earnings_archive_fleet.email + max_retries = 1 + + containers { + image = local.earnings_archive_image.poller + + resources { + limits = { + cpu = "1" + memory = "4Gi" + } + } + + # The poller targets the Task-1 archive bucket (dedup + enqueue). + env { + name = "EARNINGS_ARCHIVE_BUCKET" + value = google_storage_bucket.earnings_archive.name + } + } + } + } + + depends_on = [google_project_service.enabled] +} + +# ===================================================================== +# Daily poller scheduler + dedicated invoker SA (35-06 Task 2) +# ===================================================================== +# Cloud Scheduler is the monorepo precedent (scheduler.tf; Claude's-discretion +# choice per 35-CONTEXT resolved to Cloud Scheduler). Mirrors the SHAPE of the +# weather_incremental_daily block: a dedicated least-privilege invoker SA that +# holds run.invoker on the ONE poller Job only (T-35-09), and a daily cron that +# POSTs the Cloud Run Admin :run URI for that Job with an oauth_token minted for +# the invoker SA. Both the :run URI and the invoker binding reference the +# declared google_cloud_run_v2_job.earnings_archive_poller — no bare-name string, +# no dangling reference. +resource "google_service_account" "sched_earnings_archive" { + project = google_project.ingest.project_id + account_id = "sched-earnings-archive" + display_name = "Earnings archive daily poller scheduler invoker SA" + + depends_on = [google_project_service.enabled] +} + +resource "google_cloud_run_v2_job_iam_member" "sched_earnings_archive_invoker" { + project = google_project.ingest.project_id + location = var.serving_region + name = google_cloud_run_v2_job.earnings_archive_poller.name + role = "roles/run.invoker" + member = "serviceAccount:${google_service_account.sched_earnings_archive.email}" +} + +resource "google_cloud_scheduler_job" "earnings_archive_daily" { + project = google_project.ingest.project_id + region = var.serving_region + name = "earnings-archive-daily" + schedule = var.earnings_archive_poll_schedule + time_zone = "Etc/UTC" + + http_target { + http_method = "POST" + uri = "https://${var.serving_region}-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/${google_project.ingest.project_id}/jobs/${google_cloud_run_v2_job.earnings_archive_poller.name}:run" + + oauth_token { + service_account_email = google_service_account.sched_earnings_archive.email + } + } +} diff --git a/infra/outputs.tf b/infra/outputs.tf index d69b304b..22b4a16a 100644 --- a/infra/outputs.tf +++ b/infra/outputs.tf @@ -103,3 +103,23 @@ output "r2_bucket" { description = "The single platform R2 bucket derived parquet is written to / served from. Set as R2_BUCKET for the ingest deploy workflows." value = var.r2_bucket } + +# --- Phase 35 earnings-call audio archive (35-06) --- +# The ingest-side private archive bucket, its least-priv fleet SA, and the daily +# poller Job. The archive repo's harvest + transcription tooling reads these +# (EARNINGS_ARCHIVE_BUCKET / the fleet SA); the operator seeds the poller image +# at the AR path before the first apply. +output "earnings_archive_bucket" { + description = "Ingest-side private earnings-call audio archive bucket (35-06). Audio firewall: public access prevention enforced, NO serving-project IAM. Set as EARNINGS_ARCHIVE_BUCKET for the archive repo's harvest + transcription fleet." + value = google_storage_bucket.earnings_archive.name +} + +output "earnings_archive_fleet_sa_email" { + description = "Least-privilege harvest + transcription fleet SA email (35-06) — objectAdmin on the earnings-archive bucket ONLY." + value = google_service_account.earnings_archive_fleet.email +} + +output "earnings_archive_poller_job" { + description = "Daily earnings-archive poller Cloud Run Job name (35-06), triggered by the earnings-archive-daily scheduler. Operator-seeded image (Phase 28 precedent)." + value = google_cloud_run_v2_job.earnings_archive_poller.name +} diff --git a/infra/variables.tf b/infra/variables.tf index b0b12a5f..a68385b8 100644 --- a/infra/variables.tf +++ b/infra/variables.tf @@ -374,6 +374,12 @@ variable "image_weather_incremental" { default = "weather-incremental" } +variable "image_earnings_archive_poller" { + description = "Earnings archive daily poller image name (35-06/35-07). OPERATOR-SEEDED: no Dockerfile is committed in the SDK monorepo; the poller image is built + pushed from the separate archive repo to the reused Artifact Registry before the first apply (Phase 28 operator-seeded-image convention)." + type = string + default = "earnings-archive-poller" +} + # --------------------------------------------------------------------------- # STT L4 GPU tuning (28-11). # --------------------------------------------------------------------------- @@ -397,3 +403,20 @@ variable "serving_global_rps_ceiling" { type = number default = 50 } + +# --------------------------------------------------------------------------- +# Phase 35 — earnings-call audio archive (35-06). Private ingest-side bucket + +# daily poller cadence. The archive lands on the INGEST side of the audio +# firewall; the serving project gets NO IAM on it (35-01 GATE / D-27.9 amend). +# --------------------------------------------------------------------------- +variable "earnings_archive_bucket_name" { + description = "Optional override for the ingest-side earnings-audio archive bucket name (35-06). Empty = mostlyright-earnings-archive-, matching the backfill_progress - naming convention. GCS bucket names are globally unique; set this if the default collides." + type = string + default = "" +} + +variable "earnings_archive_poll_schedule" { + description = "Cloud Scheduler cron for the daily earnings-archive source poller (35-06). Default 07:00 UTC daily — after the incremental weather trigger (06:00). time_zone is pinned to Etc/UTC in earnings_archive.tf." + type = string + default = "0 7 * * *" +} From 76599b815bb7d0cf676c5490bea50690ced33609 Mon Sep 17 00:00:00 2001 From: helloiamvu Date: Tue, 14 Jul 2026 10:37:37 +0200 Subject: [PATCH 2/4] feat(35-07): flip Phase 27 retention seam to archive-copy (flag-gated, default OFF) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D-27.9 amendment (Phase 35 gate 35-01): when EARNINGS_ARCHIVE_RETENTION is truthy and EARNINGS_ARCHIVE_BUCKET is set, the transient captured-audio handoff object is COPIED (server-side copy_blob) to the ingest-side archive bucket (earnings/audio/{ticker}/{call_id}.opus) and THEN the transient is deleted, instead of the Phase 27 delete-after-transcript default. Ordering invariant preserved: archive/delete happens ONLY after the transcript ledger write is durable. A ledger-write failure raises before the branch, so NEITHER archive nor delete runs (audio kept for retry). A copy failure raises so the transient handoff is NOT deleted (no silent drop of the retained artifact). Flag defaults OFF: the flag-unset AND flag-explicitly-off paths are byte-identical to Phase 27. earnings/ledger.py is unchanged (ledger stays audio-free). Tests (services/earnings/tests/test_stt_handoff_delete.py): - pre-existing test_handoff_deleted_after_ledger_write / _kept_when_ledger_write_fails stay green unchanged (flag unset) - new test_handoff_archived_when_retention_on (archive before delete, post-ledger, key carries ticker) - new test_handoff_deleted_when_retention_off (explicit flag-off regression pin) - new test_handoff_kept_when_ledger_write_fails_under_retention (flag ON + ledger fails -> neither archived nor deleted) Task 2 (archive-repo poller + Dockerfile.poller) is out of scope here — handled in the separate earnings-archive repo wave. Co-Authored-By: Claude Fable 5 --- services/earnings/jobs/stt.py | 94 +++++++++++++- .../earnings/tests/test_stt_handoff_delete.py | 119 ++++++++++++++++-- 2 files changed, 202 insertions(+), 11 deletions(-) diff --git a/services/earnings/jobs/stt.py b/services/earnings/jobs/stt.py index 3eabff3d..ff849609 100644 --- a/services/earnings/jobs/stt.py +++ b/services/earnings/jobs/stt.py @@ -37,6 +37,12 @@ * Live-publish (all three required to enable): ``EARNINGS_STREAMING_ENABLED``, ``EARNINGS_STREAMING_PROJECT``, and optionally ``EARNINGS_STREAMING_TOPIC`` (default ``earnings-streaming``). +* Audio retention (D-27.9 amendment, Phase 35 gate 35-01) — DEFAULT OFF: + ``EARNINGS_ARCHIVE_RETENTION`` (truthy to enable) + ``EARNINGS_ARCHIVE_BUCKET`` + (the ingest-side archive bucket). When BOTH are set, the transient handoff audio + is COPIED into the archive bucket (``earnings/audio/{ticker}/{call_id}.opus``) + AFTER the durable ledger write and BEFORE the transient handoff is deleted. Unset + (or falsy) keeps the Phase 27 delete-after-transcript behavior byte-identical. **Region.** This image runs on an L4 GPU in us-central1 (no Cloud Run GPU in eu-west3) — see the Dockerfile. @@ -181,6 +187,58 @@ def _delete_handoff_source(bucket: str, key: str) -> None: ) +#: Object-key layout for the retained archive copy on the INGEST-side archive +#: bucket, mirroring the partition-mirrors-path convention used across the ingest +#: sinks (`earnings/audio/{ticker}/{call_id}.opus`). The call_id is the ledger +#: partition key, so an archived audio object lines up 1:1 with its transcript. +_ARCHIVE_AUDIO_KEY_FMT = "earnings/audio/{ticker}/{call_id}.opus" + + +def _archive_handoff_source( + bucket: str, + key: str, + *, + archive_bucket: str, + ticker: str, + call_id: str, +) -> None: + """COPY the transient handoff object into the ingest-side archive bucket. + + The D-27.9 retention amendment (Phase 35 gate 35-01) flips the Phase 27 + "delete-after-transcript" seam: when retention is ON, the raw earnings audio is + RETAINED as an internal archive artifact instead of being discarded. This does a + SERVER-SIDE ``copy_blob`` (no local download) from the transient + ``AUDIO_HANDOFF_BUCKET`` object into ``EARNINGS_ARCHIVE_BUCKET`` under the + ``earnings/audio/{ticker}/{call_id}.opus`` layout, so the archived audio lines + up 1:1 with its transcript ledger partition. + + Called by :func:`transcribe_call` ONLY after ``TranscriptLedger.replace`` + succeeds (same ordering guarantee as the delete seam it replaces). On a copy + failure this RAISES — the caller must NOT then delete the transient handoff, so + the audio stays in place for a retry (a lost archive copy plus a deleted handoff + would silently drop the retained artifact). The archive bucket lives ONLY in the + ingest project and has no serving-project IAM (audio never reachable from + serving; 35-06 Terraform enforces the firewall). + """ + storage = _load_gcs_storage() + + dst_key = _ARCHIVE_AUDIO_KEY_FMT.format(ticker=ticker, call_id=call_id) + client = storage.Client() + src_bucket = client.bucket(bucket) + src_blob = src_bucket.blob(key) + dst_bucket = client.bucket(archive_bucket) + # Server-side copy (no local download) — the byte stays inside GCS on the + # ingest side. Raises on failure so the caller keeps the transient audio. + src_bucket.copy_blob(src_blob, dst_bucket, dst_key) + _LOG.info( + "stt: ARCHIVED handoff audio gs://%s/%s -> gs://%s/%s (D-27.9 retention, post-ledger)", + bucket, + key, + archive_bucket, + dst_key, + ) + + #: R2 key namespace for the durable transcript parquet the STT stage publishes so #: the SEPARATE role/fact Cloud Run Job (isolated ephemeral disk) can read it. The #: role/fact job downloads ``earnings/transcripts//.parquet`` back @@ -320,6 +378,8 @@ def transcribe_call( streaming_topic: str = "earnings-streaming", handoff_bucket: str | None = None, r2_bucket: str | None = None, + retention_enabled: bool = False, + archive_bucket: str | None = None, ) -> dict[str, object]: """Transcribe one call's transient audio → transcript ledger (+ optional publish). @@ -392,10 +452,29 @@ def transcribe_call( ) # The transcript is now DURABLY written (and any live publish done), so it is - # safe to delete the transient source handoff object. Deleting it BEFORE the - # ledger write would strand a failed call with no retryable audio (Codex R4). + # safe to archive/delete the transient source handoff object. Doing EITHER before + # the ledger write would strand a failed call with no retryable audio (Codex R4). + # This ordering guarantee is load-bearing and MUST NOT move above the ledger + # write: a ledger failure raises before this point, so NEITHER branch runs. if source is not None: - _delete_handoff_source(*source) + if retention_enabled and archive_bucket: + # D-27.9 retention ON (35-01 gate): RETAIN the audio to the ingest-side + # archive instead of discarding it. Copy-then-delete keeps the transient + # handoff bucket clean (matching the prior loud-delete intent) while the + # durable copy lives in the archive bucket. If the copy RAISES, the + # transient handoff is NOT deleted (audio kept for retry) — a lost copy + # plus a deleted handoff would silently drop the retained artifact. + _archive_handoff_source( + *source, + archive_bucket=archive_bucket, + ticker=ticker, + call_id=call_id, + ) + _delete_handoff_source(*source) + else: + # D-27.9 retention OFF (Phase 27 default): byte-identical to Phase 27 — + # delete the transient handoff object after the durable ledger write. + _delete_handoff_source(*source) return { "ticker": ticker, @@ -428,6 +507,13 @@ def main(argv: list[str] | None = None) -> int: enabled = optional_env("EARNINGS_STREAMING_ENABLED") publish_live = bool(enabled) and enabled.lower() not in ("0", "false", "no") + # D-27.9 retention flag (Phase 35 gate 35-01). DEFAULT OFF: an unset/falsy flag + # keeps the Phase 27 delete-after-ledger behavior byte-identical. When ON, the + # transient handoff audio is COPIED to EARNINGS_ARCHIVE_BUCKET (ingest side) + # before the transient is deleted. Read by NAME (never inlined). + retention_raw = optional_env("EARNINGS_ARCHIVE_RETENTION") + retention_enabled = bool(retention_raw) and retention_raw.lower() not in ("0", "false", "no") + transcribe_call( audio_path, ticker=ticker, @@ -442,6 +528,8 @@ def main(argv: list[str] | None = None) -> int: or "earnings-streaming", handoff_bucket=optional_env("AUDIO_HANDOFF_BUCKET"), r2_bucket=optional_env("R2_BUCKET"), + retention_enabled=retention_enabled, + archive_bucket=optional_env("EARNINGS_ARCHIVE_BUCKET"), ) return 0 diff --git a/services/earnings/tests/test_stt_handoff_delete.py b/services/earnings/tests/test_stt_handoff_delete.py index 9825dbb8..3d279e2d 100644 --- a/services/earnings/tests/test_stt_handoff_delete.py +++ b/services/earnings/tests/test_stt_handoff_delete.py @@ -1,10 +1,19 @@ -"""Phase 28 (28-11) — the STT handoff-object delete ordering (Codex R4 P1). +"""Phase 28 (28-11) + Phase 35 (35-07) — the STT handoff-object ordering (Codex R4 P1). -The transient source audio object in AUDIO_HANDOFF_BUCKET must be deleted ONLY -AFTER the transcript is durably written to the ledger — never before. Deleting it +The transient source audio object in AUDIO_HANDOFF_BUCKET must be archived/deleted +ONLY AFTER the transcript is durably written to the ledger — never before. Doing so at transcription time would strand a call whose ledger write then fails with no -retryable audio. These tests fake google-cloud-storage + the transcriber + the -ledger to assert the ordering and the skip-on-failure behavior. +retryable audio. + +Phase 28 seam: delete the transient handoff after the durable ledger write. +Phase 35 (35-07, D-27.9 amendment) seam: when the ``EARNINGS_ARCHIVE_RETENTION`` +flag is ON (and ``EARNINGS_ARCHIVE_BUCKET`` is set), COPY the transient handoff to +the ingest-side archive bucket (``earnings/audio/{ticker}/{call_id}.opus``) BEFORE +deleting it — still only after the durable ledger write. The flag defaults OFF, so +the Phase 27/28 behavior stays byte-identical. + +These tests fake google-cloud-storage + the transcriber + the ledger to assert the +ordering, the retention branch, and the skip-on-failure behavior. """ from __future__ import annotations @@ -23,8 +32,13 @@ def __init__(self) -> None: def _install_fakes(monkeypatch, events: list[str], *, ledger_raises: bool) -> dict: - """Wire fake storage + transcriber + ledger that record an ordered event log.""" - state: dict = {"deleted": False} + """Wire fake storage + transcriber + ledger that record an ordered event log. + + The fake ``copy_blob`` records an ``archive`` event and the destination key, so a + retention-on test can assert the archive happens (a) after the ledger write and + (b) before any delete, and that the destination key carries the ticker. + """ + state: dict = {"deleted": False, "archived": False, "archive_key": None} class _FakeBlob: def __init__(self, name: str) -> None: @@ -40,12 +54,24 @@ def delete(self) -> None: state["deleted"] = True class _FakeBucket: + def __init__(self, name: str) -> None: + self._name = name + def blob(self, name: str) -> _FakeBlob: return _FakeBlob(name) + def copy_blob(self, src_blob, dst_bucket, new_name): + # Server-side copy: record the archive event + destination identity so + # tests can assert ordering (archive before any delete) and the key. + events.append("archive") + state["archived"] = True + state["archive_key"] = new_name + state["archive_bucket"] = dst_bucket._name + return _FakeBlob(new_name) + class _FakeClient: def bucket(self, name: str) -> _FakeBucket: - return _FakeBucket() + return _FakeBucket(name) monkeypatch.setitem( sys.modules, "google.cloud.storage", types.SimpleNamespace(Client=lambda: _FakeClient()) @@ -104,3 +130,80 @@ def test_handoff_kept_when_ledger_write_fails(monkeypatch) -> None: # The ledger write failed, so the source audio is KEPT for a retry (NOT deleted). assert state["deleted"] is False assert "delete" not in events + + +def test_handoff_archived_when_retention_on(monkeypatch) -> None: + """D-27.9 retention ON: the audio is ARCHIVED (copied) then deleted, post-ledger.""" + from services.earnings.jobs.stt import transcribe_call + + events: list[str] = [] + state = _install_fakes(monkeypatch, events, ledger_raises=False) + + out = transcribe_call( + "gs://handoff-bkt/handoff/CHWY/evt-3.wav", + ticker="CHWY", + call_id="evt-3", + device="cpu", + retention_enabled=True, + archive_bucket="mostlyright-earnings-archive", + ) + assert out["segments"] == 1 + # The archive copy happened AFTER the durable ledger write and BEFORE the delete. + assert events == ["download", "transcribe", "ledger", "archive", "delete"] + assert state["archived"] is True + assert state["deleted"] is True + # The archive target key mirrors the partition layout and carries the ticker. + assert state["archive_key"] == "earnings/audio/CHWY/evt-3.opus" + assert "CHWY" in state["archive_key"] + assert state["archive_bucket"] == "mostlyright-earnings-archive" + + +def test_handoff_deleted_when_retention_off(monkeypatch) -> None: + """Explicit flag-OFF regression pin: delete fires, NO archive event. + + Distinct from ``test_handoff_deleted_after_ledger_write`` (the default-behavior + test): this one sets the retention flag EXPLICITLY falsy so a future default flip + cannot silently regress the flag-off path. + """ + from services.earnings.jobs.stt import transcribe_call + + events: list[str] = [] + state = _install_fakes(monkeypatch, events, ledger_raises=False) + + out = transcribe_call( + "gs://handoff-bkt/handoff/CHWY/evt-4.wav", + ticker="CHWY", + call_id="evt-4", + device="cpu", + retention_enabled=False, + archive_bucket="mostlyright-earnings-archive", + ) + assert out["segments"] == 1 + # Flag OFF: byte-identical to Phase 27 — delete, and NO archive copy. + assert events == ["download", "transcribe", "ledger", "delete"] + assert state["deleted"] is True + assert state["archived"] is False + assert "archive" not in events + + +def test_handoff_kept_when_ledger_write_fails_under_retention(monkeypatch) -> None: + """Retention ON + ledger write raises: NEITHER archived NOR deleted (kept).""" + from services.earnings.jobs.stt import transcribe_call + + events: list[str] = [] + state = _install_fakes(monkeypatch, events, ledger_raises=True) + + with pytest.raises(RuntimeError, match="ledger write failed"): + transcribe_call( + "gs://handoff-bkt/handoff/CHWY/evt-5.wav", + ticker="CHWY", + call_id="evt-5", + device="cpu", + retention_enabled=True, + archive_bucket="mostlyright-earnings-archive", + ) + # The ledger write failed before the archive/delete branch — the audio is KEPT. + assert state["archived"] is False + assert state["deleted"] is False + assert "archive" not in events + assert "delete" not in events From 2cd35a7b5dc7fe2e68f6534f76c5a3e0b1fe95c4 Mon Sep 17 00:00:00 2001 From: minereda <84080887+minereda@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:02:00 +0200 Subject: [PATCH 3/4] fix(35 review-iter-1): wire retention through stt_server + STT archive IAM + fail-loud half-config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round 1 (codex gpt-5.6 high + opus adversarial subagent) on PR #119: P1 (codex): the deployed HTTP path never passed the retention flags — stt_server.py called transcribe_call without retention_enabled/archive_bucket, so EARNINGS_ARCHIVE_RETENTION=1 on the Cloud Run STT service silently did nothing and live audio kept being deleted un-archived. Now derived from the service env with payload override, mirroring the streaming-flag shape. P1 (codex): the STT service runtime SA (earnings_stt) had no archive-bucket grant — every live retention copy would 403. Added google_storage_bucket_iam_member.earnings_archive_stt_rw (objectAdmin on the one bucket; earnings_stt is ingest-side, T-35-07 firewall intact). P1/P2 (codex+opus): retention_enabled with no archive bucket silently fell through to the delete branch — audio the operator believes is retained was discarded. transcribe_call now raises ValueError up front (before any download/GPU work). P2 (codex+opus): archive key hardcoded .opus but capture emits WAV and copy_blob never transcodes — the suffix is now preserved from the source handoff object. P2 (opus): no test pinned the param default against a flip to ON (masked by archive_bucket=None). Added test_default_is_retention_off_even_with_bucket_available + fail-loud + server env-derivation/default-off tests (4 new tests). P3 (opus): bucket location was coupled to var.weather_region; now var.stt_region (same us-central1 default, keeps the copy_blob same-region with the handoff bucket it copies from). Kept as-designed (both reviewers agree data-safe): archive-copy failure raises post-ledger so the handoff survives for retry — a flag-gated cost, never a loss. Targeted suites green: test_stt_handoff_delete + test_stt_server + test_jobs_entrypoints (52 tests), ruff clean. Co-Authored-By: Claude Fable 5 --- infra/earnings_archive.tf | 22 +++++- services/earnings/jobs/stt.py | 39 +++++++--- services/earnings/jobs/stt_server.py | 19 +++++ .../earnings/tests/test_stt_handoff_delete.py | 74 +++++++++++++++++-- services/earnings/tests/test_stt_server.py | 52 +++++++++++++ 5 files changed, 184 insertions(+), 22 deletions(-) diff --git a/infra/earnings_archive.tf b/infra/earnings_archive.tf index dd798111..3c3b1d50 100644 --- a/infra/earnings_archive.tf +++ b/infra/earnings_archive.tf @@ -38,14 +38,15 @@ locals { # INGEST project (google_project.ingest) — never the serving project — and # public_access_prevention = "enforced" + uniform access + NO serving-project IAM # member is the structural firewall (T-35-07): the internet-facing serving -# surface has no path to a stored audio object. Co-located with the ingest audio -# toolchain (upper(var.weather_region) = US-CENTRAL1), matching the -# backfill_progress bucket's location convention. force_destroy = false: this is +# surface has no path to a stored audio object. Co-located with the STT audio +# pipeline it receives copies from (upper(var.stt_region) = US-CENTRAL1, the same +# var the handoff bucket uses — keeping the D-27.9 server-side copy_blob +# same-region; review-iter-1 P3, opus). force_destroy = false: this is # a retained archive, not a scratch bucket. resource "google_storage_bucket" "earnings_archive" { project = google_project.ingest.project_id name = var.earnings_archive_bucket_name != "" ? var.earnings_archive_bucket_name : "mostlyright-earnings-archive-${google_project.ingest.number}" - location = upper(var.weather_region) + location = upper(var.stt_region) uniform_bucket_level_access = true force_destroy = false public_access_prevention = "enforced" @@ -87,6 +88,19 @@ resource "google_storage_bucket_iam_member" "earnings_archive_rw" { member = "serviceAccount:${google_service_account.earnings_archive_fleet.email}" } +# The DEPLOYED live-capture STT service (cloud_run.tf `stt`, runtime SA +# `earnings_stt`) writes the D-27.9 retention copy (_archive_handoff_source in +# services/earnings/jobs/stt.py) into this bucket when EARNINGS_ARCHIVE_RETENTION +# is enabled — without this grant every live archive copy 403s (review-iter-1 P1, +# codex). Same objectAdmin-on-this-one-bucket shape as the fleet grant above. +# earnings_stt is an INGEST-project SA (service_accounts.tf:28), so the bucket +# still has ZERO serving-project principals — the T-35-07 audio firewall holds. +resource "google_storage_bucket_iam_member" "earnings_archive_stt_rw" { + bucket = google_storage_bucket.earnings_archive.name + role = "roles/storage.objectAdmin" + member = "serviceAccount:${google_service_account.earnings_stt.email}" +} + # ===================================================================== # Daily archive POLLER — Cloud Run Job, ingest project (35-06 Task 2) # ===================================================================== diff --git a/services/earnings/jobs/stt.py b/services/earnings/jobs/stt.py index ff849609..21e7b60e 100644 --- a/services/earnings/jobs/stt.py +++ b/services/earnings/jobs/stt.py @@ -39,10 +39,12 @@ and optionally ``EARNINGS_STREAMING_TOPIC`` (default ``earnings-streaming``). * Audio retention (D-27.9 amendment, Phase 35 gate 35-01) — DEFAULT OFF: ``EARNINGS_ARCHIVE_RETENTION`` (truthy to enable) + ``EARNINGS_ARCHIVE_BUCKET`` - (the ingest-side archive bucket). When BOTH are set, the transient handoff audio - is COPIED into the archive bucket (``earnings/audio/{ticker}/{call_id}.opus``) - AFTER the durable ledger write and BEFORE the transient handoff is deleted. Unset - (or falsy) keeps the Phase 27 delete-after-transcript behavior byte-identical. + (the ingest-side archive bucket). When enabled, the transient handoff audio is + COPIED into the archive bucket (``earnings/audio/{ticker}/{call_id}.``, + the SOURCE object's suffix — ``copy_blob`` never transcodes) AFTER the durable + ledger write and BEFORE the transient handoff is deleted. Unset (or falsy) keeps + the Phase 27 delete-after-transcript behavior byte-identical. Enabling the flag + WITHOUT a bucket is a loud config error (never a silent un-archived delete). **Region.** This image runs on an L4 GPU in us-central1 (no Cloud Run GPU in eu-west3) — see the Dockerfile. @@ -189,9 +191,13 @@ def _delete_handoff_source(bucket: str, key: str) -> None: #: Object-key layout for the retained archive copy on the INGEST-side archive #: bucket, mirroring the partition-mirrors-path convention used across the ingest -#: sinks (`earnings/audio/{ticker}/{call_id}.opus`). The call_id is the ledger +#: sinks (`earnings/audio/{ticker}/{call_id}.`). The call_id is the ledger #: partition key, so an archived audio object lines up 1:1 with its transcript. -_ARCHIVE_AUDIO_KEY_FMT = "earnings/audio/{ticker}/{call_id}.opus" +#: The suffix is PRESERVED from the source handoff object: capture emits WAV and +#: ``copy_blob`` is a byte-for-byte server-side copy (no transcode), so a +#: hardcoded ``.opus`` name would lie about the bytes and mis-key any consumer +#: that picks a decoder off the extension (review-iter-1 P2, codex + opus). +_ARCHIVE_AUDIO_KEY_FMT = "earnings/audio/{ticker}/{call_id}{suffix}" def _archive_handoff_source( @@ -209,7 +215,8 @@ def _archive_handoff_source( RETAINED as an internal archive artifact instead of being discarded. This does a SERVER-SIDE ``copy_blob`` (no local download) from the transient ``AUDIO_HANDOFF_BUCKET`` object into ``EARNINGS_ARCHIVE_BUCKET`` under the - ``earnings/audio/{ticker}/{call_id}.opus`` layout, so the archived audio lines + ``earnings/audio/{ticker}/{call_id}.`` layout (suffix preserved from + the source object — the copy never transcodes), so the archived audio lines up 1:1 with its transcript ledger partition. Called by :func:`transcribe_call` ONLY after ``TranscriptLedger.replace`` @@ -222,7 +229,9 @@ def _archive_handoff_source( """ storage = _load_gcs_storage() - dst_key = _ARCHIVE_AUDIO_KEY_FMT.format(ticker=ticker, call_id=call_id) + # Preserve the SOURCE suffix (capture emits .wav; copy_blob never transcodes). + suffix = os.path.splitext(key)[1] or ".bin" + dst_key = _ARCHIVE_AUDIO_KEY_FMT.format(ticker=ticker, call_id=call_id, suffix=suffix) client = storage.Client() src_bucket = client.bucket(bucket) src_blob = src_bucket.blob(key) @@ -391,8 +400,17 @@ def transcribe_call( a GCS reference is downloaded to an ephemeral temp file that is deleted after transcription. Returns a small audio-free summary dict (segment/row counts, language, duration) — NEVER audio. Raises on a transcription / ledger failure - (fail loud). No audio reaches the ledger or wire. + (fail loud). No audio reaches the ledger or wire. ``retention_enabled`` with + no ``archive_bucket`` raises up front (loud half-config guard) — the silent + alternative would delete audio the operator believes is being retained. """ + if retention_enabled and not archive_bucket: + raise ValueError( + "EARNINGS_ARCHIVE_RETENTION is enabled but no archive bucket is set " + "(EARNINGS_ARCHIVE_BUCKET unset/empty): refusing to run — the handoff " + "audio would be deleted UN-archived while the operator believes it is " + "retained (D-27.9). Set the bucket or disable the flag." + ) with _resolve_audio_reference(audio_path, handoff_bucket=handoff_bucket) as ( local_audio, source, @@ -457,7 +475,8 @@ def transcribe_call( # This ordering guarantee is load-bearing and MUST NOT move above the ledger # write: a ledger failure raises before this point, so NEITHER branch runs. if source is not None: - if retention_enabled and archive_bucket: + # The up-front half-config guard guarantees archive_bucket is set here. + if retention_enabled: # D-27.9 retention ON (35-01 gate): RETAIN the audio to the ingest-side # archive instead of discarding it. Copy-then-delete keeps the transient # handoff bucket clean (matching the prior loud-delete intent) while the diff --git a/services/earnings/jobs/stt_server.py b/services/earnings/jobs/stt_server.py index 54c05e9f..d0378a99 100644 --- a/services/earnings/jobs/stt_server.py +++ b/services/earnings/jobs/stt_server.py @@ -97,6 +97,23 @@ def transcribe(payload: Annotated[dict, Body(...)]) -> dict[str, object]: or "earnings-streaming" ) + # D-27.9 retention seam (35-07): default from the SERVICE env, because this + # HTTP path is what the deployed Cloud Run STT service actually runs — main() + # is only the operator/CLI fallback. Without this, the operator-handoff + # instruction "set EARNINGS_ARCHIVE_RETENTION=1 where the STT job runs" would + # silently do nothing and live audio would keep being deleted un-archived + # (review-iter-1 P1, codex). Same env-default/payload-override shape as the + # streaming flags above. transcribe_call raises loudly on the half-config + # (flag ON, bucket unset) — never a silent un-archived delete. + env_retention = os.environ.get("EARNINGS_ARCHIVE_RETENTION") + env_retention_on = bool(env_retention) and env_retention.strip().lower() not in ( + "0", + "false", + "no", + ) + retention_enabled = bool(payload.get("retention_enabled", env_retention_on)) + archive_bucket = payload.get("archive_bucket") or os.environ.get("EARNINGS_ARCHIVE_BUCKET") + try: return transcribe_call( str(audio_path), @@ -111,6 +128,8 @@ def transcribe(payload: Annotated[dict, Body(...)]) -> dict[str, object]: streaming_topic=streaming_topic, handoff_bucket=handoff_bucket, r2_bucket=r2_bucket, + retention_enabled=retention_enabled, + archive_bucket=archive_bucket, ) except FileNotFoundError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc diff --git a/services/earnings/tests/test_stt_handoff_delete.py b/services/earnings/tests/test_stt_handoff_delete.py index 3d279e2d..47283edc 100644 --- a/services/earnings/tests/test_stt_handoff_delete.py +++ b/services/earnings/tests/test_stt_handoff_delete.py @@ -8,9 +8,10 @@ Phase 28 seam: delete the transient handoff after the durable ledger write. Phase 35 (35-07, D-27.9 amendment) seam: when the ``EARNINGS_ARCHIVE_RETENTION`` flag is ON (and ``EARNINGS_ARCHIVE_BUCKET`` is set), COPY the transient handoff to -the ingest-side archive bucket (``earnings/audio/{ticker}/{call_id}.opus``) BEFORE -deleting it — still only after the durable ledger write. The flag defaults OFF, so -the Phase 27/28 behavior stays byte-identical. +the ingest-side archive bucket (``earnings/audio/{ticker}/{call_id}.``, +source suffix preserved — the copy never transcodes) BEFORE deleting it — still +only after the durable ledger write. The flag defaults OFF, so the Phase 27/28 +behavior stays byte-identical, and flag-ON with NO bucket refuses loudly up front. These tests fake google-cloud-storage + the transcriber + the ledger to assert the ordering, the retention branch, and the skip-on-failure behavior. @@ -152,8 +153,10 @@ def test_handoff_archived_when_retention_on(monkeypatch) -> None: assert events == ["download", "transcribe", "ledger", "archive", "delete"] assert state["archived"] is True assert state["deleted"] is True - # The archive target key mirrors the partition layout and carries the ticker. - assert state["archive_key"] == "earnings/audio/CHWY/evt-3.opus" + # The archive target key mirrors the partition layout, carries the ticker, and + # PRESERVES the source suffix (.wav here — copy_blob never transcodes, so a + # hardcoded .opus name would lie about the bytes; review-iter-1 P2). + assert state["archive_key"] == "earnings/audio/CHWY/evt-3.wav" assert "CHWY" in state["archive_key"] assert state["archive_bucket"] == "mostlyright-earnings-archive" @@ -161,9 +164,10 @@ def test_handoff_archived_when_retention_on(monkeypatch) -> None: def test_handoff_deleted_when_retention_off(monkeypatch) -> None: """Explicit flag-OFF regression pin: delete fires, NO archive event. - Distinct from ``test_handoff_deleted_after_ledger_write`` (the default-behavior - test): this one sets the retention flag EXPLICITLY falsy so a future default flip - cannot silently regress the flag-off path. + Distinct from ``test_handoff_deleted_after_ledger_write`` (the no-kwargs test): + this one sets the retention flag EXPLICITLY falsy, pinning the explicit-off + contract. The param DEFAULT itself is pinned separately by + ``test_default_is_retention_off_even_with_bucket_available``. """ from services.earnings.jobs.stt import transcribe_call @@ -186,6 +190,60 @@ def test_handoff_deleted_when_retention_off(monkeypatch) -> None: assert "archive" not in events +def test_default_is_retention_off_even_with_bucket_available(monkeypatch) -> None: + """Pins the PARAM DEFAULT itself: no retention kwarg + a bucket in hand → delete. + + Unlike the explicit-``False`` pin below, this test FAILS if the + ``retention_enabled`` default ever flips to ``True``: the archive bucket is + supplied, so a flipped default cannot hide behind ``archive_bucket=None`` + (the masking gap flagged in review-iter-1, opus P2). + """ + from services.earnings.jobs.stt import transcribe_call + + events: list[str] = [] + state = _install_fakes(monkeypatch, events, ledger_raises=False) + + out = transcribe_call( + "gs://handoff-bkt/handoff/CHWY/evt-6.wav", + ticker="CHWY", + call_id="evt-6", + device="cpu", + archive_bucket="mostlyright-earnings-archive", # available, but flag unset + ) + assert out["segments"] == 1 + # DEFAULT is Phase 27 behavior: delete, NO archive copy, despite the bucket. + assert events == ["download", "transcribe", "ledger", "delete"] + assert state["archived"] is False + assert state["deleted"] is True + + +def test_retention_on_without_bucket_fails_loud(monkeypatch) -> None: + """Half-config (flag ON, bucket unset) refuses LOUDLY before any work. + + The silent alternative — falling through to the delete branch — discards audio + the operator believes is being retained (review-iter-1, codex P1 / opus P2). + The guard fires before download/transcribe, so no GPU time is burned and the + handoff object is untouched. + """ + from services.earnings.jobs.stt import transcribe_call + + events: list[str] = [] + state = _install_fakes(monkeypatch, events, ledger_raises=False) + + with pytest.raises(ValueError, match="EARNINGS_ARCHIVE_BUCKET"): + transcribe_call( + "gs://handoff-bkt/handoff/CHWY/evt-7.wav", + ticker="CHWY", + call_id="evt-7", + device="cpu", + retention_enabled=True, # no archive_bucket + ) + # Refused up front: nothing downloaded, transcribed, archived, or deleted. + assert events == [] + assert state["archived"] is False + assert state["deleted"] is False + + def test_handoff_kept_when_ledger_write_fails_under_retention(monkeypatch) -> None: """Retention ON + ledger write raises: NEITHER archived NOR deleted (kept).""" from services.earnings.jobs.stt import transcribe_call diff --git a/services/earnings/tests/test_stt_server.py b/services/earnings/tests/test_stt_server.py index 12854dfe..2d498ce4 100644 --- a/services/earnings/tests/test_stt_server.py +++ b/services/earnings/tests/test_stt_server.py @@ -108,6 +108,58 @@ def _fake_transcribe_call(audio_path, **kwargs): assert captured["publish_live"] is False +def test_transcribe_derives_retention_from_env(monkeypatch) -> None: + """The STT SERVICE derives the D-27.9 retention flags from its env (35-07). + + This HTTP path is what the deployed Cloud Run STT service runs — the deployed + capture->STT trigger posts only {audio_path, ticker, call_id}, so without the + env-derived default, "set EARNINGS_ARCHIVE_RETENTION=1 where the STT job runs" + (the 35-08 operator handoff) would silently do nothing and live audio would + keep being deleted un-archived (review-iter-1 P1, codex).""" + import services.earnings.jobs.stt_server as server + + captured: dict = {} + + def _fake_transcribe_call(audio_path, **kwargs): + captured.update(kwargs) + return {"ticker": kwargs["ticker"], "call_id": kwargs["call_id"], "segments": 0} + + monkeypatch.setattr(server, "transcribe_call", _fake_transcribe_call) + monkeypatch.setenv("EARNINGS_ARCHIVE_RETENTION", "1") + monkeypatch.setenv("EARNINGS_ARCHIVE_BUCKET", "mostlyright-earnings-archive") + + client = TestClient(server.app) + resp = client.post( + "/transcribe", json={"audio_path": "/tmp/a.wav", "ticker": "GIS", "call_id": "c1"} + ) + assert resp.status_code == 200 + assert captured["retention_enabled"] is True + assert captured["archive_bucket"] == "mostlyright-earnings-archive" + + +def test_transcribe_no_retention_env_defaults_off(monkeypatch) -> None: + """No retention env -> flag stays OFF through the HTTP path (Phase 27 default).""" + import services.earnings.jobs.stt_server as server + + captured: dict = {} + + def _fake_transcribe_call(audio_path, **kwargs): + captured.update(kwargs) + return {"ticker": "GIS", "call_id": "c1", "segments": 0} + + monkeypatch.setattr(server, "transcribe_call", _fake_transcribe_call) + monkeypatch.delenv("EARNINGS_ARCHIVE_RETENTION", raising=False) + monkeypatch.delenv("EARNINGS_ARCHIVE_BUCKET", raising=False) + + client = TestClient(server.app) + resp = client.post( + "/transcribe", json={"audio_path": "/tmp/a.wav", "ticker": "GIS", "call_id": "c1"} + ) + assert resp.status_code == 200 + assert captured["retention_enabled"] is False + assert captured["archive_bucket"] is None + + def test_transcribe_request_overrides_env_streaming(monkeypatch) -> None: """An explicit request field still overrides the env-derived default.""" import services.earnings.jobs.stt_server as server From 5f94896b9a9d9f512e6cd63d0d49d9428951120f Mon Sep 17 00:00:00 2001 From: minereda <84080887+minereda@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:12:50 +0200 Subject: [PATCH 4/4] fix(35 review-iter-2): null-safe retention override + 400 on half-config + strip parity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 (codex verify + opus verify, both confirmed all iter-1 fixes): P1 (codex, opus P3): a present-but-null JSON retention_enabled was treated as an explicit False override — bool(payload.get(key, env_default)) — so a typed client serializing unset optionals as null silently disabled env-enabled retention and audio was deleted un-archived. null now means ABSENT (env default applies); explicit false still overrides. P2 (codex, opus P3): the half-config guard surfaced as a retryable 500 ("transcription failed: ...") — capture would NACK and loop full recaptures on a permanent misconfig. The server now answers an explicit 400 before transcribe_call is reached; the ValueError guard stays for the CLI/MIG path. P3 (opus): main()'s env parse now .strip()s like the server — the CLI/MIG fallback and the deployed service read the same value identically. 3 new server tests: null-falls-back-to-env, explicit-false override, half-config 400 (transcribe_call never reached). services/earnings suite green, ruff clean. Co-Authored-By: Claude Fable 5 --- services/earnings/jobs/stt.py | 8 +- services/earnings/jobs/stt_server.py | 23 +++++- services/earnings/tests/test_stt_server.py | 86 ++++++++++++++++++++++ 3 files changed, 115 insertions(+), 2 deletions(-) diff --git a/services/earnings/jobs/stt.py b/services/earnings/jobs/stt.py index 21e7b60e..2ff86cf3 100644 --- a/services/earnings/jobs/stt.py +++ b/services/earnings/jobs/stt.py @@ -531,7 +531,13 @@ def main(argv: list[str] | None = None) -> int: # transient handoff audio is COPIED to EARNINGS_ARCHIVE_BUCKET (ingest side) # before the transient is deleted. Read by NAME (never inlined). retention_raw = optional_env("EARNINGS_ARCHIVE_RETENTION") - retention_enabled = bool(retention_raw) and retention_raw.lower() not in ("0", "false", "no") + # .strip() matches the stt_server parse exactly — the CLI/MIG fallback and the + # deployed service must read the SAME env value the same way (review-iter-2 P3). + retention_enabled = bool(retention_raw) and retention_raw.strip().lower() not in ( + "0", + "false", + "no", + ) transcribe_call( audio_path, diff --git a/services/earnings/jobs/stt_server.py b/services/earnings/jobs/stt_server.py index d0378a99..c3c00f84 100644 --- a/services/earnings/jobs/stt_server.py +++ b/services/earnings/jobs/stt_server.py @@ -111,9 +111,30 @@ def transcribe(payload: Annotated[dict, Body(...)]) -> dict[str, object]: "false", "no", ) - retention_enabled = bool(payload.get("retention_enabled", env_retention_on)) + # A present-but-null "retention_enabled" is ABSENT (env default applies), NOT an + # override: typed-model clients serialize unset optionals as null, and treating + # null as False would silently disable env-enabled retention and delete audio + # un-archived (review-iter-2 P1, codex). An explicit JSON false still overrides + # env-on to OFF. + payload_retention = payload.get("retention_enabled") + retention_enabled = env_retention_on if payload_retention is None else bool(payload_retention) archive_bucket = payload.get("archive_bucket") or os.environ.get("EARNINGS_ARCHIVE_BUCKET") + # Half-config (retention ON, no bucket anywhere) is a NON-RETRYABLE config / + # client error: answer 400 here, not the blanket 500 — a 500 makes capture NACK + # and Pub/Sub redeliver, looping full recaptures on a permanent misconfiguration + # (review-iter-2 P2 codex / P3 opus). transcribe_call keeps the same loud guard + # for the CLI/MIG fallback path. + if retention_enabled and not archive_bucket: + raise HTTPException( + status_code=400, + detail=( + "retention is enabled (env or request) but no archive bucket is set " + "(EARNINGS_ARCHIVE_BUCKET env / 'archive_bucket' field): refusing — " + "the handoff audio would be deleted un-archived (D-27.9)." + ), + ) + try: return transcribe_call( str(audio_path), diff --git a/services/earnings/tests/test_stt_server.py b/services/earnings/tests/test_stt_server.py index 2d498ce4..aa26424d 100644 --- a/services/earnings/tests/test_stt_server.py +++ b/services/earnings/tests/test_stt_server.py @@ -160,6 +160,92 @@ def _fake_transcribe_call(audio_path, **kwargs): assert captured["archive_bucket"] is None +def test_transcribe_null_retention_falls_back_to_env(monkeypatch) -> None: + """A present-but-null retention_enabled is ABSENT, not an override (iter-2 P1). + + Typed-model clients serialize unset optionals as null; treating null as False + would silently disable env-enabled retention and delete audio un-archived.""" + import services.earnings.jobs.stt_server as server + + captured: dict = {} + + def _fake_transcribe_call(audio_path, **kwargs): + captured.update(kwargs) + return {"ticker": "GIS", "call_id": "c1", "segments": 0} + + monkeypatch.setattr(server, "transcribe_call", _fake_transcribe_call) + monkeypatch.setenv("EARNINGS_ARCHIVE_RETENTION", "1") + monkeypatch.setenv("EARNINGS_ARCHIVE_BUCKET", "mostlyright-earnings-archive") + + client = TestClient(server.app) + resp = client.post( + "/transcribe", + json={ + "audio_path": "/tmp/a.wav", + "ticker": "GIS", + "call_id": "c1", + "retention_enabled": None, + }, + ) + assert resp.status_code == 200 + assert captured["retention_enabled"] is True # env default survives the null + + +def test_transcribe_request_overrides_env_retention(monkeypatch) -> None: + """An explicit JSON false still overrides env-enabled retention to OFF.""" + import services.earnings.jobs.stt_server as server + + captured: dict = {} + + def _fake_transcribe_call(audio_path, **kwargs): + captured.update(kwargs) + return {"ticker": "GIS", "call_id": "c1", "segments": 0} + + monkeypatch.setattr(server, "transcribe_call", _fake_transcribe_call) + monkeypatch.setenv("EARNINGS_ARCHIVE_RETENTION", "1") + monkeypatch.setenv("EARNINGS_ARCHIVE_BUCKET", "mostlyright-earnings-archive") + + client = TestClient(server.app) + resp = client.post( + "/transcribe", + json={ + "audio_path": "/tmp/a.wav", + "ticker": "GIS", + "call_id": "c1", + "retention_enabled": False, + }, + ) + assert resp.status_code == 200 + assert captured["retention_enabled"] is False + + +def test_transcribe_half_config_retention_returns_400(monkeypatch) -> None: + """Retention ON with no bucket anywhere → 400 (non-retryable), never a 500. + + A 500 would make capture NACK → Pub/Sub redelivery → repeated full recaptures + on a PERMANENT misconfiguration (iter-2, codex P2 / opus P3). transcribe_call + must not even be reached.""" + import services.earnings.jobs.stt_server as server + + called: dict = {} + + def _fake_transcribe_call(audio_path, **kwargs): + called["reached"] = True + return {"ticker": "GIS", "call_id": "c1", "segments": 0} + + monkeypatch.setattr(server, "transcribe_call", _fake_transcribe_call) + monkeypatch.setenv("EARNINGS_ARCHIVE_RETENTION", "1") + monkeypatch.delenv("EARNINGS_ARCHIVE_BUCKET", raising=False) + + client = TestClient(server.app) + resp = client.post( + "/transcribe", json={"audio_path": "/tmp/a.wav", "ticker": "GIS", "call_id": "c1"} + ) + assert resp.status_code == 400 + assert "archive bucket" in resp.json()["detail"] + assert "reached" not in called + + def test_transcribe_request_overrides_env_streaming(monkeypatch) -> None: """An explicit request field still overrides the env-derived default.""" import services.earnings.jobs.stt_server as server