Merge pull request #116 from NVIDIA-NeMo/condensed-traces

feat(tracing): add portable journal file exporter

Author: sklinglernv

examples/quickstart/06_tracing.py

@@ -9,12 +9,18 @@
 Workflow:
   1. nooa start-dev   # start trace viewer and OTLP receiver
   2. uv run python examples/quickstart/06_tracing.py
-  3. View traces at http://localhost:5001
+  3. nooa import-traces traces/quickstart-06-journal
+  4. View traces at http://localhost:5001
 """
 
+from pathlib import Path
+
 from nooa import hidden
+from nooa.tracing import enable_tracing, exporters
 from nooa.util.quickstart import *
 
+TRACE_DIR = Path("traces/quickstart-06-journal")
+
 
 class MathAgent(Agent, llm=llm):
     """Agent that performs calculations with full tracing."""
@@ -42,8 +48,9 @@ async def _format(self, value: float) -> str:
 
 @autorun
 async def main():
-    # Tracing is auto-enabled when `nooa start-dev` is running.
-    # To write JSONL files instead, call enable_tracing(trace_dir="./traces") explicitly.
+    # The journal file keeps message bodies content-addressed instead of
+    # repeating the full LLM conversation on every OTLP span.
+    enable_tracing(exporters=[exporters.journal_file(TRACE_DIR)])
 
     agent = MathAgent()
     result = await agent.run("(10 + 5) * 2")
@@ -56,5 +63,6 @@ async def main():
     print("  - explain()    ellipsis/LLM method, child of run()")
     print("  - _format()    private helper (also traced)")
     print("=" * 80)
-    print("\nTrace Viewer: http://localhost:5001")
+    print(f"\nJournal trace: {TRACE_DIR}")
+    print(f"Import with: nooa import-traces {TRACE_DIR}")
     print("=" * 80)

packages/nooa-cli/src/nooa_cli/commands/_otlp_helpers.py

@@ -8,6 +8,10 @@
 
 import click
 
+JOURNAL_ENVELOPE_KEY = "nooaJournal"
+JOURNAL_FORMAT = "nooa.message_journal"
+JOURNAL_VERSION = 1
+
 
 def validate_endpoint(endpoint: str) -> None:
     """Validate that the endpoint uses an HTTP(S) scheme."""
@@ -102,6 +106,57 @@ def post_annotations(endpoint: str, annotations: list[dict]) -> int:
     return imported
 
 
+def get_journal_record(body: dict) -> dict | None:
+    """Return a validated NOOA journal envelope, or ``None`` for other lines."""
+    record = body.get(JOURNAL_ENVELOPE_KEY)
+    if not isinstance(record, dict):
+        return None
+    if record.get("format") != JOURNAL_FORMAT or record.get("version") != JOURNAL_VERSION:
+        return None
+    return record
+
+
+def post_journal_record(endpoint: str, record: dict, session_id: str) -> bool:
+    """POST one portable-file journal record to the viewer.
+
+    ``session_id`` is authoritative so renaming a trace file and Harbor's
+    trial-name remapping affect OTLP spans and their journal sideband equally.
+    Manifest records are accepted as no-ops.
+    """
+    record_type = record.get("type")
+    if record_type == "manifest":
+        return True
+    if record_type == "blocks":
+        payload = record.get("blocks")
+        if not isinstance(payload, list):
+            return False
+        path = "/v1/journal/blocks"
+        headers = {"Content-Type": "application/json", "X-Session-Id": session_id}
+    elif record_type == "call":
+        source = record.get("call")
+        if not isinstance(source, dict):
+            return False
+        payload = dict(source)
+        payload["session_id"] = session_id
+        path = "/v1/journal/calls"
+        headers = {"Content-Type": "application/json"}
+    else:
+        return False
+
+    data = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+    request = urllib.request.Request(
+        f"{endpoint.rstrip('/')}{path}",
+        data=data,
+        headers=headers,
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            return response.status < 300
+    except Exception:
+        return False
+
+
 def session_exists(endpoint: str, session_id: str) -> bool:
     """Check whether a session already exists in the viewer."""
     url = f"{endpoint.rstrip('/')}/api/trace-count?session_id={urllib.parse.quote(session_id)}"

packages/nooa-cli/src/nooa_cli/commands/import_harbor.py

@@ -1,6 +1,6 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
-"""Import NVIDIA OO Agents OTLP traces from a Harbor job directory into the viewer.
+"""Import NOOA OTLP or portable journal traces from Harbor into the viewer.
 
 Walks a Harbor job directory (or any directory containing one), finds all
 traces under ``artifacts/traces/*.jsonl``, enriches them with Harbor metadata
@@ -24,7 +24,9 @@
 
 from ._otlp_helpers import (
     check_endpoint_reachable,
+    get_journal_record,
     inject_resource_attrs,
+    post_journal_record,
     post_traces_batch,
     session_exists,
     validate_endpoint,
@@ -34,7 +36,7 @@
 
 
 def _find_harbor_traces(root: Path) -> list[Path]:
-    """Find all OTLP trace files nested under Harbor artifact directories.
+    """Find all OTLP or portable journal trace files under Harbor artifacts.
 
     Harbor copies the container's ``/logs/artifacts/`` to ``trial_dir/artifacts/``
     on the host. The agent decides the layout within that directory — a common
@@ -310,7 +312,7 @@ def _import_trace_file(
     batch_lines: int,
     batch_bytes: int,
 ) -> tuple[bool, list[str]]:
-    """Import one OTLP JSONL file, posting its lines in batches.
+    """Import one OTLP or portable journal JSONL file.
 
     Accumulates OTLP bodies and flushes them in batches: many ``resourceSpans``
     envelopes are merged into one POST, avoiding one HTTP request per line. A flush
@@ -346,6 +348,12 @@ def flush() -> None:
                 body = json.loads(raw_line)
             except json.JSONDecodeError:
                 continue
+            journal_record = get_journal_record(body)
+            if journal_record is not None:
+                session_id = str(resource_attrs["session.id"])
+                if not post_journal_record(endpoint, journal_record, session_id):
+                    errors.append(f"{jsonl_path.name}: failed to post journal record")
+                continue
             if "resourceSpans" not in body:
                 continue
 

packages/nooa-cli/src/nooa_cli/commands/import_traces.py

@@ -1,6 +1,6 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
-"""Import OTLP trace .jsonl files into the viewer.
+"""Import OTLP or portable NOOA journal .jsonl files into the viewer.
 
 Usage:
     nooa import-traces ./traces/
@@ -18,16 +18,18 @@
 
 from ._otlp_helpers import (
     check_endpoint_reachable,
+    get_journal_record,
     inject_resource_attrs,
     post_annotations,
+    post_journal_record,
     post_trace,
     session_exists,
     validate_endpoint,
 )
 
 NAME = "import-traces"
 
-TRACE_EXTENSIONS = {".jsonl"}
+TRACE_EXTENSIONS = (".nooa.jsonl", ".jsonl")
 
 
 def _find_trace_files(path: Path) -> list[Path]:
@@ -80,7 +82,7 @@ def _session_id_from_filename(path: Path) -> str:
     help="Batch ID for this import (default: auto-generated).",
 )
 def command(path: str, endpoint: str, batch_id: str | None):
-    """Import OTLP trace .jsonl files into the viewer."""
+    """Import OTLP and portable NOOA journal .jsonl files into the viewer."""
     target = Path(path)
     files = _find_trace_files(target)
 
@@ -136,6 +138,12 @@ def command(path: str, endpoint: str, batch_id: str | None):
                 except json.JSONDecodeError:
                     continue
 
+                journal_record = get_journal_record(body)
+                if journal_record is not None:
+                    if not post_journal_record(endpoint, journal_record, session_id):
+                        errors.append(f"{file.name}:{line_num}: failed to post journal record")
+                    continue
+
                 # Handle annotation lines from exported traces
                 if "annotations" in body and "resourceSpans" not in body:
                     anns = body["annotations"]

src/nooa/tracing/__init__.py

@@ -16,6 +16,9 @@
 
     # JSONL only (no viewer required)
     enable_tracing(exporters=[exporters.jsonl()])
+
+    # Compact portable file (OTLP spans + content-addressed message journal)
+    enable_tracing(exporters=[exporters.journal_file("./traces")])
 """
 
 from __future__ import annotations

src/nooa/tracing/_journal_exporter.py

@@ -79,7 +79,10 @@ def _install(self) -> MessageJournalCallback:
 
         with _INSTALL_LOCK:
             for cb in litellm.callbacks:
-                if isinstance(cb, MessageJournalCallback):
+                # FileMessageJournalCallback subclasses MessageJournalCallback
+                # to reuse normalization, but is a different sink.  Match the
+                # exact class so exporter construction order cannot mix them.
+                if isinstance(cb, MessageJournalCallback) and type(cb) is MessageJournalCallback:
                     cb.add_destination(self._base_url)
                     return cb