Add execution latency observations and trace correlation

docs/getting-started/operations.md

@@ -193,3 +193,11 @@ Mutating `oac` commands hold `.oac.lock`. If another command holds it, retry aft
 Web signs administrators in with the Core key, checks the origin of every request, and forwards signed-in `/core/v1` requests to Core with the Core key, which stays on the server. It forwards `/v1` and `/api/v1` to Core unchanged, with the caller's own credential, serves only the non-secret node payload at `/node-install/`, and has no Docker or KVM access. Machine routes under `/api/v1` use their own enrollment and connection credentials. No service receives a Docker socket.
 
 Sandboxes are the isolation boundary ([Runtime and outer isolation](../concepts.md#runtime-and-outer-isolation)). Docker sandboxes share the node's kernel, and a Docker node is [root-equivalent](./nodes.md#what-the-installer-sets-up) on its host; microsandbox gives each sandbox a microVM with an explicit [network policy](./nodes.md#what-the-installer-sets-up). Core itself has no Docker socket or KVM access.
+
+## Execution latency
+
+Use the `environment input reserved` log to connect the submitting HTTP trace to `reservation_id` and `execution_trace_id`. The worker derives its diagnostic execution trace from the durable reservation identity, so observer disconnection and owner restart do not require a process-local trace map. This is an explicit link between the HTTP and execution traces, not continuation of the original HTTP span. Follow `environment input selected`, `execution preparation requested` and `environment input admitted` to connect the reservation to the Session, Environment, device, preparation request, Executor and Turn. `preparation_request_id` identifies the control request and is distinct from an HTTP `request_id`. For inputs included in Session creation, join `session initial input origin` to the worker reservation with the same `session_id` and `is_initial=true`; this covers both regular and streaming creation without an extra store query.
+
+`execution stage` records `stage`, `duration_ms` and `status` (`ok`, `error`, `cancelled` or `timeout`). Stages cover input validation/ownership, reservation and admission wait, execution configuration and admission promotion, lifecycle gate wait, Runtime observation/provisioning, provider create/resume and Environment initialization. Runtime connection logs record confirmed observed connection transitions, not the exact socket connection instant. Use Environment, allocation and node IDs to join background Runtime records to an input; a background record without the input trace is not an unbroken request span. Stage durations use a local monotonic clock. `queue_age_ms` instead compares selection time to the database reservation creation time and may reflect clock skew; it is not a measured scheduler-only interval.
+
+`control_ready_ms` starts after scheduling, Runtime readiness and configuration assembly. `start_control_ms` describes the control acknowledgement. `input_to_first_text_ms` starts immediately before execution delivery, includes start-control time and is not model-only TTFT. The admission, control and first-text intervals overlap; do not sum them as independent durations. A successful provider call describes that operation, not completed daemon connection or a completed Turn. Model request start, response headers, first model text and retries remain unavailable unless the selected native adapter provides those observations. Logs contain correlation IDs and finite status categories, not input content, credentials or raw provider errors.

docs/zh/getting-started/operations.md

@@ -1,7 +1,7 @@
 ---
 title: "管理你的安装"
 source: docs/getting-started/operations.md
-source_hash: e60a6e96cd61692b6adc7664874ba0ed2ce489982e7da2fe2347631ee2a94c0a
+source_hash: 263a48e467c9aeec1f8357f0e11b40c6f1b329dd6162f65eda3a1038acd4f551
 ---
 
 安装运维人员负责 Core 主机、存储和可用性。节点主机运行各自的服务；参阅[节点](nodes.md)。设置见[配置参考](../configuration.md)。
@@ -196,3 +196,11 @@ cd && rm -rf ~/.oac/core
 Web 使用 Core 密钥让管理员登录，检查每个请求来源，并用保留在服务器上的 Core 密钥将已登录的 `/core/v1` 请求转发到 Core。它把 `/v1` 和 `/api/v1` 原样转发给 Core，使用调用方自己的凭据；Web 仅在 `/node-install/` 提供不含密钥的节点文件，没有 Docker 或 KVM 访问权限。`/api/v1` 机器路由使用独立注册和连接凭据。没有服务持有 Docker 套接字。
 
 沙箱是隔离边界（[Runtime 与外层隔离](../concepts.md#runtime-and-outer-isolation)）。Docker 沙箱共享节点内核，Docker 节点在主机上[等同于 root 权限](nodes.md#what-the-installer-sets-up)；microsandbox 为每个沙箱提供具有显式[网络策略](nodes.md#what-the-installer-sets-up)的 microVM。Core 自身无 Docker 套接字或 KVM 访问权限。
+
+## 执行延迟 {#execution-latency}
+
+通过 `environment input reserved` 日志将提交输入的 HTTP Trace 与 `reservation_id`、`execution_trace_id` 关联。worker 从持久化预留标识派生执行诊断 Trace，观察连接断开或 owner 重启都不依赖进程内 Trace 映射。这是 HTTP Trace 与执行 Trace 之间的明确关联，不是原 HTTP span 的延续。沿 `environment input selected`、`execution preparation requested`、`environment input admitted` 将预留与 Session、Environment、device、准备请求、Executor、Turn 关联。`preparation_request_id` 标识控制准备请求，与 HTTP `request_id` 不同。创建 Session 时附带输入的情况，通过相同 `session_id` 和 `is_initial=true` 将 `session initial input origin` 与 worker 的预留关联；普通和流式创建都支持，不增加 store 查询。
+
+`execution stage` 记录 `stage`、`duration_ms` 与 `status`（`ok`、`error`、`cancelled`、`timeout`）。阶段覆盖输入验证/ownership、预留和等待 admission、执行配置与 admission 提升、生命周期 gate 等待、Runtime 观察和供给、provider create/resume、Environment 初始化。Runtime 连接日志记录已确认观察到的连接变化，不是 socket 建立的精确时间。通过 Environment、allocation、node ID 将后台 Runtime 记录与输入关联；缺少输入 Trace 的后台记录不能视为连续的请求 span。阶段耗时使用本进程单调时钟。`queue_age_ms` 则比较选择时间与数据库预留创建时间，可能受到时钟偏差影响，不能当作单独测量的调度等待。
+
+`control_ready_ms` 从调度、Runtime 就绪和配置组装完成后开始。`start_control_ms` 描述控制确认。`input_to_first_text_ms` 从执行投递前开始，包含 start-control 时间，不是纯模型 TTFT。admission、control、首字区间存在重叠，不要当作独立耗时相加。provider 调用成功表示该操作成功，不等于 daemon 连接完成或 Turn 完成。除非原生适配器提供相应观测，模型请求开始、响应 headers、模型首字、重试仍为不可观测。日志仅记录关联 ID 与有限状态分类，不记录输入内容、凭据或 provider 原始错误。

services/core/internal/execution/delivery.go

@@ -7,6 +7,8 @@ import (
 	"strconv"
 	"time"
 
+	obslog "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log"
+
 	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimegateway"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
@@ -37,7 +39,11 @@ func requestCancellation(ctx context.Context, peer *runtimegateway.Session, runI
 }
 
 func send(ctx context.Context, peer *runtimegateway.Session, kind, runID string, payload any) error {
-	env, err := proto.NewEnvelope(kind, runID, payload)
+	trace := ""
+	if carrier, ok := obslog.TraceFromContext(ctx); ok {
+		trace = carrier.String()
+	}
+	env, err := proto.NewEnvelopeWithTrace(kind, runID, payload, trace)
 	if err != nil {
 		return err
 	}

services/core/internal/execution/directory_preparation.go

@@ -29,7 +29,7 @@ func (d *Dispatcher) withPreparedWorkspace(owner context.Context, peer *runtimeg
 	if err := d.configurePreparedEnvironment(session, environment, bound, &req); err != nil {
 		return ErrExecutionUnavailable
 	}
-	prepared, err := newPreparedStart(peer)
+	prepared, err := newPreparedStart(owner, peer)
 	if err != nil {
 		return ErrExecutionUnavailable
 	}

services/core/internal/execution/environment_admission.go

@@ -7,6 +7,8 @@ import (
 	"slices"
 	"time"
 
+	obslog "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log"
+
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions"
 )
 
@@ -72,11 +74,17 @@ func (w *Worker) validateCreation(ctx context.Context, input sessions.CreateSess
 	return nil
 }
 
-func (w *Worker) submitEnvironmentInputs(ctx context.Context, session sessions.Session, key string, inputs []sessions.Input) ([]sessions.InputReceipt, error) {
-	if err := w.validateEnvironmentAdmission(ctx, session.Engine, session.Configuration); err != nil {
+func (w *Worker) submitEnvironmentInputs(ctx context.Context, session sessions.Session, key string, inputs []sessions.Input) (receipts []sessions.InputReceipt, err error) {
+	checkAt := time.Now()
+	err = w.validateEnvironmentAdmission(ctx, session.Engine, session.Configuration)
+	observeExecutionStage(ctx, "input_validate", checkAt, err, "session_id", session.ID)
+	if err != nil {
 		return nil, err
 	}
-	if err := w.checkAdmissionOwnership(ctx); err != nil {
+	checkAt = time.Now()
+	err = w.checkAdmissionOwnership(ctx)
+	observeExecutionStage(ctx, "input_ownership", checkAt, err, "session_id", session.ID)
+	if err != nil {
 		return nil, err
 	}
 	kind := ""
@@ -96,12 +104,21 @@ func (w *Worker) submitEnvironmentInputs(ctx context.Context, session sessions.S
 	}
 	changed, unsubscribe := w.dispatcher.notifications.subscribe(session.TenantID, session.ID)
 	defer unsubscribe()
+	reservedAt := time.Now()
 	reserve, cancel := context.WithTimeout(ctx, 5*time.Second)
 	reservation, err := w.admission.ReserveEnvironmentInput(reserve, session.TenantID, session.ID, key, inputs)
 	cancel()
+	observeExecutionStage(ctx, "input_reserve", reservedAt, err, "session_id", session.ID)
 	if err != nil {
 		return nil, err
 	}
+	obslog.Info(ctx, "environment input reserved", "session_id", session.ID,
+		"reservation_id", reservation.ID, "execution_trace_id", reservationTraceID(reservation.ID).String(), "state", reservation.State)
+	admittedAt := time.Now()
+	defer func() {
+		observeExecutionStage(ctx, "input_admission_wait", admittedAt, err,
+			"session_id", session.ID, "reservation_id", reservation.ID, "state", reservation.State)
+	}()
 	w.wakeScheduler()
 	if reservation.State == sessions.EnvironmentInputPending && !reservation.IsInitial {
 		w.hintRuntimeWake(ctx, session)

services/core/internal/execution/execution_observation.go

@@ -0,0 +1,46 @@
+package execution
+
+import (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"time"
+
+	obslog "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log"
+)
+
+// A reservation survives observer disconnects and owner restarts. Its opaque
+// identity anchors the execution trace; the submitting HTTP trace is linked in
+// the reservation log rather than retained in a process-local map.
+func reservationTraceID(reservation string) obslog.TraceID {
+	digest := sha256.Sum256([]byte("oac.environment-input:" + reservation))
+	var id obslog.TraceID
+	copy(id[:], digest[:len(id)])
+	return id
+}
+
+func reservationTrace(ctx context.Context, reservation string) context.Context {
+	return obslog.WithTrace(ctx, obslog.Carrier{Trace: reservationTraceID(reservation), Span: obslog.NewSpanID()})
+}
+
+// Stage intervals use a local monotonic clock. Error prose and request content
+// are deliberately excluded: provider errors may contain private configuration.
+func observeExecutionStage(ctx context.Context, stage string, started time.Time, err error, attrs ...any) {
+	status := "ok"
+	switch {
+	case errors.Is(err, context.Canceled):
+		status = "cancelled"
+	case errors.Is(err, context.DeadlineExceeded):
+		status = "timeout"
+	case err != nil:
+		status = "error"
+	}
+	fields := []any{"stage", stage, "duration_ms", float64(time.Since(started).Microseconds()) / 1000, "status", status}
+	obslog.Info(ctx, "execution stage", append(fields, attrs...)...)
+}
+
+// Initial input is committed by Session creation. Joining its origin to the
+// worker's is_initial reservation avoids an additional diagnostic store query.
+func recordInitialInputOrigin(ctx context.Context, session string) {
+	obslog.Info(ctx, "session initial input origin", "session_id", session)
+}

services/core/internal/execution/execution_observation_test.go

@@ -0,0 +1,178 @@
+package execution
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
+	obslog "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/runtimegateway"
+)
+
+func captureExecutionLogs(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var out bytes.Buffer
+	previous := slog.Default()
+	slog.SetDefault(slog.New(obslog.NewContextHandler(slog.NewJSONHandler(&out, nil))))
+	t.Cleanup(func() { slog.SetDefault(previous) })
+	return &out
+}
+
+func TestReservationTraceSurvivesRecoveryWithoutRetainingObserver(t *testing.T) {
+	parent, cancel := context.WithCancel(t.Context())
+	first := reservationTrace(parent, "reservation-one")
+	recovered := reservationTrace(t.Context(), "reservation-one")
+	a, _ := obslog.TraceFromContext(first)
+	b, _ := obslog.TraceFromContext(recovered)
+	if a.Trace != b.Trace || a.Span == b.Span || a.Trace == reservationTraceID("reservation-two") {
+		t.Fatal("reservation recovery did not retain identity with a distinct attempt span")
+	}
+	if _, err := obslog.ParseTraceparent(a.String()); err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	if !errors.Is(first.Err(), context.Canceled) || recovered.Err() != nil {
+		t.Fatal("diagnostic context changed cancellation ownership")
+	}
+}
+
+func TestExecutionStageStatusAndCredentialRedaction(t *testing.T) {
+	out := captureExecutionLogs(t)
+	ctx := reservationTrace(t.Context(), "reservation-one")
+	for _, sample := range []struct {
+		err    error
+		status string
+	}{
+		{nil, "ok"}, {context.Canceled, "cancelled"}, {context.DeadlineExceeded, "timeout"},
+		{errors.New("private provider URL and credential must never appear"), "error"},
+	} {
+		out.Reset()
+		observeExecutionStage(ctx, "provider_create", time.Now().Add(-time.Millisecond), sample.err, "allocation_id", "allocation-one")
+		var entry map[string]any
+		if err := json.Unmarshal(out.Bytes(), &entry); err != nil {
+			t.Fatal(err)
+		}
+		if entry["status"] != sample.status || entry["stage"] != "provider_create" || entry["trace_id"] != reservationTraceID("reservation-one").String() || entry["duration_ms"].(float64) < 1 {
+			t.Fatal("stage identity/status/duration lost", entry)
+		}
+		if strings.Contains(out.String(), "credential") || strings.Contains(out.String(), "private provider") {
+			t.Fatal("raw provider error leaked")
+		}
+	}
+}
+
+func TestExecutorControlLogsDistinguishPreparationFromHTTP(t *testing.T) {
+	out := captureExecutionLogs(t)
+	ctx := obslog.WithRequestID(reservationTrace(t.Context(), "reservation-one"), "http-one")
+	prepared := &preparedStart{ctx: ctx, requestID: "prepare-one", executorID: "executor-one", createdAt: time.Now(), startSentAt: time.Now()}
+	recordExecutorReadiness(prepared, proto.PreparationStatusPayload{ExecutorID: "executor-one", Reused: true})
+	recordExecutorStart(prepared, "turn-one")
+	for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) {
+		var entry map[string]any
+		if err := json.Unmarshal(line, &entry); err != nil {
+			t.Fatal(err)
+		}
+		if entry["request_id"] != "http-one" || entry["preparation_request_id"] != "prepare-one" || entry["trace_id"] != reservationTraceID("reservation-one").String() {
+			t.Fatal("control log confused HTTP and preparation identity", entry)
+		}
+	}
+}
+
+func TestLifecycleGateObservationPreservesCancellation(t *testing.T) {
+	out := captureExecutionLogs(t)
+	r := &runtimeLifecycle{ctx: t.Context(), gate: make(chan struct{}, 1), nodeID: "node-one"}
+	r.gate <- struct{}{}
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	if err := r.lock(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if len(r.gate) != 1 || !strings.Contains(out.String(), `"status":"cancelled"`) {
+		t.Fatal("cancelled waiter changed gate or log status")
+	}
+}
+
+type executionTraceConn struct {
+	writes chan []byte
+	closed chan struct{}
+	once   sync.Once
+}
+
+func (c *executionTraceConn) ReadMessage() (int, []byte, error) { <-c.closed; return 0, nil, io.EOF }
+func (c *executionTraceConn) WriteMessage(_ int, data []byte) error {
+	c.writes <- append([]byte(nil), data...)
+	return nil
+}
+func (c *executionTraceConn) WriteControl(int, []byte, time.Time) error { return nil }
+func (c *executionTraceConn) SetReadLimit(int64)                        {}
+func (c *executionTraceConn) SetReadDeadline(time.Time) error           { return nil }
+func (c *executionTraceConn) SetWriteDeadline(time.Time) error          { return nil }
+func (c *executionTraceConn) Close() error                              { c.once.Do(func() { close(c.closed) }); return nil }
+
+func TestExecutionSendPreservesPayloadAndCarriesTrace(t *testing.T) {
+	for _, traced := range []bool{false, true} {
+		conn := &executionTraceConn{writes: make(chan []byte, 2), closed: make(chan struct{})}
+		peer := runtimegateway.NewSession(conn, "device-one", "workspace-one", "test", nil, nil)
+		peer.Start()
+		ctx := t.Context()
+		if traced {
+			ctx = reservationTrace(ctx, "reservation-one")
+		}
+		payload := proto.ExecutionStartPayload{Handle: "handle-one", ExecutorID: "executor-one", RunID: "turn-one", Input: proto.TextInput("private input preserved on wire")}
+		if err := send(ctx, peer, proto.TypeExecutionStart, "prepare-one", payload); err != nil {
+			peer.Close("test done")
+			t.Fatal(err)
+		}
+		select {
+		case data := <-conn.writes:
+			var env proto.Envelope
+			if err := json.Unmarshal(data, &env); err != nil {
+				t.Fatal(err)
+			}
+			var got proto.ExecutionStartPayload
+			if err := env.DecodePayload(&got); err != nil {
+				t.Fatal(err)
+			}
+			expected, _ := json.Marshal(payload)
+			actual, _ := json.Marshal(got)
+			if env.ID != "prepare-one" || env.Type != proto.TypeExecutionStart || !bytes.Equal(expected, actual) {
+				t.Fatal("trace changed execution identity or payload")
+			}
+			if traced {
+				carrier, _ := obslog.TraceFromContext(ctx)
+				if env.Trace != carrier.String() {
+					t.Fatal("execution trace did not reach actual gateway wire")
+				}
+			} else if env.Trace != "" {
+				t.Fatal("missing trace changed into a fabricated caller trace")
+			}
+		case <-time.After(time.Second):
+			t.Fatal("gateway did not send")
+		}
+		peer.Close("test done")
+	}
+}
+
+func TestInitialInputOriginKeepsHTTPTraceAndSessionIdentity(t *testing.T) {
+	out := captureExecutionLogs(t)
+	ctx, origin := obslog.StartBackgroundTrace(t.Context(), "http")
+	recordInitialInputOrigin(ctx, "session-one")
+	var entry map[string]any
+	if err := json.Unmarshal(out.Bytes(), &entry); err != nil {
+		t.Fatal(err)
+	}
+	if entry["session_id"] != "session-one" || entry["trace_id"] != origin.Trace.String() {
+		t.Fatal("initial input origin lost its HTTP/session bridge", entry)
+	}
+	if _, present := entry["reservation_id"]; present {
+		t.Fatal("creation fabricated a reservation identity")
+	}
+}