Fix stream spacing and encrypted abort previews

.changeset/upset-peaches-lick.md

@@ -0,0 +1,5 @@
+---
+'@workflow/web': patch
+---
+
+Recognize unframed serialized abort packets in the Streams viewer so encrypted packets can be decrypted, and keep stream-list rows at least 40px tall.

packages/web/app/components/run-detail-view.tsx

@@ -805,13 +805,13 @@ export function RunDetailView({
                     <div className="min-h-0 flex-1 divide-y divide-gray-400 overflow-y-auto">
                       {streamsLoading ? (
                         <div className="flex flex-col divide-y divide-gray-400">
-                          <div className="flex h-10 items-center pl-4 pr-2">
+                          <div className="flex h-10 min-h-10 items-center pl-4 pr-2">
                             <Skeleton className="h-3.5 w-24" />
                           </div>
-                          <div className="flex h-10 items-center pl-4 pr-2">
+                          <div className="flex h-10 min-h-10 items-center pl-4 pr-2">
                             <Skeleton className="h-3.5 w-32" />
                           </div>
-                          <div className="flex h-10 items-center pl-4 pr-2">
+                          <div className="flex h-10 min-h-10 items-center pl-4 pr-2">
                             <Skeleton className="h-3.5 w-20" />
                           </div>
                         </div>
@@ -820,7 +820,7 @@ export function RunDetailView({
                           {streamsError.message}
                         </div>
                       ) : streams.length === 0 ? (
-                        <div className="flex h-10 items-center px-4 text-label-12 text-gray-900">
+                        <div className="flex h-10 min-h-10 items-center px-4 text-label-12 text-gray-900">
                           No streams found for this run
                         </div>
                       ) : (
@@ -829,7 +829,7 @@ export function RunDetailView({
                             key={streamId}
                             type="button"
                             onClick={() => setSelectedStreamId(streamId)}
-                            className={`flex h-10 w-full min-w-0 items-center truncate pl-4 pr-2 text-left text-label-14 transition-colors ${
+                            className={`flex h-10 min-h-10 w-full min-w-0 items-center truncate pl-4 pr-2 text-left text-label-14 transition-colors ${
                               selectedStreamId === streamId
                                 ? 'bg-gray-100 text-gray-1000 hover:bg-gray-200 active:bg-gray-200'
                                 : 'bg-transparent text-gray-900 hover:bg-gray-100 hover:text-gray-1000 focus-visible:bg-gray-100 focus-visible:text-gray-1000'

packages/web/app/lib/hooks/use-stream-reader.test.ts

@@ -1,5 +1,14 @@
-import { renderHook, waitFor } from '@testing-library/react';
-import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { webcrypto } from 'node:crypto';
+import { cleanup, renderHook, waitFor } from '@testing-library/react';
+import { importKey } from '@workflow/core/encryption';
+import {
+  deriveRunPayloadKeys,
+  encodeWithFormatPrefix,
+  encryptEnvelope,
+  SerializationFormat,
+  sealTo,
+} from '@workflow/core/serialization-format';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import type { StreamResponse } from '~/lib/client/workflow-streams';
 import { readStream } from '~/lib/workflow-api-client';
 import { useStreamReader } from './use-stream-reader';
@@ -9,6 +18,62 @@ vi.mock('~/lib/workflow-api-client', () => ({
 }));
 
 const env = {};
+const RUN_KEY = Uint8Array.from({ length: 32 }, (_, index) => index);
+const ABORT_PACKET = { aborted: true, reason: 'Cancelled\nby the user' };
+const ENCRYPTION_MODES = ['symmetric', 'sealed'] as const;
+
+function serializeAbortPacket(): Uint8Array {
+  return encodeWithFormatPrefix(
+    SerializationFormat.DEVALUE_V1,
+    Uint8Array.from(
+      new TextEncoder().encode(
+        JSON.stringify([{ aborted: 1, reason: 2 }, true, ABORT_PACKET.reason])
+      )
+    )
+  ) as Uint8Array;
+}
+
+async function encryptedPacket(mode: (typeof ENCRYPTION_MODES)[number]) {
+  const key =
+    mode === 'symmetric'
+      ? await importKey(RUN_KEY)
+      : sealTo((await deriveRunPayloadKeys(RUN_KEY)).keyPair.publicKey);
+  return (await encryptEnvelope(serializeAbortPacket(), key)) as Uint8Array;
+}
+
+function frame(payload: Uint8Array): Uint8Array {
+  const framed = new Uint8Array(4 + payload.byteLength);
+  new DataView(framed.buffer).setUint32(0, payload.byteLength, false);
+  framed.set(payload, 4);
+  return framed;
+}
+
+function mockStreamResponse(payload: Uint8Array) {
+  vi.mocked(readStream).mockResolvedValue({
+    body: new ReadableStream<Uint8Array>({
+      start(controller) {
+        // Split both the format/length prefix and the encrypted payload.
+        for (const part of [
+          payload.slice(0, 2),
+          payload.slice(2, 7),
+          payload.slice(7, 19),
+          payload.slice(19),
+        ]) {
+          controller.enqueue(part);
+        }
+        controller.close();
+      },
+    }),
+    cursor: null,
+    done: true,
+  });
+}
+
+function renderReader(key?: Uint8Array, streamId = 'strm_test_system_abort') {
+  return renderHook(() =>
+    useStreamReader(env, streamId, 'wrun_test', key, 'completed')
+  );
+}
 
 function emptyStreamResponse(done: boolean): StreamResponse {
   return {
@@ -22,6 +87,92 @@ function emptyStreamResponse(done: boolean): StreamResponse {
   };
 }
 
+describe('useStreamReader payloads', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.stubGlobal('crypto', webcrypto);
+  });
+
+  afterEach(() => {
+    cleanup();
+    vi.unstubAllGlobals();
+  });
+
+  it.each(
+    ENCRYPTION_MODES
+  )('recognizes an unframed %s abort packet as encrypted', async (mode) => {
+    mockStreamResponse(await encryptedPacket(mode));
+    const { result } = renderReader();
+
+    await waitFor(() => expect(result.current.isInitialLoading).toBe(false));
+
+    expect(result.current.error).toContain('Click Decrypt');
+    expect(result.current.chunks).toEqual([]);
+  });
+
+  it.each(
+    ENCRYPTION_MODES.flatMap((mode) => [
+      { mode, framed: false },
+      { mode, framed: true },
+    ])
+  )('decrypts a $mode packet (framed: $framed)', async ({ mode, framed }) => {
+    const packet = await encryptedPacket(mode);
+    mockStreamResponse(framed ? frame(packet) : packet);
+    const { result } = renderReader(
+      RUN_KEY,
+      framed ? 'strm_test_user' : 'strm_test_system_abort'
+    );
+
+    await waitFor(() => expect(result.current.isInitialLoading).toBe(false));
+
+    expect(result.current.error).toBeNull();
+    expect(result.current.chunks.map(({ value }) => value)).toEqual([
+      ABORT_PACKET,
+    ]);
+  });
+
+  it.each([
+    'encrypted text',
+    'encp is plain text',
+    'devl is plain text',
+  ])('keeps legacy user text readable: %s', async (text) => {
+    mockStreamResponse(new TextEncoder().encode(`${text}\nnext line`));
+    const { result } = renderReader(undefined, 'strm_test_user');
+    await waitFor(() => expect(result.current.isInitialLoading).toBe(false));
+    expect(result.current.error).toBeNull();
+    expect(result.current.chunks.map(({ value }) => value)).toEqual([
+      text,
+      'next line',
+    ]);
+  });
+
+  it('hydrates an unframed plaintext serialized abort packet', async () => {
+    mockStreamResponse(serializeAbortPacket());
+    const { result } = renderReader();
+
+    await waitFor(() => expect(result.current.isInitialLoading).toBe(false));
+
+    expect(result.current.chunks.map(({ value }) => value)).toEqual([
+      ABORT_PACKET,
+    ]);
+  });
+
+  it('continues to read legacy JSON lines and plain text', async () => {
+    mockStreamResponse(
+      new TextEncoder().encode('{"index":1}\nplain text\n{"index":3}')
+    );
+    const { result } = renderReader();
+
+    await waitFor(() => expect(result.current.isInitialLoading).toBe(false));
+
+    expect(result.current.chunks.map(({ value }) => value)).toEqual([
+      { index: 1 },
+      'plain text',
+      { index: 3 },
+    ]);
+  });
+});
+
 describe('useStreamReader status', () => {
   beforeEach(() => {
     vi.clearAllMocks();

packages/web/app/lib/hooks/use-stream-reader.ts

@@ -4,6 +4,7 @@ import {
   hydrateData,
   isEncryptedData,
   type PayloadKey,
+  SerializationFormat,
 } from '@workflow/core/serialization-format';
 import { getWebRevivers } from '@workflow/web-shared';
 import type { WorkflowRunStatus } from '@workflow/world';
@@ -35,13 +36,24 @@ const YIELD_EVERY_N_FRAMES = 64;
  *
  * - **framed**: Current format — 4-byte big-endian length + format-prefixed
  *   payload (`devl`/`encr`). The first uint32 is a plausible frame size.
+ * - **unframed**: One format-prefixed payload, such as an internal abort
+ *   packet. Network chunks must be joined before decoding the payload.
  * - **legacy**: Older SDK versions used newline-delimited devalue strings
  *   with no binary framing.
  */
-type StreamEncoding = 'framed' | 'legacy';
+type StreamEncoding = 'framed' | 'unframed' | 'legacy';
 
-function detectEncoding(data: Uint8Array): StreamEncoding {
+function detectEncoding(data: Uint8Array, streamId: string): StreamEncoding {
   if (data.length < FRAME_HEADER_SIZE) return 'framed';
+  const prefix = new TextDecoder().decode(data.subarray(0, FRAME_HEADER_SIZE));
+  // Only internal abort streams store unframed serialized packets. A user
+  // text stream may begin with e.g. "encrypted", which shares the encr prefix.
+  if (
+    streamId.endsWith('_system_abort') &&
+    Object.values(SerializationFormat).some((format) => format === prefix)
+  ) {
+    return 'unframed';
+  }
   const view = new DataView(data.buffer, data.byteOffset, data.byteLength);
   const len = view.getUint32(0, false);
   if (len > 0 && len <= 10 * 1024 * 1024) return 'framed';
@@ -212,7 +224,14 @@ export function useStreamReader(
 
         const { value, done } = await reader.read();
         if (done) {
-          if (encoding === 'legacy' && textRemainder.trim()) {
+          if (encoding === 'unframed' && buffer.length > 0) {
+            frameIndex++;
+            if (frameIndex > skipFrames) {
+              const result = await processFrame(buffer, cryptoKey, revivers);
+              if (result.encrypted) return { encrypted: true };
+              targetBuffer.push(result.chunk);
+            }
+          } else if (encoding === 'legacy' && textRemainder.trim()) {
             frameIndex++;
             if (frameIndex > skipFrames) {
               targetBuffer.push(parseLegacyLine(textRemainder.trim()));
@@ -225,7 +244,7 @@ export function useStreamReader(
         if (encoding === null) {
           appendToBuffer(value);
           if (buffer.length >= FRAME_HEADER_SIZE) {
-            encoding = detectEncoding(buffer);
+            encoding = detectEncoding(buffer, streamId);
           }
           if (encoding === 'legacy') {
             textRemainder = decoder.decode(buffer, { stream: true });
@@ -252,6 +271,8 @@ export function useStreamReader(
           continue;
         }
 
+        if (encoding === 'unframed') continue;
+
         let offset = 0;
         let framesInBatch = 0;