Observe Java execution

Start with pipeline IDs, stage statuses and execution logs. Add lifecycle observers for agent-specific events and OpenTelemetry when your application already exports traces or metrics.

Inspect execution without an extra integration

PipelineResponse exposes status, stages and typed context. Each StageResponse exposes status, result output, attempt information and error codes. The demo's CreatePipelineExample get <id> and wait <id> use these typed models.

Inside a handler, use context.logInfo, logWarning and logError. The SDK returns these logs with the stage result. Context provides pipelineId, stageId, executionId, attempt, traceId and propagated trace fields.

Prefer IDs and structured outcomes to dumping entire inputs or context. putSensitive marks data for protected handling, but it does not sanitize a message you deliberately write to a log.

Observe agent lifecycle events

The agent module includes TypedLifecycleObserver. Implement only the callbacks you need:

package example;

import io.pipelogiq.sdk.agent.AgentLifecycleEvents;
import io.pipelogiq.sdk.agent.TypedLifecycleObserver;

public final class AuditObserver implements TypedLifecycleObserver {
  private static final System.Logger LOG = System.getLogger(AuditObserver.class.getName());

  @Override
  public void onToolExecuted(AgentLifecycleEvents.ToolExecuted event) {
    LOG.log(System.Logger.Level.INFO, "Tool {0}, success={1}, session={2}",
        event.toolName(), event.isSuccess(), event.sessionId());
  }

  @Override
  public void onBudgetExceeded(AgentLifecycleEvents.BudgetExceeded event) {
    LOG.log(System.Logger.Level.WARNING, "Budget reached: session={0}, estimated USD={1}",
        event.sessionId(), event.spentUsd());
  }
}

Register it with .observer(new AuditObserver()) on an AgentRuntime builder. Other callbacks include approval requested, session completed and critic reviewed. Repeated builder .observer(...) calls compose observers; the existing runtime's .observer(...) setter replaces its observer.

Ordinary observer failures do not change the agent's business decision. Events may repeat on retries, so an audit destination requiring deduplication must provide it. Cancellation remains cancellation.

Install the optional telemetry adapter

Build the local artifact from the matching source root:

./mvnw -B -pl sdk-observability -am install -DskipTests

Add this dependency to the application:

<dependency>
  <groupId>com.pipelogiq</groupId>
  <artifactId>pipelogiq-sdk-observability</artifactId>
  <version>0.5.0</version>
</dependency>

The module depends on the OpenTelemetry API. It does not install exporters or choose a collector. If your application already owns a configured OpenTelemetry instance, pass it to new OpenTelemetryAdapter(telemetry).

adapter.install() installs a process-wide trace bridge and returns a closeable registration token. The newest active bridge is used; closing it restores the previous active bridge. It does not select a different provider automatically for each worker in one JVM, and does not close your OpenTelemetry instance.

Configure an OTLP exporter when needed

For an application without existing telemetry setup, this example adds an OpenTelemetry SDK and an OTLP/gRPC exporter. The versions match the adapter's source configuration:

<dependency>
  <groupId>io.opentelemetry</groupId>
  <artifactId>opentelemetry-sdk</artifactId>
  <version>1.66.0</version>
</dependency>
<dependency>
  <groupId>io.opentelemetry</groupId>
  <artifactId>opentelemetry-exporter-otlp</artifactId>
  <version>1.66.0</version>
</dependency>

Create src/main/java/example/Telemetry.java:

package example;

import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
import io.opentelemetry.sdk.resources.Resource;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;

public final class Telemetry {
  private Telemetry() {}

  public static OpenTelemetrySdk create(String endpoint) {
    var resource = Resource.getDefault().merge(Resource.create(
        Attributes.of(AttributeKey.stringKey("service.name"), "orders-java")));
    var spans = OtlpGrpcSpanExporter.builder().setEndpoint(endpoint).build();
    var metrics = OtlpGrpcMetricExporter.builder().setEndpoint(endpoint).build();
    var traces = SdkTracerProvider.builder().setResource(resource)
        .addSpanProcessor(BatchSpanProcessor.builder(spans).build()).build();
    var meters = SdkMeterProvider.builder().setResource(resource)
        .registerMetricReader(PeriodicMetricReader.builder(metrics).build()).build();
    return OpenTelemetrySdk.builder().setTracerProvider(traces).setMeterProvider(meters).build();
  }
}

This creates exporters; it does not start a collector. The following entry point uses the getting-started order stages and an already-running collector at http://localhost:4317:

package example;

import io.pipelogiq.sdk.Pipelogiq;
import io.pipelogiq.sdk.observability.OpenTelemetryAdapter;

public final class TelemetryOrders {
  public static void main(String[] args) {
    try (var telemetry = Telemetry.create("http://localhost:4317")) {
      var adapter = new OpenTelemetryAdapter(telemetry);
      try (var tracing = adapter.install()) {
        Pipelogiq.worker().config("pipelogiq.yml")
            .stages(ValidateOrder.class, RecordOrder.class).run();
      }
    }
  }
}

Stop the worker before closing the bridge and the application-owned telemetry SDK. When using an existing framework's OpenTelemetry instance, let that framework close it.

Enable agent metrics separately

Trace installation and agent metrics registration are separate choices. Add .observer(adapter.metricsObserver()) to the agent builder from the agent guide. Register that runtime on the same worker facade before startup.

The observer exports:

MetricsLifecycleObserver can also maintain a local snapshot() without OpenTelemetry or accept an application-defined sink. Estimated cost is runtime accounting, not a provider invoice.

Understand propagation and verification

With the bridge installed, pipeline submission captures an active external OpenTelemetry span. sendAsync() captures context when constructing its request. Worker execution creates a real pipelogiq.stage span, uses its ID for nested submissions and restores the previous context after execution.

Failed stages are marked as errors. Exception recording uses the exception type rather than messages or stack traces that might contain credentials or payloads. Waiting for approval remains a successful waiting state.

Application-created executor tasks still require explicit context propagation: capture TraceContext.current() before scheduling and activate it inside the task when non-null, or use OpenTelemetry context wrapping. Automatic HTTP/JDBC instrumentation is outside this adapter.

In-memory exporter tests verify span relationships and emitted instruments. Actual export depends on sampling, the configured SDK, collector and network. Verify one submitted pipeline in your collector before treating export as operational; a successful workflow alone does not prove telemetry delivery.