[NOTICKET] fix: split report floor from report body; scope analysis reads by user_id

Two defects with one root cause, plus the security pair from the 2026-07-23 review.

Report correctness. Planner recipes R2/R2b make the analyze_* step optional (R2 is
"ONE grouped retrieve_data IR (+/- analyze_aggregate)"; R2b, added yesterday, is
explicitly "NO analyze_* step"), so an aggregate answered entirely inside one
retrieve_data IR is a correct, complete analysis that uses no analyze_* tool. One
predicate gated both "is this session worth a report" and "should this run appear in
it", so those runs were classed non-substantive, dropped from the report body, and
their business question rendered "Unanswered" while the chat had answered it
correctly (analysis 966224d4, two of six runs, both threshold questions, real findings).

- has_reportable_result (BODY): plan HAS an analysis step -> it must have succeeded;
plan has NO analysis step -> a successful non-check_* task is enough. A run whose
analysis step FAILED still stays in `unresolved`.
- has_successful_analysis (FLOOR) additionally clears on a retrieve_data that returned
rows, fixing a hard 409 when every question in a session is R2/R2b-shaped. Not a
relaxation: empty retrievals, check_*-only and fully-failed runs all still fail it.
- Deliberate asymmetry: a zero-row retrieval fails the floor but passes the body --
"no units missed both targets" is a real answer.
- GET .../records `substantive` repointed to the body predicate, so the curation list
stops contradicting the artifact it curates.
- CK5b: CK5 only inspected analyze_* tasks, so on an analyze-free plan an all-null
aggregate column reached the answer unflagged. check_* excluded.
- The dropped-runs path now logs; its silence is why this needed a bug report.

Security. Go makes no outbound call to Python (verified in the Go source), so the
"Go fronts Python" premise in the code comments is not wired and the surface is
reachable directly.

- Service-secret gate (X-Dataeyond-Service-Secret, constant-time) as a router-level
dependency, so a new endpoint cannot skip it. Blank env var = disabled, so this is
inert until dataeyond__service__secret is set; boot logs which mode is active.
- CatalogStore.get_by_analysis filtered on analysis_id alone where Go filters on
analysis_id AND user_id. The catalog payload carries the owner's user_id, so
DbExecutor's ownership check compared the victim's id against itself and passed: a
caller knowing another tenant's analysis_id could execute SQL against that tenant's
database. user_id was already a parameter of AnalysisScopedCatalogReader.read and
simply never passed down. Scoping added to all six analysis-keyed reads; the reports
read tolerates user_id IS NULL for pre-pr/18 rows. A denied read that would have
matched unscoped logs loudly.
- These predicates are defence-in-depth, NOT the control: while user_id is
caller-supplied an attacker can read it from the unauthenticated GET /traceability
response. The gate is what closes the door.

ruff src/: 238 errors, unchanged from baseline (zero new, measured per touched file).
Suite: 394 passed, 0 failed, 7 skipped. import main: 0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

API_CONTRACT_BE_PYTHON.md

@@ -36,6 +36,20 @@ The frontend uses this service during the analysis conversation flow:
 | `GET` | `/api/v1/traceability` | Retrieve provenance for one assistant answer. |
 | `GET` | `/api/v1/charts` | Retrieve chart(s) produced by `render_chart` for one assistant answer (added 2026-07-13). |
 
+## Authentication (added 2026-07-23)
+
+Every endpoint in the table above sits behind a **shared service secret**. Callers send it as a header on every request:
+
+```
+X-Dataeyond-Service-Secret: <secret>
+```
+
+A missing or incorrect value returns `401` with `{"detail": "Missing or invalid service credentials."}`. `GET /` and `GET /health` are exempt (deployment health probes).
+
+**Rollout state — currently inert.** The gate is **disabled unless** the service is deployed with `dataeyond__service__secret` set. Until that env var is configured, requests without the header succeed exactly as before, so no frontend change is required to keep working today. Send the header as soon as the secret is issued; it is safe to send before the gate is armed (an unconfigured service ignores it).
+
+**What this is and is not.** It authenticates the *calling service*, not the end user — it does not identify which user is making the request, and it is not a substitute for the frontend's own auth. It exists because this service currently accepts `user_id` and `analysis_id` as ordinary request fields with no verification, and is reachable directly. Per-user authorization will move to a verified identity forwarded by the Go service; when it does, `user_id` will come from the verified token rather than the request body, and the request fields documented below may become redundant.
+
 ## Common Concepts
 
 ### Identifiers
