feat(upstream-link): link agent spans to upstream trace context (#166)

* feat(upstream-link): link agent spans to upstream trace context

Opt-in via LOONGSUITE_PILOT_UPSTREAM_LINK (default off). Upstream
traceparent is delivered through two schemes into a shared acp-correlate
store; a central TraceLinker stamps trace_id (all turn events) +
parent_span_id (the user-input "other" event) at collection time, so
convertEventLogToTrace reparents each turn's span tree under the upstream span.

- Scheme 2 (ACP, per-turn): adapter writes {sessionId, contentHash, contentPrefix, traceparent}
- Scheme 1 (env, session first turn): hook/plugin reads TRACEPARENT, writes a session record
- content match: exact hash or prefix (handles agent @file rewrite); consume-once by file order
- wired: claude-code / codex / qwen / qoder hooks + opencode plugin
- TTL cleanup for acp-correlate; fully fail-open; turn grouping preserved
- unit tests + real integrated OTLP verification (qwen / qoder-cli / opencode)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* docs(readme): document upstream trace linking toggle

Add an "Upstream Trace Linking" section to README (EN + zh-CN): the
LOONGSUITE_PILOT_UPSTREAM_LINK switch (values/default), the TTL var, the
config.json equivalents, and the two schemes (correlation file / env).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* docs(readme): clarify upstream linking is protocol-agnostic

The correlation-file scheme is not ACP-specific; the real requirement is
that the record's sessionId matches the gen_ai.session.id Pilot collects
and the content aligns. ACP is the primary case because session/new
provides that id up front.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* test(upstream-link): add synthetic load/stress test for stamp path

Covers correctness under interleaved cross-batch, cross-session traffic
(N sessions x M turns, mixed exact/prefix/duplicate/miss records) with
consume-once uniqueness assertions, plus measurements of resolveTurn
per-session scan cost and cache retention.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* perf(upstream-link): index turn records by contentHash for exact match

resolveTurn scanned all of a session's turn records on every lookup,
giving O(turns^2) per session under many turns. Index records by
contentHash for O(1) amortized exact match (the common path); keep
contentPrefix as a linear fallback for prompt-rewrite turns. Consume-once
file order preserved. Load test ratio for 2x turns dropped 2.68 -> 1.57.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(upstream-link): address PR review (hot-path sleep, ttl clamp, memory)

- [High] resolveWithRetry short-circuits when the session has no correlation
  file (and skips retry for empty other text), so an enabled-but-unwired setup
  no longer burns retries×retryDelayMs (~300ms) per new turn on the collect path.
- [Medium] clamp upstreamLink.ttlMs to a positive value; 0/negative previously
  made the retention cutoff now/future and deleted all fresh correlation files.
- [Medium] evict idle per-session state (CorrelationStore.states, TraceLinker
  cache/firstTurnBySession) on the retention cadence with the same TTL, bounding
  memory in a long-running daemon.
- [Low] treat an empty-string enable env as unset, not "true".
- [Low] log the resolved acp-correlate dir when linking is enabled, so a
  diverging config.json dataDir is diagnosable.

Adds unit coverage for all of the above. Full suite green (1843 passed).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: fangxiu-wf

README.md

@@ -113,6 +113,21 @@ Start with the guide that matches what you want to change:
 | See global config loading and retention settings | [Configuration Guide](docs/configuration.md)     |
 
 
+### Upstream Trace Linking (optional)
+
+Link collected agent spans to an **upstream** trace so each turn's span tree reparents under the upstream span. Disabled by default, and fully fail-open (never affects normal collection/reporting).
+
+| Setting | Values | Default |
+| ------- | ------ | ------- |
+| `LOONGSUITE_PILOT_UPSTREAM_LINK` (env) · `upstreamLink.enabled` (config.json) | `true` / `1` to enable; unset, `false`, or `0` to disable | disabled |
+| `LOONGSUITE_PILOT_UPSTREAM_LINK_TTL_MS` (env) · `upstreamLink.ttlMs` (config.json) | cleanup TTL in ms for `acp-correlate` files | `86400000` (24h) |
+
+When enabled, the upstream `traceparent` reaches Pilot via one of two schemes and is stamped onto collected records (`trace_id` on the turn, `parent_span_id` on the user-input event):
+
+- **Correlation file** (per-turn): the caller writes `{sessionId, contentHash, contentPrefix, traceparent}` to `~/.loongsuite-pilot/acp-correlate/<sessionId>.jsonl` when it sends a prompt. Linking is protocol-agnostic — the only requirement is that `sessionId` matches the `gen_ai.session.id` Pilot collects for that turn, and the content (hash or prefix) matches the collected user text. ACP clients satisfy this naturally (the `session/new` id flows into collection), so ACP is the primary case.
+- **Environment** (`TRACEPARENT` on the agent process): applied to the session's first turn, via the agent's hook. Use this when the caller cannot obtain a per-turn `sessionId` up front.
+
+
 ## Output Data
 
 

README.zh-CN.md

@@ -96,6 +96,20 @@ loongsuite-pilot info
 | 输出前进行密钥脱敏 | [数据脱敏](docs/zh-CN/masking.md) |
 | 查看全局配置加载顺序和保留策略 | [配置总览](docs/zh-CN/configuration.md) |
 
+### 上游 Trace 串联(可选)
+
+把采集到的 agent span 挂到**上游** trace 下,使每一轮的 span 树重挂到上游 span。默认关闭,且全程 fail-open(绝不影响正常采集/上报)。
+
+| 配置项 | 取值 | 默认 |
+| ------ | ---- | ---- |
+| `LOONGSUITE_PILOT_UPSTREAM_LINK`(环境变量)· `upstreamLink.enabled`(config.json) | `true` / `1` 开启;不设、`false` 或 `0` 关闭 | 关闭 |
+| `LOONGSUITE_PILOT_UPSTREAM_LINK_TTL_MS`(环境变量)· `upstreamLink.ttlMs`(config.json) | `acp-correlate` 文件清理 TTL(毫秒) | `86400000`(24 小时) |
+
+开启后,上游 `traceparent` 经以下两种方案之一到达 Pilot,并在采集时 stamp 到记录(turn 打 `trace_id`、用户输入事件打 `parent_span_id`):
+
+- **关联文件**(per-turn):调用方在发送 prompt 时,把 `{sessionId, contentHash, contentPrefix, traceparent}` 写入 `~/.loongsuite-pilot/acp-correlate/<sessionId>.jsonl`。串联与协议无关——唯一要求是 `sessionId` 等于 Pilot 采集该 turn 时的 `gen_ai.session.id`,且内容(hash 或前缀)能匹配采集到的用户文本。ACP client 天然满足(`session/new` 的 id 会贯穿采集),故 ACP 是主要场景。
+- **环境变量**(agent 进程上的 `TRACEPARENT`):经 agent 的 hook 作用于该会话的第一个 turn。适用于调用方无法预先拿到 per-turn `sessionId` 的情况。
+
 ## 输出数据
 
 | 后端 | 用途 |

assets/hooks/claude-code-hook-processor.mjs

@@ -33,6 +33,7 @@ import {
   writeJsonlRecords,
 } from './shared/event-emitter.mjs';
 import { logHookError } from './shared/error-logger.mjs';
+import { recordUpstreamContextOnce } from './shared/upstream-context.mjs';
 import {
   sanitizeObject,
   toJsonValue,
@@ -260,6 +261,9 @@ async function cmdStop() {
   const sessionId = requireSessionId(event, 'cmd');
   if (!sessionId) return;
 
+  // 方案1(env):首个 turn 读 TRACEPARENT 写 session 级关联记录(fail-open, 每 session 一次)
+  recordUpstreamContextOnce({ agentId: AGENT_ID, sessionId, dataDir: pilotDataDir() });
+
   const state = loadState(sessionId);
   if (!state.transcript_path && event.transcript_path) {
     state.transcript_path = event.transcript_path;

assets/hooks/codex-hook-processor.mjs

@@ -14,6 +14,7 @@ import os from 'node:os';
 import path from 'node:path';
 import crypto from 'node:crypto';
 import { logHookError } from './shared/error-logger.mjs';
+import { recordUpstreamContextOnce } from './shared/upstream-context.mjs';
 import {
   collectResourceAttributesFromEnv,
 } from './shared/resource-context.mjs';
@@ -52,6 +53,10 @@ function safePathPart(value) {
 function writeWakeupMarker(input) {
   const sessionId = typeof input.session_id === 'string' ? input.session_id : '';
   if (!sessionId) return;
+
+  // 方案1(env):首个 turn 读 TRACEPARENT 写 session 级关联记录(fail-open, 每 session 一次)
+  recordUpstreamContextOnce({ agentId: AGENT_ID, sessionId, dataDir: pilotDataDir() });
+
   const directory = path.join(pilotDataDir(), 'state', 'codex', 'transcript-wakeups');
   const marker = path.join(directory, `${safePathPart(sessionId)}.json`);
   const temporary = path.join(directory, `.${safePathPart(sessionId)}.${process.pid}.${crypto.randomUUID()}.tmp`);

assets/hooks/qoder-hook-processor.mjs

@@ -35,6 +35,7 @@ import {
   agentBaseFieldPatch,
   collectResourceAttributesFromEnv,
 } from './shared/resource-context.mjs';
+import { recordUpstreamContextOnce } from './shared/upstream-context.mjs';
 
 const RESOURCE_ATTRIBUTES = collectResourceAttributesFromEnv(process.env, { agentId: 'qoder' });
 const RESOURCE_BASE_FIELD_PATCH = agentBaseFieldPatch(RESOURCE_ATTRIBUTES);
@@ -211,6 +212,12 @@ async function main() {
   if (!payload) return;
 
   const { transcriptPath, sessionId, cwd } = payload;
+
+  // 方案1(env):首个 turn 读 TRACEPARENT 写 session 级关联记录(fail-open, 每 session 一次)
+  if (sessionId) {
+    recordUpstreamContextOnce({ agentId, sessionId, dataDir: path.dirname(LOONGSUITE_PILOT_LOGS_BASE_DIR) });
+  }
+
   const runtimeConfig = loadHookRuntimeConfig(path.join(HOOKS_DIR, '..'));
 
   const range = getLineRange(agentId, transcriptPath, sessionId);

assets/hooks/qwen-code-cli-hook-processor.mjs

@@ -36,6 +36,7 @@ import {
   writeJsonlRecords,
 } from './shared/event-emitter.mjs';
 import { logHookError } from './shared/error-logger.mjs';
+import { recordUpstreamContextOnce } from './shared/upstream-context.mjs';
 import {
   sanitizeObject,
   loadHookRuntimeConfig,
@@ -167,6 +168,9 @@ async function cmdStop() {
   const sessionId = requireSessionId(event, 'stop');
   if (!sessionId) return;
 
+  // 方案1(env):首个 turn 读 TRACEPARENT 写 session 级关联记录(fail-open, 每 session 一次)
+  recordUpstreamContextOnce({ agentId: AGENT_ID, sessionId, dataDir: pilotDataDir() });
+
   const state = loadState(sessionId);
   if (!state.transcript_path && event.transcript_path) {
     state.transcript_path = event.transcript_path;

assets/hooks/shared/upstream-context.mjs

@@ -0,0 +1,70 @@
+// Copyright 2026 Alibaba Group Holding Limited
+// SPDX-License-Identifier: Apache-2.0
+
+/**
+ * upstream-context.mjs — 方案1(env 注入)hook 侧 helper。
+ *
+ * 非交互式 CLI(headless)或交互式会话首个 turn,读进程环境变量 TRACEPARENT
+ * (由上游/适配层注入到 agent 子进程 env,hook 子进程继承),写一条 session 级
+ * 关联记录到 ${dataDir}/acp-correlate/<sessionId>.jsonl,供 pilot 归一阶段 stamp。
+ *
+ * 每个 session 只写一次:用 <sessionId>.env.lock 的 O_CREAT|O_EXCL(fs 'wx')抢锁,
+ * 已存在(EEXIST)即视为已写过,直接返回。
+ *
+ * 全程 fail-open:任何异常都不得影响宿主 agent。
+ */
+
+import fs from 'node:fs';
+import path from 'node:path';
+
+const TRACEPARENT_RE = /^00-([0-9a-f]{32})-([0-9a-f]{16})-[0-9a-f]{2}$/i;
+const ZERO_TRACE = '0'.repeat(32);
+const ZERO_SPAN = '0'.repeat(16);
+
+function safeName(value) {
+  return path.basename(String(value)).replace(/[^a-zA-Z0-9_-]/g, '_') || 'unknown';
+}
+
+function isValidTraceparent(tp) {
+  if (typeof tp !== 'string') return false;
+  const m = TRACEPARENT_RE.exec(tp.trim());
+  if (!m) return false;
+  return m[1].toLowerCase() !== ZERO_TRACE && m[2].toLowerCase() !== ZERO_SPAN;
+}
+
+/**
+ * @param {object} opts
+ * @param {string} opts.agentId    agent 标识(如 "claude-code")
+ * @param {string} opts.sessionId  当前会话 id
+ * @param {string} opts.dataDir    pilot 数据目录(各 processor 的 pilotDataDir())
+ */
+export function recordUpstreamContextOnce({ agentId, sessionId, dataDir }) {
+  try {
+    if (!sessionId || !dataDir) return;
+    const tp = (process.env.TRACEPARENT || '').trim();
+    if (!isValidTraceparent(tp)) return;
+
+    const dir = path.join(dataDir, 'acp-correlate');
+    fs.mkdirSync(dir, { recursive: true });
+
+    const base = safeName(sessionId);
+    const lock = path.join(dir, `${base}.env.lock`);
+    // O_CREAT|O_EXCL:抢锁成功者才写;已存在(EEXIST)= 已写过,直接返回。
+    try {
+      fs.closeSync(fs.openSync(lock, 'wx'));
+    } catch (err) {
+      if (err && err.code === 'EEXIST') return; // 正常路径,非错误
+      throw err;
+    }
+
+    const record = { type: 'session', sessionId, traceparent: tp, ts: new Date().toISOString() };
+    fs.appendFileSync(path.join(dir, `${base}.jsonl`), JSON.stringify(record) + '\n', 'utf-8');
+  } catch (err) {
+    // fail-open: 记录标记失败绝不能影响宿主 agent
+    try {
+      process.stderr.write(`[${agentId || 'hook'}] upstream_correlate skip: ${String((err && err.message) || err)}\n`);
+    } catch {
+      // ignore
+    }
+  }
+}

assets/plugins/opencode/plugin.mjs

@@ -372,13 +372,38 @@ function buildOutputMessages(pendingParts, finishReason) {
 // Event handlers
 // ---------------------------------------------------------------------------
 
+// 方案1(env):首个 turn 读 process.env.TRACEPARENT,写 session 级关联记录到
+// acp-correlate/<sessionId>.jsonl,每 session 只写一次(O_CREAT|O_EXCL 锁)。fail-open。
+const UPSTREAM_TP_RE = /^00-([0-9a-f]{32})-([0-9a-f]{16})-[0-9a-f]{2}$/i;
+function recordUpstreamEnvOnce(sessionID) {
+  try {
+    const tp = (process.env.TRACEPARENT || "").trim();
+    const m = UPSTREAM_TP_RE.exec(tp);
+    if (!m || m[1].toLowerCase() === "0".repeat(32) || m[2].toLowerCase() === "0".repeat(16)) return;
+    const dir = path.join(resolveDataDir(), "acp-correlate");
+    fs.mkdirSync(dir, { recursive: true });
+    const base = path.basename(String(sessionID)).replace(/[^a-zA-Z0-9_-]/g, "_") || "unknown";
+    try {
+      fs.closeSync(fs.openSync(path.join(dir, `${base}.env.lock`), "wx"));
+    } catch (e) {
+      if (e && e.code === "EEXIST") return; // 已写过, 正常返回
+      throw e;
+    }
+    const rec = { type: "session", sessionId: sessionID, traceparent: tp, ts: new Date().toISOString() };
+    fs.appendFileSync(path.join(dir, `${base}.jsonl`), JSON.stringify(rec) + "\n", "utf-8");
+  } catch {
+    // fail-open: 绝不影响 opencode
+  }
+}
+
 function handleChatMessage(inp, out, userId) {
   const sessionID = inp.sessionID;
   if (!sessionID) return;
 
   const session = getSession(sessionID);
 
   session.turnSeq += 1;
+  if (session.turnSeq === 1) recordUpstreamEnvOnce(sessionID);
   const turnId = `${sessionID}:t${session.turnSeq}`;
   const traceId = generateTraceId();
 

src/core/config-loader.ts

@@ -19,6 +19,7 @@ import type {
   SlsEndpoint,
   SlsMode,
   StatusBarConfig,
+  UpstreamLinkConfig,
 } from '../types/index.js';
 import { readJsonFile, resolveHome } from '../utils/fs-utils.js';
 import { createLogger } from '../utils/logger.js';
@@ -112,6 +113,11 @@ export interface ConfigFile {
   collectTrace?: boolean;
   serviceNamePrefix?: string;
 
+  upstreamLink?: {
+    enabled?: boolean;
+    ttlMs?: number;
+  };
+
   mask?: {
     mode?: string;
     types?: string[];
@@ -176,7 +182,7 @@ function env(key: string): string | undefined {
 
 function envBool(key: string, fallback: boolean): boolean {
   const v = env(key);
-  if (v === undefined) return fallback;
+  if (v === undefined || v.trim() === '') return fallback; // empty string == unset, not "true"
   return v !== 'false' && v !== '0';
 }
 
@@ -242,10 +248,21 @@ export async function loadConfig(): Promise<AnalyticsConfig> {
     fileCollection: buildFileCollectionConfig(file),
     pipeline: buildPipelineConfig(file),
     statusBar: buildStatusBarConfig(file),
+    upstreamLink: buildUpstreamLinkConfig(file),
     globalSpanAttributes: resolveGlobalSpanAttributes(file),
   };
 }
 
+function buildUpstreamLinkConfig(file: ConfigFile | null): UpstreamLinkConfig {
+  const ttlMs = envInt('LOONGSUITE_PILOT_UPSTREAM_LINK_TTL_MS', file?.upstreamLink?.ttlMs ?? 86_400_000); // 24h
+  return {
+    enabled: envBool('LOONGSUITE_PILOT_UPSTREAM_LINK', file?.upstreamLink?.enabled ?? false),
+    // Clamp: ttlMs <= 0 would make the retention cutoff Date.now() (or the future),
+    // deleting all freshly-written correlation files and silently breaking linking.
+    ttlMs: ttlMs > 0 ? ttlMs : 86_400_000,
+  };
+}
+
 /**
  * User-defined global span attributes: config.json `globalSpanAttributes` merged
  * with the `OTEL_SPAN_ATTRIBUTES` env (key1=value1,key2=value2). Env wins over

src/core/input-manager.ts

@@ -14,6 +14,7 @@ import { applyAgentContentPolicy } from '../normalization/agent-content-policy.j
 import { maskAgentActivityEntry } from '../mask/entry-masker.js';
 import { loadEnabledRules } from '../mask/rule-loader.js';
 import type { CompiledMaskRule } from '../mask/types.js';
+import type { TraceLinker } from './upstream-link/trace-linker.js';
 
 const logger = createLogger('InputManager');
 
@@ -48,6 +49,7 @@ export class InputManager extends EventEmitter {
   private agentsConfig: AgentsConfig = {};
   private maskConfig: MaskConfig = { mode: 'none', types: [] };
   private maskRules: CompiledMaskRule[] = [];
+  private traceLinker: TraceLinker | null = null;
 
   setFlusher(flusher: BaseFlusher): void {
     this.flusher = flusher;
@@ -74,6 +76,10 @@ export class InputManager extends EventEmitter {
     this.maskRules = loadEnabledRules(config);
   }
 
+  setTraceLinker(linker: TraceLinker): void {
+    this.traceLinker = linker;
+  }
+
   registerInput(input: BaseInput): void {
     if (this.inputs.has(input.id)) {
       logger.warn('input already registered', { id: input.id });
@@ -223,6 +229,16 @@ export class InputManager extends EventEmitter {
       }
     }
 
+    // Upstream trace linking: stamp trace_id / parent_span_id from correlation
+    // store so agent spans reparent under the upstream span. Fully fail-open.
+    if (this.traceLinker) {
+      try {
+        await this.traceLinker.stamp(entries);
+      } catch (err) {
+        logger.warn('trace linker stamp failed (skipped)', { inputId, error: String(err) });
+      }
+    }
+
     const policyAppliedEntries = entries.map(entry =>
       applyAgentContentPolicy(entry, this.agentsConfig),
     );