sysadminandClaude Opus 4.8
25bc2a3291
Phase 1 child of the Web Console epic #631. Adds a durable, versioned WorkflowEvent schema with per-source adapters and a read-only query API so operators can browse a unified timeline of workflow events, decisions, tool calls, and handoffs instead of scattered evidence. - webui/timeline.py (new): versioned WorkflowEvent schema; control-plane event adapter and Gitea CTH handoff-comment adapter; read-only mode=ro control-plane reader; conjunctive filter by issue/PR/session; stable (timestamp, source_rank, event_key) ordering; bounded pagination; fail-soft per-source status; redaction at the boundary, fail closed. - webui/app.py: GET /api/v1/timeline read-only route with thread-scoped, fail-soft handoff comment source. - tests/test_webui_timeline.py (new): schema, adapters, redaction of secret-like payloads, filter/sort/pagination, scoped CP reader, fail-soft composition, and API integration. - docs/webui-local-dev.md: timeline route and field-authority notes. Read-only Phase 1; no mutation of historical events; no full chat replay; no unredacted tool-argument storage. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
540 lines
19 KiB
Python
540 lines
19 KiB
Python
"""Workflow-event and conversation timeline model (#637, Phase 1).
|
|
|
|
Operators cannot browse a unified timeline of workflow events, decisions,
|
|
tool calls, and handoffs: the evidence is scattered across control-plane
|
|
events, Gitea canonical handoff comments, and local logs. This module defines
|
|
one durable, versioned event schema and per-source adapters that normalise
|
|
those scattered records into a single ``WorkflowEvent`` stream, plus a
|
|
read-only query layer (filter by issue / PR / session, stable ordering,
|
|
pagination) that the ``/api/v1/timeline`` route serves.
|
|
|
|
Design rules honoured here:
|
|
|
|
- **Read-only.** Sources are read; nothing is mutated. The control-plane
|
|
database is opened through a ``mode=ro`` URI so a missing or unwritable DB
|
|
degrades to a reason instead of creating directories or running migrations.
|
|
- **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.
|
|
- **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.
|
|
- **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.
|
|
|
|
Non-goals (from the issue): no full chat replay, no mutation of historical
|
|
events, no unredacted tool-argument storage.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, Iterable
|
|
|
|
import control_plane_db
|
|
from webui import console_redaction
|
|
|
|
# The schema is versioned so consumers can branch on shape. Bump on any
|
|
# breaking change to WorkflowEvent's serialized form.
|
|
TIMELINE_SCHEMA_VERSION = 1
|
|
|
|
# Known event sources and their deterministic ordering rank. When two events
|
|
# carry the same timestamp, the source rank breaks the tie before the
|
|
# per-source event key, so a control-plane event and a handoff comment minted
|
|
# in the same second always sort in a fixed order.
|
|
SOURCE_CONTROL_PLANE = "control_plane"
|
|
SOURCE_GITEA_HANDOFF = "gitea_handoff"
|
|
_SOURCE_RANK = {
|
|
SOURCE_CONTROL_PLANE: 0,
|
|
SOURCE_GITEA_HANDOFF: 1,
|
|
}
|
|
|
|
# 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"
|
|
|
|
|
|
def _parse_ts(value: str | None) -> str | None:
|
|
"""Normalise a timestamp to ``...Z`` UTC, or None when unparseable."""
|
|
if not value:
|
|
return None
|
|
text = str(value).strip()
|
|
if not text:
|
|
return None
|
|
candidate = text[:-1] + "+00:00" if text.endswith("Z") else text
|
|
try:
|
|
parsed = datetime.fromisoformat(candidate)
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def _redact(value: Any) -> Any:
|
|
"""Redact a single free-text field, failing closed to the placeholder."""
|
|
if value is None:
|
|
return None
|
|
return console_redaction.redact_text(str(value))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkflowEvent:
|
|
"""One normalised timeline event.
|
|
|
|
Every field is optional except ``source``/``event_type``/``event_key``
|
|
because sources carry different subsets. The class is frozen so an adapted
|
|
event is an immutable record; a consumer that needs a variant builds a new
|
|
one rather than mutating history.
|
|
"""
|
|
|
|
source: str
|
|
event_type: str
|
|
event_key: str
|
|
timestamp: str | None = None
|
|
actor: str | None = None
|
|
role: str | None = None
|
|
issue_number: int | None = None
|
|
pr_number: int | None = None
|
|
session_id: str | None = None
|
|
tool_name: str | None = None
|
|
decision: str | None = None
|
|
message: str | None = None
|
|
correlation_id: str | None = None
|
|
evidence_refs: tuple[str, ...] = ()
|
|
sensitive: bool = False
|
|
|
|
def sort_key(self) -> tuple[str, int, str]:
|
|
return (
|
|
self.timestamp or _MISSING_TS_SORT,
|
|
_SOURCE_RANK.get(self.source, 99),
|
|
self.event_key,
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"source": self.source,
|
|
"event_type": self.event_type,
|
|
"event_key": self.event_key,
|
|
"timestamp": self.timestamp,
|
|
"actor": self.actor,
|
|
"role": self.role,
|
|
"issue_number": self.issue_number,
|
|
"pr_number": self.pr_number,
|
|
"session_id": self.session_id,
|
|
"tool_name": self.tool_name,
|
|
"decision": self.decision,
|
|
"message": self.message,
|
|
"correlation_id": self.correlation_id,
|
|
"evidence_refs": list(self.evidence_refs),
|
|
"sensitive": self.sensitive,
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Adapters — pure functions from a source's raw records to WorkflowEvents. #
|
|
# Each is total: a malformed record is skipped, never raised on. #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
# Event types whose payload is treated as sensitive and always redaction-hard
|
|
# (they can carry lease/session provenance or tool arguments).
|
|
_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")
|
|
|
|
|
|
def _kind_to_numbers(kind: str | None, number: int | None) -> tuple[int | None, int | None]:
|
|
"""Map a control-plane work-item (kind, number) to (issue_no, pr_no)."""
|
|
if number is None:
|
|
return (None, None)
|
|
if kind == "pr":
|
|
return (None, int(number))
|
|
if kind == "issue":
|
|
return (int(number), None)
|
|
return (None, None)
|
|
|
|
|
|
def _correlation_for(kind: str | None, number: int | None) -> str | None:
|
|
if number is None or kind not in ("issue", "pr"):
|
|
return None
|
|
return f"{kind}#{number}"
|
|
|
|
|
|
def _extract_evidence_refs(*texts: str | None) -> tuple[str, ...]:
|
|
refs: list[str] = []
|
|
for text in texts:
|
|
if not text:
|
|
continue
|
|
for match in _EVIDENCE_REF_RE.finditer(text):
|
|
token = f"#{match.group(1)}"
|
|
if token not in refs:
|
|
refs.append(token)
|
|
for match in _SHA_RE.finditer(text):
|
|
token = match.group(0)
|
|
if token not in refs:
|
|
refs.append(token)
|
|
return tuple(refs)
|
|
|
|
|
|
def adapt_cp_events(rows: Iterable[dict[str, Any]]) -> list[WorkflowEvent]:
|
|
"""Adapt control-plane ``events`` rows (joined to work_items) into events.
|
|
|
|
Each row is expected to carry ``event_id``, ``event_type``, ``message``,
|
|
``created_at`` and the joined work-item ``kind``/``number``. Rows missing
|
|
an id or type are skipped so a partially written table never raises.
|
|
"""
|
|
events: list[WorkflowEvent] = []
|
|
for row in rows or []:
|
|
try:
|
|
event_id = row.get("event_id")
|
|
event_type = (row.get("event_type") or "").strip()
|
|
if event_id is None or not event_type:
|
|
continue
|
|
kind = row.get("kind")
|
|
number = row.get("number")
|
|
issue_no, pr_no = _kind_to_numbers(kind, number)
|
|
sensitive = any(hint in event_type.lower() for hint in _SENSITIVE_EVENT_HINTS)
|
|
events.append(
|
|
WorkflowEvent(
|
|
source=SOURCE_CONTROL_PLANE,
|
|
event_type=event_type,
|
|
event_key=f"cp:{event_id}",
|
|
timestamp=_parse_ts(row.get("created_at")),
|
|
issue_number=issue_no,
|
|
pr_number=pr_no,
|
|
session_id=(row.get("session_id") or None),
|
|
message=_redact(row.get("message")),
|
|
correlation_id=_correlation_for(kind, number),
|
|
sensitive=sensitive,
|
|
)
|
|
)
|
|
except Exception:
|
|
# A single malformed row must not sink the whole adaptation.
|
|
continue
|
|
return events
|
|
|
|
|
|
def adapt_cth_comments(
|
|
comments: Iterable[dict[str, Any]],
|
|
*,
|
|
kind: str,
|
|
number: int,
|
|
) -> list[WorkflowEvent]:
|
|
"""Adapt Gitea Canonical Thread Handoff (CTH) comments into events.
|
|
|
|
Only comments that parse as a CTH (``canonical_thread_handoff.parse_cth_comment``)
|
|
become events; ordinary comments are ignored. ``kind``/``number`` scope the
|
|
events to the issue or PR the comments belong to.
|
|
"""
|
|
# Imported lazily so this module has no import-time dependency on the
|
|
# handoff parser when only the control-plane adapter is used.
|
|
from canonical_thread_handoff import parse_cth_comment
|
|
|
|
issue_no, pr_no = _kind_to_numbers(kind, number)
|
|
correlation = _correlation_for(kind, number)
|
|
events: list[WorkflowEvent] = []
|
|
for comment in comments or []:
|
|
try:
|
|
body = comment.get("body") or ""
|
|
parsed = parse_cth_comment(body)
|
|
if not parsed:
|
|
continue
|
|
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")
|
|
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,
|
|
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")),
|
|
correlation_id=correlation,
|
|
evidence_refs=_extract_evidence_refs(proof, decision),
|
|
sensitive=False,
|
|
)
|
|
)
|
|
except Exception:
|
|
continue
|
|
return events
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Read-only control-plane event source. #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
_CP_EVENTS_QUERY = """
|
|
SELECT e.event_id AS event_id,
|
|
e.event_type AS event_type,
|
|
e.message AS message,
|
|
e.created_at AS created_at,
|
|
w.kind AS kind,
|
|
w.number AS number
|
|
FROM events e
|
|
JOIN work_items w ON e.work_item_id = w.work_item_id
|
|
WHERE w.remote = ? AND w.org = ? AND w.repo = ?
|
|
"""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SourceStatus:
|
|
"""Fail-soft status for one timeline source."""
|
|
|
|
name: str
|
|
ok: bool
|
|
reason: str | None = None
|
|
count: int = 0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {"name": self.name, "ok": self.ok, "reason": self.reason, "count": self.count}
|
|
|
|
|
|
def read_cp_events(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
db_path: str | None = None,
|
|
) -> tuple[list[WorkflowEvent], SourceStatus]:
|
|
"""Read scoped control-plane events read-only. Never creates the DB.
|
|
|
|
Opens the SQLite file through a ``mode=ro`` URI: a health/timeline read
|
|
must never create directories or run the schema migration that
|
|
``ControlPlaneDB()`` performs on construction. A missing or unreadable DB
|
|
degrades to a status with a reason.
|
|
"""
|
|
path = (db_path or control_plane_db.default_db_path()).strip()
|
|
conn: sqlite3.Connection | None = None
|
|
try:
|
|
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
|
|
conn.row_factory = sqlite3.Row
|
|
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}"))
|
|
except sqlite3.Error as exc:
|
|
return ([], SourceStatus(SOURCE_CONTROL_PLANE, 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)))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Filter, sort, paginate. #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def filter_events(
|
|
events: Iterable[WorkflowEvent],
|
|
*,
|
|
issue: int | None = None,
|
|
pr: int | None = None,
|
|
session: str | None = None,
|
|
) -> list[WorkflowEvent]:
|
|
"""Filter events by issue number, PR number, and/or session id.
|
|
|
|
Filters are conjunctive. A filter that names a dimension an event does not
|
|
carry excludes that event (an issue filter excludes PR-only events).
|
|
"""
|
|
out: list[WorkflowEvent] = []
|
|
for ev in events:
|
|
if issue is not None and ev.issue_number != issue:
|
|
continue
|
|
if pr is not None and ev.pr_number != pr:
|
|
continue
|
|
if session is not None and ev.session_id != session:
|
|
continue
|
|
out.append(ev)
|
|
return out
|
|
|
|
|
|
def sort_events(events: Iterable[WorkflowEvent]) -> list[WorkflowEvent]:
|
|
"""Return events in stable timeline order (ascending)."""
|
|
return sorted(events, key=lambda ev: ev.sort_key())
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TimelinePage:
|
|
"""One page of the sorted, filtered timeline."""
|
|
|
|
events: tuple[WorkflowEvent, ...]
|
|
total: int
|
|
limit: int
|
|
offset: int
|
|
|
|
@property
|
|
def next_offset(self) -> int | None:
|
|
nxt = self.offset + len(self.events)
|
|
return nxt if nxt < self.total else None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"events": [ev.to_dict() for ev in self.events],
|
|
"pagination": {
|
|
"total": self.total,
|
|
"limit": self.limit,
|
|
"offset": self.offset,
|
|
"returned": len(self.events),
|
|
"next_offset": self.next_offset,
|
|
"has_more": self.next_offset is not None,
|
|
},
|
|
}
|
|
|
|
|
|
_MAX_LIMIT = 500
|
|
_DEFAULT_LIMIT = 50
|
|
|
|
|
|
def _coerce_bounds(limit: int | None, offset: int | None) -> tuple[int, int]:
|
|
try:
|
|
lim = int(limit) if limit is not None else _DEFAULT_LIMIT
|
|
except (TypeError, ValueError):
|
|
lim = _DEFAULT_LIMIT
|
|
try:
|
|
off = int(offset) if offset is not None else 0
|
|
except (TypeError, ValueError):
|
|
off = 0
|
|
lim = max(1, min(lim, _MAX_LIMIT))
|
|
off = max(0, off)
|
|
return (lim, off)
|
|
|
|
|
|
def paginate(events: list[WorkflowEvent], *, limit: int | None, offset: int | None) -> TimelinePage:
|
|
lim, off = _coerce_bounds(limit, offset)
|
|
window = events[off : off + lim]
|
|
return TimelinePage(events=tuple(window), total=len(events), limit=lim, offset=off)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Composition — load_timeline aggregates all sources, fail-soft. #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
# A comment source is a callable that, given (kind, number), returns the raw
|
|
# Gitea comment list for that issue/PR. The route supplies a live fail-soft
|
|
# fetcher; tests supply a fixture. When None, the handoff source is reported as
|
|
# not-run (never silently empty-and-healthy).
|
|
CommentSource = Callable[[str, int], list[dict[str, Any]]]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TimelineSnapshot:
|
|
schema_version: int
|
|
remote: str
|
|
org: str
|
|
repo: str
|
|
filters: dict[str, Any]
|
|
page: TimelinePage
|
|
sources: tuple[SourceStatus, ...]
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": self.schema_version,
|
|
"scope": {"remote": self.remote, "org": self.org, "repo": self.repo},
|
|
"filters": self.filters,
|
|
"sources": [s.to_dict() for s in self.sources],
|
|
**self.page.to_dict(),
|
|
}
|
|
|
|
|
|
def load_timeline(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue: int | None = None,
|
|
pr: int | None = None,
|
|
session: str | None = None,
|
|
limit: int | None = None,
|
|
offset: int | None = None,
|
|
db_path: str | None = None,
|
|
comment_source: CommentSource | None = None,
|
|
) -> TimelineSnapshot:
|
|
"""Aggregate every timeline source into one filtered, paginated snapshot.
|
|
|
|
Sources are read independently and fail soft: an unavailable source
|
|
contributes a ``SourceStatus`` with ``ok=False`` and a reason, and never
|
|
collapses the whole timeline. The handoff source only runs when a specific
|
|
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.
|
|
"""
|
|
all_events: list[WorkflowEvent] = []
|
|
statuses: list[SourceStatus] = []
|
|
|
|
cp_events, cp_status = read_cp_events(remote=remote, org=org, repo=repo, db_path=db_path)
|
|
all_events.extend(cp_events)
|
|
statuses.append(cp_status)
|
|
|
|
# Gitea handoff comments are thread-scoped: only fetch when the caller
|
|
# narrowed to one issue or PR, and only when a source was provided.
|
|
handoff_target: tuple[str, int] | None = None
|
|
if pr is not None:
|
|
handoff_target = ("pr", pr)
|
|
elif issue is not None:
|
|
handoff_target = ("issue", issue)
|
|
|
|
if handoff_target is None:
|
|
statuses.append(
|
|
SourceStatus(
|
|
SOURCE_GITEA_HANDOFF,
|
|
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,
|
|
ok=False,
|
|
reason="not run: no comment source configured for this timeline read",
|
|
)
|
|
)
|
|
else:
|
|
kind, number = handoff_target
|
|
try:
|
|
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)))
|
|
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}")
|
|
)
|
|
|
|
filtered = filter_events(all_events, issue=issue, pr=pr, session=session)
|
|
ordered = sort_events(filtered)
|
|
page = paginate(ordered, limit=limit, offset=offset)
|
|
|
|
return TimelineSnapshot(
|
|
schema_version=TIMELINE_SCHEMA_VERSION,
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
filters={"issue": issue, "pr": pr, "session": session},
|
|
page=page,
|
|
sources=tuple(statuses),
|
|
)
|
|
|
|
|
|
def snapshot_to_dict(snapshot: TimelineSnapshot) -> dict[str, Any]:
|
|
return snapshot.to_dict()
|