diff --git a/skills-seed/github-gitlab/SKILL.md b/skills-seed/github-gitlab/SKILL.md
index 20fcd8d3..676a284d 100644
--- a/skills-seed/github-gitlab/SKILL.md
+++ b/skills-seed/github-gitlab/SKILL.md
@@ -12,9 +12,20 @@ requiredCapabilities:
 Use this skill when the user asks to inspect repos, issues, pull requests, merge
 requests, code history, branches, or to make a small code change in a hosted repo.
 
-This is a resident-machine-auth connector. Prefer the native CLIs (`gh`, `glab`) and
-`git`, using the agent computer's logged-in state. Do not ask the user to paste tokens,
-and do not rely on proxy bearer-token injection.
+Prefer the native CLIs (`gh`, `glab`) and `git`. GitHub supports either the requesting
+user's product OAuth token or resident machine auth. When
+`$VAULT_TOKEN_API_GITHUB_COM` is present, pass it to `gh` only through `GH_TOKEN` for
+the command being run:
+
+```bash
+GH_TOKEN="$VAULT_TOKEN_API_GITHUB_COM" gh auth status
+GH_TOKEN="$VAULT_TOKEN_API_GITHUB_COM" gh repo view OWNER/REPO --json name,description,url,defaultBranchRef
+```
+
+Never print either variable, persist it in a file, or use another principal's token. If
+the vault variable is absent in a direct DM, the user has not connected GitHub through
+the product; use resident login only when the computer profile says durable process
+sessions are supported. Do not ask the user to paste a token.
 
 One exception: if the system prompt lists a shared org credential for a Git remote, the
 token is broker-only and never appears on the computer. For clone/fetch/push, use the
@@ -24,7 +35,8 @@ server-side.
 
 ## Logging in
 
