diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index c06eb88..bde9379 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -348,14 +348,46 @@ and `offset` for pagination; `remote`, `org`, `repo` to override the default registry-project scope. Events sort ascending by `(timestamp, source_rank, event_key)`; missing timestamps sort last. +### Filter authority, and refusing what cannot be answered + +A filter dimension is only meaningful for a source whose records carry it. +Each source declares its own support in `_SOURCE_FILTER_SUPPORT` and reports it +per response as `supported_filters` / `unsupported_filters`: + +| Source | issue | pr | session | +|---|---|---|---| +| `control_plane` | yes | yes | **no** — the `events` table is `(event_id, work_item_id, event_type, message, created_at)` and records no session | +| `gitea_handoff` | yes | yes | yes — a CTH comment declares its own `Session:` field | + +`session_id` is read only from that declared CTH field. It is never inferred +from a work item, an actor, or message text, and a value that is +redaction-altering or bare-secret-shaped is dropped rather than emitted. + +When **no source that ran** can carry a requested dimension, the request is +refused rather than answered: the response is `422` with `ok:false` and a +structured `error` naming `unsupported_filters` and the per-source reason. A +`200` with zero events would tell an operator that no such activity exists, +which is a stronger — and false — claim than "this cannot be answered here". +A source that *can* answer the dimension and simply matched nothing still +returns `200` with `ok:true` and an empty page. + ### Redaction -Every free-text field (event messages, decision/proof text, roles) is passed -through the console redaction policy (`webui.console_redaction`, backed by -`gitea_audit.redact`) before it leaves the module, failing closed to the +Every free-text field (event messages, decision/proof text, roles, actors) is +passed through the console redaction policy (`webui.console_redaction`, backed +by `gitea_audit.redact`) before it leaves the module, failing closed to the placeholder. No unredacted tool arguments or secrets are ever emitted, and a generation error never drops raw data to a caller or a log. +Redaction also runs *before* any structured value is derived from free text. +`evidence_refs` are extracted from already-redacted proof/decision text, and a +commit reference is recognised only where the text declares one (`commit`, +`head`, `base`, `sha`, …). An undeclared 40-character hex run has the exact +shape of a Gitea access token, so it is never lifted out of prose into a +structured field. Every reference is then independently revalidated against an +allowed shape and a second redaction pass immediately before serialization; +anything unproven is dropped and the event is flagged `sensitive`. + ### Tests ```bash diff --git a/tests/test_webui_timeline.py b/tests/test_webui_timeline.py index 35384c0..2464e74 100644 --- a/tests/test_webui_timeline.py +++ b/tests/test_webui_timeline.py @@ -4,6 +4,7 @@ Covers the acceptance criteria: versioned schema, adaptation of control-plane events and Gitea handoff comments, filter by issue/PR/session, redaction of secret-like payloads, and stable pagination. """ +import json import os import sqlite3 import sys @@ -302,6 +303,243 @@ class TestLoadTimeline(unittest.TestCase): self.assertIn("network down", handoff["reason"]) +# A fabricated 40-character lowercase hex value with the shape of a Gitea +# personal access token. Never a real credential — its only job is to prove it +# cannot reach any part of a serialized timeline payload. +SYNTHETIC_SECRET_40_HEX = "a3f9c17be44d2058e6b17c9d0f5321ab77c4e9d1" + + +def _cth_comment(comment_id, *, created_at, session=None, decision="d", proof="none", **kw): + """Build one real CTH comment record, optionally declaring a session.""" + extra = {"Session": session} if session is not None else None + body = format_cth_body( + cth_type=kw.pop("cth_type", "Author Handoff"), + status=kw.pop("status", "ready"), + next_owner=kw.pop("next_owner", "reviewer"), + decision=decision, + proof=proof, + next_action=kw.pop("next_action", "review"), + ready_to_paste_prompt=kw.pop("ready_to_paste_prompt", "Review PR #1 now"), + extra_fields=extra, + ) + return { + "id": comment_id, + "body": body, + "created_at": created_at, + "user": {"login": kw.pop("login", "jcwalker3")}, + } + + +class TestSessionFilterThroughAdapter(unittest.TestCase): + """F1: the session dimension must be real, or explicitly refused. + + These drive the filter through the CTH adapter and the composed + ``load_timeline``/API path, never through a hand-built ``WorkflowEvent``. + """ + + def _source(self, comments): + return lambda kind, number: list(comments) + + def test_adapter_populates_declared_session(self): + events = timeline.adapt_cth_comments( + [_cth_comment(1, created_at="2026-07-23T05:00:00Z", session="sess-alpha")], + kind="issue", + number=637, + ) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].session_id, "sess-alpha") + + def test_session_filter_matches_through_adapter(self): + import tempfile + + comments = [ + _cth_comment(1, created_at="2026-07-23T09:00:00Z", session="sess-alpha"), + _cth_comment(2, created_at="2026-07-23T08:00:00Z", session="sess-alpha"), + _cth_comment(3, created_at="2026-07-23T10:00:00Z", session="sess-beta"), + ] + with tempfile.TemporaryDirectory() as tmp: + db = os.path.join(tmp, "cp.sqlite3") + _seed_db(db) + snap = timeline.load_timeline( + remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + issue=637, session="sess-alpha", db_path=db, + comment_source=self._source(comments), + ) + d = snap.to_dict() + self.assertTrue(d["ok"]) + self.assertIsNone(d["error"]) + keys = [e["event_key"] for e in d["events"]] + # Only the two sess-alpha events, still in ascending timestamp order. + self.assertEqual(keys, ["cth:issue:637:2", "cth:issue:637:1"]) + self.assertTrue(all(e["session_id"] == "sess-alpha" for e in d["events"])) + self.assertEqual(d["pagination"]["total"], 2) + + def test_session_filter_paginates_and_keeps_order(self): + import tempfile + + comments = [ + _cth_comment(i, created_at=f"2026-07-23T0{i}:00:00Z", session="sess-alpha") + for i in range(1, 4) + ] + with tempfile.TemporaryDirectory() as tmp: + db = os.path.join(tmp, "cp.sqlite3") + _seed_db(db) + kwargs = dict( + remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + issue=637, session="sess-alpha", db_path=db, + comment_source=self._source(comments), + ) + page1 = timeline.load_timeline(limit=2, offset=0, **kwargs).to_dict() + page2 = timeline.load_timeline(limit=2, offset=2, **kwargs).to_dict() + self.assertEqual( + [e["event_key"] for e in page1["events"]], + ["cth:issue:637:1", "cth:issue:637:2"], + ) + self.assertEqual(page1["pagination"]["next_offset"], 2) + self.assertEqual([e["event_key"] for e in page2["events"]], ["cth:issue:637:3"]) + self.assertIsNone(page2["pagination"]["next_offset"]) + + def test_unknown_session_is_honestly_empty_when_supported(self): + """A source that *can* answer the dimension may legitimately match nothing.""" + import tempfile + + comments = [_cth_comment(1, created_at="2026-07-23T09:00:00Z", session="sess-alpha")] + with tempfile.TemporaryDirectory() as tmp: + db = os.path.join(tmp, "cp.sqlite3") + _seed_db(db) + d = timeline.load_timeline( + remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + issue=637, session="sess-nope", db_path=db, + comment_source=self._source(comments), + ).to_dict() + self.assertTrue(d["ok"]) + self.assertEqual(d["events"], []) + + def test_session_filter_refused_when_no_source_can_answer(self): + """The F1 defect: an empty-and-healthy page for an unanswerable filter.""" + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + db = os.path.join(tmp, "cp.sqlite3") + _seed_db(db) + d = timeline.load_timeline( + remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + issue=637, session="sess-alpha", db_path=db, comment_source=None, + ).to_dict() + self.assertFalse(d["ok"]) + self.assertEqual(d["error"]["code"], "filter_not_supported") + self.assertEqual(d["error"]["unsupported_filters"], ["session"]) + self.assertEqual(d["events"], []) + self.assertEqual(d["pagination"]["total"], 0) + # The refusal states which source could not answer, and why. + explained = {r["source"] for r in d["error"]["sources"]} + self.assertEqual(explained, {"control_plane", "gitea_handoff"}) + + def test_control_plane_declares_session_unsupported(self): + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + db = os.path.join(tmp, "cp.sqlite3") + _seed_db(db) + d = timeline.load_timeline( + remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + issue=637, session="sess-alpha", db_path=db, + comment_source=self._source([]), + ).to_dict() + cp = [s for s in d["sources"] if s["name"] == "control_plane"][0] + handoff = [s for s in d["sources"] if s["name"] == "gitea_handoff"][0] + self.assertNotIn("session", cp["supported_filters"]) + self.assertEqual(cp["unsupported_filters"], ["session"]) + self.assertIn("session", handoff["supported_filters"]) + self.assertEqual(handoff["unsupported_filters"], []) + + def test_secret_shaped_session_value_is_dropped(self): + events = timeline.adapt_cth_comments( + [_cth_comment(1, created_at="2026-07-23T05:00:00Z", session=SYNTHETIC_SECRET_40_HEX)], + kind="issue", + number=637, + ) + self.assertIsNone(events[0].session_id) + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(events[0].to_dict())) + + +class TestEvidenceRefRedaction(unittest.TestCase): + """F2: evidence_refs must not be a hole in the redaction boundary.""" + + def _refs_for(self, *, proof="none", decision="d"): + events = timeline.adapt_cth_comments( + [_cth_comment(1, created_at="2026-07-23T05:00:00Z", proof=proof, decision=decision)], + kind="issue", + number=637, + ) + self.assertEqual(len(events), 1) + return events[0] + + def test_assigned_secret_never_reaches_evidence_refs(self): + ev = self._refs_for(proof=f"authenticated with token={SYNTHETIC_SECRET_40_HEX}") + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.evidence_refs) + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict())) + + def test_bare_secret_shaped_value_never_reaches_evidence_refs(self): + # An undeclared hex run in proof text is not evidence of anything, and + # proof itself is never serialized — so the value has no way out. + ev = self._refs_for(proof=f"proof {SYNTHETIC_SECRET_40_HEX}", decision="rotate the credential") + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.evidence_refs) + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict())) + + def test_secret_absent_from_complete_serialized_payload(self): + import tempfile + + comments = [ + _cth_comment( + 1, + created_at="2026-07-23T09:00:00Z", + proof=f"lease token={SYNTHETIC_SECRET_40_HEX} and bare {SYNTHETIC_SECRET_40_HEX}", + decision=f"rotate api_key={SYNTHETIC_SECRET_40_HEX}", + ) + ] + with tempfile.TemporaryDirectory() as tmp: + db = os.path.join(tmp, "cp.sqlite3") + _seed_db(db) + snap = timeline.load_timeline( + remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + issue=637, db_path=db, comment_source=lambda k, n: comments, + ) + payload = json.dumps(snap.to_dict()) + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, payload) + # And the surface is clean by the redaction policy's own detectors. + from webui import console_redaction + + self.assertEqual(console_redaction.scan_for_secrets(snap.to_dict()), []) + + def test_legitimate_references_still_usable(self): + head = "4f3a464a1c455b6ecaaae9b6eef496c8f8ed451a" + ev = self._refs_for( + proof=f"closes #637, PR #849, commit abc1234, at head {head}", + decision="none", + ) + for token in ("#637", "#849", "abc1234", head): + self.assertIn(token, ev.evidence_refs) + + def test_undeclared_hex_words_are_not_references(self): + ev = self._refs_for(proof="the record was defaced and the facade decayed") + self.assertEqual(ev.evidence_refs, ()) + + def test_refs_revalidated_independently_before_serialization(self): + """Extraction is not trusted: the validator drops anything unproven.""" + safe, dropped = timeline._validated_evidence_refs( + ["#637", "abc1234", "not-a-ref", f"token={SYNTHETIC_SECRET_40_HEX}", ""] + ) + self.assertEqual(safe, ("#637", "abc1234")) + self.assertTrue(dropped) + self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(list(safe))) + + def test_clean_event_is_not_marked_sensitive(self): + ev = self._refs_for(proof="closes #637") + self.assertEqual(ev.evidence_refs, ("#637",)) + self.assertFalse(ev.sensitive) + + class TestTimelineApi(unittest.TestCase): def setUp(self): self._prev_db = os.environ.get(control_plane_db.DB_PATH_ENV) @@ -359,6 +597,22 @@ class TestTimelineApi(unittest.TestCase): resp = self.client.get("/api/v1/timeline?issue=637") self.assertNotIn("ghs_ABCDEF1234567890abcdef", resp.text) + def test_api_refuses_unanswerable_session_filter(self): + """No handoff source is configured offline, so nothing can carry a session.""" + resp = self.client.get("/api/v1/timeline?issue=637&session=sess-alpha") + self.assertEqual(resp.status_code, 422) + body = resp.json() + self.assertFalse(body["ok"]) + self.assertEqual(body["error"]["code"], "filter_not_supported") + self.assertEqual(body["error"]["unsupported_filters"], ["session"]) + self.assertEqual(body["events"], []) + self.assertEqual(body["pagination"]["total"], 0) + + def test_api_unfiltered_read_stays_ok(self): + body = self.client.get("/api/v1/timeline?issue=637").json() + self.assertTrue(body["ok"]) + self.assertIsNone(body["error"]) + if __name__ == "__main__": unittest.main() diff --git a/webui/app.py b/webui/app.py index 75e3f1b..c049707 100644 --- a/webui/app.py +++ b/webui/app.py @@ -503,7 +503,10 @@ async def api_v1_timeline(request: Request) -> JSONResponse: offset=_query_int(request, "offset"), comment_source=comment_source, ) - return JSONResponse(timeline_snapshot_to_dict(snapshot)) + # A filter no surviving source can carry is refused, not answered empty: + # a 200 with zero events would tell the operator no such activity exists. + status_code = 200 if snapshot.ok else 422 + return JSONResponse(timeline_snapshot_to_dict(snapshot), status_code=status_code) async def method_not_allowed(request: Request, _exc: Exception) -> Response: diff --git a/webui/timeline.py b/webui/timeline.py index e4597af..ded93c8 100644 --- a/webui/timeline.py +++ b/webui/timeline.py @@ -16,11 +16,19 @@ Design rules honoured here: - **Fail-soft per source.** An unavailable source degrades to a status with a reason rather than raising, and a source that could not run is never rendered as an empty-and-healthy timeline. +- **Answerable filters only.** Each source declares which filter dimensions it + can actually answer. A filter dimension no source that ran can carry is + refused with an explicit reason rather than silently matching nothing: an + empty page from an unanswerable filter reads to an operator as "no such + activity", which is a different — and false — statement. - **Redaction at the boundary, fail closed.** Every free-text field (event messages, redacted tool arguments, decision/proof text) is run through the - console redaction policy before it leaves this module. An unredactable value - becomes the placeholder — an unredacted payload is never emitted, and a - generation error never drops raw data to a caller or a log. + console redaction policy before it leaves this module, and *before* any + structured value is derived from it — evidence references are extracted from + redacted text, then independently revalidated before serialization. An + unredactable value becomes the placeholder, and a value that cannot be proven + safe is dropped — an unredacted payload is never emitted, and a generation + error never drops raw data to a caller or a log. - **Stable ordering.** Events sort by ``(timestamp, source_rank, event_key)`` with a deterministic tiebreak, so pagination is stable across calls and events with equal or missing timestamps keep a fixed order. @@ -33,7 +41,7 @@ from __future__ import annotations import re import sqlite3 -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timezone from typing import Any, Callable, Iterable @@ -55,6 +63,32 @@ _SOURCE_RANK = { SOURCE_GITEA_HANDOFF: 1, } +# The filter dimensions the query layer accepts. +FILTER_ISSUE = "issue" +FILTER_PR = "pr" +FILTER_SESSION = "session" + +# Which dimensions each source can actually answer. This is a property of the +# underlying records, not of the query code: the control-plane ``events`` table +# is (event_id, work_item_id, event_type, message, created_at) and carries no +# session identity at all, so no control-plane event can ever match a session +# filter. A CTH handoff comment can declare its session as a field, so the +# handoff source answers all three. Filtering on a dimension the surviving +# sources cannot carry is refused in ``load_timeline`` rather than answered +# with an empty page. +_SOURCE_FILTER_SUPPORT: dict[str, tuple[str, ...]] = { + SOURCE_CONTROL_PLANE: (FILTER_ISSUE, FILTER_PR), + SOURCE_GITEA_HANDOFF: (FILTER_ISSUE, FILTER_PR, FILTER_SESSION), +} + +# Why a source cannot answer a dimension, for the refusal reason an operator reads. +_SOURCE_FILTER_LIMITS: dict[tuple[str, str], str] = { + (SOURCE_CONTROL_PLANE, FILTER_SESSION): ( + "control-plane events carry no session identity " + "(the events table has no session column)" + ), +} + # A timestamp far in the future so events with no parseable timestamp sort # last (after everything real) instead of first, without raising. _MISSING_TS_SORT = "9999-12-31T23:59:59Z" @@ -148,7 +182,26 @@ _SENSITIVE_EVENT_HINTS = ("lease", "capability", "token", "auth", "secret") # Reference tokens (issue/PR/comment ids) and SHAs parsed out of proof text. _EVIDENCE_REF_RE = re.compile(r"(?:#|PR\s*#?|issue\s*#?|comment\s*#?)(\d+)", re.IGNORECASE) -_SHA_RE = re.compile(r"\b[0-9a-f]{7,40}\b") + +# A commit reference is only recognised when the text *declares* it as one. +# A bare lowercase hex run is not evidence of anything: at 40 characters it is +# exactly the shape of a Gitea personal access token, and at 7 it also matches +# ordinary words such as "defaced". Requiring an anchoring keyword keeps real +# references ("commit abc1234", "at head a209756...", "base caaae9b6") usable +# while refusing to lift an undeclared secret-shaped run out of free text. +_SHA_RE = re.compile( + r"(?i:\b(?:commit|sha|head|base|parent|revision|rev|merge[- ]base)\b[\s:=@#]*)" + r"([0-9a-f]{7,40})\b" +) + +# Shapes a serialized evidence reference is allowed to take. Anything else is +# dropped rather than emitted. +_REF_ISSUE_SHAPE = re.compile(r"^#[0-9]{1,9}$") +_REF_SHA_SHAPE = re.compile(r"^[0-9a-f]{7,40}$") + +# A long undelimited hex run with no declaring context is treated as credential +# material wherever it appears, never as an identifier. +_BARE_SECRET_SHAPE = re.compile(r"^[0-9a-f]{32,}$") def _kind_to_numbers(kind: str | None, number: int | None) -> tuple[int | None, int | None]: @@ -169,6 +222,14 @@ def _correlation_for(kind: str | None, number: int | None) -> str | None: def _extract_evidence_refs(*texts: str | None) -> tuple[str, ...]: + """Extract issue/PR and declared-commit references from **redacted** text. + + Callers must pass text that has already been through :func:`_redact`; this + function derives a structured field from its input, so extracting ahead of + redaction would republish whatever redaction was about to remove. Every + reference is revalidated by :func:`_validated_evidence_refs` before it is + serialized. + """ refs: list[str] = [] for text in texts: if not text: @@ -178,12 +239,64 @@ def _extract_evidence_refs(*texts: str | None) -> tuple[str, ...]: if token not in refs: refs.append(token) for match in _SHA_RE.finditer(text): - token = match.group(0) + token = match.group(1) if token not in refs: refs.append(token) return tuple(refs) +def _validated_evidence_refs(refs: Iterable[str]) -> tuple[tuple[str, ...], bool]: + """Independently revalidate references immediately before serialization. + + Extraction is not trusted on its own. A reference survives only when it has + a known reference shape and is unchanged by a second redaction pass — a + value the redaction policy would alter is credential material that must not + be emitted as a structured field. A full 40-character SHA stays usable + because extraction only accepts a hex run the source text explicitly + declared as a commit. Returns ``(safe_refs, dropped_any)``; ``dropped_any`` + marks the event sensitive so the drop is visible rather than silent. + """ + safe: list[str] = [] + dropped = False + for ref in refs or (): + try: + token = str(ref).strip() + if not token: + continue + recognised = bool(_REF_ISSUE_SHAPE.match(token) or _REF_SHA_SHAPE.match(token)) + if not recognised: + dropped = True + continue + if _redact(token) != token: + dropped = True + continue + if token not in safe: + safe.append(token) + except Exception: + # Fail closed: a reference that cannot be proven safe is dropped. + dropped = True + continue + return (tuple(safe), dropped) + + +def _safe_session_id(value: Any) -> str | None: + """Return a session identifier only when it is safe to emit. + + The value is authoritative source data — a session the record names for + itself — but it is still free text. It is dropped when redaction alters it + or when it is a bare secret-shaped hex run, so a credential parked in a + session field can never reach the payload or be echoed back by a filter. + """ + if value is None: + return None + text = str(value).strip() + if not text: + return None + if _BARE_SECRET_SHAPE.match(text): + return None + return text if _redact(text) == text else None + + def adapt_cp_events(rows: Iterable[dict[str, Any]]) -> list[WorkflowEvent]: """Adapt control-plane ``events`` rows (joined to work_items) into events. @@ -210,7 +323,13 @@ def adapt_cp_events(rows: Iterable[dict[str, Any]]) -> list[WorkflowEvent]: timestamp=_parse_ts(row.get("created_at")), issue_number=issue_no, pr_number=pr_no, - session_id=(row.get("session_id") or None), + # No session_id: the control-plane events table is + # (event_id, work_item_id, event_type, message, created_at) + # and records no session. Inventing one from the work item + # or the message text would be a guess, so this source + # declares the session dimension unsupported instead + # (_SOURCE_FILTER_SUPPORT) and the query layer refuses a + # session filter it cannot honestly answer. message=_redact(row.get("message")), correlation_id=_correlation_for(kind, number), sensitive=sensitive, @@ -250,25 +369,34 @@ def adapt_cth_comments( fields = parsed.get("fields") or {} cth_type = parsed.get("cth_type") or "handoff" comment_id = comment.get("id") - actor = (comment.get("user") or {}).get("login") - decision = fields.get("decision") - proof = fields.get("proof") - next_action = fields.get("next action") + # Redaction runs first, and every derived value is taken from the + # redacted text — deriving evidence refs from the raw proof would + # re-emit exactly what redaction was about to remove. + decision = _redact(fields.get("decision")) + proof = _redact(fields.get("proof")) + next_action = _redact(fields.get("next action")) + refs, refs_dropped = _validated_evidence_refs( + _extract_evidence_refs(proof, decision) + ) events.append( WorkflowEvent( source=SOURCE_GITEA_HANDOFF, event_type=f"handoff:{cth_type}", event_key=f"cth:{kind}:{number}:{comment_id}", timestamp=_parse_ts(comment.get("created_at")), - actor=actor, + actor=_redact((comment.get("user") or {}).get("login")), role=_redact(fields.get("next owner")), issue_number=issue_no, pr_number=pr_no, - decision=_redact(decision), - message=_redact(next_action or fields.get("status")), + # A CTH names its own session when the producer records one; + # it is read from that declared field, never inferred from + # unrelated text. + session_id=_safe_session_id(fields.get("session")), + decision=decision, + message=next_action or _redact(fields.get("status")), correlation_id=correlation, - evidence_refs=_extract_evidence_refs(proof, decision), - sensitive=False, + evidence_refs=refs, + sensitive=refs_dropped, ) ) except Exception: @@ -295,15 +423,50 @@ WHERE w.remote = ? AND w.org = ? AND w.repo = ? @dataclass(frozen=True) class SourceStatus: - """Fail-soft status for one timeline source.""" + """Fail-soft status for one timeline source. + + ``supported_filters`` states which filter dimensions this source's records + can carry; ``unsupported_filters`` names the requested dimensions it cannot, + so an operator can see *why* a source contributed nothing rather than being + left to read an empty list as an absence of activity. + """ name: str ok: bool reason: str | None = None count: int = 0 + supported_filters: tuple[str, ...] = () + unsupported_filters: tuple[str, ...] = () def to_dict(self) -> dict[str, Any]: - return {"name": self.name, "ok": self.ok, "reason": self.reason, "count": self.count} + return { + "name": self.name, + "ok": self.ok, + "reason": self.reason, + "count": self.count, + "supported_filters": list(self.supported_filters), + "unsupported_filters": list(self.unsupported_filters), + } + + +def _cp_status(*, ok: bool, reason: str | None = None, count: int = 0) -> SourceStatus: + return SourceStatus( + SOURCE_CONTROL_PLANE, + ok=ok, + reason=reason, + count=count, + supported_filters=_SOURCE_FILTER_SUPPORT[SOURCE_CONTROL_PLANE], + ) + + +def _handoff_status(*, ok: bool, reason: str | None = None, count: int = 0) -> SourceStatus: + return SourceStatus( + SOURCE_GITEA_HANDOFF, + ok=ok, + reason=reason, + count=count, + supported_filters=_SOURCE_FILTER_SUPPORT[SOURCE_GITEA_HANDOFF], + ) def read_cp_events( @@ -328,14 +491,14 @@ def read_cp_events( cursor = conn.execute(_CP_EVENTS_QUERY, (remote, org, repo)) rows = [dict(r) for r in cursor.fetchall()] except sqlite3.OperationalError as exc: - return ([], SourceStatus(SOURCE_CONTROL_PLANE, ok=False, reason=f"control-plane DB unavailable: {exc}")) + return ([], _cp_status(ok=False, reason=f"control-plane DB unavailable: {exc}")) except sqlite3.Error as exc: - return ([], SourceStatus(SOURCE_CONTROL_PLANE, ok=False, reason=f"control-plane read failed: {exc}")) + return ([], _cp_status(ok=False, reason=f"control-plane read failed: {exc}")) finally: if conn is not None: conn.close() events = adapt_cp_events(rows) - return (events, SourceStatus(SOURCE_CONTROL_PLANE, ok=True, count=len(events))) + return (events, _cp_status(ok=True, count=len(events))) # --------------------------------------------------------------------------- # @@ -437,6 +600,14 @@ CommentSource = Callable[[str, int], list[dict[str, Any]]] @dataclass(frozen=True) class TimelineSnapshot: + """One answered timeline query. + + ``ok`` is False when the query could not be answered as asked — currently + when a requested filter dimension no surviving source can carry was + supplied. The page is then empty *and* the snapshot says so, because an + ``ok`` empty page is a claim that no such activity exists. + """ + schema_version: int remote: str org: str @@ -444,9 +615,13 @@ class TimelineSnapshot: filters: dict[str, Any] page: TimelinePage sources: tuple[SourceStatus, ...] + ok: bool = True + error: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: return { + "ok": self.ok, + "error": self.error, "schema_version": self.schema_version, "scope": {"remote": self.remote, "org": self.org, "repo": self.repo}, "filters": self.filters, @@ -455,6 +630,28 @@ class TimelineSnapshot: } +def _unanswerable_reasons( + statuses: Iterable[SourceStatus], unanswerable: Iterable[str] +) -> list[dict[str, str]]: + """Explain, per source, why each unanswerable dimension went unanswered.""" + out: list[dict[str, str]] = [] + for status in statuses: + for dim in unanswerable: + if dim not in status.supported_filters: + reason = _SOURCE_FILTER_LIMITS.get( + (status.name, dim), f"this source's records carry no {dim} identity" + ) + elif not status.ok: + reason = ( + f"this source can carry {dim} but did not run: " + f"{status.reason or 'unavailable'}" + ) + else: + continue + out.append({"source": status.name, "filter": dim, "reason": reason}) + return out + + def load_timeline( *, remote: str, @@ -476,6 +673,11 @@ def load_timeline( issue or PR is requested (a handoff comment belongs to one thread) and a ``comment_source`` is available; otherwise it is reported as ``not run`` rather than as an empty-and-healthy source. + + A filter dimension that no surviving source can carry — a ``session`` + filter when the only source that ran is the control plane, whose events + record no session — is refused with ``ok=False`` and a structured error + instead of being answered with an empty page. """ all_events: list[WorkflowEvent] = [] statuses: list[SourceStatus] = [] @@ -494,16 +696,14 @@ def load_timeline( if handoff_target is None: statuses.append( - SourceStatus( - SOURCE_GITEA_HANDOFF, + _handoff_status( ok=False, reason="not run: handoff comments are thread-scoped; filter by issue or pr to include them", ) ) elif comment_source is None: statuses.append( - SourceStatus( - SOURCE_GITEA_HANDOFF, + _handoff_status( ok=False, reason="not run: no comment source configured for this timeline read", ) @@ -514,11 +714,56 @@ def load_timeline( comments = comment_source(kind, number) or [] handoff_events = adapt_cth_comments(comments, kind=kind, number=number) all_events.extend(handoff_events) - statuses.append(SourceStatus(SOURCE_GITEA_HANDOFF, ok=True, count=len(handoff_events))) + statuses.append(_handoff_status(ok=True, count=len(handoff_events))) except Exception as exc: # fail soft: a fetch/parse error degrades this source only - statuses.append( - SourceStatus(SOURCE_GITEA_HANDOFF, ok=False, reason=f"handoff source failed: {exc}") - ) + statuses.append(_handoff_status(ok=False, reason=f"handoff source failed: {exc}")) + + requested = tuple( + name + for name, value in ((FILTER_ISSUE, issue), (FILTER_PR, pr), (FILTER_SESSION, session)) + if value is not None + ) + statuses = [ + replace( + status, + unsupported_filters=tuple( + dim for dim in requested if dim not in status.supported_filters + ), + ) + for status in statuses + ] + filters = {"issue": issue, "pr": pr, "session": session} + + # A dimension is answerable only if a source that actually ran can carry it. + # If none can, refuse: an empty page would assert "no such activity", which + # is a claim this timeline is not in a position to make. + answerable: set[str] = set() + for status in statuses: + if status.ok: + answerable.update(status.supported_filters) + unanswerable = tuple(dim for dim in requested if dim not in answerable) + + if unanswerable: + return TimelineSnapshot( + schema_version=TIMELINE_SCHEMA_VERSION, + remote=remote, + org=org, + repo=repo, + filters=filters, + page=paginate([], limit=limit, offset=offset), + sources=tuple(statuses), + ok=False, + error={ + "code": "filter_not_supported", + "unsupported_filters": list(unanswerable), + "detail": ( + "no timeline source that ran can answer " + + ", ".join(f"'{dim}'" for dim in unanswerable) + + "; the result is refused rather than returned empty" + ), + "sources": _unanswerable_reasons(statuses, unanswerable), + }, + ) filtered = filter_events(all_events, issue=issue, pr=pr, session=session) ordered = sort_events(filtered) @@ -529,7 +774,7 @@ def load_timeline( remote=remote, org=org, repo=repo, - filters={"issue": issue, "pr": pr, "session": session}, + filters=filters, page=page, sources=tuple(statuses), )