import fs from "node:fs/promises";
import { afterEach, describe, expect, test, vi } from "vitest";
import {
  AGENT_MESSAGE_RESPONSE_EVENT_TYPE,
  AGENT_MESSAGE_SOURCE_EVENT_TYPE,
  appendCoAgentMessage,
  parseAgentMessageResponseEvent,
} from "../src/agents/agent-messages.js";
import { declarationFingerprint } from "../src/agents/declarations.js";
import { OBSERVATION_OUTPUT_CONTRACT } from "../src/agents/output-contracts.js";
import { ThoughtAgentRuntime } from "../src/agents/runtime.js";
import {
  buildSubscribedAgentConversationContextPacket,
  type AgentContextPacket,
} from "../src/agents/context.js";
import { sha256 } from "../src/core/json.js";
import type { AgentRunner, ThoughtAgentDeclaration } from "../src/agents/types.js";
import type { EventCandidate } from "../src/events/types.js";
import { TelegramChannelDispatcher } from "../src/bridges/telegram-dispatcher.js";
import type { TelegramBotClient } from "../src/connectors/telegram-bot.js";
import type { JazzThoughtStore } from "../src/jazz/store.js";
import { temporaryProject, testStore } from "./helpers.js";

const stores: JazzThoughtStore[] = [];
const roots: string[] = [];

afterEach(async () => {
  await Promise.all(stores.splice(0).map((store) => store.close()));
  await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));
});

