feat: annotate traces with de.* attributes via metadata channel + procedural classifier

packages/opencode/src/altimate/observability/de-attributes.ts

@@ -56,6 +56,8 @@ export const DE_WAREHOUSE = {
   ROWS_RETURNED: "de.warehouse.rows_returned",
   /** Rows affected (INSERT/UPDATE/DELETE) */
   ROWS_AFFECTED: "de.warehouse.rows_affected",
+  /** Total rows in a table (from schema_inspect, not query result) */
+  ROWS_TOTAL: "de.warehouse.rows_total",
   /** Query ID from the warehouse (for linking to warehouse query history) */
   QUERY_ID: "de.warehouse.query_id",
   /** Whether the query hit a warehouse cache (Snowflake result cache, BQ cache) */
@@ -102,6 +104,8 @@ export const DE_SQL = {
 export const DE_DBT = {
   /** dbt command: run, test, build, compile, seed, snapshot */
   COMMAND: "de.dbt.command",
+  /** dbt project layer derived from model path: staging, intermediate, dim, fact, agg, mart, source, seed, macro, test */
+  LAYER: "de.dbt.layer",
   /** Model unique_id (e.g., model.my_project.stg_orders) */
   MODEL_UNIQUE_ID: "de.dbt.model.unique_id",
   /** Model short name */
@@ -206,6 +210,79 @@ export const DE_COST = {
   ATTRIBUTION_PROJECT: "de.cost.attribution.project",
 } as const
 
+// ---------------------------------------------------------------------------
+// Workflow & session classification (Layer 6) — populated by derived annotator
+// ---------------------------------------------------------------------------
+
+export const DE_WORKFLOW = {
+  /** Session workflow type: dbt_develop, dbt_troubleshoot, dbt_test, dbt_docs, sql_analysis, warehouse_exploration, project_scan, generic_file_edit */
+  TYPE: "de.workflow.type",
+  /** Session intent verb-phrase: "create model", "fix error", "add tests", "refactor model", "inspect schema", "run query" */
+  INTENT: "de.workflow.intent",
+  /** Confidence of the classifier (0.0-1.0) */
+  TYPE_CONFIDENCE: "de.workflow.type_confidence",
+} as const
+
+// ---------------------------------------------------------------------------
+// Session outcome (Layer 7) — populated by derived annotator
+// ---------------------------------------------------------------------------
+
+export const DE_OUTCOME = {
+  /** success, failure, interrupted, provider_routing_failure, validation_failure, no_op, partial_fix */
+  CLASS: "de.outcome.class",
+  /** Whether a build/test/run was actually executed */
+  EXECUTED: "de.outcome.executed",
+  /** Whether a fix or change was applied */
+  CHANGE_APPLIED: "de.outcome.change_applied",
+} as const
+
+// ---------------------------------------------------------------------------
+// Touched artifacts (Layer 8) — populated by derived annotator
+// ---------------------------------------------------------------------------
+
+export const DE_ARTIFACTS = {
+  /** File paths read (JSON array) */
+  FILES_READ: "de.artifacts.files_read",
+  /** File paths written or edited (JSON array) */
+  FILES_EDITED: "de.artifacts.files_edited",
+  /** dbt model names mentioned in the user prompt (JSON array) */
+  MODELS_MENTIONED: "de.artifacts.models_mentioned",
+  /** Fully-qualified tables referenced by executed SQL (JSON array) */
+  TABLES_REFERENCED: "de.artifacts.tables_referenced",
+} as const
+
+// ---------------------------------------------------------------------------
+// Environment capabilities (Layer 9) — populated by derived annotator
+// ---------------------------------------------------------------------------
+
+export const DE_ENV = {
+  /** Whether dbt is present in the working directory */
+  DBT_PRESENT: "de.env.dbt_present",
+  /** Whether dbt manifest.json exists */
+  DBT_MANIFEST_PRESENT: "de.env.dbt_manifest_present",
+  /** Warehouse adapter detected: snowflake, bigquery, postgres, duckdb, databricks, redshift */
+  WAREHOUSE_TYPE: "de.env.warehouse_type",
+  /** Orchestrators / data-quality / lint tools detected (JSON array): airflow, dagster, prefect, soda, sqlmesh, great_expectations, sqlfluff */
+  TOOLS_DETECTED: "de.env.tools_detected",
+} as const
+
+// ---------------------------------------------------------------------------
+// Generic tool classification (Layer 10) — populated by derived annotator
+// ---------------------------------------------------------------------------
+
+export const DE_TOOL = {
+  /** Tool category: warehouse, sql, dbt, schema, lineage, quality, finops, fs, exec, planning, generic */
+  CATEGORY: "de.tool.category",
+  /** Finer-grained subcategory (e.g., "sql.execute", "dbt.build", "fs.read") */
+  SUBCATEGORY: "de.tool.subcategory",
+  /** Vendor or framework the tool wraps: snowflake, bigquery, duckdb, dbt, altimate-core, etc. */
+  VENDOR: "de.tool.vendor",
+  /** bash-command intent when kind=tool name=bash: dbt, altimate_dbt, python_sql, sql, fs, vcs, install, other */
+  BASH_INTENT: "de.tool.bash_intent",
+  /** Concrete CLI invoked in a bash command: dbt, altimate-dbt, python3, psql, sqlfluff, etc. */
+  BASH_INVOKED: "de.tool.bash_invoked",
+} as const
+
 // ---------------------------------------------------------------------------
 // Convenience namespace
 // ---------------------------------------------------------------------------
@@ -217,4 +294,9 @@ export const DE = {
   DBT: DE_DBT,
   QUALITY: DE_QUALITY,
   COST: DE_COST,
+  WORKFLOW: DE_WORKFLOW,
+  OUTCOME: DE_OUTCOME,
+  ARTIFACTS: DE_ARTIFACTS,
+  ENV: DE_ENV,
+  TOOL: DE_TOOL,
 } as const

packages/opencode/src/altimate/observability/tracing.ts

@@ -22,6 +22,9 @@ import path from "path"
 import { Global } from "../../global"
 import { randomUUIDv7 } from "bun"
 import { Log } from "../../util/log"
+// altimate_change start — trace augmentation: procedural classifier
+import { annotateToolSpan, annotateSession } from "./annotator"
+// altimate_change end
 
 // ---------------------------------------------------------------------------
 // Trace data types — v2 schema
@@ -841,12 +844,14 @@ export class Trace {
           status: "completed"
           input: Record<string, unknown>
           output: string
+          metadata?: Record<string, unknown>
           time: { start: number; end: number }
         }
       | {
           status: "error"
           input: Record<string, unknown>
           error: string
+          metadata?: Record<string, unknown>
           time: { start: number; end: number }
         }
   }) {
@@ -874,6 +879,78 @@ export class Trace {
         safeInput = { _serialization_error: "Input contained circular references or non-serializable data" }
       }
 
+      // altimate_change start — trace augmentation: lift structured tool metadata
+      // (carried via ToolStateCompleted/Error.metadata in message-v2) onto the
+      // span. Only keys matching the `de.` semantic-convention prefix are
+      // promoted — other metadata stays internal to the tool framework
+      // (e.g., `truncated`, `outputPath`, `findings`, `success`).
+      //
+      // This is the "metadata channel": tools surface structured fields by
+      // setting `de.*` keys on their returned metadata object, without ever
+      // importing the observability layer.
+      //
+      // Values must be JSON-compatible (Date / BigInt / class instances are
+      // silently dropped or stringified, matching the trace file's serialization
+      // contract). Per-value and total byte caps below prevent runaway payloads
+      // from ballooning snapshots and HTTP exports — both truncation layers
+      // upstream (`tool.ts` output truncation, `tracing.ts` 10 KB output slice)
+      // operate on `output: string`, not on `metadata`.
+      //
+      // Byte size is computed as UTF-8 byte length, not JS string length
+      // (which counts UTF-16 code units): non-ASCII payloads (e.g.,
+      // CJK identifiers, emoji in error messages) would otherwise bypass
+      // the cap and balloon trace exports.
+      const ATTR_VALUE_MAX_BYTES = 10_000
+      const ATTR_TOTAL_MAX_BYTES = 32_000
+      const spanAttributes: Record<string, unknown> = {}
+      let totalBytes = 0
+
+      // Layer 1: tool-provided structured metadata (high fidelity — driver
+      // values, parser output). Filtered to the de.* prefix.
+      const rawMetadata = state.metadata
+      if (rawMetadata && typeof rawMetadata === "object") {
+        for (const [k, v] of Object.entries(rawMetadata)) {
+          if (typeof k !== "string" || !k.startsWith("de.") || v === undefined) continue
+          try {
+            const serialized = JSON.stringify(v)
+            if (serialized === undefined) continue
+            const valueBytes = Buffer.byteLength(serialized, "utf8")
+            if (valueBytes > ATTR_VALUE_MAX_BYTES) continue
+            if (totalBytes + valueBytes > ATTR_TOTAL_MAX_BYTES) continue
+            // Store original value (matches setSpanAttributes() at line ~1135 for
+            // consistent overwrite semantics if both paths target the same key).
+            spanAttributes[k] = v
+            totalBytes += valueBytes
+          } catch {
+            // Bad metadata value must never break the tracer
+          }
+        }
+      }
+
+      // Layer 2: derived classification from (name, input, output). Best-effort
+      // procedural — taxonomy lookup, bash intent, dbt layer from path, etc.
+      // Tool-provided metadata (Layer 1) wins on conflicts.
+      try {
+        const derived = annotateToolSpan(toolName, safeInput, isError ? errorStr : outputStr)
+        for (const [k, v] of Object.entries(derived)) {
+          if (v === undefined || k in spanAttributes) continue
+          try {
+            const serialized = JSON.stringify(v)
+            if (serialized === undefined) continue
+            const valueBytes = Buffer.byteLength(serialized, "utf8")
+            if (valueBytes > ATTR_VALUE_MAX_BYTES) continue
+            if (totalBytes + valueBytes > ATTR_TOTAL_MAX_BYTES) continue
+            spanAttributes[k] = v
+            totalBytes += valueBytes
+          } catch {
+            // best-effort
+          }
+        }
+      } catch {
+        // Annotator must never break the tracer
+      }
+      // altimate_change end
+
       this.spans.push({
         spanId: randomUUIDv7(),
         parentSpanId: this.currentGenerationSpanId ?? this.rootSpanId,
@@ -889,6 +966,7 @@ export class Trace {
         },
         input: safeInput,
         output: isError ? { error: errorStr } : outputStr.slice(0, 10000),
+        ...(Object.keys(spanAttributes).length > 0 && { attributes: spanAttributes }),
       })
       this.toolCallCount++
 
@@ -1233,6 +1311,34 @@ export class Trace {
 
     const trace = this.buildTraceFile(error)
 
+    // altimate_change start — trace augmentation: session-level rollup.
+    // Pure-function classifier attaches workflow / outcome / artifacts / env
+    // attributes to the root (session) span. Runs over the snapshotted trace
+    // (post buildTraceFile) so it sees the final state. Best-effort.
+    //
+    // Merge semantics: derived attributes fill only ABSENT keys. Anything an
+    // upstream caller set explicitly via setSpanAttributes(..., "session")
+    // wins — Layer 1 (caller-provided) > Layer 2 (derived), same rule as
+    // logToolCall.
+    try {
+      const sessionAttrs = annotateSession(trace)
+      if (Object.keys(sessionAttrs).length > 0) {
+        const rootSnapshotSpan = trace.spans.find((s) => s.spanId === this.rootSpanId)
+        if (rootSnapshotSpan) {
+          rootSnapshotSpan.attributes = { ...sessionAttrs, ...(rootSnapshotSpan.attributes ?? {}) }
+        }
+        // Also mirror onto the live root span so a subsequent snapshot()
+        // (e.g., via flushSync) sees the same attributes.
+        const liveRoot = this.spans.find((s) => s.spanId === this.rootSpanId)
+        if (liveRoot) {
+          liveRoot.attributes = { ...sessionAttrs, ...(liveRoot.attributes ?? {}) }
+        }
+      }
+    } catch {
+      // Session annotation must never break the trace
+    }
+    // altimate_change end
+
     // altimate_change start — trace: post-session summary (narrative, loops, topTools)
     try {
       // Top tools by call count
@@ -1346,6 +1452,24 @@ export class Trace {
       }
       const trace = this.buildTraceFile(error || "Process exited before trace completed")
       trace.summary.status = "crashed"
+
+      // altimate_change start — trace augmentation: session-level rollup on crash.
+      // Without this, crashed/interrupted sessions get per-tool attributes but
+      // no root rollup (workflow/outcome/artifacts/env). Pure-function call,
+      // best-effort, must never throw.
+      try {
+        const sessionAttrs = annotateSession(trace)
+        if (Object.keys(sessionAttrs).length > 0) {
+          const rootSnapshotSpan = trace.spans.find((s) => s.spanId === this.rootSpanId)
+          if (rootSnapshotSpan) {
+            rootSnapshotSpan.attributes = { ...sessionAttrs, ...(rootSnapshotSpan.attributes ?? {}) }
+          }
+        }
+      } catch {
+        // best-effort
+      }
+      // altimate_change end
+
       const safeId = (this.sessionId || "unknown").replace(/[/\\.:]/g, "_") || "unknown"
       const filePath = path.join(this.snapshotDir, `${safeId}.json`)
       // Must be synchronous — async writes won't complete before signal handler exits

packages/opencode/src/altimate/observability/viewer.ts

@@ -516,7 +516,7 @@ function showDetail(span) {
   }
   // DE attributes grouped
   var a = span.attributes || {};
-  var groups = [['de.warehouse.','Warehouse','cyan'],['de.sql.','SQL','secondary'],['de.dbt.','dbt','orange'],['de.quality.','Quality','green'],['de.cost.','Cost','orange']];
+  var groups = [['de.warehouse.','Warehouse','cyan'],['de.sql.','SQL','secondary'],['de.dbt.','dbt','orange'],['de.quality.','Quality','green'],['de.cost.','Cost','orange'],['de.workflow.','Workflow','accent'],['de.outcome.','Outcome','green'],['de.artifacts.','Artifacts','secondary'],['de.env.','Environment','cyan'],['de.tool.','Tool','accent']];
   var used = {};
   groups.forEach(function(g) {
     var entries = Object.keys(a).filter(function(k){return k.indexOf(g[0])===0;});