feat: add text_label bulk labeling tool for the analyst, surface provider stream errors

.opencode/skills/data-question/SKILL.md

@@ -139,9 +139,9 @@ Dialect references: [DuckDB query_table](https://www.duckdb.org/docs/current/gui
 
 Treat inferred labels as new data, not ground truth. Define a fixed rubric, allowed labels and uncertainty policy before classifying. For synthetic `messages(message_id, body)`, a rubric might classify a request to reverse a charge as `refund`, a question about shipping progress as `delivery`, and ambiguous text as `uncertain`. Specify tie-breaking for multi-topic messages.
 
-Inspect text, then define the complete population of stable IDs. If `task` and read-only `data-explorer` are available, assign disjoint ID ranges or catalog-defined batches with the same rubric/version. Each child returns exactly one `{message_id, label, rubric_version}` record per assigned ID, including `uncertain`; request records, not a guessed count. Otherwise label manageable batches directly using the same contract. Review uncertain cases consistently; batching makes counting reproducible, not classification inherently correct.
+Inspect text, then define the complete population of stable IDs. When `text_label` is available, materialize that population in `scratch` with `sql_materialize`, then call `text_label` with SQL over scratch, the stable ID and text column names, a fixed label set with a description for each label, and any rubric/tie-breaking instructions. It classifies every row through the session's LLM in bounded batches and persists the results in scratch; do not substitute regex or keyword guesses for semantic labels. `text_label` accepts at most 30,000 rows per call. For a larger population, split it into disjoint, deterministic id ranges (one call per range, each into a distinct result table), combine the persisted result tables in SQL, and verify that combined coverage equals the population count before claiming results. Check `rows_in`, `labeled`, `failed` and `unparseable`, and retrieve failed IDs before claiming complete coverage. Optional spot-check agreement measures repeatability, not correctness. Review uncertain cases consistently; batching makes counting reproducible, not classification inherently correct.
 
-Use an available permitted ingestion path to turn these records into a relation `labels(message_id, label, rubric_version)`. When `scratch` and `sql_materialize` are available, small audited annotation batches can be expressed as a SELECT relation and materialized there. For example (DuckDB/SQLite SELECT syntax):
+Use the persisted `text_label` output for SQL counts. It retains the source ID column name, so use `message_id` as the ID column for this example. To reuse the checks below, select `message_id`, `label` and your declared rubric version as `rubric_version` in a SELECT/CTE named `labels`; `population` must cover the same intended IDs. If `text_label` is unavailable, small audited annotation batches can use an available permitted ingestion path to form `labels(message_id, label, rubric_version)`. When `scratch` and `sql_materialize` are available, these small batches can be expressed as a SELECT relation and materialized there. For example (DuckDB/SQLite SELECT syntax):
 
 ```sql
 WITH annotations(message_id, label, rubric_version) AS (
@@ -152,7 +152,7 @@ WITH annotations(message_id, label, rubric_version) AS (
 SELECT * FROM annotations;
 ```
 
-These literals are newly inferred annotations, not a copied intermediate database result. Do not scale this into hand-copying thousands of IDs. For large jobs require a permitted way to ingest complete returned annotation records; `sql_materialize` alone does not imply an arbitrary file writer. If unavailable, report the coverage limitation. With no persistence tool, a small complete annotation CTE can still be checked and counted in a read-only query.
+These literals are newly inferred annotations, not a copied intermediate database result. Do not scale this into hand-copying thousands of IDs. Use `text_label` for large labeling jobs; if it is unavailable and there is no permitted way to ingest complete returned annotation records, report the coverage limitation. `sql_materialize` alone does not imply an arbitrary file writer. Never claim labels or counts you did not compute and persist; with no persistence tool, a small complete annotation CTE can still be checked and counted in a read-only query.
 
 **Prerequisites for the following SQL:** DuckDB/SQLite; `population(message_id)` contains exactly the intended unique IDs and `labels` contains the annotation records in the same engine. They may be scratch tables or CTEs. Run each SELECT separately as needed:
 

packages/opencode/src/agent/agent.ts

@@ -281,6 +281,9 @@ export const layer = Layer.effect(
           tool_lookup: "allow",
           // Materialized scratch tables for multi-step analysis (tool added separately; key is harmless if absent)
           sql_materialize: "allow",
+          // altimate_change start — model labels are persisted only in session-local scratch tables.
+          text_label: "allow",
+          // altimate_change end
           bash: "deny",
         } as const
         // Cap every configured ruleset at the subagent's read-only allow-list.

packages/opencode/src/altimate/enhance-prompt.ts

@@ -6,6 +6,7 @@ import { Config } from "@/config/config"
 import { Log } from "@/altimate/util/log"
 import { MessageV2 } from "@/session/message-v2"
 import { MessageID, SessionID } from "@/session/schema"
+import { readStreamText } from "./llm/stream-text"
 
 const ENHANCE_NAME = "enhance-prompt"
 const ENHANCE_TIMEOUT_MS = 15_000
@@ -132,15 +133,7 @@ export async function enhancePrompt(text: string): Promise<string> {
       ],
     })
 