describe("private agent messages", () => {
  test("keeps an exact Co-to-Stream thread separate and settles typed response receipts", async () => {
    const project = await temporaryProject("thoughtstream-agent-message-");
    roots.push(project);
    const store = await testStore(project);
    stores.push(store);
    const contexts = new Map<string, AgentContextPacket>();
    const runner: AgentRunner = {
      mode: "deterministic",
      run: vi.fn(async (input) => {
        contexts.set(String(input.event.payload.messageId), input.context);
        return {
          summary: `Stream reply: ${String(input.event.payload.text)}`,
          tags: ["conversation"],
          importance: "normal" as const,
          confidence: 0.5,
        };
      }),
    };
    const declaration = agentMessageDeclaration();
    const runtime = new ThoughtAgentRuntime(store, [runner]);

    const first = await appendCoAgentMessage(store, {
      messageId: "co-first",
      threadId: "co-stream-test",
      text: "Hello Stream.",
      occurredAt: "2026-08-04T02:00:00.000Z",
    });
    expect(first.inserted).toBe(true);
    const replay = await appendCoAgentMessage(store, {
      messageId: "co-first",
      threadId: "co-stream-test",
      text: "Hello Stream.",
      occurredAt: "2026-08-04T02:01:00.000Z",
    });
    expect(replay).toMatchObject({ inserted: false, event: { id: first.event.id } });
    await expect(appendCoAgentMessage(store, {
      messageId: "co-first",
      threadId: "co-stream-test",
      text: "Divergent reuse.",
    })).rejects.toThrow("divergent content");

    const [firstProcessed] = await runtime.consumeBacklog([declaration]);
    const firstRun = await store.getRun(firstProcessed!.runId);
    const firstOutput = await store.getEvent(firstRun!.outputEventIds[0]!);
    const firstResponse = parseAgentMessageResponseEvent(firstOutput!);
    expect(firstRun).toMatchObject({ status: "completed", agentId: "stream-agent-conversation" });
    expect(firstOutput).toMatchObject({
      type: AGENT_MESSAGE_RESPONSE_EVENT_TYPE,
      source: "agent:stream-agent-conversation",
      actor: "stream-agent-conversation",
      parentEventId: first.event.id,
      rootEventId: first.event.id,
      correlationId: "co-stream-test",
      privacy: "sensitive",
    });
    expect(firstResponse).toMatchObject({
      inReplyToMessageId: "co-first",
      threadId: "co-stream-test",
      senderAgentId: "stream-agent-conversation",
      recipientAgentId: "co",
      summary: "Stream reply: Hello Stream.",
    });
    const dispatcher = new TelegramChannelDispatcher({
      id: "telegram-notifier:agent-message-deny",
      client: { sendMessage: vi.fn(() => { throw new Error("agent response reached Telegram"); }) } as unknown as TelegramBotClient,
      chatId: "123456789",
      allowedSources: ["agent-message:co"],
      allowedActors: ["agent:co"],
      directReplyAgentIds: ["stream-agent-conversation"],
      runStatuses: ["completed", "failed"],
    });
    await expect(dispatcher.sendPending(store, { includeNormal: true })).resolves.toMatchObject({
      pending: 1,
      eligible: 0,
      delivered: 0,
    });
    expect(contexts.get("co-first")).toMatchObject({
      text: "Hello Stream.",
      manifest: {
        contextStrategy: "agent-conversation",
        agentMessage: {
          senderAgentId: "co",
          recipientAgentId: "stream-agent-conversation",
          threadId: "co-stream-test",
        },
      },
    });
    expect(contexts.get("co-first")?.messages).toBeUndefined();

    await appendCoAgentMessage(store, {
      messageId: "co-other-thread",
      threadId: "other-thread",
      text: "This is another thread.",
      occurredAt: "2026-08-04T02:02:00.000Z",
    });
    await runtime.consumeBacklog([declaration]);
    expect(contexts.get("co-other-thread")?.messages).toBeUndefined();

    await appendCoAgentMessage(store, {
      messageId: "co-second",
      threadId: "co-stream-test",
      text: "Do you remember me?",
      occurredAt: "2026-08-04T02:03:00.000Z",
    });
    await runtime.consumeBacklog([declaration]);
    expect(contexts.get("co-second")?.messages).toEqual([
      { role: "user", content: "Hello Stream." },
      { role: "assistant", content: "Stream reply: Hello Stream." },
    ]);
    expect(JSON.stringify(contexts.get("co-second")?.messages)).not.toContain("another thread");
  });

  test("skips a source-row route forgery before runner invocation and advances progress", async () => {
    const project = await temporaryProject("thoughtstream-agent-message-skip-");
    roots.push(project);
    const store = await testStore(project);
    stores.push(store);
    const runner: AgentRunner = {
      mode: "deterministic",
      run: vi.fn(async () => ({
        summary: "should not run",
        tags: [],
        importance: "low" as const,
        confidence: 0,
      })),
    };
    const runtime = new ThoughtAgentRuntime(store, [runner]);
    const declaration = agentMessageDeclaration();
    const forged: EventCandidate = {
      type: AGENT_MESSAGE_SOURCE_EVENT_TYPE,
      schemaVersion: 1,
      source: "agent-message:co",
      sourceKind: "agent",
      externalId: "forged-recipient",
      idempotencyKey: "forged-recipient",
      occurredAt: "2026-08-04T02:00:00.000Z",
      actor: "agent:co",
      correlationId: "co-stream-test",
      privacy: "sensitive",
      payload: {
        messageId: "forged-recipient",
        threadId: "co-stream-test",
        senderAgentId: "co",
        recipientAgentId: "other-agent",
        text: "Do not admit this.",
      },
    };
    const appended = await store.appendEvent(forged);
    const [processed] = await runtime.consumeBacklog([declaration]);
    const run = await store.getRun(processed!.runId);
    expect(runner.run).not.toHaveBeenCalled();
    expect(run).toMatchObject({
      status: "skipped",
      triggerEventId: appended.event.id,
      errorText: "Agent message is outside the endpoint route",
      contextManifest: {
        skip: { code: "agent-message-not-admitted" },
      },
    });
    expect(await store.getConsumerProgress(
      `consumer-progress:stream-agent-conversation:1:agent-message%3Aco`,
    )).toMatchObject({ lastSequence: appended.event.sourceSequence });
  });

  test("freezes operator documents and exact thread context for retries", async () => {
    const project = await temporaryProject("thoughtstream-agent-message-snapshot-");
    roots.push(project);
    const store = await testStore(project);
    stores.push(store);
    await putCurrentDocument(store, "identity", "identity.md", "# Stream identity v1\n", "identity-v1");
    await putCurrentDocument(store, "memory", "memory.md", "# Stream memory\n", "memory-v1");
    const declaration: ThoughtAgentDeclaration = {
      ...agentMessageDeclaration(),
      contextDocumentMaxChars: 4_000,
      contextDocumentSubscriptions: [{
        source: "filesystem:telegram-agent-context",
        paths: ["identity.md", "memory.md"],
        required: true,
      }],
    };
    declaration.declarationFingerprint = declarationFingerprint(declaration);
    const appended = await appendCoAgentMessage(store, {
      messageId: "co-snapshot",
      threadId: "co-stream-snapshot",
      text: "Freeze this turn.",
      occurredAt: "2026-08-04T02:00:00.000Z",
    });
    const first = await buildSubscribedAgentConversationContextPacket(declaration, appended.event, store);
    await putCurrentDocument(store, "identity", "identity.md", "# Stream identity v2\n", "identity-v2");
    const retried = await buildSubscribedAgentConversationContextPacket(declaration, appended.event, store);
    expect(retried).toEqual(first);
    expect(first.systemText).toContain("Stream identity v1");
    expect(first.systemText).not.toContain("Stream identity v2");
    expect(first.systemText).toContain("current private correspondent is agent:co");
    const snapshot = first.manifest.contextSnapshot as { id: string };
    expect(await store.getDocumentVersion(snapshot.id)).toMatchObject({
      source: "context:stream-agent-conversation",
      path: `agent-subscribed-context/${appended.event.id}.json`,
      contentType: "application/vnd.thoughtstream.agent-context+json",
    });
  });

  test("discovers the first Co message after an exact replay-now endpoint starts", async () => {
    const project = await temporaryProject("thoughtstream-agent-message-live-");
    roots.push(project);
    const store = await testStore(project);
    stores.push(store);
    const runner: AgentRunner = {
      mode: "deterministic",
      run: async () => ({
        summary: "I hear Co.",
        tags: ["conversation"],
        importance: "normal",
        confidence: 0.5,
      }),
    };
    const declaration = { ...agentMessageDeclaration(), initialReplay: "now" as const };
    declaration.declarationFingerprint = declarationFingerprint(declaration);
    const runtime = new ThoughtAgentRuntime(store, [runner], { reconcileIntervalMs: 25 });
    const consumers = await runtime.startConsumers([declaration]);
    const appended = await appendCoAgentMessage(store, {
      messageId: "co-after-start",
      threadId: "co-stream-live",
      text: "Can you hear me?",
    });
    await waitFor(async () => (await store.getRunsForTriggerEvents([appended.event.id]))
      .some((run) => run.status === "completed"), 5_000);
    await consumers.stop();
    const [response] = (await store.listEvents({ types: [AGENT_MESSAGE_RESPONSE_EVENT_TYPE] }))
      .filter((event) => event.parentEventId === appended.event.id);
    expect(parseAgentMessageResponseEvent(response!)).toMatchObject({ summary: "I hear Co." });
  });
});

