feat(slack): support native agent streaming

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

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 = {

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;
             }

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}`;

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) => {

src/runs/turn-stream.ts

@@ -9,13 +9,15 @@ 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;
   subscribe(runId: string, listener: TurnStreamListener): () => void;
 }
 
 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);
+        });
       };
     },
   };

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 } : {}),

src/slack/lib.ts

@@ -139,4 +139,6 @@ export {
   renderTaskList,
   type TaskListPresenter,
   createTaskListPresenter,
+  type NativeAgentPresenter,
+  createNativeAgentPresenter,
 } from "./presenters.ts";

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,