-If `gh auth status` (or `glab auth status`) fails, log in with the native command:
+When no product OAuth token is available and `gh auth status` (or `glab auth status`)
+fails, log in with the native command only on a computer with durable process sessions:
 
 ```bash
 gh auth login
diff --git a/skills-seed/slack-drafts/scripts/slack_drafts.py b/skills-seed/slack-drafts/scripts/slack_drafts.py
index 982bc8b4..83294ca1 100755
--- a/skills-seed/slack-drafts/scripts/slack_drafts.py
+++ b/skills-seed/slack-drafts/scripts/slack_drafts.py
@@ -15,9 +15,10 @@
 import json
 import os
 import re
-import subprocess
 import sys
+import urllib.error
 import urllib.parse
+import urllib.request
 import uuid
 
 API = "https://slack.com/api"
@@ -28,18 +29,21 @@ def call(method: str, body: dict | None = None, query: dict | None = None):
     if not tok:
         sys.exit("no Slack token: ask the user to connect Slack")
     url = f"{API}/{method}" + (f"?{urllib.parse.urlencode(query, doseq=True)}" if query else "")
-    cmd = ["curl", "-sS", "--fail-with-body", "--max-time", "60",
-           "-H", f"Authorization: Bearer {tok}", url]
+    headers = {"Authorization": f"Bearer {tok}"}
+    data = None
     if body is not None:
-        cmd += ["-H", "Content-Type: application/json; charset=utf-8", "--data-binary", "@-"]
-    proc = subprocess.run(cmd, input=json.dumps(body) if body is not None else None,
-                          capture_output=True, text=True)
-    if proc.returncode != 0:
-        sys.exit(f"slack api unreachable on {method}: {proc.stderr.strip()[:300]}")
+        headers["Content-Type"] = "application/json; charset=utf-8"
+        data = json.dumps(body).encode("utf-8")
+    request = urllib.request.Request(url, data=data, headers=headers)
     try:
-        payload = json.loads(proc.stdout)
+        with urllib.request.urlopen(request, timeout=60) as response:
+            response_text = response.read().decode("utf-8")
+    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as error:
+        sys.exit(f"slack api unreachable on {method}: {str(error)[:300]}")
+    try:
+        payload = json.loads(response_text)
     except ValueError:
-        sys.exit(f"slack api returned non-JSON on {method}: {proc.stdout[:300]}")
+        sys.exit(f"slack api returned non-JSON on {method}: {response_text[:300]}")
     if not payload.get("ok"):
         err = payload.get("error")
         hints = {
diff --git a/src/api/slack-core-client.ts b/src/api/slack-core-client.ts
index 67f88aa9..6e979b55 100644
--- a/src/api/slack-core-client.ts
+++ b/src/api/slack-core-client.ts
@@ -28,6 +28,7 @@ import { resolveRuntimeChoiceDurable, type RuntimeChoice } from "../harness/harn
 import { modelDisplayName } from "../model/pi-models.ts";
 
 interface SlackRunHooks {
+  onDelta?(delta: string): void | Promise<void>;
   onFirstBlock?(text: string): void;
   onSurfacePosted?(): void;
   onTasks?(tasks: Array<{ id: string; title: string; status: TaskStatus }>): void | Promise<void>;
@@ -187,6 +188,7 @@ export function createSlackCoreClient(deps: SlackCoreClientDeps): SlackCoreClien
       const waiters = terminalWaiters.get(runId) ?? new Set();
       terminalWaiters.set(runId, waiters);
       const unsubscribe = deps.turnStream.subscribe(runId, {
+        onDelta: (delta) => hooks.onDelta?.(delta),
         onFirstBlock: signalFirstBlock,
         onSurfacePosted: signalSurface,
       });
@@ -221,6 +223,7 @@ export function createSlackCoreClient(deps: SlackCoreClientDeps): SlackCoreClien
             if (isTerminal(run.status)) {
               const view = await deps.app.getRun(runId);
               await emitTasks().catch(swallowAs("slack-core-client: terminal task refresh", undefined));
+              await deps.turnStream.drain(runId);
               if (view?.surfacePosted) signalSurface();
               return (view?.result as TurnResult | null | undefined) ?? null;
             }
diff --git a/src/core/orchestrator.ts b/src/core/orchestrator.ts
index 373b496a..c1f1f5c7 100644
--- a/src/core/orchestrator.ts
+++ b/src/core/orchestrator.ts
@@ -36,6 +36,7 @@ import {
   configuredConnectorProviders,
   connectorStatusIsStale,
   refreshConnectorStatus,
+  usableConnectorProviders,
 } from "../credentials/connector-status.ts";
 import { renderComputerBlock, renderResidentLoginsBlock, renderConnectedAppsBlock } from "./environment-facts.ts";
 import { PROVIDERS } from "../connectors/oauth.ts";
@@ -826,11 +827,28 @@ export function createOrchestrator(deps: OrchestratorDeps): Orchestrator {
           : "";
 
       await deps.skillsReady;
-      const configuredProviders = deps.resolveConnectorClient
+      let connectorStatus = null;
+      if (!strictReadOnly && conversation.kind === "dm") {
+        try {
+          connectorStatus = deps.connectorStatusCache ? await deps.connectorStatusCache.get(actor.id) : null;
+          if (
+            deps.connectorTokens &&
+            deps.connectorStatusCache &&
+            connectorStatusIsStale(connectorStatus, Date.now())
+          ) {
+            connectorStatus = await refreshConnectorStatus(deps.connectorTokens, actor.id, Date.now());
+            await deps.connectorStatusCache.put(connectorStatus);
+          }
+        } catch (e) {
+          swallow("orchestrator: connected-app status", e);
+        }
+      }
+      const oauthConfiguredProviders = deps.resolveConnectorClient
         ? await configuredConnectorProviders(deps.resolveConnectorClient).catch(
             swallowAs("orchestrator: configured connector providers", []),
           )
         : [];
+      const configuredProviders = usableConnectorProviders(oauthConfiguredProviders, connectorStatus);
       const visibleSkillsForTurn = async (): Promise<SkillResolution[]> =>
         filterConnectorSkills((await deps.skills?.visibleFor(skillScopes)) ?? [], configuredProviders);
       const visibleSkills = await visibleSkillsForTurn();
@@ -1487,18 +1505,8 @@ export function createOrchestrator(deps: OrchestratorDeps): Orchestrator {
           }
         }
         if (!strictReadOnly && deps.resolveConnectorClient && conversation.kind === "dm") {
-          let status = null;
-          try {
-            status = deps.connectorStatusCache ? await deps.connectorStatusCache.get(actor.id) : null;
-            if (deps.connectorTokens && deps.connectorStatusCache && connectorStatusIsStale(status, Date.now())) {
-              status = await refreshConnectorStatus(deps.connectorTokens, actor.id, Date.now());
-              await deps.connectorStatusCache.put(status);
-            }
-          } catch (e) {
-            swallow("orchestrator: connected-app status", e);
-          }
           const connectionsUrl = deps.publicWebUrl ? `${deps.publicWebUrl.replace(/\/$/, "")}/keychain` : undefined;
-          systemPrompt += `\n\n${renderConnectedAppsBlock(status, configuredProviders, connectionsUrl)}`;
+          systemPrompt += `\n\n${renderConnectedAppsBlock(connectorStatus, configuredProviders, connectionsUrl)}`;
         }
         systemPrompt += memoryBlock;
         if (onboardingBlock) systemPrompt += `\n\n${onboardingBlock}`;
diff --git a/src/credentials/connector-status.ts b/src/credentials/connector-status.ts
index 5effbeaa..dd1d3403 100644
--- a/src/credentials/connector-status.ts
+++ b/src/credentials/connector-status.ts
@@ -106,6 +106,17 @@ export function connectorLabel(name: string): string {
   return PROVIDER_LABELS[name] ?? name.charAt(0).toUpperCase() + name.slice(1);
 }
 
+export function usableConnectorProviders(
+  configuredProviders: readonly string[],
+  status: ConnectorStatusRecord | null,
+): string[] {
+  const usable = new Set(configuredProviders);
+  for (const [provider, entry] of Object.entries(status?.providers ?? {})) {
+    if (entry.connected && !entry.needsReconnect) usable.add(provider);
+  }
+  return [...usable];
+}
+
 export async function configuredConnectorProviders(resolveClient: OAuthClientResolver): Promise<string[]> {
   const configured = await Promise.all(
     Object.keys(PROVIDERS).map(async (provider) => {
diff --git a/src/runs/turn-stream.ts b/src/runs/turn-stream.ts
index 26373579..97d3587d 100644
--- a/src/runs/turn-stream.ts
+++ b/src/runs/turn-stream.ts
@@ -9,6 +9,7 @@ export interface TurnStream {
   markSurfacePosted(runId: string): void;
   surfacePosted(runId: string): boolean;
   snapshot(runId: string): string | null;
+  drain(runId: string): Promise<void>;
   markReplyDone(runId: string): void;
   isReplyDone(runId: string): boolean;
   end(runId: string): void;
@@ -16,6 +17,7 @@ export interface TurnStream {
 }
 
 interface TurnStreamListener {
+  onDelta?(delta: string): void | Promise<void>;
   onFirstBlock?(text: string): void;
   onSurfacePosted?(): void;
 }
@@ -32,6 +34,11 @@ interface Entry {
   timer: ReturnType<typeof setTimeout> | null;
 }
 
+interface ListenerDelivery {
+  tail: Promise<void>;
+  error?: unknown;
+}
+
 export interface TurnStreamOptions {
   maxChars?: number;
   graceMs?: number;
@@ -62,6 +69,19 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream {
   const graceMs = opts.graceMs ?? DEFAULT_GRACE_MS;
   const runs = new Map<string, Entry>();
   const listeners = new Map<string, Set<TurnStreamListener>>();
+  const deliveries = new Map<string, Map<TurnStreamListener, ListenerDelivery>>();
+
+  const enqueueDelta = (runId: string, listener: TurnStreamListener, delta: string): void => {
+    const delivery = deliveries.get(runId)?.get(listener);
+    if (!delivery || !listener.onDelta) return;
+    delivery.tail = delivery.tail.then(async () => {
+      try {
+        await listener.onDelta?.(delta);
+      } catch (error) {
+        delivery.error ??= error;
+      }
+    });
+  };
 
   const ensure = (runId: string): Entry => {
     let entry = runs.get(runId);
@@ -101,6 +121,7 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream {
       if (entry.firstBlockOpen && entry.firstBlock.length < FIRST_BLOCK_MAX_CHARS)
         entry.firstBlock = (entry.firstBlock + delta).slice(0, FIRST_BLOCK_MAX_CHARS);
       if (entry.text.length < maxChars) entry.text = (entry.text + delta).slice(0, maxChars);
+      for (const l of listeners.get(runId) ?? []) enqueueDelta(runId, l, delta);
     },
 
     publishBlockStart(runId) {
@@ -143,6 +164,13 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream {
       return text ? text : null;
     },
 
+    async drain(runId) {
+      const pending = [...(deliveries.get(runId)?.values() ?? [])];
+      await Promise.all(pending.map((delivery) => delivery.tail));
+      const failed = pending.find((delivery) => delivery.error !== undefined);
+      if (failed) throw failed.error;
+    },
+
     markReplyDone(runId) {
       const entry = runs.get(runId);
       if (entry) entry.replyDone = true;
@@ -170,9 +198,22 @@ export function createTurnStream(opts: TurnStreamOptions = {}): TurnStream {
         listeners.set(runId, set);
       }
       set.add(listener);
+      let deliveryMap = deliveries.get(runId);
+      if (!deliveryMap) {
+        deliveryMap = new Map();
+        deliveries.set(runId, deliveryMap);
+      }
+      deliveryMap.set(listener, { tail: Promise.resolve() });
+      const buffered = runs.get(runId)?.text;
+      if (buffered) enqueueDelta(runId, listener, buffered);
       return () => {
         set.delete(listener);
         if (set.size === 0 && listeners.get(runId) === set) listeners.delete(runId);
+        const delivery = deliveryMap.get(listener);
+        void delivery?.tail.finally(() => {
+          if (deliveryMap.get(listener) === delivery) deliveryMap.delete(listener);
+          if (deliveryMap.size === 0 && deliveries.get(runId) === deliveryMap) deliveries.delete(runId);
+        });
       };
     },
   };
diff --git a/src/slack/core-bridge.ts b/src/slack/core-bridge.ts
index 5946cb8b..811a0e8b 100644
--- a/src/slack/core-bridge.ts
+++ b/src/slack/core-bridge.ts
@@ -10,6 +10,7 @@ interface CoreCallHooks {
   /** The turn was folded into a run that was ALREADY live (a mid-turn steer), so this handler
    *  owns nothing: the envelope is durably accepted, but the reply belongs to the run's owner. */
   onSteered?: (runId: string) => void;
+  onDelta?: (delta: string) => void | Promise<void>;
   onFirstBlock?: (text: string) => void;
   onSurfacePosted?: () => void;
   onTasks?: (tasks: RunTaskView[]) => void;
@@ -140,6 +141,7 @@ export function createCoreBridge(core: SlackCoreClient): CoreBridge {
     let result: TurnResult | null;
     try {
       result = await core.waitRun(runId, {
+        ...(hooks.onDelta ? { onDelta: hooks.onDelta } : {}),
         ...(hooks.onFirstBlock ? { onFirstBlock: hooks.onFirstBlock } : {}),
         ...(hooks.onSurfacePosted ? { onSurfacePosted: hooks.onSurfacePosted } : {}),
         ...(hooks.onTasks ? { onTasks: hooks.onTasks } : {}),
diff --git a/src/slack/lib.ts b/src/slack/lib.ts
index d0d33980..831f24fe 100644
--- a/src/slack/lib.ts
+++ b/src/slack/lib.ts
@@ -139,4 +139,6 @@ export {
   renderTaskList,
   type TaskListPresenter,
   createTaskListPresenter,
+  type NativeAgentPresenter,
+  createNativeAgentPresenter,
 } from "./presenters.ts";
diff --git a/src/slack/messaging.ts b/src/slack/messaging.ts
index 3e7321b7..1795129e 100644
--- a/src/slack/messaging.ts
+++ b/src/slack/messaging.ts
@@ -74,6 +74,64 @@ export function stripSlackDirectives(text: string): string {
   return stripAgentRequestDirectives(stripReactionDirectives(text));
 }
 
+export interface StreamingReplyFilter {
+  push(delta: string): string;
+  flush(): string;
+}
+
+export function createStreamingReplyFilter(): StreamingReplyFilter {
+  const directiveStarts = ["[[react:", "[[ask-agent:"];
+  let pending = "";
+  let insideDirective = false;
+  const take = (flush: boolean): string => {
+    let visible = "";
+    for (;;) {
+      if (insideDirective) {
+        const end = pending.indexOf("]]");
+        if (end === -1) {
+          if (flush) pending = "";
+          return visible;
+        }
+        pending = pending.slice(end + 2);
+        insideDirective = false;
+        continue;
+      }
+      const possibleStart = pending.indexOf("[[");
+      if (possibleStart === -1) {
+        const held = !flush && pending.endsWith("[") ? 1 : 0;
+        visible += pending.slice(0, pending.length - held);
+        pending = pending.slice(pending.length - held);
+        return visible;
+      }
+      visible += pending.slice(0, possibleStart);
+      pending = pending.slice(possibleStart);
+      const lower = pending.toLowerCase();
+      if (directiveStarts.some((start) => lower.startsWith(start))) {
+        insideDirective = true;
+        continue;
+      }
+      if (directiveStarts.some((start) => start.startsWith(lower))) {
+        if (flush) {
+          visible += pending;
+          pending = "";
+        }
+        return visible;
+      }
+      visible += pending[0];
+      pending = pending.slice(1);
+    }
+  };
+  return {
+    push(delta) {
+      pending += delta;
+      return take(false);
+    },
+    flush() {
+      return take(true);
+    },
+  };
+}
+
 export async function applyAndLogReactions(
   client: any,
   channel: string,
diff --git a/src/slack/presenters.ts b/src/slack/presenters.ts
index 25150f86..b94a74f9 100644
--- a/src/slack/presenters.ts
+++ b/src/slack/presenters.ts
@@ -175,6 +175,181 @@ export interface TaskListPresenter {
   settle(): Promise<void>;
 }
 
+type NativeAgentChunk =
+  | { type: "markdown_text"; text: string }
+  | {
+      type: "task_update";
+      id: string;
+      title: string;
+      status: "pending" | "in_progress" | "complete" | "error";
+    };
+
+const NATIVE_STREAM_BUFFER_SIZE = 256;
+const NATIVE_STREAM_CHUNK_SIZE = 12_000;
+
+export interface NativeAgentPresenter {
+  begin(): Promise<void>;
+  onDelta(delta: string): Promise<boolean>;
+  onTasks(tasks: RunTaskView[]): Promise<boolean>;
+  finalize(): Promise<boolean>;
+  settle(): Promise<void>;
+  ownsSurface(): boolean;
+}
+
+export function createNativeAgentPresenter(deps: {
+  setStatus(status: string): Promise<void>;
+  start(chunks: NativeAgentChunk[]): Promise<string | undefined>;
+  append(ts: string, chunks: NativeAgentChunk[]): Promise<void>;
+  stop(ts: string): Promise<void>;
+  remove(ts: string): Promise<void>;
+  checkpoint(ts: string): Promise<void>;
+  onSurfacePosted(): void;
+  onError?(error: unknown): void;
+}): NativeAgentPresenter {
+  let state: "idle" | "starting" | "active" | "disabled" | "orphaned" | "stopped" = "idle";
+  let messageTs: string | undefined;
+  let chain = Promise.resolve();
+  let statusStarted = false;
+  let statusCleared = false;
+  let pendingText = "";
+  const nativeTaskStatus = (status: RunTaskStatus): "pending" | "in_progress" | "complete" | "error" => {
+    if (status === "completed" || status === "skipped") return "complete";
+    if (status === "failed") return "error";
+    return status;
+  };
+  const clearStatus = async (): Promise<void> => {
+    if (!statusStarted || statusCleared) return;
+    statusCleared = true;
+    await deps.setStatus("").catch((error) => deps.onError?.(error));
+  };
+  const enqueue = (operation: () => Promise<void>): Promise<void> => {
+    chain = chain.then(operation);
+    return chain;
+  };
+  const send = async (chunks: NativeAgentChunk[]): Promise<boolean> => {
+    if (state === "disabled" || state === "stopped") return false;
+    if (!messageTs && state === "idle") state = "starting";
+    await enqueue(async () => {
+      if (state === "disabled" || state === "stopped") return;
+      let uncheckpointedTs: string | undefined;
+      try {
+        if (!messageTs) {
+          const ts = await deps.start(chunks);
+          if (!ts) throw new Error("chat.startStream returned no message timestamp");
+          uncheckpointedTs = ts;
+          await deps.checkpoint(ts);
+          uncheckpointedTs = undefined;
+          messageTs = ts;
+          state = "active";
+          deps.onSurfacePosted();
+          return;
+        }
+        await deps.append(messageTs, chunks);
+      } catch (error) {
+        const failedStreamTs = uncheckpointedTs ?? messageTs;
+        let cleanupFailed = false;
+        if (failedStreamTs) {
+          try {
+            await deps.remove(failedStreamTs);
+            messageTs = undefined;
+          } catch (removeError) {
+            cleanupFailed = true;
+            deps.onError?.(removeError);
+          }
+        }
+        state = cleanupFailed ? "orphaned" : "disabled";
+        deps.onError?.(error);
+      }
+    });
+    return state === "active";
+  };
+  const flushPendingText = async (): Promise<boolean> => {
+    if (!pendingText) return state === "starting" || state === "active";
+    while (pendingText) {
+      const text = pendingText.slice(0, NATIVE_STREAM_CHUNK_SIZE);
+      pendingText = pendingText.slice(NATIVE_STREAM_CHUNK_SIZE);
+      if (!(await send([{ type: "markdown_text", text }]))) return false;
+    }
+    return true;
+  };
+  return {
+    async begin() {
+      statusStarted = true;
+      await enqueue(async () => {
+        try {
+          await deps.setStatus("Thinking…");
+        } catch (error) {
+          deps.onError?.(error);
+        }
+      });
+    },
+    async onDelta(delta) {
+      if (!delta) return state === "starting" || state === "active";
+      if (state === "disabled" || state === "stopped") return false;
+      pendingText += delta;
+      if (state === "idle") state = "starting";
+      if (pendingText.length < NATIVE_STREAM_BUFFER_SIZE) return true;
+      return flushPendingText();
+    },
+    async onTasks(tasks) {
+      if (!tasks.length) return state === "starting" || state === "active";
+      if (pendingText && !(await flushPendingText())) return false;
+      const chunks: NativeAgentChunk[] = tasks.slice(0, 20).map((task) => ({
+        type: "task_update",
+        id: task.id,
+        title: task.title
+          .replace(/[\r\n\t]+/g, " ")
+          .replace(/\s+/g, " ")
+          .trim()
+          .slice(0, 120),
+        status: nativeTaskStatus(task.status),
+      }));
+      return send(chunks);
+    },
+    async finalize() {
+      if (pendingText) await flushPendingText();
+      await chain;
+      if (state === "orphaned") {
+        await clearStatus();
+        return true;
+      }
+      if (state !== "active" || !messageTs) {
+        await clearStatus();
+        return false;
+      }
+      try {
+        await deps.stop(messageTs);
+        state = "stopped";
+        await clearStatus();
+        return true;
+      } catch (error) {
+        state = "orphaned";
+        deps.onError?.(error);
+        await clearStatus();
+        return true;
+      }
+    },
+    async settle() {
+      pendingText = "";
+      await chain;
+      if (state === "active" && messageTs) {
+        try {
+          await deps.remove(messageTs);
+          messageTs = undefined;
+          state = "stopped";
+        } catch (error) {
+          state = "orphaned";
+          deps.onError?.(error);
+        }
+      }
+      await clearStatus();
+    },
+    ownsSurface() {
+      return state === "starting" || state === "active" || state === "orphaned" || state === "stopped";
+    },
+  };
+}
+
 export function createTaskListPresenter(deps: {
   post(text: string, blocks: Array<Record<string, unknown>>): Promise<string | undefined>;
   update(ts: string, text: string, blocks: Array<Record<string, unknown>>): Promise<void>;
diff --git a/src/slack/turn-handler.ts b/src/slack/turn-handler.ts
index 0214c17a..7a10dc66 100644
--- a/src/slack/turn-handler.ts
+++ b/src/slack/turn-handler.ts
@@ -7,6 +7,7 @@ import {
   type OverheardMessage,
   type ReactionTally,
   type RunTaskView,
+  type NativeAgentPresenter,
   type SlackFile,
   type TaskListPresenter,
   DEFAULT_ACK_REACTIONS,
@@ -16,6 +17,7 @@ import {
   buildReactionTurnText,
   createAckPresenter,
   createDeduper,
+  createNativeAgentPresenter,
   createTaskListPresenter,
   createThreadTracker,
   decodeSlackEntities,
@@ -257,6 +259,7 @@ export function createTurnHandler(deps: {
 
     let queuedRunId: string | undefined;
     let taskList: TaskListPresenter | undefined;
+    let nativeAgent: NativeAgentPresenter | undefined;
     const ack = inc.unprompted
       ? undefined
       : createAckPresenter({
@@ -292,6 +295,7 @@ export function createTurnHandler(deps: {
     }
     const settleAck = async (): Promise<void> => {
       await ack?.settle().catch(swallowAs("slack: ack settle", undefined));
+      await nativeAgent?.settle().catch(swallowAs("slack: native agent settle", undefined));
     };
 
     if (inc.kind === "channel") {
@@ -403,6 +407,34 @@ export function createTurnHandler(deps: {
 
     if (inc.unprompted && !text.trim() && attachments.length === 0) return;
 
+    if (!inc.unprompted) {
+      const nativeThreadTs = replyThreadTs ?? inc.ts;
+      nativeAgent = createNativeAgentPresenter({
+        setStatus: (status) =>
+          client.assistant.threads
+            .setStatus({ channel_id: inc.channel, thread_ts: nativeThreadTs, status })
+            .then(() => {}),
+        start: async (chunks) => {
+          const response = await client.chat.startStream({
+            channel: inc.channel,
+            thread_ts: nativeThreadTs,
+            chunks,
+            task_display_mode: "timeline",
+            ...(inc.kind === "channel" ? { recipient_team_id: ids.ownTeamId, recipient_user_id: inc.userId } : {}),
+          });
+          return typeof response.ts === "string" ? response.ts : undefined;
+        },
+        append: (ts, chunks) => client.chat.appendStream({ channel: inc.channel, ts, chunks }).then(() => {}),
+        stop: (ts) => client.chat.stopStream({ channel: inc.channel, ts }).then(() => {}),
+        remove: (ts) => client.chat.delete({ channel: inc.channel, ts }).then(() => {}),
+        checkpoint: async (ts) => {
+          if (queuedRunId) await checkpointRunEditRef(queuedRunId, ts);
+        },
+        onSurfacePosted: () => ack?.onSurfacePosted(),
+        onError: (error) => console.error("[slack-plugin] native agent update failed:", errMessage(error)),
+      });
+      void nativeAgent.begin();
+    }
     const turn: Omit<CoreTurnBody, "approval"> = {
       actor,
       conversation: {
@@ -453,18 +485,12 @@ export function createTurnHandler(deps: {
           // Folded into a live run: the envelope is durably accepted just the same, but the run
           // stays pinned to its own handler — claiming it here would unpin it on the way out.
           onSteered: () => inc.ackGate?.persisted(),
-          ...(ack
-            ? {
-                onFirstBlock: (blockText: string) => {
-                  ack.onFirstBlock(cleanAgentReplyForSlack(blockText).text);
-                },
-                onSurfacePosted: () => ack.onSurfacePosted(),
-              }
-            : {}),
+          ...(ack ? { onSurfacePosted: () => ack.onSurfacePosted() } : {}),
           ...(taskList
             ? {
                 onTasks: async (tasks: RunTaskView[]) => {
                   await ack?.drain();
+                  if (await nativeAgent?.onTasks(tasks)) return;
                   await taskList?.onTasks(tasks);
                 },
               }
@@ -523,6 +549,10 @@ export function createTurnHandler(deps: {
       const postText = reply;
       const tDeliverStart = performance.now();
       let finalizedTaskList = false;
+      if (postText && (queuedRunId || nativeAgent?.ownsSurface())) {
+        await nativeAgent?.onDelta(postText);
+      }
+      const finalizedNative = (await nativeAgent?.finalize()) ?? false;
       if (result.attachments?.length) {
         let uploadError: unknown;
         try {
@@ -539,13 +569,13 @@ export function createTurnHandler(deps: {
           console.error("[slack-plugin] file upload failed:", (err as Error).message);
         }
         await settleAck();
-        if (postText) finalizedTaskList = (await taskList?.finalize(postText)) ?? false;
-        if (postText && !finalizedTaskList) await postReply(postText);
+        if (postText && !finalizedNative) finalizedTaskList = (await taskList?.finalize(postText)) ?? false;
+        if (postText && !finalizedNative && !finalizedTaskList) await postReply(postText);
         if (uploadError) await postReply(uploadFailureNote(uploadError));
       } else {
         await settleAck();
-        if (postText) finalizedTaskList = (await taskList?.finalize(postText)) ?? false;
-        if (postText && !finalizedTaskList) await postReply(postText);
+        if (postText && !finalizedNative) finalizedTaskList = (await taskList?.finalize(postText)) ?? false;
+        if (postText && !finalizedNative && !finalizedTaskList) await postReply(postText);
       }
       if (queuedRunId) {
         reportTurnMetrics(queuedRunId, {
diff --git a/test/connector-status.test.ts b/test/connector-status.test.ts
index 306d1577..aa9ff2f2 100644
--- a/test/connector-status.test.ts
+++ b/test/connector-status.test.ts
@@ -6,6 +6,7 @@ import {
   connectorStatusIsStale,
   createConnectorStatusCache,
   refreshConnectorStatus,
+  usableConnectorProviders,
   type ConnectorStatusRecord,
 } from "../src/credentials/connector-status.ts";
 import type { ConnectorTokenStore, OAuthTokenStatus } from "../src/credentials/keychain.ts";
@@ -135,3 +136,15 @@ test("labels: known providers get friendly names; unknown is capitalized", () =>
   assert.equal(connectorLabel("github"), "GitHub");
   assert.equal(connectorLabel("acme"), "Acme");
 });
+
+test("usable providers include valid manual tokens without requiring an OAuth client", () => {
+  const status: ConnectorStatusRecord = {
+    principalId: "U1",
+    checkedAt: 1,
+    providers: {
+      linear: { connected: true },
+      github: { connected: true, needsReconnect: true },
+    },
+  };
+  assert.deepEqual(usableConnectorProviders(["google"], status), ["google", "linear"]);
+});
diff --git a/test/skill-conformance.test.ts b/test/skill-conformance.test.ts
index c67b8f57..b2144f11 100644
--- a/test/skill-conformance.test.ts
+++ b/test/skill-conformance.test.ts
@@ -21,3 +21,10 @@ test("every seed SKILL.md parses with a name, description, and body", () => {
     assert.ok(m.name && m.description && m.body.trim(), `${path} is missing a required field`);
   }
 });
+
+test("Slack draft OAuth tokens never enter subprocess arguments", () => {
+  const script = readFileSync(join(SEED_DIR, "slack-drafts", "scripts", "slack_drafts.py"), "utf8");
+  assert.doesNotMatch(script, /subprocess|\["curl"/);
+  assert.match(script, /urllib\.request\.Request/);
+  assert.match(script, /"Authorization": f"Bearer \{tok\}"/);
+});
diff --git a/test/skills-seed.test.ts b/test/skills-seed.test.ts
index 02d56e09..cb934d3f 100644
--- a/test/skills-seed.test.ts
+++ b/test/skills-seed.test.ts
@@ -206,6 +206,15 @@ test("a fresh app advertises and materializes only admin-enabled connector skill
   } as TurnRequest);
   assert.match(drive.reply ?? "", /Google Drive \/ Docs \/ Sheets \/ Slides/);
   assert.match(drive.reply ?? "", /sheets\.googleapis\.com/);
+
+  const github = await app.turn({
+    surface: "test",
+    actor,
+    conversation: { kind: "dm", threadRef: "dm:U1:seeded-github-read" },
+    text: "!read skills/github-gitlab/SKILL.md",
+  } as TurnRequest);
+  assert.match(github.reply ?? "", /VAULT_TOKEN_API_GITHUB_COM/);
+  assert.match(github.reply ?? "", /GH_TOKEN/);
 });
 
 function fakeSandbox() {
diff --git a/test/slack-index.integration.test.ts b/test/slack-index.integration.test.ts
index a5bc2a65..753e6c76 100644
--- a/test/slack-index.integration.test.ts
+++ b/test/slack-index.integration.test.ts
@@ -18,6 +18,10 @@ class FakeSlackClient {
   readonly deletes: any[] = [];
   readonly reactionsAdded: any[] = [];
   readonly reactionsRemoved: any[] = [];
+  readonly statuses: any[] = [];
+  readonly streamsStarted: any[] = [];
+  readonly streamsAppended: any[] = [];
+  readonly streamsStopped: any[] = [];
   readonly usersById = new Map<string, any>();
   readonly channelsById = new Map<string, any>();
   readonly membersByChannel = new Map<string, string[]>();
@@ -87,6 +91,26 @@ class FakeSlackClient {
       this.deletes.push(body);
       return { ok: true };
     },
+    startStream: async (body: any) => {
+      this.streamsStarted.push(body);
+      return { ok: true, ts: "stream-1" };
+    },
+    appendStream: async (body: any) => {
+      this.streamsAppended.push(body);
+      return { ok: true, ts: body.ts };
+    },
+    stopStream: async (body: any) => {
+      this.streamsStopped.push(body);
+      return { ok: true, ts: body.ts };
+    },
+  };
+  readonly assistant = {
+    threads: {
+      setStatus: async (body: any) => {
+        this.statuses.push(body);
+        return { ok: true };
+      },
+    },
   };
   readonly reactions = {
     add: async (body: any) => {
@@ -199,6 +223,8 @@ class FakeCore implements SlackCoreClient {
   private runGate: Promise<void> | undefined;
   private releaseRun: (() => void) | undefined;
   readonly modelChangeListeners: Array<(scope: any) => void> = [];
+  streamDeltas: string[] = [];
+  streamTasks: Array<{ id: string; title: string; status: "pending" | "in_progress" | "completed" }> = [];
 
   async externalSlackParticipants(): Promise<boolean> {
     return this.externalParticipants;
@@ -238,9 +264,11 @@ class FakeCore implements SlackCoreClient {
     }
     return this.result;
   }
-  async waitRun(runId: string): Promise<TurnResult | null> {
+  async waitRun(runId: string, hooks: any = {}): Promise<TurnResult | null> {
     this.polled.push(runId);
     if (this.runGate) await this.runGate;
+    for (const delta of this.streamDeltas) await hooks.onDelta?.(delta);
+    if (this.streamTasks.length) await hooks.onTasks?.(this.streamTasks);
     return this.result;
   }
   /** Enqueue `runId` on the first submit and hold waitRun open; every later submit is a
@@ -367,11 +395,12 @@ test("a mid-turn message that STEERS the live run does not post the reply twice"
     f.core.finishRun({ status: "ok", reply: "agent reply" });
     await Promise.all([first, steer]);
 
-    assert.equal(
-      f.client.posts.filter((p) => p.text === "agent reply").length,
-      1,
-      "the shared run's reply is posted once, by the handler that owns it",
-    );
+    const deliveredReplies = [
+      ...f.client.posts.map((post) => post.text),
+      ...f.client.streamsStarted.flatMap((request) => request.chunks.map((chunk: { text?: string }) => chunk.text)),
+      ...f.client.streamsAppended.flatMap((request) => request.chunks.map((chunk: { text?: string }) => chunk.text)),
+    ].filter((text) => text === "agent reply");
+    assert.equal(deliveredReplies.length, 1, "the shared run's reply is delivered once, by the handler that owns it");
   } finally {
     await f.stop();
   }
@@ -401,6 +430,120 @@ test("a DM becomes one scoped live turn and one Slack reply", async () => {
   }
 });
 
+test("a queued DM keeps model text private until terminal approval and uses native task updates", async () => {
+  const f = await fixture();
+  try {
+    f.core.queuedRunId = "R-stream";
+    f.core.streamDeltas = ["Checking ", "now."];
+    f.core.streamTasks = [{ id: "lookup", title: "Inspect deployment", status: "in_progress" }];
+    f.core.result = { status: "ok", reply: "Checking now." };
+
+    await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "check it", ts: "100.9" });
+
+    assert.deepEqual(f.client.statuses, [
+      { channel_id: "D1", thread_ts: "100.9", status: "Thinking…" },
+      { channel_id: "D1", thread_ts: "100.9", status: "" },
+    ]);
+    assert.deepEqual(f.client.streamsStarted, [
+      {
+        channel: "D1",
+        thread_ts: "100.9",
+        chunks: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }],
+        task_display_mode: "timeline",
+      },
+    ]);
+    assert.deepEqual(f.client.streamsAppended, [
+      {
+        channel: "D1",
+        ts: "stream-1",
+        chunks: [{ type: "markdown_text", text: "Checking now." }],
+      },
+    ]);
+    assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]);
+    assert.equal(f.client.posts.length, 0);
+  } finally {
+    await f.stop();
+  }
+});
+
+test("a rejected queued turn deletes its provisional native task stream", async () => {
+  const f = await fixture();
+  try {
+    f.core.queuedRunId = "R-silent";
+    f.core.streamDeltas = ["must never be shown"];
+    f.core.streamTasks = [{ id: "lookup", title: "Inspect deployment", status: "in_progress" }];
+    f.core.result = { status: "silent" };
+
+    await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "check it", ts: "100.12" });
+
+    assert.equal(
+      f.client.streamsStarted.some((request) =>
+        request.chunks.some((chunk: { type: string; text?: string }) => chunk.text === "must never be shown"),
+      ),
+      false,
+    );
+    assert.deepEqual(f.client.deletes, [{ channel: "D1", ts: "stream-1" }]);
+    assert.deepEqual(f.client.streamsStopped, []);
+    assert.equal(f.client.posts.length, 0);
+  } finally {
+    await f.stop();
+  }
+});
+
+test("a task-only native stream appends the terminal reply before stopping", async () => {
+  const f = await fixture();
+  try {
+    f.core.queuedRunId = "R-task-only";
+    f.core.streamTasks = [{ id: "lookup", title: "Inspect deployment", status: "in_progress" }];
+    f.core.result = { status: "ok", reply: "The deployment is healthy." };
+
+    await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "check it", ts: "100.10" });
+
+    assert.deepEqual(f.client.streamsStarted[0]?.chunks, [
+      { type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" },
+    ]);
+    assert.deepEqual(f.client.streamsAppended, [
+      {
+        channel: "D1",
+        ts: "stream-1",
+        chunks: [{ type: "markdown_text", text: "The deployment is healthy." }],
+      },
+    ]);
+    assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]);
+    assert.equal(f.client.posts.length, 0);
+  } finally {
+    await f.stop();
+  }
+});
+
+test("native Slack streaming splits large model deltas at the API chunk limit", async () => {
+  const f = await fixture();
+  try {
+    const reply = "x".repeat(24_001);
+    f.core.queuedRunId = "R-large-delta";
+    f.core.streamDeltas = [reply];
+    f.core.result = { status: "ok", reply };
+
+    await f.app.emitMessage({ channel: "D1", channel_type: "im", user: "U1", text: "write it", ts: "100.11" });
+
+    const texts = [
+      ...f.client.streamsStarted.flatMap((request) => request.chunks),
+      ...f.client.streamsAppended.flatMap((request) => request.chunks),
+    ]
+      .filter((chunk) => chunk.type === "markdown_text")
+      .map((chunk) => chunk.text);
+    assert.deepEqual(
+      texts.map((text) => text.length),
+      [12_000, 12_000, 1],
+    );
+    assert.equal(texts.join(""), reply);
+    assert.deepEqual(f.client.streamsStopped, [{ channel: "D1", ts: "stream-1" }]);
+    assert.equal(f.client.posts.length, 0);
+  } finally {
+    await f.stop();
+  }
+});
+
 test("a human's DM sets the conversation header to the serving model + web surface", async () => {
   const f = await fixture({ webUiPublicUrl: "https://claw.example.dev" });
   try {
diff --git a/test/slack-messaging.test.ts b/test/slack-messaging.test.ts
new file mode 100644
index 00000000..aed0e523
--- /dev/null
+++ b/test/slack-messaging.test.ts
@@ -0,0 +1,39 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { createStreamingReplyFilter } from "../src/slack/messaging.ts";
+
+test("streaming reply filter emits prose incrementally and never exposes internal directives", () => {
+  const filter = createStreamingReplyFilter();
+  const visible = [
+    filter.push("Here is the answer. "),
+    filter.push("[[rea"),
+    filter.push("ct: eyes]]"),
+    filter.push(" Done."),
+    filter.flush(),
+  ].join("");
+
+  assert.equal(visible, "Here is the answer.  Done.");
+  assert.equal(visible.includes("[[react"), false);
+});
+
+test("streaming reply filter holds a split agent request directive out of Slack", () => {
+  const filter = createStreamingReplyFilter();
+  const visible = [
+    filter.push("I need help. [[ask-"),
+    filter.push("agent: <@U2> | inspect token=secret"),
+    filter.push("]] Thanks."),
+    filter.flush(),
+  ].join("");
+
+  assert.equal(visible, "I need help.  Thanks.");
+  assert.equal(visible.includes("token=secret"), false);
+});
+
+test("streaming reply filter never leaks a long directive delivered one character at a time", () => {
+  const filter = createStreamingReplyFilter();
+  const raw = "Public. [[react: internal-tool-argument-secret-1234567890]] Done.";
+  const visible = [...raw].map((character) => filter.push(character)).join("") + filter.flush();
+
+  assert.equal(visible, "Public.  Done.");
+  assert.equal(visible.includes("internal-tool-argument"), false);
+});
diff --git a/test/slack-presenters.test.ts b/test/slack-presenters.test.ts
index f6aa2b5c..97caa69a 100644
--- a/test/slack-presenters.test.ts
+++ b/test/slack-presenters.test.ts
@@ -1,6 +1,335 @@
 import { test } from "node:test";
 import assert from "node:assert/strict";
-import { renderTaskList, createTaskListPresenter, createAckPresenter, stripAckPrefix } from "../src/slack/lib.ts";
+import {
+  renderTaskList,
+  createTaskListPresenter,
+  createAckPresenter,
+  createNativeAgentPresenter,
+  stripAckPrefix,
+} from "../src/slack/lib.ts";
+
+test("native agent presenter streams text and public task progress into one message", async () => {
+  const calls: Array<{ method: string; body: unknown }> = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push({ method: "status", body: status });
+    },
+    start: async (chunks) => {
+      calls.push({ method: "start", body: chunks });
+      return "171.1";
+    },
+    append: async (ts, chunks) => {
+      calls.push({ method: `append:${ts}`, body: chunks });
+    },
+    stop: async (ts) => {
+      calls.push({ method: `stop:${ts}`, body: null });
+    },
+    remove: async (ts) => {
+      calls.push({ method: `remove:${ts}`, body: null });
+    },
+    checkpoint: async (ts) => {
+      calls.push({ method: "checkpoint", body: ts });
+    },
+    onSurfacePosted: () => calls.push({ method: "surface", body: null }),
+  });
+
+  await presenter.begin();
+  assert.equal(await presenter.onDelta("Checking "), true);
+  assert.equal(await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]), true);
+  assert.equal(await presenter.onDelta("now."), true);
+  assert.equal(await presenter.finalize(), true);
+
+  assert.deepEqual(calls, [
+    { method: "status", body: "Thinking…" },
+    { method: "start", body: [{ type: "markdown_text", text: "Checking " }] },
+    { method: "checkpoint", body: "171.1" },
+    { method: "surface", body: null },
+    {
+      method: "append:171.1",
+      body: [{ type: "task_update", id: "lookup", title: "Inspect deployment", status: "in_progress" }],
+    },
+    { method: "append:171.1", body: [{ type: "markdown_text", text: "now." }] },
+    { method: "stop:171.1", body: null },
+    { method: "status", body: "" },
+  ]);
+});
+
+test("native agent presenter coalesces token deltas into Slack-sized stream updates", async () => {
+  const calls: Array<{ method: string; text?: string }> = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async () => {},
+    start: async (chunks) => {
+      calls.push({ method: "start", text: chunks[0]?.type === "markdown_text" ? chunks[0].text : undefined });
+      return "171.6";
+    },
+    append: async (_ts, chunks) => {
+      calls.push({ method: "append", text: chunks[0]?.type === "markdown_text" ? chunks[0].text : undefined });
+    },
+    stop: async () => {
+      calls.push({ method: "stop" });
+    },
+    remove: async () => {},
+    checkpoint: async () => {},
+    onSurfacePosted: () => {},
+  });
+
+  for (const character of "x".repeat(600)) assert.equal(await presenter.onDelta(character), true);
+  assert.equal(await presenter.finalize(), true);
+
+  assert.deepEqual(
+    calls.map(({ method, text }) => ({ method, length: text?.length })),
+    [
+      { method: "start", length: 256 },
+      { method: "append", length: 256 },
+      { method: "append", length: 88 },
+      { method: "stop", length: undefined },
+    ],
+  );
+});
+
+test("native agent presenter falls back cleanly when Slack streaming is unavailable", async () => {
+  const calls: string[] = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push(`status:${status}`);
+    },
+    start: async () => {
+      calls.push("start");
+      throw new Error("unknown_method");
+    },
+    append: async () => {
+      calls.push("append");
+    },
+    stop: async () => {
+      calls.push("stop");
+    },
+    remove: async () => {
+      calls.push("remove");
+    },
+    checkpoint: async () => {
+      calls.push("checkpoint");
+    },
+    onSurfacePosted: () => calls.push("surface"),
+    onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`),
+  });
+
+  await presenter.begin();
+  assert.equal(await presenter.onDelta("x".repeat(256)), false);
+  assert.equal(await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]), false);
+  assert.equal(await presenter.finalize(), false);
+  assert.deepEqual(calls, ["status:Thinking…", "start", "error:unknown_method", "status:"]);
+});
+
+test("native agent presenter removes an uncheckpointed stream before falling back", async () => {
+  const calls: string[] = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push(`status:${status}`);
+    },
+    start: async () => {
+      calls.push("start");
+      return "171.7";
+    },
+    append: async () => {
+      calls.push("append");
+    },
+    stop: async () => {
+      calls.push("stop");
+    },
+    remove: async (ts) => {
+      calls.push(`remove:${ts}`);
+    },
+    checkpoint: async () => {
+      calls.push("checkpoint");
+      throw new Error("core unavailable");
+    },
+    onSurfacePosted: () => calls.push("surface"),
+    onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`),
+  });
+
+  await presenter.begin();
+  assert.equal(await presenter.onDelta("x".repeat(256)), false);
+  assert.equal(await presenter.finalize(), false);
+  assert.deepEqual(calls, [
+    "status:Thinking…",
+    "start",
+    "checkpoint",
+    "remove:171.7",
+    "error:core unavailable",
+    "status:",
+  ]);
+});
+
+test("native agent presenter deletes a partial stream before fallback after append fails", async () => {
+  const calls: string[] = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push(`status:${status}`);
+    },
+    start: async () => {
+      calls.push("start");
+      return "171.8";
+    },
+    append: async () => {
+      calls.push("append");
+      throw new Error("stream disconnected");
+    },
+    stop: async () => {
+      calls.push("stop");
+    },
+    remove: async (ts) => {
+      calls.push(`remove:${ts}`);
+    },
+    checkpoint: async () => {
+      calls.push("checkpoint");
+    },
+    onSurfacePosted: () => calls.push("surface"),
+    onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`),
+  });
+
+  await presenter.begin();
+  assert.equal(await presenter.onDelta("x".repeat(256)), true);
+  assert.equal(await presenter.onDelta("y".repeat(256)), false);
+  assert.equal(await presenter.finalize(), false);
+  assert.deepEqual(calls, [
+    "status:Thinking…",
+    "start",
+    "checkpoint",
+    "surface",
+    "append",
+    "remove:171.8",
+    "error:stream disconnected",
+    "status:",
+  ]);
+});
+
+test("native agent presenter clears status after a delayed begin", async () => {
+  const calls: string[] = [];
+  let releaseStatus: (() => void) | undefined;
+  const statusGate = new Promise<void>((resolve) => {
+    releaseStatus = resolve;
+  });
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push(`status:${status}`);
+      if (status) await statusGate;
+    },
+    start: async () => undefined,
+    append: async () => {},
+    stop: async () => {},
+    remove: async () => {},
+    checkpoint: async () => {},
+    onSurfacePosted: () => {},
+  });
+
+  const beginning = presenter.begin();
+  const finalizing = presenter.finalize();
+  releaseStatus?.();
+  await beginning;
+  assert.equal(await finalizing, false);
+  assert.deepEqual(calls, ["status:Thinking…", "status:"]);
+});
+
+test("native agent presenter suppresses fallback when a failed partial stream cannot be removed", async () => {
+  const presenter = createNativeAgentPresenter({
+    setStatus: async () => {},
+    start: async () => "171.9",
+    append: async () => {
+      throw new Error("stream disconnected");
+    },
+    stop: async () => {},
+    remove: async () => {
+      throw new Error("delete failed");
+    },
+    checkpoint: async () => {},
+    onSurfacePosted: () => {},
+  });
+
+  await presenter.begin();
+  assert.equal(await presenter.onDelta("x".repeat(256)), true);
+  assert.equal(await presenter.onDelta("y".repeat(256)), false);
+  assert.equal(await presenter.finalize(), true);
+});
+
+test("native agent presenter preserves approved text when stopping its stream fails", async () => {
+  const calls: string[] = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push(`status:${status}`);
+    },
+    start: async () => {
+      calls.push("start");
+      return "171.11";
+    },
+    append: async () => {},
+    stop: async (ts) => {
+      calls.push(`stop:${ts}`);
+      throw new Error("stop failed");
+    },
+    remove: async (ts) => {
+      calls.push(`remove:${ts}`);
+    },
+    checkpoint: async () => {
+      calls.push("checkpoint");
+    },
+    onSurfacePosted: () => calls.push("surface"),
+    onError: (error) => calls.push(`error:${error instanceof Error ? error.message : String(error)}`),
+  });
+
+  assert.equal(await presenter.onDelta("approved".repeat(40)), true);
+  assert.equal(await presenter.finalize(), true);
+  await presenter.settle();
+
+  assert.deepEqual(calls, [
+    "start",
+    "checkpoint",
+    "surface",
+    "stop:171.11",
+    "error:stop failed",
+  ]);
+});
+
+test("native agent presenter deletes a provisional stream when the turn is abandoned", async () => {
+  const calls: string[] = [];
+  const presenter = createNativeAgentPresenter({
+    setStatus: async (status) => {
+      calls.push(`status:${status}`);
+    },
+    start: async () => {
+      calls.push("start");
+      return "171.10";
+    },
+    append: async () => {
+      calls.push("append");
+    },
+    stop: async () => {
+      calls.push("stop");
+    },
+    remove: async (ts) => {
+      calls.push(`remove:${ts}`);
+    },
+    checkpoint: async () => {
+      calls.push("checkpoint");
+    },
+    onSurfacePosted: () => calls.push("surface"),
+  });
+
+  await presenter.begin();
+  assert.equal(
+    await presenter.onTasks([{ id: "lookup", title: "Inspect deployment", status: "in_progress" }]),
+    true,
+  );
+  await presenter.settle();
+
+  assert.deepEqual(calls, [
+    "status:Thinking…",
+    "start",
+    "checkpoint",
+    "surface",
+    "remove:171.10",
+    "status:",
+  ]);
+});
 
 test("renderTaskList renders every terminal state", () => {
   assert.equal(
diff --git a/test/turn-stream.test.ts b/test/turn-stream.test.ts
index b4687307..95cf9c93 100644
--- a/test/turn-stream.test.ts
+++ b/test/turn-stream.test.ts
@@ -6,6 +6,7 @@ import { mkdtempSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { createTurnStream } from "../src/runs/turn-stream.ts";
+import { createStreamingReplyFilter } from "../src/slack/messaging.ts";
 import { buildApp } from "../src/wiring.ts";
 import { testConfig } from "./support/test-config.ts";
 
@@ -21,6 +22,75 @@ test("accumulates deltas per run and isolates runs", () => {
   assert.equal(s.snapshot("r2"), "world");
 });
 
+test("subscribers receive public reply deltas in order", async () => {
+  const s = createTurnStream();
+  const deltas: string[] = [];
+  const unsubscribe = s.subscribe("r1", {
+    onDelta: (delta) => {
+      deltas.push(delta);
+    },
+  });
+  s.publish("r1", "Hel");
+  s.publish("r1", "lo");
+  s.publish("r2", "private to another run");
+  unsubscribe();
+  s.publish("r1", "!");
+  await s.drain("r1");
+  assert.deepEqual(deltas, ["Hel", "lo"]);
+});
+
+test("a late subscriber receives the buffered prefix before live deltas", async () => {
+  const s = createTurnStream();
+  s.publish("r1", "Hel");
+  const deltas: string[] = [];
+  s.subscribe("r1", {
+    onDelta: (delta) => {
+      deltas.push(delta);
+    },
+  });
+  s.publish("r1", "lo");
+  await s.drain("r1");
+  assert.deepEqual(deltas, ["Hel", "lo"]);
+});
+
+test("drain waits for asynchronous delta listeners and preserves delivery order", async () => {
+  const s = createTurnStream();
+  const deltas: string[] = [];
+  let releaseFirst: (() => void) | undefined;
+  const firstGate = new Promise<void>((resolve) => (releaseFirst = resolve));
+  s.subscribe("r1", {
+    onDelta: async (delta) => {
+      if (delta === "Hel") await firstGate;
+      deltas.push(delta);
+    },
+  });
+
+  s.publish("r1", "Hel");
+  s.publish("r1", "lo");
+  const draining = s.drain("r1");
+  await Promise.resolve();
+  assert.deepEqual(deltas, []);
+  releaseFirst?.();
+  await draining;
+  assert.deepEqual(deltas, ["Hel", "lo"]);
+});
+
+test("a directive split across buffered and live deltas stays private", async () => {
+  const s = createTurnStream();
+  const filter = createStreamingReplyFilter();
+  const publicDeltas: string[] = [];
+  s.publish("r1", "Public. [[ask-agent:");
+  s.subscribe("r1", {
+    onDelta: (delta) => {
+      publicDeltas.push(filter.push(delta));
+    },
+  });
+  s.publish("r1", " <@U2> | private context]] Done.");
+  await s.drain("r1");
+  publicDeltas.push(filter.flush());
+  assert.equal(publicDeltas.join(""), "Public.  Done.");
+});
+
 test("ignores empty deltas", () => {
   const s = createTurnStream();
   s.publish("r1", "");