feat(timeline): kb.timeline — entity chronological trajectory (#313)

CHANGELOG.md

@@ -49,6 +49,16 @@ All notable changes to vouch are documented here. Format follows
 ## [1.1.0] — 2026-07-03
 
 ### Added
+- `kb.timeline` / `vouch timeline <entity>` — a read-only chronological
+  trajectory of an entity's approved claims and relations (#313). orders them
+  along a time axis oldest-first: `--order effective` uses artifact `created_at`,
+  `--order decided` recovers approval time from the audit log. `--since` /
+  `--until` / `--types` / `--limit` filters; `--json` for the machine shape.
+  superseded/archived claims still appear, flagged by current status; relations
+  carry `status = null`; pending proposals never appear. pure read — no write
+  path is reachable from it. registered at all four surfaces (mcp/jsonl/
+  capabilities/cli) and attaches `_meta.vouch_salience` when a `session_id` is
+  passed.
 - auto-capture: claude code sessions are harvested via hooks and filed as a
   single pending session-summary proposal for human approval. a `PostToolUse`
   hook (`vouch capture observe`) appends compact tool-use observations to an

src/vouch/capabilities.py

@@ -21,6 +21,7 @@
     "kb.digest",
     "kb.search",
     "kb.neighbors",
+    "kb.timeline",
     "kb.context",
     "kb.synthesize",
     "kb.read_page",

src/vouch/cli.py

@@ -39,6 +39,7 @@
 from . import stats as stats_mod
 from . import sync as sync_mod
 from . import synthesize as synth
+from . import timeline as timeline_mod
 from . import trust as trust_mod
 from . import vault_sync as vault_sync_mod
 from . import verify as verify_mod
@@ -790,6 +791,87 @@ def pages_cmd(
         click.echo("no matching pages")
 
 
+# --- timeline -------------------------------------------------------------
+
+
+@cli.command()
+@click.argument("entity_id")
+@click.option(
+    "--order",
+    type=click.Choice(["effective", "decided"]),
+    default="effective",
+    show_default=True,
+    help="Time axis: `effective` (artifact created_at) or `decided` (approval "
+    "time from the audit log).",
+)
+@click.option(
+    "--since", default=None, help="Lower bound (iso date/datetime or a duration like 30d)."
+)
+@click.option("--until", default=None, help="Upper bound (same formats as --since).")
+@click.option(
+    "--types",
+    default=None,
+    help="Comma-separated filter on claim type (fact,decision,…), relation type, "
+    "or the literal `claim` / `relation`.",
+)
+@click.option("--limit", default=None, type=int, help="Keep only the most recent N entries.")
+@click.option("--json", "as_json", is_flag=True, help="Emit the machine shape instead of a table.")
+def timeline(
+    entity_id: str,
+    order: str,
+    since: str | None,
+    until: str | None,
+    types: str | None,
+    limit: int | None,
+    as_json: bool,
+) -> None:
+    """Chronological trajectory of an entity's claims and relations (#313).
+
+    \b
+    Orders an entity's approved claims + relations along a time axis, oldest
+    first. Read-only — pending proposals never appear.
+
+    \b
+    Examples:
+      vouch timeline person:alice-example
+      vouch timeline repo:acme --order decided --since 2026-01-01
+      vouch timeline concept:auth --types decision,fact --json
+    """
+    store = _load_store()
+    type_list = [t.strip() for t in types.split(",")] if types else None
+    try:
+        result = timeline_mod.build_timeline(
+            store,
+            entity_id,
+            since=metrics_mod.parse_since(since),
+            until=metrics_mod.parse_since(until),
+            order=order,
+            types=type_list,
+            limit=limit,
+        )
+    except timeline_mod.TimelineError as e:
+        raise click.ClickException(str(e)) from e
+    except (ArtifactNotFoundError, KeyError) as e:
+        raise click.ClickException(f"entity not found: {entity_id}") from e
+
+    if as_json:
+        _emit_json(result)
+        return
+
+    ent = result["entity"]
+    click.echo(f"timeline: {ent['name']}  [{ent['type']}]  ({ent['id']})")
+    click.echo(f"  order={result['order']}  {result['count']} of {result['total']} entries")
+    click.echo("")
+    if not result["entries"]:
+        click.echo("  (no claims or relations for this entity in range)")
+        return
+    for entry in result["entries"]:
+        when = (entry["when"] or "—")[:19]
+        status = f"  <{entry['status']}>" if entry["status"] else ""
+        click.echo(f"  {when}  {entry['kind']:<8} {entry['id']}{status}")
+        click.echo(f"    {entry['summary'][:100]}")
+
+
 # --- proposals ------------------------------------------------------------
 
 

src/vouch/jsonl_server.py

@@ -195,6 +195,26 @@ def _h_neighbors(p: dict) -> dict:
     )
 
 
+def _h_timeline(p: dict) -> dict:
+    from .metrics import parse_since
+    from .timeline import build_timeline
+
+    store = _store()
+    cfg = _load_cfg(store)
+    session_id = p.get("session_id")
+    limit = p.get("limit")
+    result = build_timeline(
+        store,
+        p["entity_id"],
+        since=parse_since(p.get("since")),
+        until=parse_since(p.get("until")),
+        order=p.get("order", "effective"),
+        types=p.get("types"),
+        limit=int(limit) if limit is not None else None,
+    )
+    return salience_mod.attach_salience(result, store, session_id, cfg)
+
+
 def _h_context(p: dict) -> dict:
     store = _store()
     query = p["task"]
@@ -700,6 +720,7 @@ def _h_propose_theme(p: dict) -> dict:
     "kb.digest": _h_digest,
     "kb.search": _h_search,
     "kb.neighbors": _h_neighbors,
+    "kb.timeline": _h_timeline,
     "kb.context": _h_context,
     "kb.synthesize": _h_synthesize,
     "kb.read_page": _h_read_page,

src/vouch/server.py

@@ -226,6 +226,39 @@ def kb_neighbors(
         raise ValueError(str(e)) from e
 
 
+@mcp.tool()
+def kb_timeline(
+    entity_id: str,
+    order: str = "effective",
+    since: str | None = None,
+    until: str | None = None,
+    types: list[str] | None = None,
+    limit: int | None = None,
+    session_id: str | None = None,
+) -> dict[str, Any]:
+    """Chronological trajectory of an entity's approved claims and relations.
+
+    ``order`` is ``effective`` (artifact ``created_at``) or ``decided`` (approval
+    time recovered from the audit log). Read-only; pending proposals never
+    appear. Attaches ``_meta.vouch_salience`` when a ``session_id`` is given.
+    """
+    from .metrics import parse_since
+    from .timeline import TimelineError, build_timeline
+
+    store = _store()
+    try:
+        result = build_timeline(
+            store, entity_id,
+            since=parse_since(since), until=parse_since(until),
+            order=order, types=types, limit=limit,
+        )
+    except ArtifactNotFoundError as e:
+        raise ValueError(str(e)) from e
+    except TimelineError as e:
+        raise ValueError(str(e)) from e
+    return salience_mod.attach_salience(result, store, session_id, _load_cfg(store))
+
+
 @mcp.tool()
 def kb_context(
     task: str,

src/vouch/timeline.py

@@ -0,0 +1,172 @@
+"""``kb.timeline`` — chronological trajectory of an entity (vouchdev/vouch#313).
+
+``kb.read_entity`` returns an entity and its claims as a flat set; ``kb.neighbors``
+expands graph adjacency at a point in time. Neither answers "what did the KB
+learn about this entity, *in what order*?" — a trajectory. The raw material is
+already on disk: ``Claim`` and ``Relation`` carry ``created_at``, and decision
+time is recoverable from the append-only ``audit.log.jsonl``. This orders an
+entity's approved claims and relations along that time axis.
+
+Pure read. It never proposes, approves, or writes — a viewport over
+already-reviewed artifacts, exactly like ``read_entity`` / ``neighbors``. Only
+*approved* durable artifacts are read (``list_claims`` / ``list_relations``);
+pending proposals never appear. All ordering logic lives here, not in
+``storage.py``, which stays pure I/O.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from .audit import read_events
+from .metrics import _APPROVE_RE
+from .models import Claim, Relation
+from .storage import KBStore
+
+# Ordering axes. ``effective`` orders by the artifact's own ``created_at`` (when
+# the fact entered the KB); ``decided`` orders by the moment the proposal that
+# produced it was approved, recovered from the audit log.
+ORDER_EFFECTIVE = "effective"
+ORDER_DECIDED = "decided"
+ORDERS = (ORDER_EFFECTIVE, ORDER_DECIDED)
+
+
+class TimelineError(ValueError):
+    """User-visible bad input (e.g. an unknown ``order``)."""
+
+
+def _as_utc(dt: datetime | None) -> datetime | None:
+    if dt is None:
+        return None
+    return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC)
+
+
+def _iso(dt: datetime | None) -> str | None:
+    return dt.isoformat() if dt is not None else None
+
+
+def _decided_map(store: KBStore) -> dict[str, datetime]:
+    """artifact id -> approval time, from the authoritative audit stream.
+
+    ``proposals.approve`` logs an approve event with ``object_ids = [proposal_id,
+    result_id]``; the event's ``created_at`` is the decision time. Artifacts
+    written outside the proposal path (rare, e.g. a seeded starter KB) simply
+    have no entry and fall back to their ``created_at``.
+    """
+    out: dict[str, datetime] = {}
+    for ev in read_events(store.kb_dir):
+        if _APPROVE_RE.match(ev.event) and len(ev.object_ids) >= 2:
+            ts = _as_utc(ev.created_at)
+            if ts is not None:
+                out[ev.object_ids[1]] = ts
+    return out
+
+
+def _claim_summary(c: Claim) -> str:
+    return str(c.text).strip()[:120] or "—"
+
+
+def _relation_summary(r: Relation) -> str:
+    return f"{r.source} {r.relation.value} {r.target}"
+
+
+def build_timeline(
+    store: KBStore,
+    entity_id: str,
+    *,
+    since: datetime | None = None,
+    until: datetime | None = None,
+    order: str = ORDER_EFFECTIVE,
+    types: list[str] | None = None,
+    limit: int | None = None,
+) -> dict[str, Any]:
+    """Order an entity's approved claims + relations along a time axis.
+
+    Entries are ``{when, kind, id, summary, status}``, oldest first
+    (most-recent-last). ``status`` is the claim's current :class:`ClaimStatus`
+    (a superseded/archived claim still appears, flagged); relations have no
+    status, so it is ``None``. When ``limit`` is set, the most recent ``limit``
+    entries are kept, still in chronological order. Raises
+    :class:`~vouch.storage.ArtifactNotFoundError` if the entity does not exist.
+    """
+    if order not in ORDERS:
+        raise TimelineError(f"order must be one of {ORDERS}, got {order!r}")
+    if limit is not None and limit < 0:
+        raise TimelineError("limit must be >= 0")
+    since = _as_utc(since)
+    until = _as_utc(until)
+
+    entity = store.get_entity(entity_id)  # raises ArtifactNotFoundError
+
+    want = {t.strip() for t in types if t.strip()} if types else None
+    decided = _decided_map(store) if order == ORDER_DECIDED else {}
+
+    def when_for(artifact_id: str, created: datetime | None) -> datetime | None:
+        eff = _as_utc(created)
+        if order == ORDER_DECIDED:
+            return decided.get(artifact_id) or eff
+        return eff
+
+    rows: list[dict[str, Any]] = []
+
+    for c in store.list_claims():
+        if entity_id not in c.entities:
+            continue
+        # type filter: a claim matches on its ClaimType, or the generic "claim".
+        if want is not None and c.type.value not in want and "claim" not in want:
+            continue
+        rows.append(
+            {
+                "_when": when_for(c.id, c.created_at),
+                "kind": "claim",
+                "id": c.id,
+                "summary": _claim_summary(c),
+                "status": c.status.value,
+            }
+        )
+
+    for r in store.list_relations():
+        if entity_id not in (r.source, r.target):
+            continue
+        # type filter: a relation matches on its RelationType, or "relation".
+        if want is not None and r.relation.value not in want and "relation" not in want:
+            continue
+        rows.append(
+            {
+                "_when": when_for(r.id, r.created_at),
+                "kind": "relation",
+                "id": r.id,
+                "summary": _relation_summary(r),
+                "status": None,
+            }
+        )
+
+    # window filter on the chosen timestamp
+    if since is not None:
+        rows = [e for e in rows if e["_when"] is not None and e["_when"] >= since]
+    if until is not None:
+        rows = [e for e in rows if e["_when"] is not None and e["_when"] <= until]
+
+    # oldest first; id tie-break for deterministic output on identical stamps.
+    # entries with no recoverable timestamp sort to the front (epoch).
+    _epoch = datetime(1970, 1, 1, tzinfo=UTC)
+    rows.sort(key=lambda e: (e["_when"] or _epoch, e["id"]))
+
+    total = len(rows)
+    if limit is not None and limit < total:
+        rows = rows[total - limit :]  # keep the most recent `limit`, still chronological
+
+    entries = [
+        {"when": _iso(e["_when"]), "kind": e["kind"], "id": e["id"],
+         "summary": e["summary"], "status": e["status"]}
+        for e in rows
+    ]
+
+    return {
+        "entity": {"id": entity.id, "name": entity.name, "type": entity.type.value},
+        "order": order,
+        "count": len(entries),
+        "total": total,
+        "entries": entries,
+    }