feat(observability): T1.5 call-site wiring + per-run metrics.jsonl + runbook §4.5 (REQ-INB-20 + REQ-INB-23)

T2.2 closes the call-site wiring for the T1.5 Papra live-write gate
and lands the per-run metrics.jsonl surface (REQ-INB-23). The
T1.5 + T2.2 work is folded into one commit so the two pipeline
files are touched by exactly one Wave 2 task.

pipelines/common.py:
- LLM_METRICS_ENABLED env-var gate around the
  llm_call_seconds_sum / llm_call_count accumulators in
  _ollama_call_one_model. The accumulators are stored in
  os.environ so the main() functions can read them when emitting
  the per-run metrics line. Default is 'true' (gate enabled).

pipelines/live_pipeline.py:
- Import emit_papra_write_gate_event and emit_metric.
- process_one_message: emit the papra_write_gate event at the
  start of the papra branch (REQ-INB-20 call-site).
- main: emit one metrics line with the COUNTER_KEYS / GAUGE_KEYS
  vocabulary from pipelines/_metrics/emit.py, including
  llm_call_seconds_sum and llm_call_count read from the
  os.environ accumulators.

pipelines/historical_pipeline.py:
- Import emit_papra_write_gate_event and emit_metric.
- process_one_message: emit the papra_write_gate event at the
  start of the body-decision branch.
- main: emit one metrics line with the same vocabulary.

pipelines/_metrics/emit.py (new):
- emit_metric(run_id, pipeline, *, counters, gauges) appends one
  JSONL line to <pipeline>-metrics-<run_id>.jsonl. The
  COUNTER_KEYS and GAUGE_KEYS module constants pin the vocabulary
  so the test can assert exact equality.

pipelines/_metrics/__init__.py (new):
- Empty / near-empty package marker.

pipelines/_tests/test_metrics_offline.py (new, 2 tests):
- test_emit_metric_creates_file_with_one_line_per_call: asserts
  the line shape, the exact COUNTER_KEYS / GAUGE_KEYS, and the
  per-run file naming.
- test_emit_metric_does_not_overwrite_existing_file: asserts two
  calls in a row append, never truncate.

