Add serve-origin marker + parent-task attribution to usage records

ARCHITECTURE.md

@@ -217,7 +217,10 @@ adapters' optional `run_stream` capability): the view primes the pump — the fi
 pulled before any headers commit, so connect-time failures stay plain JSON errors — then the
 handler writes one flushed SSE event per delta (close-delimited body). Backends that can't
 stream (cli kinds, hence the router path) deliver the completed text as a single chunk. The
-routing core is untouched — measurement and both paid-API gates behave exactly as for a CLI run.
+routing core is untouched — measurement and both paid-API gates behave exactly as for a CLI run,
+with two attribution extras (#74): records tag `origin: "serve"` (vs `cli`/`gui`), and the
+optional `X-TangleBrain-Parent-Task` request header is sanitized and recorded as
+`parent_task_id` — metadata only, never routed on.
 
 Like the panel, it binds `127.0.0.1` only and is deliberately keyless: the `Authorization` header
 is never read (local callers need no credential), and the loopback bind is what keeps an

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **Serve-origin marker + parent-task attribution (#74)** — usage records now carry an
+  `origin` field (`cli` | `gui` | `serve`) so serve-mode traffic is distinguishable from CLI and
+  panel runs, and `tanglebrain --stats` shows the per-origin split (records predating the field
+  roll up as `untagged`, never guessed at). OpenAI-compat callers can additionally send an
+  optional `X-TangleBrain-Parent-Task` header carrying their own task/session identity —
+  trimmed, capped at 128 chars, recorded onto the usage record as `parent_task_id` for
+  cross-system attribution, never routed on. Both are additive record fields; old records and
+  readers are unaffected.
+
 ## [0.19.0] - 2026-07-04
 
 ### Added

README.md

@@ -232,6 +232,13 @@ field reports **which backend actually served**; the requested directive and rou
 in a `tanglebrain` extension field, and `usage` carries the same `chars/4` estimate the
 measurement log uses (served requests are metered exactly like CLI runs).
 
+Served requests are attributed in the usage log: each record carries `origin: "serve"` (CLI runs
+tag `cli`, panel runs `gui`), and `tanglebrain --stats` shows the per-origin split. A caller can
+additionally send an optional `X-TangleBrain-Parent-Task` header carrying its own task/session
+identity — trimmed, capped at 128 chars, recorded onto the usage record as `parent_task_id` for
+cross-system attribution, and never routed on. The reverse linkage already exists: the response's
+completion id is `chatcmpl-<task_id>`, the same task id the usage record carries.
+
 Caveats, by design:
 
 - **Streaming is real where the backend can stream.** `stream: true` delivers incremental

tanglebrain/cli.py

@@ -150,6 +150,8 @@ def run_once(
     task: str | None = None,
     return_served: bool = False,
     gate: bool | None = None,
+    origin: str = "cli",
+    parent_task_id: str | None = None,
 ):
     """Route a single prompt to a roster tier and return the response text.
 
@@ -180,6 +182,10 @@ def run_once(
         gate: Override for the classifier gate on the default path. ``None`` (default) uses the
             ``classifier_gate_enabled`` setting; ``True``/``False`` force the gate on/off for this
             call. Ignored when ``model`` or ``local`` is set.
+        origin: Which surface this call entered through, recorded on the usage record (#74) —
+            ``"cli"`` (default), ``"gui"``, or ``"serve"``. Attribution only; routing unaffected.
+        parent_task_id: Optional external caller identity recorded on the usage record (#74:
+            the serve endpoint's ``X-TangleBrain-Parent-Task`` header). Attribution only.
 
     Returns:
         The response text (``str``), or ``(text, served)`` when ``return_served`` is ``True``.
@@ -219,12 +225,21 @@ def run_once(
             text = router.route(prompt, task=task, opts=opts)
             entry = router.last_served
 
-    record_task(path=path, entry=entry, prompt=prompt, response=text, task_id=task_id)
+    record_task(
+        path=path, entry=entry, prompt=prompt, response=text, task_id=task_id,
+        origin=origin, parent_task_id=parent_task_id,
+    )
     return (text, _served(path, entry, task_id)) if return_served else text
 
 
 def _recording_stream(
-    deltas: Iterator[str], path: str, entry, prompt: str, task_id: str
+    deltas: Iterator[str],
+    path: str,
+    entry,
+    prompt: str,
+    task_id: str,
+    origin: str,
+    parent_task_id: str | None,
 ) -> Iterator[str]:
     """Wrap a delta stream so the task is metered exactly once, however the stream ends.
 
@@ -245,6 +260,8 @@ def _recording_stream(
         entry: The serving roster entry.
         prompt: The routed prompt (for the usage estimate).
         task_id: The task id minted for this run.
+        origin: The entry surface recorded on the usage record (#74).
+        parent_task_id: Optional external caller identity recorded on the usage record (#74).
 
     Yields:
         The fragments of ``deltas``, unchanged.
@@ -258,7 +275,8 @@ def _record(require_text: bool) -> None:
             return
         recorded = True
         record_task(
-            path=path, entry=entry, prompt=prompt, response="".join(pieces), task_id=task_id
+            path=path, entry=entry, prompt=prompt, response="".join(pieces), task_id=task_id,
+            origin=origin, parent_task_id=parent_task_id,
         )
 
     try:
@@ -284,6 +302,8 @@ def run_once_stream(
     local: bool = False,
     task: str | None = None,
     gate: bool | None = None,
+    origin: str = "cli",
+    parent_task_id: str | None = None,
 ) -> tuple[Iterator[str], dict | None]:
     """Route a single prompt like :func:`run_once`, delivering the response as a delta stream.
 
@@ -310,6 +330,8 @@ def run_once_stream(
         local: Force the free local tier instead of the frontier-first router.
         task: Optional task-fit hint for the router (a ``good_at`` tag).
         gate: Classifier-gate override for the default path, as in :func:`run_once`.
+        origin: Which surface this call entered through, recorded on the usage record (#74).
+        parent_task_id: Optional external caller identity recorded on the usage record (#74).
 
     Returns:
         ``(deltas, served)`` — ``deltas`` yields response text fragments in order (joined, they
@@ -344,18 +366,26 @@ def run_once_stream(
             router = Router(roster)
             text = router.route(prompt, task=task, opts=opts)
             entry = router.last_served
-            record_task(path="router", entry=entry, prompt=prompt, response=text, task_id=task_id)
+            record_task(
+                path="router", entry=entry, prompt=prompt, response=text, task_id=task_id,
+                origin=origin, parent_task_id=parent_task_id,
+            )
             return iter([text]), _served("router", entry, task_id)
 
     adapter = build_adapter(entry)
     run_stream = getattr(adapter, "run_stream", None)
     if run_stream is None:
         # Per-backend emulation: no streaming capability — run blocking, frame as one delta.
         text = adapter.run(prompt, opts)
-        record_task(path=path, entry=entry, prompt=prompt, response=text, task_id=task_id)
+        record_task(
+            path=path, entry=entry, prompt=prompt, response=text, task_id=task_id,
+            origin=origin, parent_task_id=parent_task_id,
+        )
         return iter([text]), _served(path, entry, task_id)
 
-    deltas = _recording_stream(run_stream(prompt, opts), path, entry, prompt, task_id)
+    deltas = _recording_stream(
+        run_stream(prompt, opts), path, entry, prompt, task_id, origin, parent_task_id
+    )
     return deltas, _served(path, entry, task_id)
 
 

tanglebrain/gui/static/index.html

@@ -109,10 +109,16 @@ <h2>Pricing reference</h2>
     const d = await getJSON("/api/stats");
     const s = d.summary || {};
     const tiers = Object.entries(s.by_tier || {}).map(([k, v]) => `${esc(k)} ${v}`).join(", ") || "—";
+    // Origin split (#74): shown only once some record is actually tagged — all-untagged history
+    // adds no signal (mirrors the CLI's format_rollup stance).
+    const byOrigin = s.by_origin || {};
+    const hasOrigin = Object.keys(byOrigin).some((k) => k !== "untagged");
+    const origins = Object.entries(byOrigin).map(([k, v]) => `${esc(k)} ${v}`).join(", ");
     let html = `<div class="stat-grid">
       <div class="stat"><div class="label">Tasks routed</div><div class="value">${s.tasks || 0}</div></div>
       <div class="stat"><div class="label">Spend avoided</div><div class="value big">${money(s.spend_avoided_usd)}</div></div>
       <div class="stat"><div class="label">By tier</div><div class="value mono" style="font-size:.95rem">${tiers}</div></div>
+      ${hasOrigin ? `<div class="stat"><div class="label">By origin</div><div class="value mono" style="font-size:.95rem">${origins}</div></div>` : ""}
       <div class="stat"><div class="label">Est. tokens (in / out)</div><div class="value mono" style="font-size:.95rem">${(s.in_tokens_est||0).toLocaleString()} / ${(s.out_tokens_est||0).toLocaleString()}</div></div>
     </div>`;
     if (d.is_placeholder) html += `<div class="caveat">⚠ pricing: PLACEHOLDER — figures illustrative until the anchor is ratified.</div>`;

tanglebrain/gui/views.py

@@ -128,7 +128,9 @@ def run_prompt(payload: dict) -> dict:
 
     try:
         # return_served gives us the served tier/model directly — no usage-log re-read, no race.
-        text, served = run_once(str(prompt), model=model, local=local, task=task, return_served=True)
+        text, served = run_once(
+            str(prompt), model=model, local=local, task=task, return_served=True, origin="gui"
+        )
     except _RUN_ERRORS as exc:
         return {"ok": False, "error": str(exc)}
 

tanglebrain/measurement.py

@@ -303,6 +303,7 @@ def record_task(
     kind: str = "task",
     task_id: str | None = None,
     parent_task_id: str | None = None,
+    origin: str | None = None,
     log_path: str | os.PathLike[str] | None = None,
     pricing: Pricing | None = None,
 ) -> None:
@@ -325,7 +326,13 @@ def record_task(
             can be linked back to it). Omitted from the record when ``None``.
         parent_task_id: For a delegated sub-call, the id of the top-level task that spawned it (read
             from :data:`PARENT_TASK_ID_ENV`). Omitted from the record when ``None`` — e.g. a delegate
-            invoked outside a propagated task, which rolls up as ``unlinked``.
+            invoked outside a propagated task, which rolls up as ``unlinked``. For a top-level task,
+            an external caller's own task/session identity (#74: the serve endpoint's
+            ``X-TangleBrain-Parent-Task`` header) — pure attribution metadata; the delegate tree's
+            ``by_parent`` rollup reads it only off ``delegate`` records.
+        origin: Which surface the work entered through — ``"cli"`` | ``"gui"`` | ``"serve"``
+            (#74). Omitted from the record when ``None``; records without it roll up as
+            ``untagged`` (pre-#74 history is never guessed at).
         log_path: Override the usage-log path (tests inject a temp path). Defaults to
             :func:`default_log_path`.
         pricing: Override the pricing. Defaults to :func:`load_pricing`.
@@ -359,6 +366,8 @@ def record_task(
             record["task_id"] = str(task_id)
         if parent_task_id is not None:
             record["parent_task_id"] = str(parent_task_id)
+        if origin is not None:
+            record["origin"] = str(origin)
         target = Path(log_path) if log_path is not None else default_log_path()
         target.parent.mkdir(parents=True, exist_ok=True)
         with _LOG_LOCK:
@@ -420,8 +429,10 @@ def rollup(records: list[dict]) -> dict:
         records: The records from :func:`read_records`.
 
     Returns:
-        A dict with: ``tasks`` (int), ``by_tier`` (tier → count), ``in_tokens_est`` /
-        ``out_tokens_est`` (summed estimates), and ``cloud_equiv_usd`` / ``spend_avoided_usd``
+        A dict with: ``tasks`` (int), ``by_tier`` (tier → count), ``by_origin`` (origin → count,
+        where a record without an ``origin`` field counts as ``untagged`` — pre-#74 history is
+        never guessed at), ``in_tokens_est`` / ``out_tokens_est`` (summed estimates), and
+        ``cloud_equiv_usd`` / ``spend_avoided_usd``
         (summed dollars) — all over **top-level tasks only** — plus ``delegates``, a separate
         sub-rollup of delegated sub-calls ``{count, by_backend: {model: {count, in_tokens_est,
         out_tokens_est}}, by_parent: {parent_task_id: {count, by_backend: {model: count}}},
@@ -434,6 +445,7 @@ def rollup(records: list[dict]) -> dict:
     summary: dict = {
         "tasks": 0,
         "by_tier": {},
+        "by_origin": {},
         "in_tokens_est": 0,
         "out_tokens_est": 0,
         "cloud_equiv_usd": 0.0,
@@ -473,6 +485,8 @@ def rollup(records: list[dict]) -> dict:
         summary["tasks"] += 1
         tier = str(r.get("tier", "unknown"))
         summary["by_tier"][tier] = summary["by_tier"].get(tier, 0) + 1
+        origin = str(r.get("origin") or "untagged")
+        summary["by_origin"][origin] = summary["by_origin"].get(origin, 0) + 1
         summary["in_tokens_est"] += in_tok
         summary["out_tokens_est"] += out_tok
         summary["cloud_equiv_usd"] += _as_float(r.get("cloud_equiv_usd"))
@@ -503,6 +517,11 @@ def format_rollup(summary: dict, pricing: Pricing) -> str:
     if by_tier:
         tiers = ", ".join(f"{k} {v}" for k, v in sorted(by_tier.items()))
         lines.append(f"  By tier:        {tiers}")
+    by_origin = summary.get("by_origin") or {}
+    # Show the origin split only once it says something — all-untagged history adds no signal.
+    if any(k != "untagged" for k in by_origin):
+        origins = ", ".join(f"{k} {v}" for k, v in sorted(by_origin.items()))
+        lines.append(f"  By origin:      {origins}")
     lines.append(
         f"  Est. tokens:    in {summary.get('in_tokens_est', 0):,} / "
         f"out {summary.get('out_tokens_est', 0):,}"

tanglebrain/serve/server.py

@@ -21,10 +21,12 @@
 
 from tanglebrain.serve.views import (
     DEFAULT_PORT,
+    PARENT_TASK_HEADER,
     error_envelope,
     handle_chat_completion,
     handle_chat_completion_stream,
     list_models,
+    sanitize_parent_task,
     wants_stream,
 )
 
@@ -38,7 +40,11 @@ def _json_response(status: int, obj: object) -> tuple[int, str, bytes]:
 
 
 def dispatch(
-    method: str, path: str, body: bytes = b"", content_type: str = "application/json"
+    method: str,
+    path: str,
+    body: bytes = b"",
+    content_type: str = "application/json",
+    parent_task: str | None = None,
 ) -> tuple[int, str, bytes | Iterator[bytes]]:
     """Route one request to a view and return ``(status, content_type, body)``.
 
@@ -61,6 +67,8 @@ def dispatch(
         body: Raw request body bytes (for ``POST``).
         content_type: The request's ``Content-Type`` header value (POST only; defaults to JSON
             so socket-free tests needn't supply it).
+        parent_task: The raw ``X-TangleBrain-Parent-Task`` header value (or ``None``) — sanitized
+            here and recorded onto the usage record for cross-system attribution (#74).
 
     Returns:
         ``(status_code, content_type, body_bytes)``.
@@ -94,13 +102,14 @@ def dispatch(
                 return _json_response(
                     400, error_envelope("request body must be a JSON object", "invalid_request_error")
                 )
+            caller_task = sanitize_parent_task(parent_task)
             try:
                 if wants_stream(payload):
-                    status, result = handle_chat_completion_stream(payload)
+                    status, result = handle_chat_completion_stream(payload, caller_task)
                     if status == 200:
                         return 200, _SSE, result  # Iterator[bytes] — pump already primed
                     return _json_response(status, result)
-                status, obj = handle_chat_completion(payload)
+                status, obj = handle_chat_completion(payload, caller_task)
             except Exception as exc:  # noqa: BLE001 — any escape must be clean JSON, never a
                 # dropped connection (e.g. a malformed settings.yaml raising SettingsError on the
                 # auto path). Typed, expected failures are already mapped inside the handlers,
@@ -119,7 +128,8 @@ class Handler(BaseHTTPRequestHandler):
     ``Authorization`` is deliberately never consulted: local callers need no key, and any dummy
     bearer a client insists on sending is simply ignored. The only headers read are the framing
     ones — ``Content-Length`` and ``Content-Type`` (see :func:`dispatch` for why the latter is
-    enforced).
+    enforced) — plus the optional ``X-TangleBrain-Parent-Task`` attribution header (#74), which
+    is recorded onto the usage record and never routed on.
     """
 
     def do_GET(self) -> None:  # noqa: N802 (stdlib naming)
@@ -138,7 +148,13 @@ def do_POST(self) -> None:  # noqa: N802 (stdlib naming)
             )
             return
         body = self.rfile.read(length) if length else b""
-        self._respond(*dispatch("POST", self.path, body, self.headers.get("Content-Type", "")))
+        self._respond(
+            *dispatch(
+                "POST", self.path, body,
+                self.headers.get("Content-Type", ""),
+                self.headers.get(PARENT_TASK_HEADER),
+            )
+        )
 
     def _respond(self, status: int, content_type: str, body: bytes | "Iterator[bytes]") -> None:
         """Write a complete HTTP response — buffered bytes, or a streamed body.