function agentMessageDeclaration(): ThoughtAgentDeclaration {
  const declaration: ThoughtAgentDeclaration = {
    id: "stream-agent-conversation",
    version: 1,
    name: "Stream",
    description: "fixture private agent conversation",
    mode: "deterministic",
    role: "standard",
    outputContract: { ...OBSERVATION_OUTPUT_CONTRACT.identity },
    outputMode: "conversation-text",
    eventTypes: [AGENT_MESSAGE_SOURCE_EVENT_TYPE],
    compiledEventTypes: [AGENT_MESSAGE_SOURCE_EVENT_TYPE],
    sourcePatterns: ["agent-message:co"],
    acceptedPrivacy: ["sensitive"],
    initialReplay: "beginning",
    outputEventType: AGENT_MESSAGE_RESPONSE_EVENT_TYPE,
    emit: [AGENT_MESSAGE_RESPONSE_EVENT_TYPE],
    promptRef: "fixture-agent-message.md",
    systemPrompt: "Reply to Co.",
    enabled: true,
    maxEvents: 20,
    maxInputChars: 8_000,
    contextStrategy: "agent-conversation",
    maxOutputTokens: 2_000,
    timeoutMs: 60_000,
    tools: [],
    proposals: [],
    externalActions: false,
  };
  declaration.declarationFingerprint = declarationFingerprint(declaration);
  return declaration;
}

async function putCurrentDocument(
  store: JazzThoughtStore,
  documentId: string,
  documentPath: string,
  content: string,
  versionId: string,
): Promise<void> {
  const source = "filesystem:telegram-agent-context";
  const digest = sha256(content);
  const at = "2026-08-04T01:00:00.000Z";
  await store.appendDocumentVersion({
    id: versionId,
    source,
    documentId,
    path: documentPath,
    contentType: "text/markdown",
    sha256: digest,
    content,
    sizeBytes: Buffer.byteLength(content),
    mtimeMs: Date.parse(at),
    createdAt: at,
  });
  await store.upsertCurrentDocument({
    id: `${source}:${documentId}`,
    source,
    documentId,
    path: documentPath,
    versionId,
    sha256: digest,
    contentType: "text/markdown",
    sizeBytes: Buffer.byteLength(content),
    mtimeMs: Date.parse(at),
    deleted: false,
    updatedAt: at,
  });
}

async function waitFor(predicate: () => Promise<boolean>, timeoutMs: number): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (await predicate()) return;
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  throw new Error("Timed out waiting for agent-message consumer");
}