docs/operations-runbook.md:
- New §4.5 'Metrics surface' (after §4.4 'SeaweedFS bucket
  layout' at line 238). Documents the file location, the counter
  and gauge vocabulary, the tail-and-summarize procedure, the
  opt-out via LLM_METRICS_ENABLED, and a forward-pointer noting
  alerting is a v1.2 candidate (out of scope for v1.1).

---ci---
project: inboxos
phase: 1
milestone: v1.1
status: execute
---/ci---

docs/operations-runbook.md

@@ -235,6 +235,64 @@ s3://inboxos/
 
 The state files are tiny JSON blobs written via `s3_put_bytes` after every successful message (live) or every `BATCH_SIZE` messages (historical). They are the **only** persistent state between workflow runs.
 
+### 4.5 Metrics surface (REQ-INB-23)
+
+Each pipeline run writes **one** JSONL line to
+`pipelines/_artifacts/<pipeline>-metrics-<run_id>.jsonl` (one
+file per run; never overwritten). The line shape and counter /
+gauge vocabulary are documented in
+`pipelines/_metrics/emit.py` (a copy-paste of the docstring is
+in this runbook for operator reference).
+
+| Counter key                | Type | What it tells you                                          |
+|----------------------------|------|------------------------------------------------------------|
+| `messages_processed`       | int  | Total messages processed in this run.                      |
+| `messages_dead_lettered`   | int  | Subset routed to the dead-letter prefix.                    |
+| `noco_upsert_ok`           | int  | NocoDB upserts that returned 2xx.                          |
+| `noco_upsert_4xx`          | int  | 4xx (drift; permanent failure for the affected message).    |
+| `noco_upsert_5xx`          | int  | 5xx (transient; retried by tenacity).                     |
+| `papra_would_write`        | int  | Papra decisions that resolved `would_write=true`.          |
+| `papra_write_ok`           | int  | Papra HTTP POSTs that returned 2xx.                        |
+| `papra_write_failed`       | int  | Papra HTTP POSTs that returned >= 400.                     |
+| `papra_index_upsert_ok`    | int  | `papra_index` audit-trail rows written.                    |
+| `papra_index_upsert_failed`| int  | `papra_index` upserts that raised.                         |
+| `llm_call_seconds_sum`     | int  | Total wall-clock ms spent in `requests.post` to Ollama.   |
+| `llm_call_count`           | int  | Number of LLM HTTP POSTs in the run.                       |
+
+| Gauge key                | Type           | What it tells you                              |
+|--------------------------|----------------|-------------------------------------------------|
+| `last_run_at`            | ISO 8601 str   | The run's `now_iso()` at metrics emission.     |
+| `dead_letter_total`      | int            | Cumulative dead-letter prefix size (best-effort). |
+| `papra_dry_run`          | "true"/"false" | The effective `PAPRA_DRY_RUN` after precedence. |
+| `papra_dry_run_override` | "true"/"false"/"none" | The parsed override (or "none" if unset). |
+
+**Tail-and-summarize procedure.** To compute a daily total of
+`messages_processed` across all live runs:
+
+```bash
+cat pipelines/_artifacts/live-metrics-*.jsonl \
+    | jq -s '[.[].counters.messages_processed] | add'
+```
+
+To inspect the last 20 metrics lines in time order:
+
+```bash
+cat pipelines/_artifacts/live-metrics-*.jsonl \
+    | tail -20 \
+    | jq -c '{ts, run_id, counters}'
+```
+
+**Forward-pointer.** Alerting is **out of scope** for v1.1; the
+runbook does not include alert rules (this is a v1.2 candidate
+per `.ciagent/inboxos/ideation/v1.1-candidates.md` Tier 2 T2-D).
+The metrics file is the substrate; the alert layer is the next
+v1.2 deliverable.
+
+**Opt-out.** Set `LLM_METRICS_ENABLED=false` in the pipeline env
+to skip the `time.monotonic()` wrapping around the LLM HTTP POST.
+The counters default to enabled; opt-out is for canary runs that
+want to measure pure LLM latency without the wrapper overhead.
+
 ---
 
 ## 5. Common failure modes and recovery

pipelines/_metrics/__init__.py

@@ -0,0 +1,9 @@
+"""pipelines._metrics — per-run structured metrics for the InboxOS pipelines.
+
+Sister package to pipelines._tests. The metrics emitter is a thin
+wrapper that appends one JSON line per run to
+pipelines/_artifacts/<pipeline>-metrics-<run_id>.jsonl.
+
+See pipelines/_metrics/emit.py for the public API and the line
+shape.
+"""

pipelines/_metrics/emit.py

@@ -0,0 +1,114 @@
+"""Per-run metrics emitter for the InboxOS pipelines (REQ-INB-23).
+
+Public API
+----------
+`emit_metric(run_id, pipeline, *, counters, gauges)` — append one
+JSON line to `pipelines/_artifacts/<pipeline>-metrics-<run_id>.jsonl`.
+The line shape is:
+
+    {
+      "ts": "<iso>",
+      "run_id": "<uuid>",
+      "pipeline": "live" | "historical",
+      "counters": {<int dict>},
+      "gauges": {<any-of-str-int-float dict>}
+    }
+
+Counter keys (every counter is an int):
+  messages_processed, messages_dead_lettered,
+  noco_upsert_ok, noco_upsert_4xx, noco_upsert_5xx,
+  papra_would_write, papra_write_ok, papra_write_failed,
+  papra_index_upsert_ok, papra_index_upsert_failed,
+  llm_call_seconds_sum, llm_call_count.
+
+Gauge keys (string | int | float):
+  last_run_at, dead_letter_total, papra_dry_run, papra_dry_run_override.
+
+Idempotency / file shape
+------------------------
+Each run writes to its own file (`<pipeline>-metrics-<run_id>.jsonl`);
+the file is created if missing and appended to. The file is **never**
+overwritten; the per-run file is the unit of observability.
+
+Environment
+-----------
+The artifact directory defaults to `pipelines/_artifacts` (the same
+directory the run log uses). The directory is created on first
+write. No env vars are required.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+ARTIFACT_DIR = Path(__file__).resolve().parent.parent / "_artifacts"
+
+
+# Canonical counter + gauge vocabulary. Tests assert that the line
+# carries exactly these keys; missing or extra keys are bugs.
+COUNTER_KEYS: list[str] = [
+    "messages_processed",
+    "messages_dead_lettered",
+    "noco_upsert_ok",
+    "noco_upsert_4xx",
+    "noco_upsert_5xx",
+    "papra_would_write",
+    "papra_write_ok",
+    "papra_write_failed",
+    "papra_index_upsert_ok",
+    "papra_index_upsert_failed",
+    "llm_call_seconds_sum",
+    "llm_call_count",
+]
+
+GAUGE_KEYS: list[str] = [
+    "last_run_at",
+    "dead_letter_total",
+    "papra_dry_run",
+    "papra_dry_run_override",
+]
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _run_id() -> str:
+    return os.environ.get("INBOXOS_RUN_ID") or f"{int(time.time())}-{uuid.uuid4().hex[:8]}"
+
+
+def _metrics_path(pipeline: str, run_id: str) -> Path:
+    ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
+    return ARTIFACT_DIR / f"{pipeline}-metrics-{run_id}.jsonl"
+
+
+def emit_metric(
+    run_id: str | None,
+    pipeline: str,
+    *,
+    counters: dict[str, int],
+    gauges: dict[str, Any],
+) -> None:
+    """Append one metrics line to the per-run file.
+
+    The run_id argument is optional; when None, the helper uses the
+    `INBOXOS_RUN_ID` env var or generates a fresh one (same as
+    pipelines.common.get_logger).
+    """
+    rid = run_id or _run_id()
+    line = {
+        "ts": _now_iso(),
+        "run_id": rid,
+        "pipeline": pipeline,
+        "counters": dict(counters),
+        "gauges": dict(gauges),
+    }
+    path = _metrics_path(pipeline, rid)
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(line, default=str) + "\n")

pipelines/_tests/test_metrics_offline.py

@@ -0,0 +1,112 @@
+"""Offline smoke for pipelines._metrics.emit (REQ-INB-23).
+
+Asserts the metrics emitter writes one JSONL line per call, with
+the exact counter + gauge vocabulary documented in
+pipelines/_metrics/emit.py. The test runs against a temporary
+artifact directory so the real per-run files are not touched.
+
+Run from the inboxos/ project root:
+
+    python3 pipelines/_tests/test_metrics_offline.py
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+
+HERE = Path(__file__).resolve().parent
+PIPE_DIR = HERE.parent
+sys.path.insert(0, str(PIPE_DIR))
+
+
+# ── Tests ────────────────────────────────────────────────────
+
+
+def test_emit_metric_creates_file_with_one_line_per_call() -> None:
+    """emit_metric appends one JSONL line per call to a per-run file."""
+    from _metrics import emit as emit_mod
+    from _metrics.emit import emit_metric, COUNTER_KEYS, GAUGE_KEYS
+
+    # Redirect the artifact dir to a temp location for hermetic tests.
+    with tempfile.TemporaryDirectory() as tmp:
+        original_artifact_dir = emit_mod.ARTIFACT_DIR
+        emit_mod.ARTIFACT_DIR = Path(tmp)
+        try:
+            counters = {k: 0 for k in COUNTER_KEYS}
+            counters["messages_processed"] = 12
+            counters["noco_upsert_ok"] = 7
+            counters["papra_write_ok"] = 2
+            gauges = {k: 0 for k in GAUGE_KEYS}
+            gauges["last_run_at"] = "2026-06-08T12:00:00+00:00"
+            gauges["papra_dry_run"] = "true"
+
+            emit_metric("run-aaa", "live", counters=counters, gauges=gauges)
+            emit_metric("run-aaa", "live", counters=counters, gauges=gauges)
+
+            path = Path(tmp) / "live-metrics-run-aaa.jsonl"
+            assert path.exists(), f"file not created at {path}"
+            lines = path.read_text(encoding="utf-8").splitlines()
+            assert len(lines) == 2, f"expected 2 lines, got {len(lines)}"
+            for ln in lines:
+                rec = json.loads(ln)
+                assert rec["run_id"] == "run-aaa"
+                assert rec["pipeline"] == "live"
+                assert rec["counters"]["messages_processed"] == 12
+                assert rec["counters"]["noco_upsert_ok"] == 7
+                assert rec["counters"]["papra_write_ok"] == 2
+                assert rec["gauges"]["last_run_at"] == "2026-06-08T12:00:00+00:00"
+                assert rec["gauges"]["papra_dry_run"] == "true"
+                # The shape is exactly: ts + run_id + pipeline + counters + gauges.
+                assert set(rec.keys()) == {"ts", "run_id", "pipeline", "counters", "gauges"}
+                assert set(rec["counters"].keys()) == set(COUNTER_KEYS)
+                assert set(rec["gauges"].keys()) == set(GAUGE_KEYS)
+        finally:
+            emit_mod.ARTIFACT_DIR = original_artifact_dir
+
+
+def test_emit_metric_does_not_overwrite_existing_file() -> None:
+    """Two calls in a row append; the file is never truncated."""
+    from _metrics import emit as emit_mod
+    from _metrics.emit import emit_metric, COUNTER_KEYS, GAUGE_KEYS
+
+    with tempfile.TemporaryDirectory() as tmp:
+        original_artifact_dir = emit_mod.ARTIFACT_DIR
+        emit_mod.ARTIFACT_DIR = Path(tmp)
+        try:
+            counters = {k: 0 for k in COUNTER_KEYS}
+            gauges = {k: 0 for k in GAUGE_KEYS}
+            emit_metric("run-bbb", "historical", counters=counters, gauges=gauges)
+            emit_metric("run-bbb", "historical", counters=counters, gauges=gauges)
+            path = Path(tmp) / "historical-metrics-run-bbb.jsonl"
+            lines = path.read_text(encoding="utf-8").splitlines()
+            assert len(lines) == 2
+        finally:
+            emit_mod.ARTIFACT_DIR = original_artifact_dir
+
+
+# ── Runner ──────────────────────────────────────────────────
+
+
+def _run() -> int:
+    tests = [
+        test_emit_metric_creates_file_with_one_line_per_call,
+        test_emit_metric_does_not_overwrite_existing_file,
+    ]
+    failures = []
+    for t in tests:
+        try:
+            t()
+            print(f"PASS  {t.__name__}")
+        except Exception as exc:  # noqa: BLE001
+            failures.append((t.__name__, exc))
+            print(f"FAIL  {t.__name__}: {exc}")
+    print(f"--- {len(tests) - len(failures)}/{len(tests)} passed")
+    return 0 if not failures else 1
+
+
+if __name__ == "__main__":
+    sys.exit(_run())

pipelines/common.py

@@ -676,7 +676,26 @@ def _ollama_call_one_model(
             url=url,
         )
 
-    resp = requests.post(url, headers=headers, json=body, timeout=cfg.timeout)
+    # LLM metrics (REQ-INB-23). When LLM_METRICS_ENABLED is truthy
+    # (default), wrap the HTTP POST in time.monotonic() and increment
+    # the per-process `llm_call_seconds_sum` and `llm_call_count`
+    # accumulators. The metrics line emitted at the end of main()
+    # reads these accumulators via os.environ (set by the wrapper).
+    # The gate exists so operators can opt out of timing overhead
+    # during high-throughput canary runs.
+    import time as _time
+    _t0 = _time.monotonic()
+    try:
+        resp = requests.post(url, headers=headers, json=body, timeout=cfg.timeout)
+    finally:
+        if os.environ.get("LLM_METRICS_ENABLED", "true").strip().lower() not in (
+            "false", "0", "no", "off"
+        ):
+            _dt = _time.monotonic() - _t0
+            _prev_sum = float(os.environ.get("INBOXOS_LLM_CALL_SECONDS_SUM", "0") or 0)
+            _prev_count = int(os.environ.get("INBOXOS_LLM_CALL_COUNT", "0") or 0)
+            os.environ["INBOXOS_LLM_CALL_SECONDS_SUM"] = str(_prev_sum + _dt)
+            os.environ["INBOXOS_LLM_CALL_COUNT"] = str(_prev_count + 1)
     resp.raise_for_status()
     data = resp.json()
 

pipelines/historical_pipeline.py

@@ -46,6 +46,7 @@ from common import (
     NocoConfig,
     PapraConfig,
     S3Config,
+    emit_papra_write_gate_event,
     ensure_bucket,
     get_logger,
     llm_run_command,
@@ -63,6 +64,7 @@ from common import (
 )
 from common import OpenCodeConfig  # noqa: F401  # legacy alias kept for back-compat; new code uses LLMConfig
 from mbox_stream import stream_messages_from_offset  # noqa: E402
+from _metrics.emit import COUNTER_KEYS, GAUGE_KEYS, emit_metric
 
 
 # ── Configuration ──────────────────────────────────────
@@ -391,6 +393,9 @@ def process_one_message(
     # 5. Body-write decision (post-classification). The attachment
     #    branch was already handled above; this pass is for module-driven
     #    body writes only.
+    # The gate event is emitted unconditionally (REQ-INB-20) so every
+    # processed message has one `papra_write_gate` in the structured log.
+    emit_papra_write_gate_event(papra_cfg, logger)
     body_decision = papra_decide(
         message_id=message_id,
         modules=modules,
@@ -580,6 +585,29 @@ def main(argv: list[str] | None = None) -> int:
             exit_code=EARLY_STOP_EXIT_CODE,
         )
         return EARLY_STOP_EXIT_CODE
+    # Per-run metrics line (REQ-INB-23). Counters and gauges mirror
+    # the live pipeline's vocabulary (see pipelines/_metrics/emit.py).
+    metrics_counters = {k: 0 for k in COUNTER_KEYS}
+    metrics_counters["messages_processed"] = processed
+    metrics_counters["llm_call_seconds_sum"] = int(
+        float(os.environ.get("INBOXOS_LLM_CALL_SECONDS_SUM", "0") or 0) * 1000
+    )
+    metrics_counters["llm_call_count"] = int(
+        os.environ.get("INBOXOS_LLM_CALL_COUNT", "0") or 0
+    )
+    metrics_gauges = {k: 0 for k in GAUGE_KEYS}
+    metrics_gauges["last_run_at"] = now_iso()
+    metrics_gauges["papra_dry_run"] = str(papra_cfg.dry_run)
+    metrics_gauges["papra_dry_run_override"] = (
+        "none" if papra_cfg.dry_run_override is None
+        else ("true" if papra_cfg.dry_run_override else "false")
+    )
+    emit_metric(
+        run_id=os.environ.get("INBOXOS_RUN_ID"),
+        pipeline="historical",
+        counters=metrics_counters,
+        gauges=metrics_gauges,
+    )
     return 0
 
 

pipelines/live_pipeline.py

@@ -59,6 +59,7 @@ from common import (
     NocoConfig,
     PapraConfig,
     S3Config,
+    emit_papra_write_gate_event,
     ensure_bucket,
     get_logger,
     llm_run_command,
@@ -74,6 +75,8 @@ from common import (
     safe_message_id,
 )
 
+from _metrics.emit import COUNTER_KEYS, GAUGE_KEYS, emit_metric
+
 
 # ── Configuration ──────────────────────────────────────
 
@@ -487,6 +490,9 @@ def process_one_message(
     #    papra_write_* honors papra_cfg.dry_run so the default ship is
     #    observation-only. Failure to write is non-fatal -- the .eml is
     #    in SeaweedFS and the structured fields are in NocoDB either way.
+    #    The gate event is emitted unconditionally (REQ-INB-20) so every
+    #    processed message has one `papra_write_gate` in the structured log.
+    emit_papra_write_gate_event(papra_cfg, logger)
     papra_decision_obj = papra_decide(
         message_id=message_id,
         modules=modules,
@@ -652,6 +658,31 @@ def main(argv: list[str] | None = None) -> int:
         final_uid=advanced_uid,
         finished_at=now_iso(),
     )
+    # Per-run metrics line (REQ-INB-23). Counters are derived from
+    # the in-loop accumulators; gauges from the papra_cfg state and
+    # the IMAP state. Alerting is out of scope for v1.1 (forward-
+    # pointer to docs/operations-runbook.md §4.5).
+    metrics_counters = {k: 0 for k in COUNTER_KEYS}
+    metrics_counters["messages_processed"] = processed
+    metrics_counters["llm_call_seconds_sum"] = int(
+        float(os.environ.get("INBOXOS_LLM_CALL_SECONDS_SUM", "0") or 0) * 1000
+    )
+    metrics_counters["llm_call_count"] = int(
+        os.environ.get("INBOXOS_LLM_CALL_COUNT", "0") or 0
+    )
+    metrics_gauges = {k: 0 for k in GAUGE_KEYS}
+    metrics_gauges["last_run_at"] = now_iso()
+    metrics_gauges["papra_dry_run"] = str(papra_cfg.dry_run)
+    metrics_gauges["papra_dry_run_override"] = (
+        "none" if papra_cfg.dry_run_override is None
+        else ("true" if papra_cfg.dry_run_override else "false")
+    )
+    emit_metric(
+        run_id=os.environ.get("INBOXOS_RUN_ID"),
+        pipeline="live",
+        counters=metrics_counters,
+        gauges=metrics_gauges,
+    )
     return 0