-    // Consume the stream explicitly to avoid potential SDK hangs where
-    // .text never resolves if the stream isn't drained (Vercel AI SDK caveat)
-    for await (const _ of stream.fullStream) {
-      // drain
-    }
-    const result = await Promise.resolve(stream.text).catch((err: unknown) => {
-      log.error("failed to enhance prompt", { error: err })
-      return undefined
-    })
+    const result = await readStreamText(stream)
 
     if (!result) return text
 

packages/opencode/src/altimate/llm/stream-text.ts

@@ -0,0 +1,14 @@
+/** Read an AI SDK text stream without losing provider errors to .text's fallback error. */
+export async function readStreamText(stream: {
+  fullStream: AsyncIterable<{ type: string; error?: unknown }>
+  readonly text: PromiseLike<string>
+}): Promise<string> {
+  const text = Promise.resolve(stream.text)
+  // .text can reject with NoOutputGeneratedError while fullStream contains the
+  // original provider error. Observe its rejection, but prefer the stream error.
+  void text.catch(() => {})
+  for await (const part of stream.fullStream) {
+    if (part.type === "error") throw part.error
+  }
+  return text
+}

packages/opencode/src/altimate/review/ai-review.ts

@@ -7,6 +7,7 @@ import { MessageID, SessionID } from "@/session/schema"
 import { Log } from "@/altimate/util/log"
 import { Dispatcher } from "../native"
 import { type Finding, type ReviewCategory, type Severity, makeFinding } from "./finding"
+import { readStreamText } from "../llm/stream-text"
 
 const log = Log.create({ service: "ai-review" })
 
@@ -120,13 +121,7 @@ export async function runAiReview(input: AiReviewInput): Promise<Finding[]> {
       retries: 1,
       messages: [{ role: "user", content: buildUserMessage({ ...input, files }) }],
     })
-    for await (const _ of stream.fullStream) {
-      // drain to avoid SDK hangs
-    }
-    const text = await Promise.resolve(stream.text).catch((err: unknown) => {
-      log.error("ai review stream failed", { error: err })
-      return undefined
-    })
+    const text = await readStreamText(stream)
     if (!text) return []
 
     // Parse + clamp in core (the prompt-injection-resistant, advisory-only

packages/opencode/src/altimate/skill-selector.ts

@@ -8,6 +8,7 @@ import { MessageID, SessionID } from "../session/schema"
 import type { Skill } from "../skill"
 import type { Fingerprint } from "./fingerprint"
 import { Tracer } from "./observability/tracing"
+import { readStreamText } from "./llm/stream-text"
 
 const log = Log.create({ service: "skill-selector" })
 
@@ -195,11 +196,7 @@ async function runWithLLM(prompt: string, validNames: string[]): Promise<string[
       ],
     })
 
-    // Drain the stream
-    for await (const _ of stream.fullStream) {
-      // drain
-    }
-    const text = await stream.text
+    const text = await readStreamText(stream)
 
     // Parse: one skill name per line, filter to valid names
     const nameSet = new Set(validNames)