fix: address review feedback on utf8 line chunking, shutdown isolation, and test contracts

Author: elkaix

packages/agent-core-v2/src/os/backends/node-local/hostFsService.ts

@@ -14,7 +14,7 @@ import { ScopeActivation, registerScopedService } from '#/_base/di/scope';
 import { decodeTextWithErrors, type TextDecodeErrors } from '#/_base/execEnv/decodeText';
 
 import { type HostDirEntry, type HostFileStat, IHostFileSystem } from '#/os/interface/hostFileSystem';
-import { toHostFsError } from '#/os/interface/hostFsErrors';
+import { toHostFsError, HostFsError, OsFsErrors } from '#/os/interface/hostFsErrors';
 import { atomicWrite } from '#/_base/utils/fs';
 
 const NEWLINE = Buffer.from([0x0a]);
@@ -38,6 +38,33 @@ function* splitLinesKeepingTerminator(text: string): Generator<string> {
   }
 }
 
+function trimToValidUtf8Boundary(buf: Buffer): Buffer {
+  let i = buf.length - 1;
+  let continuations = 0;
+  while (i >= 0 && continuations < 3) {
+    const byte = buf[i];
+    if (byte === undefined || (byte & 0xc0) !== 0x80) break;
+    continuations += 1;
+    i -= 1;
+  }
+  if (i < 0) return buf;
+  const lead = buf[i];
+  if (lead === undefined) return buf;
+  if ((lead & 0x80) === 0) {
+    return continuations === 0 ? buf : buf.subarray(0, i + 1);
+  }
+  let needed = 0;
+  if ((lead & 0xe0) === 0xc0) needed = 1;
+  else if ((lead & 0xf0) === 0xe0) needed = 2;
+  else if ((lead & 0xf8) === 0xf0) needed = 3;
+  else return buf.subarray(0, i);
+
+  if (continuations < needed) {
+    return buf.subarray(0, i);
+  }
+  return buf;
+}
+
 export class HostFileSystem implements IHostFileSystem {
   declare readonly _serviceBrand: undefined;
 
@@ -121,6 +148,9 @@ export class HostFileSystem implements IHostFileSystem {
       const errors = options?.errors ?? 'strict';
 
       if (!isUtf8Encoding(encoding)) {
+        if (options?.maxLineBytes !== undefined) {
+          throw new HostFsError(OsFsErrors.codes.OS_FS_UNKNOWN, 'maxLineBytes is only supported for UTF-8 encoding');
+        }
         const content = decodeTextWithErrors(await readFile(path), encoding, errors);
         yield* splitLinesKeepingTerminator(content);
         return;
@@ -159,8 +189,9 @@ export class HostFileSystem implements IHostFileSystem {
       const takeLine = (piece: Buffer): Buffer => {
         if (pending.length === 0 && piece.length <= maxLineBytes) return piece;
         retain(piece);
-        const parts = pendingTruncated && piece.at(-1) === 0x0a ? [...pending, NEWLINE] : pending;
-        const line = Buffer.concat(parts);
+        const kept = Buffer.concat(pending);
+        const safe = pendingTruncated ? trimToValidUtf8Boundary(kept) : kept;
+        const line = pendingTruncated && piece.at(-1) === 0x0a ? Buffer.concat([safe, NEWLINE]) : safe;
         pending = [];
         pendingBytes = 0;
         pendingTruncated = false;
@@ -190,7 +221,8 @@ export class HostFileSystem implements IHostFileSystem {
       }
 
       if (pending.length > 0) {
-        const line = Buffer.concat(pending);
+        const kept = Buffer.concat(pending);
+        const line = pendingTruncated ? trimToValidUtf8Boundary(kept) : kept;
         yield decodeTextWithErrors(line, 'utf-8', errors, pendingOffset !== 0);
       }
     } finally {

packages/agent-core-v2/test/agent/prompt/promptService.test.ts

@@ -4,12 +4,12 @@ import { Readable } from 'node:stream';
 
 import { DisposableStore } from '#/_base/di/lifecycle';
 import { createServices } from '#/_base/di/test';
-import { Event } from '#/_base/event';
+import { Emitter, Event } from '#/_base/event';
 import { IAgentBlobService } from '#/agent/blob/agentBlobService';
 import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
 import type { ContextMessage } from '#/agent/contextMemory/types';
 import type { ContentPart } from '#/kosong/contract/message';
-import { IAgentFullCompactionService } from '#/agent/fullCompaction/fullCompaction';
+import { IAgentFullCompactionService, type FullCompactionTask } from '#/agent/fullCompaction/fullCompaction';
 import { IAgentLoopService } from '#/agent/loop/loop';
 import { TurnSteer } from '#/agent/loop/turnOps';
 import { IAgentPromptService } from '#/agent/prompt/prompt';
@@ -74,13 +74,19 @@ function harness(loopOptions: StubLoopOptions = { pendingTurnResult: true }) {
     },
   });
   const loop = stubLoopWithHooks(loopOptions);
-  const fullCompaction = {
+  let activeTask: FullCompactionTask | null = null;
+  const finishCompactionEmitter = new Emitter<FullCompactionTask>();
+  disposables.add(finishCompactionEmitter);
+  const fullCompaction: IAgentFullCompactionService = {
     _serviceBrand: undefined,
-    compacting: null,
+    get compacting() {
+      return activeTask;
+    },
     begin: () => false,
+    cancel: () => {},
     hooks: createHooks(['onWillCompact']),
-    onDidFinishCompaction: Event.None,
-  } as unknown as IAgentFullCompactionService;
+    onDidFinishCompaction: finishCompactionEmitter.event,
+  };
   const intake = {
     get: vi.fn(async () => ({
       meta: {
@@ -124,7 +130,21 @@ function harness(loopOptions: StubLoopOptions = { pendingTurnResult: true }) {
   (ix.get(IEventBus) as ISessionEventBus).activateAgent(
     ix.get(IAgentScopeContext).agentContext,
   );
-  return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction, eventBus: ix.get(IEventBus), intake };
+  return {
+    prompt: ix.get(IAgentPromptService),
+    loop,
+    context,
+    fullCompaction,
+    eventBus: ix.get(IEventBus),
+    intake,
+    setCompacting: (task: FullCompactionTask | null) => {
+      activeTask = task;
+    },
+    finishCompaction: (task: FullCompactionTask) => {
+      activeTask = null;
+      finishCompactionEmitter.fire(task);
+    },
+  };
 }
 
 describe('AgentPromptService', () => {
@@ -305,14 +325,23 @@ describe('AgentPromptService', () => {
   });
 
   it('parks a queued prompt while compaction runs instead of recursing', async () => {
-    const { prompt, fullCompaction } = harness();
-    (fullCompaction as unknown as { compacting: unknown }).compacting = { promise: new Promise(() => {}), abortController: new AbortController() };
+    const { prompt, setCompacting, finishCompaction } = harness();
+    const task: FullCompactionTask = {
+      promise: new Promise(() => {}),
+      abortController: new AbortController(),
+      trigger: 'manual',
+      tokenCount: 100,
+    };
+    setCompacting(task);
     const handle = await prompt.enqueue({ id: 'parked', message: message('later') });
     expect(handle.state).toBe('pending');
-    await (prompt as unknown as { startNext(): Promise<void> }).startNext();
-    await (prompt as unknown as { startNext(): Promise<void> }).startNext();
     expect(prompt.list().pending.map((item) => item.id)).toEqual(['parked']);
     expect(prompt.list().active).toBeUndefined();
+
+    finishCompaction(task);
+    await expect(handle.launched).resolves.toBeDefined();
+    expect(prompt.list().pending).toEqual([]);
+    expect(prompt.list().active?.id).toBe('parked');
   });
 
   it('aborts pending prompts and settles completion', async () => {

packages/agent-core-v2/test/agent/toolExecutor/toolScheduler.test.ts

@@ -256,7 +256,6 @@ describe('ToolScheduler leases and budgets', () => {
     abandoned.resolve();
     await waitOneMacrotask();
 
-    expect(drained).toEqual([]);
     expect(started).toEqual(['abandoned']);
     settleEffects();
     await waitOneMacrotask();

packages/agent-core-v2/test/kosong/provider/dsml-tool-parser-conformance.test.ts

@@ -1,5 +1,6 @@
 import { readFileSync } from 'node:fs';
-import { resolve } from 'node:path';
+import { dirname, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
 
 import { describe, expect, it } from 'vitest';
 
@@ -320,8 +321,9 @@ describe('DsmlStreamParser conformance', () => {
 
   describe('implementation parity with @pymodel/kosong', () => {
     it('keeps both parser copies byte-identical except for the message import', () => {
-      const here = resolve(__dirname, '../../../src/kosong/provider/bases/openai/dsml-tool-parser.ts');
-      const legacy = resolve(__dirname, '../../../../kosong/src/providers/dsml-tool-parser.ts');
+      const dir = import.meta.dirname;
+      const here = resolve(dir, '../../../src/kosong/provider/bases/openai/dsml-tool-parser.ts');
+      const legacy = resolve(dir, '../../../../kosong/src/providers/dsml-tool-parser.ts');
       const v2 = readFileSync(here, 'utf8').replace(
         "from '#/kosong/contract/message';",
         "from '#/message';",

packages/agent-core-v2/test/os/backends/node-local/hostFsService.test.ts

@@ -87,6 +87,24 @@ describe('HostFileSystem readLines budget', () => {
 
     expect(lines).toEqual([`${long}\n`]);
   });
+
+  it('preserves complete UTF-8 code points when line budget cuts into multibyte characters', async () => {
+    const path = join(dir, 'multibyte.txt');
+    await writeFile(path, `${'€'.repeat(100)}\n`, 'utf-8');
+
+    const lines: string[] = [];
+    for await (const line of fs.readLines(path, { maxLineBytes: 16, errors: 'strict' })) lines.push(line);
+
+    expect(lines).toEqual([`${'€'.repeat(5)}\n`]);
+  });
+
+  it('rejects maxLineBytes when encoding is not UTF-8', async () => {
+    const path = join(dir, 'other.txt');
+    await writeFile(path, 'content', 'utf-8');
+
+    const iter = fs.readLines(path, { encoding: 'latin1', maxLineBytes: 4 });
+    await expect(iter.next()).rejects.toThrow('maxLineBytes is only supported for UTF-8 encoding');
+  });
 });
 
 describe('HostFileSystem stat / lstat', () => {

packages/agent-core-v2/test/session/sessionMetadata/sessionMetadata.test.ts

@@ -234,8 +234,13 @@ describe('SessionMetadata', () => {
     await meta.ready;
     await meta.registerAgent('sub-1', { homedir: '/tmp/sub-1', type: 'sub', parentAgentId: 'main' });
     const before = (await meta.read()).updatedAt;
-    await meta.unregisterAgent('sub-1');
-    await meta.unregisterAgent('missing');
+    const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(before + 10_000);
+    try {
+      await meta.unregisterAgent('sub-1');
+      await meta.unregisterAgent('missing');
+    } finally {
+      nowSpy.mockRestore();
+    }
     const after = await meta.read();
     expect(after.agents).toEqual({});
     expect(after.updatedAt).toBe(before);

packages/agent-core-v2/test/session/subagent/runAgentTurn.test.ts

@@ -21,36 +21,37 @@ function fakeHandle(totals: TokenUsage[]): IAgentScopeHandle {
     id: 1,
     signal: new AbortController().signal,
     ready: Promise.resolve(),
-    result: Promise.resolve({ type: 'completed', steps: 1, truncated: false } as never),
+    result: Promise.resolve({ type: 'completed', steps: 1, truncated: false }),
     cancel: () => true,
   };
   return {
     id: 'sub',
     kind: LifecycleScope.Agent,
     accessor: {
-      get: (serviceId: unknown) => {
+      get: <T>(serviceId: unknown): T => {
         if (serviceId === ISessionUsageService) {
           return {
             status: () => ({ total: totals[Math.min(cursor, totals.length - 1)] }),
-          };
+          } as T;
         }
         if (serviceId === IAgentPromptService) {
           return {
             enqueue: async () => {
               cursor += 1;
               return { launched: Promise.resolve(turn) };
             },
-          };
+          } as T;
         }
-        if (serviceId === IAgentLoopService) return { cancel: () => true };
+        if (serviceId === IAgentLoopService) return { cancel: () => true } as T;
         if (serviceId === IAgentContextMemoryService) {
-          return { get: () => [{ role: 'assistant', content: [{ type: 'text', text: 'done' }] }] };
+          return { get: () => [{ role: 'assistant', content: [{ type: 'text', text: 'done' }] }] } as T;
         }
-        if (serviceId === IAgentScopeContext) return { agentContext };
+        if (serviceId === IAgentScopeContext) return { agentContext } as T;
         throw new Error(`unexpected service ${String(serviceId)}`);
       },
     },
-  } as unknown as IAgentScopeHandle;
+    dispose: () => {},
+  };
 }
 
 describe('runAgentTurn usage attribution', () => {

packages/agent-gateway/src/start.ts

@@ -318,11 +318,22 @@ export async function startServer(opts: ServerStartOptions): Promise<RunningServ
       await phase('config-publisher', () => configChangedPublisher.close());
       await phase('http', () => app.close());
       await phase('subscriptions', () => {
-        configWarningSubscription.dispose();
-        pluginChangeSubscription.dispose();
-        capabilityInstallSubscription.dispose();
-        authFailureLimiter?.dispose();
-        modelCatalogRefreshScheduler.dispose();
+        for (const sub of [
+          configWarningSubscription,
+          pluginChangeSubscription,
+          capabilityInstallSubscription,
+          authFailureLimiter,
+          modelCatalogRefreshScheduler,
+        ]) {
+          try {
+            sub?.dispose();
+          } catch (error) {
+            logger.warn(
+              { err: error instanceof Error ? error.message : String(error) },
+              'subscription disposal failed; continuing',
+            );
+          }
+        }
       });
       await phase('telemetry', () => shutdownServerTelemetry(telemetry), false);
       await phase('session-metadata', () => drainSessionMetadataWrites());

packages/agent-gateway/test/wsConnectionV1.test.ts

@@ -71,11 +71,8 @@ function makeBroadcaster(): SessionEventBroadcaster {
   } as unknown as SessionEventBroadcaster;
 }
 
-function withBroadcaster(overrides: Record<string, unknown>): SessionEventBroadcaster {
-  return Object.assign(
-    makeBroadcaster() as unknown as Record<string, unknown>,
-    overrides,
-  ) as unknown as SessionEventBroadcaster;
+function withBroadcaster(overrides: Partial<SessionEventBroadcaster>): SessionEventBroadcaster {
+  return Object.assign(makeBroadcaster(), overrides);
 }
 
 function makeRegistry(): IConnectionRegistry {

packages/kosong/test/openai-legacy.test.ts

@@ -1539,7 +1539,9 @@ describe('OpenAILegacyChatProvider', () => {
         };
       }
 
-      (provider as any)._client.chat.completions.create = vi
+      (
+        provider as unknown as { _client: { chat: { completions: { create: unknown } } } }
+      )._client.chat.completions.create = vi
         .fn()
         .mockResolvedValue(mockedStream());