apps/vis/server/test/lib/wire-reader.test.ts

import { describe, it, expect, afterEach } from 'vitest';
import { mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { buildSessionFixture } from '../fixtures/build';
import { readAgentWire } from '../../src/lib/wire-reader';

describe('wire-reader', () => {
  let cleanup: (() => Promise<void>) | null = null;
  afterEach(async () => {
    if (cleanup) await cleanup();
    cleanup = null;
  });

  it('reads main agent wire and assigns line numbers', async () => {
    const { sessionDir, cleanup: c } = await buildSessionFixture('sample-main');
    cleanup = c;
    const result = await readAgentWire(join(sessionDir, 'agents', 'main', 'wire.jsonl'));
    expect(result.metadata.protocolVersion).toBe('1.1');
    expect(result.records[0]!.lineNo).toBe(2); // metadata is line 1, first record is line 2
    expect(result.records.at(-1)!.lineNo).toBe(10);
    expect(result.records.map((r) => r.data.type)).toEqual([
      'config.update',
      'tools.set_active_tools',
      'permission.set_mode',
      'turn.prompt',
      'context.append_message',
      'context.append_loop_event',
      'context.append_loop_event',
      'context.append_loop_event',
      'usage.record',
    ]);
    // No vis annotation should leak into the data/raw bodies.
    for (const entry of result.records) {
      expect(entry.data).not.toHaveProperty('_lineNo');
      expect(entry.raw as object).not.toHaveProperty('_lineNo');
    }
  });

  it('accepts v1.0 wire and migrates nested tool calls to flat shape', async () => {
    const dir = await mkdtemp(join(tmpdir(), 'vis-v10-'));
    const path = join(dir, 'wire.jsonl');
    const lines = [
      JSON.stringify({ type: 'metadata', protocol_version: '1.0', created_at: 1 }),
      JSON.stringify({
        type: 'context.append_message',
        message: {
          role: 'assistant',
          content: [{ type: 'text', text: 'calling' }],
          toolCalls: [
            {
              type: 'function',
              id: 'call_1',
              function: { name: 'Read', arguments: '{"path":"/x"}' },
            },
          ],
        },
      }),
    ];
    await writeFile(path, lines.join('\n') + '\n');
    try {
      const result = await readAgentWire(path);
      expect(result.metadata.protocolVersion).toBe('1.0');
      const entry = result.records[0]!;
      expect(entry.data.type).toBe('context.append_message');

      // `data` carries the migrated (flat) shape.
      const dataMsg = (entry.data as { message: { toolCalls: unknown[] } }).message;
      expect(dataMsg.toolCalls[0]).toEqual({
        type: 'function',
        id: 'call_1',
        name: 'Read',
        arguments: '{"path":"/x"}',
      });
      expect(dataMsg.toolCalls[0]).not.toHaveProperty('function');

      // `raw` keeps the on-disk (nested) shape — this is what the "as
      // written" view in the detail panel relies on.
      const rawMsg = (entry.raw as { message: { toolCalls: Array<{ function: unknown; name?: unknown }> } }).message;
      expect(rawMsg.toolCalls[0]).toHaveProperty('function');
      expect(rawMsg.toolCalls[0]!.function).toEqual({ name: 'Read', arguments: '{"path":"/x"}' });
      expect(rawMsg.toolCalls[0]).not.toHaveProperty('name');
    } finally {
      await rm(dir, { recursive: true, force: true });
    }
  });

  it('best-effort parses unknown protocol versions with a warning', async () => {
    const { sessionDir, cleanup: c } = await buildSessionFixture('sample-main');
    cleanup = c;
    const path = join(sessionDir, 'agents', 'main', 'wire.jsonl');
    const { writeFile, readFile } = await import('node:fs/promises');
    const lines = (await readFile(path, 'utf8')).split('\n');
    lines[0] = '{"type":"metadata","protocol_version":"0.9","created_at":1}';
    await writeFile(path, lines.join('\n'));
    const result = await readAgentWire(path);
    expect(result.metadata.protocolVersion).toBe('0.9');
    expect(result.records.length).toBeGreaterThan(0);
    expect(result.warnings.some((w) => /unrecognised protocol_version.*0\.9/i.test(w))).toBe(
      true,
    );
  });

  it('passes v1.5 records through unchanged with no warnings', async () => {
    const dir = await mkdtemp(join(tmpdir(), 'vis-v15-'));
    const path = join(dir, 'wire.jsonl');
    const records = [
      { type: 'metadata', protocol_version: '1.5', created_at: 1 },
      { type: 'token_counting.measured', agentId: 'main', length: 4, tokens: 1234, time: 2 },
      {
        type: 'cron.add',
        agentId: 'main',
        task: { id: '01ARZ3NDEKTSV4RRFFQ69G5FAV', cron: '* * * * *', prompt: 'p', createdAt: 3 },
        time: 3,
      },
    ];
    await writeFile(path, records.map((r) => JSON.stringify(r)).join('\n') + '\n');
    try {
      const result = await readAgentWire(path);
      expect(result.metadata.protocolVersion).toBe('1.5');
      expect(result.warnings).toEqual([]);
      expect(result.records).toHaveLength(2);
      // data === raw for current-protocol records (no migration applied).
      expect(result.records[0]!.data).toEqual(result.records[0]!.raw);
      expect(result.records[1]!.data).toEqual(result.records[1]!.raw);
    } finally {
      await rm(dir, { recursive: true, force: true });
    }
  });

  it('recovers a headerless journal with the same v1.4 assumption as core-v2', async () => {
    const { sessionDir, cleanup: c } = await buildSessionFixture('sample-main');
    cleanup = c;
    const path = join(sessionDir, 'agents', 'main', 'wire.jsonl');
    await writeFile(
      path,
      JSON.stringify({
        type: 'goal.create',
        agentId: 'main',
        goalId: 'goal-1',
        objective: 'ship',
        time: 40,
      }) + '\n',
    );

    const result = await readAgentWire(path);

    expect(result.metadata).toEqual({ protocolVersion: '1.4', createdAt: 0 });
    expect(result.warnings).toEqual([
      'line 1: missing metadata header — assuming protocol_version "1.4"',
    ]);
    expect(result.records[0]).toMatchObject({
      lineNo: 1,
      data: { type: 'goal.create', wallClockResumedAt: 40 },
    });
  });

  it('normalizes legacy plan revision paths to the current storage key', async () => {
    const { sessionDir, cleanup: c } = await buildSessionFixture('sample-main');
    cleanup = c;
    const path = join(sessionDir, 'agents', 'main', 'wire.jsonl');
    await writeFile(
      path,
      [
        JSON.stringify({ type: 'metadata', protocol_version: '1.5', created_at: 1 }),
        JSON.stringify({
          type: 'plan.revision',
          agentId: 'main',
          id: 'demo-plan',
          version: 2,
          path: 'sessions/workspace/session_demo/agents/main/plan/demo-plan/v2.md',
          sha256: 'abc',
          bytes: 10,
          time: 2,
        }),
      ].join('\n') + '\n',
    );

    const result = await readAgentWire(path);

    expect(result.records[0]!.data).toMatchObject({
      type: 'plan.revision',
      key: 'plan/demo-plan/v2.md',
    });
    expect(result.records[0]!.data).not.toHaveProperty('path');
    expect(result.records[0]!.raw).toHaveProperty(
      'path',
      'sessions/workspace/session_demo/agents/main/plan/demo-plan/v2.md',
    );
  });

  it('collects warnings for malformed body lines', async () => {
    const { sessionDir, cleanup: c } = await buildSessionFixture('sample-main');
    cleanup = c;
    const path = join(sessionDir, 'agents', 'main', 'wire.jsonl');
    const { appendFile } = await import('node:fs/promises');
    await appendFile(path, 'not json\n');
    const result = await readAgentWire(path);
    expect(result.warnings.length).toBe(1);
    expect(result.warnings[0]).toMatch(/line 11/);
  });
});