Claude/logging

[brian-arnold]

Logging Infrastructure Overview

What it captures

Every time a packet function executes, the infrastructure captures stdout, stderr, Python logging output, and
tracebacks (on failure). These are bundled into a CapturedLogs dataclass.

How capture works

There are two capture mechanisms depending on where the function runs:

Local execution (LocalExecutor / no executor): LocalCaptureContext sets per-task/thread ContextVar buffers.
sys.stdout/sys.stderr are globally replaced (once, at observer creation) with ContextLocalTeeStream objects
that write to both the terminal and the active buffer. A ContextVarLoggingHandler on the root logger captures
Python logging the same way. Concurrent packets never intermingle because each asyncio task/thread gets its own
ContextVar copy.

Ray execution (RayExecutor): _make_capture_wrapper() returns a self-contained closure (no orcapod imports) that
does fd-level capture (os.dup2) on the worker process. It returns a plain 6-tuple over the Ray object store;
the driver reconstructs CapturedLogs from it.

How CapturedLogs flows through the call chain

User function
↓ catches exception, builds CapturedLogs
PythonPacketFunction.direct_call() → (PacketProtocol | None, CapturedLogs)
↓ or via executor.execute_callable() which also returns (raw, CapturedLogs)
PythonPacketFunction.call() → (PacketProtocol | None, CapturedLogs)
↓
CachedPacketFunction.call() → (PacketProtocol | None, CapturedLogs)
↓ (cache hit returns empty CapturedLogs)
_FunctionPodBase.process_packet_with_capture() → (Tag, Packet | None, CapturedLogs)
↓
FunctionNode._process_packet_internal() → (Tag, Packet | None, CapturedLogs)
↓
FunctionNode.execute() — passes CapturedLogs to pkt_logger.record()

The key design decision: CapturedLogs travels as a return value, not through a ContextVar side-channel.
direct_call() catches user-function exceptions internally and returns (None, captured_failure) — it never
re-raises.

Observer/Logger interaction

The orchestrator injects an ExecutionObserverProtocol into each node's execute() call. The observer provides
lifecycle hooks (on_run_start, on_node_start, on_packet_start, etc.) and a factory method:

observer.create_packet_logger(node, tag, packet, pipeline_path=...) → PacketExecutionLoggerProtocol

FunctionNode.execute() calls this factory before each non-cached packet, then after execution:
tag_out, result, captured = self._process_packet_internal(tag, packet)
pkt_logger.record(captured) # writes to database

Concrete implementations

• NoOpObserver / NoOpLogger — Default when no observability is configured. Zero-cost no-ops.
• LoggingObserver / PacketLogger — Writes structured log rows to any ArrowDatabaseProtocol
  (InMemoryArrowDatabase, DeltaTableDatabase, etc.).

Log storage structure

LoggingObserver stores logs at pipeline-path-mirrored locations. If a function node's pipeline path is
("my_pipeline", "transform_a", "v0", ...), its logs go to ("my_pipeline", "logs", "transform_a", "v0", ...).
Each function node gets its own table. Retrieve with obs.get_logs(pipeline_path=node.pipeline_path).

Each log row has fixed columns (log_id, run_id, node_label, stdout, stderr, python_logs, traceback, success,
timestamp) plus dynamic tag columns — each tag key becomes its own queryable column rather than being
JSON-encoded.

Changes to orcapod-python

1. CapturedLogs as return values (removed ContextVar side-channel)

Core change: call(), direct_call(), async_call(), direct_async_call() on packet functions now return
tuple[PacketProtocol | None, CapturedLogs] instead of PacketProtocol | None. Similarly
executor.execute()/async_execute() return tuples. direct_call() catches user-function exceptions internally and
returns (None, captured_failure) instead of re-raising.

• Protocols — packet_function.py, executor.py, node_protocols.py updated with new return types
• Executors — base.py, local.py, ray.py pass through CapturedLogs; Ray's execute() now delegates to
  execute_callable()
• packet_function.py — All _captured_logs.set() calls removed; PythonPacketFunction, CachedPacketFunction,
  PacketFunctionWrapper return tuples
• function_pod.py — Added process_packet_with_capture() (returns 3-tuple with CapturedLogs); process_packet()
  unchanged (still 2-tuple, discards captured)
• cached_function_pod.py — Added process_packet_with_capture() preserving pod-level caching
• function_node.py — Uses process_packet_with_capture() to get CapturedLogs directly from the return value. No
  more _captured_logs ContextVar reads. Simplified execute logic (no try/except needed)

2. Removed duplicate _ray_capture_wrapper

• Deleted the module-level function from logging_capture.py (unused imports cleaned up too)
• RayExecutor._make_capture_wrapper() is the single capture mechanism for Ray

3. Pipeline-path-mirrored log storage with queryable tag columns

• create_packet_logger() accepts pipeline_path kwarg on protocol, NoOpObserver, and LoggingObserver
• Logs stored at pipeline_path[:1] + ("logs",) + pipeline_path[1:] — each function node gets its own log table
• Tag data stored as individual columns (not a single JSON "tags" column)
• get_logs(pipeline_path=...) retrieves node-specific logs

4. Error handling modes

• FunctionNode.execute() accepts error_policy: "continue" | "fail_fast"
• SyncPipelineOrchestrator and AsyncPipelineOrchestrator accept error_policy and pass it through

5. New files

• src/orcapod/pipeline/logging_capture.py — Capture infrastructure (CapturedLogs, LocalCaptureContext, tee
  streams)
• src/orcapod/pipeline/logging_observer.py — LoggingObserver + PacketLogger
• src/orcapod/protocols/observability_protocols.py — Observer/Logger protocols
• tests/test_pipeline/test_logging_capture.py — Unit tests for capture infrastructure
• tests/test_pipeline/test_logging_observer_integration.py — 9 end-to-end integration tests

6. Test updates

~20 test files updated to unpack the new tuple return types, and RecordingObserver classes updated to accept
pipeline_path kwarg.

[codecov]

Welcome to Codecov 🎉

Once you merge this PR into your default branch, you're all set! Codecov will compare coverage reports and display results in all future pull requests.

ℹ️ You can also turn on project coverage checks and project coverage reporting on Pull Request comment

Thanks for integrating Codecov - We've got you covered ☂️

[brian-arnold]

cannot change target branch in UI