@@ -426,7 +440,9 @@ Response `200`:
 ]
 ```
 
-`substantive: false` means no `analyze_*` step succeeded — that run is listed in the report's `unresolved` JSON field rather than the findings body. (Since 2026-07-09 the rendered markdown is compact and no longer includes "Attempted, Unresolved" / "Notes & Limitations" / "How This Was Analyzed" sections; the JSON fields `unresolved` / `caveats` / `open_questions` / `method_steps` are unchanged.) If no runs exist, returns `[]`.
+`substantive: true` means **the run will appear in the report's findings body**; `substantive: false` means it will be listed in the report's `unresolved` JSON field instead. (Since 2026-07-09 the rendered markdown is compact and no longer includes "Attempted, Unresolved" / "Notes & Limitations" / "How This Was Analyzed" sections; the JSON fields `unresolved` / `caveats` / `open_questions` / `method_steps` are unchanged.) If no runs exist, returns `[]`.
+
+> **Changed 2026-07-23 — behavioral, non-breaking (field name, type and position unchanged).** `substantive` previously meant the narrower "a successful `analyze_*` step exists". Planner recipes R2/R2b legitimately answer a question with a single aggregate `retrieve_data` and **no** `analyze_*` step, so under the old rule such a run was reported `substantive: false`, was dropped from the report body, and its business question rendered **"Unanswered"** in `bq_answers` — while the chat had answered it correctly. `substantive` now tracks the report body exactly. **FE impact:** none required. A run that previously showed `false` and was silently excluded may now show `true` and be included; no field was added, removed or retyped.
 
 ### `GET /api/v1/tools/report/{analysis_id}/readiness` (added 2026-07-09)
 

main.py

@@ -2,9 +2,10 @@
 
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI
 from src.middlewares.logging import configure_logging, get_logger
 from src.middlewares.cors import add_cors_middleware
+from src.middlewares.service_auth import is_enforced, require_service_secret
 from src.middlewares.rate_limit import limiter, _rate_limit_exceeded_handler
 from slowapi.errors import RateLimitExceeded
 # --- pr/5 Phase 1: unwire non-AI routers (Go owns these now). ---
@@ -41,6 +42,16 @@ async def lifespan(app: FastAPI):
         logger.info("Database initialized")
     else:
         logger.info("Skipping database initialization (SKIP_INIT_DB=true)")
+    # Make the security posture visible at boot: an unset secret means the whole
+    # surface is open, which is a deployment fact worth seeing in the logs rather
+    # than inferring from the absence of 401s.
+    if is_enforced():
+        logger.info("Service-secret gate ENFORCED on all live routers")
+    else:
+        logger.warning(
+            "Service-secret gate DISABLED (dataeyond__service__secret unset) — "
+            "every endpoint is reachable without credentials"
+        )
     yield
 
 
@@ -65,12 +76,19 @@ app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 # app.include_router(db_client_router)     # unwired: Go registers DB client
 # app.include_router(data_catalog_router)  # unwired: Go handles the catalog
 # app.include_router(chat_router)  # unwired: v2 chat replaces it (drops v1 cache ops routes)
-app.include_router(report_router)
-app.include_router(tools_router)
-app.include_router(help_router)
-app.include_router(traceability_router)  # KM-691: GET /api/v1/traceability
-app.include_router(charts_router)  # W2: GET /api/v1/charts (SPINE_V2_PLAN §4.5)
-app.include_router(chat_v2_router)     # pr/5 Phase 2: POST /api/v2/chat/stream (analysis_id)
+# F-2 (2026-07-23): every live router is mounted behind the service-secret gate.
+# Applied here rather than per route so a new endpoint cannot be added without it.
+# The dependency is a NO-OP until `dataeyond__service__secret` is set, so this is
+# inert for local dev / the current FE until the secret is deployed on both sides.
+# `/` and `/health` stay open for the HF Space health probe.
+_guard = [Depends(require_service_secret)]
+
+app.include_router(report_router, dependencies=_guard)
+app.include_router(tools_router, dependencies=_guard)
+app.include_router(help_router, dependencies=_guard)
+app.include_router(traceability_router, dependencies=_guard)  # KM-691: GET /api/v1/traceability
+app.include_router(charts_router, dependencies=_guard)  # W2: GET /api/v1/charts (§4.5)
+app.include_router(chat_v2_router, dependencies=_guard)  # pr/5 Phase 2: POST /api/v2/chat/stream
 
 
 @app.get("/")

src/agents/chat_handler.py

@@ -272,7 +272,7 @@ class ChatHandler:
         # not-ready) — the HelpAgent guard only offers generate_report when ready.
         from .report.readiness import is_report_ready
 
-        report_ready = await is_report_ready(analysis_id, state)
+        report_ready = await is_report_ready(analysis_id, state, user_id=user_id)
 
         yield {"event": "sources", "data": json.dumps([])}
         try:
@@ -525,7 +525,7 @@ class ChatHandler:
             # HelpAgent only offers `generate_report` when this says ready.
             from .report.readiness import is_report_ready
 
-            report_ready = await is_report_ready(analysis_id, state)
+            report_ready = await is_report_ready(analysis_id, state, user_id=user_id)
             # The prompt sees chat history -> masked.
             hc = tracer.callbacks(masked=True)
             hkw = {"callbacks": hc} if hc else {}

src/agents/report/generator.py

@@ -607,7 +607,12 @@ class ReportGenerator:
         user_name: str | None = None,
         exclude_record_ids: list[str] | None = None,
     ) -> AnalysisReport:
-        all_records = await self._ensure_record_store().list_for_analysis(analysis_id)
+        # Scoped to the requesting user (2026-07-23): `POST /tools/report` always
+        # supplies `user_id`, so a report can only ever be built from records the
+        # caller owns.
+        all_records = await self._ensure_record_store().list_for_analysis(
+            analysis_id, user_id
+        )
         excluded_ids = set(exclude_record_ids or [])
         excluded = [r for r in all_records if r.record_id in excluded_ids]
         kept = [r for r in all_records if r.record_id not in excluded_ids]
@@ -686,7 +691,8 @@ class ReportGenerator:
         try:
             store = self._ensure_catalog_store()
             if analysis_id:
-                cat = await store.get_by_analysis(analysis_id)
+                # Scoped to the requesting user (2026-07-23) — see CatalogStore.
+                cat = await store.get_by_analysis(analysis_id, user_id)
                 if cat is not None:
                     return cat
             return await store.get(user_id) if user_id else None

src/agents/report/readiness.py

@@ -85,23 +85,56 @@ def _is_analysis_tool(tool: str) -> bool:
     return tool.startswith("analyze") or tool == "render_chart"
 
 
+def _completed_analysis_task(record) -> bool:
+    """True if some task using an analysis tool succeeded. The original floor rule."""
+    return any(
+        t.status == "success" and any(_is_analysis_tool(tool) for tool in t.tools_used)
+        for t in record.tasks_run
+    )
+
+
+def _produced_rows(record) -> bool:
+    """True if some successful `retrieve_data` actually returned rows.
+
+    Read from `results_snapshot`, not `tasks_run` — `TaskSummary` carries tool names
+    and status but no row counts. Requiring rows (not merely a successful call) means
+    an empty retrieval can never satisfy the floor.
+    """
+    for result in record.results_snapshot.values():
+        if result.status != "success":
+            continue
+        for out in result.outputs:
+            if out.tool == "retrieve_data" and out.kind == "table" and out.rows:
+                return True
+    return False
+
+
 def has_successful_analysis(record) -> bool:
-    """True if the record has at least one *result-producing* task that succeeded.
+    """True if the record produced a real result — the report **FLOOR**.
 
     A failed run still writes findings (narrating the failure) and its data-access
     tasks (check_/retrieve_) succeed, so we can't key on findings or on "any task
     succeeded". A completed analysis tool (analyze_*) — or, since W2 charts
     (2026-07-14), a completed `render_chart`, whose viz-tail upstream necessarily
-    computed the numbers being charted — is the real "we produced a result"
-    signal. A chart-only session therefore satisfies the report floor.
-
-    This is the report **FLOOR** only. The report *body* uses the broader
-    `has_reportable_result` — see its docstring for why the two diverged.
+    computed the numbers being charted — is the classic "we produced a result"
+    signal. A chart-only session therefore satisfies the floor.
+
+    **Extended 2026-07-23.** Planner recipes R2/R2b made the `analyze_*` step
+    optional, so a correct, complete analysis can consist of exactly one aggregate
+    `retrieve_data` IR. Under the original rule such a session had NO substantive
+    record at all, and `POST /api/v1/tools/report` returned 409 — "not ready" — for a
+    session in which every business question had been answered. A successful
+    `retrieve_data` **that returned rows** therefore also clears the floor.
+
+    This is NOT a relaxation of the floor's intent (cf. the "Floor Fixer" failure
+    mode): the floor still asks "did we produce a real result", and an empty
+    retrieval, a `check_*`-only run, and a fully-failed run all still fail it. What
+    changed is that producing a result no longer requires a specific tool family.
+
+    The report *body* uses `has_reportable_result` instead — the two are deliberately
+    independent; see that docstring.
     """
-    return any(
-        t.status == "success" and any(_is_analysis_tool(tool) for tool in t.tools_used)
-        for t in record.tasks_run
-    )
+    return _completed_analysis_task(record) or _produced_rows(record)
 
 
 def has_reportable_result(record) -> bool:
@@ -128,13 +161,24 @@ def has_reportable_result(record) -> bool:
       - The plan has NO analysis step  -> a successful data-producing task is enough
         (`check_*` alone is not — that inspected the schema, it did not answer
         anything).
+
+    Deliberate asymmetry with the floor: a retrieval that returned **zero rows**
+    fails the floor (a session whose only outcome was an empty result is not worth a
+    report) but PASSES here. "No units missed both targets" is a real answer to a
+    real business question, and excluding it would recreate the original bug for
+    negative results. CK2 already flags the empty retrieve so the assembler narrates
+    it honestly. Do not "align" these two — the divergence is the point.
     """
     tasks = record.tasks_run
     plan_has_analysis = any(
         _is_analysis_tool(tool) for task in tasks for tool in task.tools_used
     )
     if plan_has_analysis:
-        return has_successful_analysis(record)
+        # Deliberately calls `_completed_analysis_task`, NOT `has_successful_analysis`:
+        # the floor also clears on a row-producing retrieve_data, and inheriting that
+        # here would let a run whose analysis step FAILED back into the body on the
+        # strength of its upstream fetch. The two predicates stay independent.
+        return _completed_analysis_task(record)
     return any(
         task.status == "success"
         and any(tool not in _CATALOG_ONLY_TOOLS for tool in task.tools_used)
@@ -147,6 +191,7 @@ async def report_floor(
     state: AnalysisState,
     *,
     record_store=None,
+    user_id: str | None = None,
 ) -> tuple[list[str], list]:
     """The report **floor**: ≥1 substantive analysis.
 
@@ -169,7 +214,10 @@ async def report_floor(
     if analysis_id:
         try:
             store = record_store or _default_record_store()
-            records = await store.list_for_analysis(analysis_id)
+            # `user_id` scopes the read to the analysis's owner when the caller has
+            # one (2026-07-23); None means "unscoped", per the ReportInputStore
+            # Protocol default.
+            records = await store.list_for_analysis(analysis_id, user_id)
             substantive = [r for r in records if has_successful_analysis(r)]
         except Exception as exc:  # noqa: BLE001 — never-throw; fail closed to not-ready
             logger.warning(
@@ -190,14 +238,16 @@ async def is_report_ready(
     *,
     record_store=None,
     report_store=None,
+    user_id: str | None = None,
 ) -> ReportReadiness:
     """Return whether a report can be generated for this analysis, and the gaps if not.
 
     `record_store` / `report_store` are injectable for tests; they default to the
-    real Postgres stores.
+    real Postgres stores. `user_id`, when the caller has one, scopes the record read
+    to the analysis's owner (2026-07-23).
     """
     missing, substantive = await report_floor(
-        analysis_id, state, record_store=record_store
+        analysis_id, state, record_store=record_store, user_id=user_id
     )
 
     if not substantive:

src/agents/report/store.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 
 import hashlib
 
-from sqlalchemy import func, select, text
+from sqlalchemy import func, or_, select, text
 
 from src.db.postgres.connection import AsyncSessionLocal
 from src.db.postgres.models import AnalysisReportRow
@@ -96,11 +96,30 @@ class ReportStore:
         )
         return report
 
-    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisReport]:
+    async def list_for_analysis(
+        self, analysis_id: str, user_id: str | None = None
+    ) -> list[AnalysisReport]:
+        """Every version for one analysis, oldest-first.
+
+        `user_id` scopes the read to the owner (2026-07-23) but **tolerates NULL**:
+        `ReportStore.save` did not write `reports.user_id` until pr/18 (2026-07-22),
+        so every report generated before that date has a NULL owner. A strict
+        equality filter would hide them, so legacy rows are matched too — the loosest-
+        deployment-shape convention (§7D). Drop the NULL branch once the column is
+        backfilled.
+        """
         async with AsyncSessionLocal() as session:
+            where = [AnalysisReportRow.analysis_id == analysis_id]
+            if user_id is not None:
+                where.append(
+                    or_(
+                        AnalysisReportRow.user_id == user_id,
+                        AnalysisReportRow.user_id.is_(None),
+                    )
+                )
             result = await session.execute(
                 select(AnalysisReportRow)
-                .where(AnalysisReportRow.analysis_id == analysis_id)
+                .where(*where)
                 .order_by(AnalysisReportRow.version.asc())
             )
             rows = result.scalars().all()

src/agents/slow_path/checkpoint.py

@@ -112,12 +112,15 @@ def _assess(run_state: RunState, task_list: TaskList) -> RunAssessment:
     # CK5 — an analyze_* consumed a table whose column(s) are entirely null.
     # Consumption is read from the PLAN (the `data`/`data_right` placeholders);
     # the runner resolves the same references at execution time.
+    consumed_refs: set[str] = set()
     for task in task_list.tasks:
         for call in task.tool_calls:
             if not call.tool.startswith("analyze_"):
                 continue
             for arg_name in ("data", "data_right"):
                 ref = _placeholder_ref(call.args.get(arg_name))
+                if ref:
+                    consumed_refs.add(ref)
                 table = _last_table_output(results.get(ref)) if ref else None
                 if table is None:
                     continue
@@ -128,6 +131,27 @@ def _assess(run_state: RunState, task_list: TaskList) -> RunAssessment:
                          "null — results based on them are meaningless",
                          repairable=True)
 
+    # CK5b — the same defect on a retrieval NO analyze_* consumed, i.e. one that goes
+    # straight to the assembler. Planner recipes R2/R2b answer a question with exactly
+    # one aggregate `retrieve_data` and no analyze_* step, so the consumer-side sweep
+    # above never runs for them and an all-null aggregate column (e.g. avg(PA) over a
+    # window where PA was never recorded) reached the answer unflagged. Found
+    # 2026-07-23, same root cause as the report-body bug. Restricted to
+    # `retrieve_data`: `check_*` outputs legitimately carry all-null columns (an
+    # uncounted table surfaces `table_row_count = None`).
+    for tid, result in results.items():
+        if tid in consumed_refs or result.status != "success":
+            continue
+        table = _last_table_output(result)
+        if table is None or table.tool != "retrieve_data":
+            continue
+        null_cols = _all_null_columns(table)
+        if null_cols:
+            flag(tid, "CK5",
+                 f"column(s) {null_cols} in this result are entirely null — "
+                 "findings based on them are meaningless",
+                 repairable=True)
+
     assessments: list[TaskAssessment] = []
     for tid, result in results.items():
         if result.status == "failure":

src/agents/slow_path/store.py

@@ -41,7 +41,9 @@ class ReportInputStore(Protocol):
 
     async def save(self, record: AnalysisRecord) -> None: ...
 
-    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisRecord]: ...
+    async def list_for_analysis(
+        self, analysis_id: str, user_id: str | None = None
+    ) -> list[AnalysisRecord]: ...
 
 
 class NullReportInputStore:
@@ -55,7 +57,9 @@ class NullReportInputStore:
             n_tasks=len(record.tasks_run),
         )
 
-    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisRecord]:
+    async def list_for_analysis(
+        self, analysis_id: str, user_id: str | None = None
+    ) -> list[AnalysisRecord]:
         return []
 
 
@@ -99,11 +103,23 @@ class PostgresReportInputStore:
                 error=str(exc),
             )
 
-    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisRecord]:
+    async def list_for_analysis(
+        self, analysis_id: str, user_id: str | None = None
+    ) -> list[AnalysisRecord]:
+        """Records for one analysis, oldest-first.
+
+        `user_id` scopes the read to the analysis's owner (2026-07-23). Optional so
+        the unthreaded call sites (`report_floor`, `GET …/records`, `GET …/readiness`
+        — none of which currently receive a user_id) keep working; those endpoints
+        gaining the parameter is the remaining half of the change.
+        """
         async with AsyncSessionLocal() as session:
+            where = [ReportInputRow.analysis_id == analysis_id]
+            if user_id is not None:
+                where.append(ReportInputRow.user_id == user_id)
             result = await session.execute(
                 select(ReportInputRow.data)
-                .where(ReportInputRow.analysis_id == analysis_id)
+                .where(*where)
                 .order_by(ReportInputRow.created_at.asc())
             )
             rows = result.scalars().all()

src/api/v1/chat.py

@@ -99,15 +99,25 @@ async def cache_response(redis, cache_key: str, response: str, sources: list):
     )
 
 
-async def load_history(db: AsyncSession, analysis_id: str, limit: int = 10) -> list:
+async def load_history(
+    db: AsyncSession, analysis_id: str, limit: int = 10, user_id: str | None = None
+) -> list:
     """Load recent conversation messages for an analysis as LangChain messages (oldest-first).
 
     Reads the dedorch `analyses_messages` table (`role ∈ user|ai`), which replaced the
     deprecated `rooms`/`chat_messages`.
+
+    `user_id` scopes the read to the analysis's owner (2026-07-23). Safe for both
+    roles: Go's own `ListByAnalysis` filters `WHERE analysis_id=$1 AND user_id=$2`
+    (`message_repo.go`), so `role='ai'` rows necessarily carry the same `user_id` —
+    otherwise Go's own history reads would lose every AI reply.
     """
+    where = [AnalysesMessageRow.analysis_id == analysis_id]
+    if user_id is not None:
+        where.append(AnalysesMessageRow.user_id == user_id)
     result = await db.execute(
         select(AnalysesMessageRow)
-        .where(AnalysesMessageRow.analysis_id == analysis_id)
+        .where(*where)
         .order_by(AnalysesMessageRow.created_at.asc())
         .limit(limit)
     )

src/api/v1/report.py

@@ -141,7 +141,7 @@ async def generate_report(
 
     state = await _load_state(analysis_id)
     floor_missing, _ = await report_floor(
-        analysis_id, state or stub_analysis_state()
+        analysis_id, state or stub_analysis_state(), user_id=user_id
     )
     if floor_missing:
         raise HTTPException(
@@ -238,8 +238,15 @@ async def list_analysis_records(analysis_id: str):
 
     The FE shows this list before generating so the user can deselect runs; the
     chosen ids go to POST /tools/report as `exclude_record_ids`.
+
+    `substantive` answers "will this run appear in the report?", so it uses
+    `has_reportable_result` — the same predicate the generator's body filter uses
+    (2026-07-23). It used to use the report FLOOR's predicate, which since planner
+    recipes R2/R2b diverged from the body: an analyze-free run showed
+    `substantive: false` here while appearing in the report, so the curation list
+    contradicted the artifact it curates.
     """
-    from src.agents.report.readiness import has_successful_analysis
+    from src.agents.report.readiness import has_reportable_result
     from src.agents.slow_path.store import PostgresReportInputStore
 
     try:
@@ -256,7 +263,7 @@ async def list_analysis_records(analysis_id: str):
             record_id=r.record_id,
             goal_restated=r.goal_restated,
             created_at=r.created_at,
-            substantive=has_successful_analysis(r),
+            substantive=has_reportable_result(r),
             findings_count=len(r.findings),
         )
         for r in records

src/api/v2/chat.py

@@ -143,7 +143,9 @@ async def chat_stream(
             await _save_empty_chat_trace(analysis_id, body.user_id, message_id)
             return EventSourceResponse(stream_direct())
 
-        history = await load_history(db, analysis_id, limit=10)
+        # Scoped to the caller (2026-07-23): history for someone else's analysis_id
+        # now comes back empty rather than leaking their conversation.
+        history = await load_history(db, analysis_id, limit=10, user_id=body.user_id)
         handler = _chat_handler
 
         async def stream_response():

src/catalog/reader.py

@@ -117,7 +117,12 @@ class AnalysisScopedCatalogReader(CatalogReader):
         # real DB names AND the room's documents (`source_type='unstructured'`),
         # unlike the user-scope rows (`postgres_<hash>` names, no documents).
         try:
-            catalog = await self._store.get_by_analysis(self._analysis_id)
+            # `user_id` scopes the read to the analysis's owner (2026-07-23). It was
+            # always a parameter of this method and simply never passed down, so a
+            # caller supplying another tenant's `analysis_id` received that tenant's
+            # catalog — and every downstream ownership check then compared the
+            # victim's id against itself. Go enforces the same pair on its own reads.
+            catalog = await self._store.get_by_analysis(self._analysis_id, user_id)
         except Exception as e:  # noqa: BLE001 — never block check on the analysis read
             logger.warning(
                 "analysis catalog read failed — returning empty",

src/catalog/store.py

@@ -49,7 +49,9 @@ class CatalogStore:
         decode_sample_values(catalog)
         return catalog
 
-    async def get_by_analysis(self, analysis_id: str) -> Catalog | None:
+    async def get_by_analysis(
+        self, analysis_id: str, user_id: str | None = None
+    ) -> Catalog | None:
         """Read the `scope_type='analysis'` catalog row for an analysis.
 
         Distinct from `get()` (which reads the user-scope row): the analysis-scope
@@ -58,15 +60,54 @@ class CatalogStore:
         `postgres_<hash>` placeholder in the user-scope row). Returns None when the
         analysis has no catalog row (legacy / not yet bound), so callers fall back
         to the user-scope catalog.
+
+        **Tenant scoping (2026-07-23).** When `user_id` is supplied the row must be
+        owned by that user. Go enforces exactly this pair on every equivalent read
+        (`catalog_repo.go`: `WHERE scope_type='analysis' AND analysis_id=$1 AND
+        user_id=$2`); Python filtered on `analysis_id` alone, so a caller who knew
+        another tenant's `analysis_id` received that tenant's catalog — and, because
+        the payload also carries the owner's `user_id`, `DbExecutor`'s ownership
+        check then compared the victim's id against itself and passed, executing SQL
+        against their database.
+
+        `user_id` is optional so the legacy/unthreaded call sites keep working, but
+        an unscoped read is logged: those call sites are the remaining work.
         """
         async with AsyncSessionLocal() as session:
+            where = [
+                CatalogRow.analysis_id == analysis_id,
+                CatalogRow.scope_type == "analysis",
+            ]
+            if user_id is not None:
+                where.append(CatalogRow.user_id == user_id)
             result = await session.execute(
-                select(CatalogRow.catalog_payload).where(
-                    CatalogRow.analysis_id == analysis_id,
-                    CatalogRow.scope_type == "analysis",
-                )
+                select(CatalogRow.catalog_payload).where(*where)
             )
             row = result.scalar_one_or_none()
+            if row is None and user_id is not None:
+                # Diagnose the miss: an owned row that we just refused is either a
+                # genuine cross-tenant attempt or a `user_id` format mismatch between
+                # what Go wrote and what the caller sent. Both need to be loud; the
+                # answer is the same either way (deny).
+                probe = await session.execute(
+                    select(CatalogRow.user_id).where(
+                        CatalogRow.analysis_id == analysis_id,
+                        CatalogRow.scope_type == "analysis",
+                    )
+                )
+                owner = probe.scalar_one_or_none()
+                if owner is not None:
+                    logger.error(
+                        "analysis catalog owner mismatch — denied",
+                        analysis_id=analysis_id,
+                        requested_by=user_id,
+                        owner=owner,
+                    )
+        if user_id is None:
+            logger.warning(
+                "analysis catalog read is UNSCOPED (no user_id) — call site needs threading",
+                analysis_id=analysis_id,
+            )
         if row is None:
             return None
         catalog = infer_foreign_keys(Catalog.model_validate(row))

src/config/settings.py

@@ -99,6 +99,16 @@ class Settings(BaseSettings):
         alias="dataeyond__db__credential__key"
     )
 
+    # Shared service secret guarding the live surface (F-2, 2026-07-23). Callers send
+    # it as the `X-Dataeyond-Service-Secret` header. **Blank by default and blank =
+    # disabled**, so nothing breaks until it is deployed on both sides; setting it is
+    # what arms the gate. Interim measure only — it authenticates the *caller service*,
+    # not the user, and is meant to be replaced by verified per-request identity once
+    # Go forwards one. See src/middlewares/service_auth.py.
+    dataeyond_service_secret: str = Field(
+        alias="dataeyond__service__secret", default=""
+    )
+
 
 # Singleton instance
 settings = Settings()

src/middlewares/service_auth.py

@@ -0,0 +1,76 @@
+"""Service-secret gate for the live Python surface (F-2, 2026-07-23).
+
+Python has no authentication of its own. The comments in `api/v1/traceability.py`
+and `api/v1/charts.py` say "No auth — Go fronts Python", but Go does not: a
+repo-wide search of the Orchestrator source finds no HTTP client pointed at this
+service and no config key for one, and the FE calls `POST /api/v2/chat/stream`
+directly. So every live endpoint is reachable by anyone who knows the URL, with
+`user_id` and `analysis_id` supplied as ordinary request fields.
+
+This is the interim control: a shared secret, carried in a header, checked before
+the route runs. It deliberately does NOT identify *which* user is calling — it only
+stops the open internet. The per-user authorization story is the tenant predicates
+in the stores (see `CatalogStore.get_by_analysis`) plus, eventually, a real
+per-request identity forwarded by Go.
+
+Design:
+  - **Off unless configured.** With `dataeyond__service__secret` unset the
+    dependency is a no-op, so local dev, tests, and the current FE keep working
+    unchanged until the secret is deployed on both sides. Setting the env var is
+    what arms it — a single, reversible switch.
+  - **Constant-time compare** so the check can't be narrowed by timing.
+  - **Applied at router mount** (`main.py`), not per route, so a new endpoint cannot
+    be added without it.
+  - `/` and `/health` stay open — the HF Space health probe has no secret.
+
+Replace with JWT verification once Go forwards a real identity; at that point
+`user_id` should come from the verified claims rather than the request body, which
+is what makes the store predicates authoritative instead of merely defensive.
+"""
+
+from __future__ import annotations
+
+import hmac
+
+from fastapi import Header, HTTPException, status
+
+from src.config.settings import settings
+from src.middlewares.logging import get_logger
+
+logger = get_logger("service_auth")
+
+# Header name kept vendor-neutral; Go/FE send the same value.
+SERVICE_SECRET_HEADER = "X-Dataeyond-Service-Secret"
+
+
+def _configured_secret() -> str:
+    return (getattr(settings, "dataeyond_service_secret", "") or "").strip()
+
+
+def is_enforced() -> bool:
+    """True when a secret is configured — i.e. the gate actually rejects."""
+    return bool(_configured_secret())
+
+
+async def require_service_secret(
+    x_dataeyond_service_secret: str | None = Header(default=None),
+) -> None:
+    """FastAPI dependency: 401 unless the caller presents the configured secret.
+
+    No-op when no secret is configured, so enabling this is a deployment decision
+    rather than a code change.
+    """
+    expected = _configured_secret()
+    if not expected:
+        return
+    presented = (x_dataeyond_service_secret or "").strip()
+    if not presented or not hmac.compare_digest(presented, expected):
+        # Deliberately terse: never echo the presented value, never distinguish
+        # "missing" from "wrong" to the caller.
+        logger.warning(
+            "service secret rejected", presented=bool(presented)
+        )
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing or invalid service credentials.",
+        )

src/traceability/store.py

@@ -39,7 +39,7 @@ class TraceabilityStore(Protocol):
     async def save(self, payload: TraceabilityPayload) -> None: ...
 
     async def get(
-        self, analysis_id: str, message_id: str
+        self, analysis_id: str, message_id: str, user_id: str | None = None
     ) -> TraceabilityPayload | None: ...
 
 
@@ -55,7 +55,7 @@ class NullTraceabilityStore:
         )
 
     async def get(
-        self, analysis_id: str, message_id: str
+        self, analysis_id: str, message_id: str, user_id: str | None = None
     ) -> TraceabilityPayload | None:
         return None
 
@@ -95,14 +95,26 @@ class PostgresTraceabilityStore:
             )
 
     async def get(
-        self, analysis_id: str, message_id: str
+        self, analysis_id: str, message_id: str, user_id: str | None = None
     ) -> TraceabilityPayload | None:
+        """One turn's payload, or None on a miss.
+
+        `user_id` scopes the read to the turn's owner (2026-07-23). Optional because
+        `GET /api/v1/traceability` does not yet accept the parameter — adding it is an
+        FE contract change. Until then this row is reachable by `(analysis_id,
+        message_id)` alone, and the payload itself carries `user_id`, so the endpoint
+        also hands an attacker the id needed to pass every other scoped read. That is
+        why the service-secret gate, not this predicate, is the control that matters.
+        """
         async with AsyncSessionLocal() as session:
+            where = [
+                MessageTraceabilityRow.message_id == message_id,
+                MessageTraceabilityRow.analysis_id == analysis_id,
+            ]
+            if user_id is not None:
+                where.append(MessageTraceabilityRow.user_id == user_id)
             result = await session.execute(
-                select(MessageTraceabilityRow.data).where(
-                    MessageTraceabilityRow.message_id == message_id,
-                    MessageTraceabilityRow.analysis_id == analysis_id,
-                )
+                select(MessageTraceabilityRow.data).where(*where)
             )
             row = result.scalar_one_or_none()
         if row is None: