fix(tasks): read output previews by range

Read only the requested task-output tail so polling stays responsive when logs are large.

Author: elkaix

.changeset/fix-task-preview-utf8.md

@@ -2,4 +2,4 @@
 "@pymodel/pythinker-code": patch
 ---
 
-Keep task output previews valid at UTF-8 byte boundaries.
+Keep task output previews responsive for large logs and valid at UTF-8 byte boundaries.

packages/agent-core-v2/src/agent/task/persist.ts

@@ -142,14 +142,38 @@ export class AgentTaskPersistence {
     taskId: string,
     maxPreviewBytes: number,
   ): Promise<AgentTaskStoredOutputSnapshot | undefined> {
-    const output = await this.readTaskOutputData(taskId);
-    if (output === undefined) return undefined;
-    const preview = utf8TailPreview(output.data, maxPreviewBytes);
+    let root = this.primaryRoot();
+    let outputSizeBytes = await this.bytes.size(this.taskOutputScope(taskId, root), OUTPUT_LOG_KEY);
+    if (outputSizeBytes === undefined) {
+      const fallbackRoot = this.fallbackRoot;
+      if (fallbackRoot === undefined) return undefined;
+      root = fallbackRoot;
+      outputSizeBytes = await this.bytes.size(this.taskOutputScope(taskId, root), OUTPUT_LOG_KEY);
+      if (outputSizeBytes === undefined) return undefined;
+    }
+
+    const previewLimit = Math.min(outputSizeBytes, Math.max(0, Math.trunc(maxPreviewBytes)));
+    const data = new Uint8Array(previewLimit);
+    let offset = 0;
+    if (previewLimit > 0) {
+      const start = outputSizeBytes - previewLimit;
+      for await (const chunk of this.bytes.readStream(
+        this.taskOutputScope(taskId, root),
+        OUTPUT_LOG_KEY,
+        { start, end: outputSizeBytes - 1 },
+      )) {
+        const slice = chunk.subarray(0, previewLimit - offset);
+        data.set(slice, offset);
+        offset += slice.byteLength;
+        if (offset === previewLimit) break;
+      }
+    }
+    const preview = utf8TailPreview(data.subarray(0, offset), previewLimit);
     return {
-      outputPath: this.taskOutputFileAt(taskId, output.root),
-      outputSizeBytes: output.data.byteLength,
+      outputPath: this.taskOutputFileAt(taskId, root),
+      outputSizeBytes,
       previewBytes: preview.bytes,
-      truncated: output.data.byteLength > preview.bytes,
+      truncated: outputSizeBytes > preview.bytes,
       preview: preview.text,
     };
   }

packages/agent-core-v2/test/agent/task/persist.test.ts

@@ -2,7 +2,7 @@ import { mkdir, rm, stat, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'pathe';
 
-import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
 import { SyncDescriptor } from '#/_base/di/descriptors';
 import { DisposableStore } from '#/_base/di/lifecycle';
@@ -59,6 +59,7 @@ beforeEach(async () => {
 });
 
 afterEach(async () => {
+  vi.restoreAllMocks();
   disposables.dispose();
   await rm(sessionDir, { recursive: true, force: true });
 });
@@ -195,6 +196,27 @@ describe('AgentTaskPersistence', () => {
       });
     });
 
+    it('reads only the requested persisted output tail for a snapshot', async () => {
+      const taskId = 'bash-tail0000';
+      const output = `${'x'.repeat(1024 * 1024)}tail`;
+      await persistence.appendTaskOutput(taskId, output);
+      const read = vi.spyOn(bytes, 'read');
+      const readStream = vi.spyOn(bytes, 'readStream');
+
+      expect(await persistence.readTaskOutputSnapshot(taskId, 4)).toMatchObject({
+        outputSizeBytes: output.length,
+        previewBytes: 4,
+        truncated: true,
+        preview: 'tail',
+      });
+      expect(read).not.toHaveBeenCalled();
+      expect(readStream).toHaveBeenCalledWith(
+        `${SESSION_SCOPE}/tasks/${taskId}`,
+        'output.log',
+        { start: output.length - 4, end: output.length - 1 },
+      );
+    });
+
     it('readTaskOutputBytes returns empty string when output.log is absent', async () => {
       expect(await persistence.readTaskOutputBytes('bash-none0001', 0, 100)).toBe('');
     });

packages/agent-core-v2/test/agent/task/taskService.test.ts

@@ -174,6 +174,7 @@ describe('AgentTaskService', () => {
       append: async () => {},
       list: async () => [],
       delete: async () => {},
+      size: async () => undefined,
       flush: async () => {},
       close: async () => {},
     });
@@ -1142,6 +1143,7 @@ describe('AgentTaskService', () => {
       },
       list: async () => [],
       delete: async () => {},
+      size: async () => undefined,
       flush: async () => {},
       close: async () => {},
     });

packages/agent-core-v2/test/agent/task/tools/task-tools.test.ts

@@ -1185,6 +1185,7 @@ describe('WaitForTool (harness)', () => {
 
       const slow = controllableProcess();
       const taskA = tasks.registerTask(new ProcessTask(slow.proc, 'sleep 30', 'slow'));
+      const waitDelivered = ctx.once('task.waitDelivered');
       const pending = executeTool(tool!, context('wait_race', { timeout: 30 }));
 
       const late = controllableProcess();
@@ -1205,6 +1206,7 @@ describe('WaitForTool (harness)', () => {
       slow.pushOutput('A-OUT\n');
       slow.resolveWait(0);
       const result = await pending;
+      await waitDelivered;
       const output = outputString(result);
 
       expect(result.isError ?? false).toBe(false);