import { registerChatTopicRoutes } from "./chat-topic-routes.js";
import { matchHistoryText } from "../sessions/history-search.js";
import { credentialRefreshCoordinator } from "../product/credential-refresh.js";
import { type Router } from "express";
import { z } from "zod";

import { type ChatSessionBridge } from "../chat/sessionBridge.js";
import { projectInteractionReplyMessages } from "../chat/interaction-message-presentation.js";
import type { InteractionRequestStore } from "../sessions/interaction-request-store.js";
import { agentModelOptions } from "../chat/models.js";
import { config } from "../config.js";
import {
  getLocalAgentContext,
  getTrustedProductCredential,
} from "../context/localContext.js";
import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js";
import { logger } from "../logger.js";
import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
import type { ProductCredential } from "../product/server-auth-client.js";
import {
  type SessionMetadataStore,
  type SessionRecord,
  type SessionPurpose,
  SessionContextMismatchError,
  readPurpose,
} from "../sessions/metadataStore.js";
import { type SessionTranscriptStore } from "../sessions/transcriptStore.js";
import { type SessionUiStateStore } from "../sessions/uiStateStore.js";
import { HistoryCursorError, historyScope, sessionHistoryPage, sessionMessagePage } from "../sessions/history-page.js";
import { type LearningOrchestrator } from "../learning/orchestrator.js";
import { toActorKey, toProjectKey } from "../utils/fileStore.js";
import { type ChatRunState } from "./chatRunState.js";
import { buildForkedSessionUiState, isGenericTaskCollaborationTitle } from "./chatSession.js";
import { type StreamSubscriber } from "./chatUiState.js";
import { agentTriggerContextSchema } from "./agentTrigger.js";
import {
  bindOperationalTaskCredential,
  clearOperationalTaskLease,
  operationalTaskClient,
  releaseOperationalTaskCredential,
} from "../product/operational-task-client.js";

const createSessionPayloadSchema = z.object({
  session_id: z.string().max(128).optional(),
  parent_session_id: z.string().max(128).optional(),
  trigger_context: agentTriggerContextSchema.optional(),
  topic_id: z.string().uuid().optional(),
});

const taskExecutionPayloadSchema = z.object({
  task_id: z.string().uuid(),
  run_id: z.string().uuid(),
  execution_request_id: z.string().uuid(),
  agent_instance_id: z.string().trim().min(1).max(128),
  workflow: z.string().trim().min(1).max(120).optional(),
});

const taskExecutionHistoryQuerySchema = z.object({
  run_id: z.string().uuid(),
  session_id: z.string().trim().min(1).max(128),
});

const taskExecutionDebugPayloadSchema = z.object({
  run_id: z.string().uuid(),
  session_id: z.string().trim().min(1).max(128),
}).strict();

export const buildTaskExecutionPrompt = (workflow?: string) =>
  `执行宿主绑定的工况任务${workflow ? `：${workflow}` : ""}。` +
  `第一步分别调用 mcp_tjwater_get_task_context 读取默认摘要和 section="skill"；` +
  `以返回的 skill.name、skill.version 和 instructions 为本轮业务流程，不沿用记忆中的旧版本。` +
  `若 Skill 返回不可用或读取失败，明确报告执行异常，不得声称已按某个版本执行。` +
  `随后按后台调查指令推进当前目标，保存证据、进度和报告；收尾遵循后台调查指令的停止条件。`;

const sessionListQuerySchema = z.object({
  purpose: z.enum(["ordinary_chat", "task_execution", "task_collaboration"]).optional(),
  task_id: z.string().uuid().optional(),
  include_collaboration: z.enum(["true", "false"]).optional(),
  q: z.string().trim().max(200).optional(),
  topic_id: z.union([z.string().uuid(), z.literal("unclassified")]).optional(),
  limit: z.coerce.number().int().min(1).max(100).optional(),
  cursor: z.string().max(2048).optional(),
});

const messageQuerySchema = z.object({
  message_limit: z.coerce.number().int().min(1).max(100).optional(),
});
const olderMessagesQuerySchema = z.object({
  limit: z.coerce.number().int().min(1).max(100).default(40),
  before: z.string().max(2048).optional(),
});

const forkPayloadSchema = z.object({
  session_id: z.string().max(128).optional(),
  keep_message_count: z.coerce.number().int().min(0),
});

type RegisterChatSessionRoutesOptions = {
  interactionRequestStore?: InteractionRequestStore;
  learningOrchestrator: LearningOrchestrator;
  frontendActionCoordinator: FrontendActionCoordinator;
  runState: ChatRunState;
  runtime: OpencodeRuntimeAdapter;
  sessionBridge: ChatSessionBridge;
  sessionMetadataStore: SessionMetadataStore;
  sessionTranscriptStore: SessionTranscriptStore;
  sessionUiStateStore: SessionUiStateStore;
};

const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
  sessionId: sessionRecord.sessionId,
  actorKey: sessionRecord.actorKey,
  projectKey: sessionRecord.projectKey,
});

const toSse = (event: string, data: Record<string, unknown>) =>
  `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;

export const registerChatSessionRoutes = (
  chatRouter: Router,
  options: RegisterChatSessionRoutesOptions,
) => {
  const {
    interactionRequestStore,
    frontendActionCoordinator,
    learningOrchestrator,
    runState,
    runtime,
    sessionBridge,
    sessionMetadataStore,
    sessionTranscriptStore,
    sessionUiStateStore,
  } = options;
  registerChatTopicRoutes(chatRouter, sessionMetadataStore);
  const { activeRuns, lastRunStatuses } = runState;
  const getSessionRunStatus = runState.getStatus;

  const withInteractionReplies = async (record: SessionRecord, messages: unknown[]) => {
    if (!interactionRequestStore) return messages;
    const ids = new Set(messages.flatMap((message) => {
      if (!message || typeof message !== "object") return [];
      const value = message as Record<string, unknown>;
      return value.role === "user" && typeof value.id === "string" ? [value.id] : [];
    }));
    const replies = await interactionRequestStore.forMessages(toSessionUiStateContext(record), ids);
    return projectInteractionReplyMessages(messages, replies);
  };

  const refreshCollaborationTitle = async (record: SessionRecord, credential: ProductCredential | null, knownTaskTitle?: string, signal?: AbortSignal) => {
    if (readPurpose(record) !== "task_collaboration" || !record.taskId
      || !isGenericTaskCollaborationTitle(record.title)) return record;
    if ((!credential && !knownTaskTitle) || signal?.aborted) return record;
    try {
      const state = await sessionUiStateStore.read(toSessionUiStateContext(record));
      if (state?.isTitleManuallyEdited) return record;
      let taskTitle = knownTaskTitle;
      if (!taskTitle && credential) {
        const titleSessionId = `task-title-${crypto.randomUUID()}`;
        bindOperationalTaskCredential(titleSessionId, credential);
        try {
          taskTitle = (await operationalTaskClient.getAgentContext(titleSessionId, record.taskId, undefined, signal)).task.title;
        } finally {
          releaseOperationalTaskCredential(titleSessionId);
          clearOperationalTaskLease(titleSessionId);
        }
      }
      if (!taskTitle?.trim()) return record;
      const latestState = await sessionUiStateStore.read(toSessionUiStateContext(record));
      if (latestState?.isTitleManuallyEdited) return record;
      return await sessionMetadataStore.touch(record, { title: taskTitle }, {
        preserveUpdatedAt: true, expectedTitle: record.title ?? "",
      });
    } catch (error) {
      if (!signal?.aborted) logger.warn({ err: error, sessionId: record.sessionId }, "failed to refresh task collaboration title");
      return record;
    }
  };

  const startTaskExecution = (
    record: SessionRecord,
    credential: NonNullable<ReturnType<typeof getTrustedProductCredential>>,
    model: string,
    workflow?: string,
  ) => {
    queueMicrotask(() => {
      const baseUrl = process.env.TJWATER_AGENT_INTERNAL_BASE_URL ?? `http://127.0.0.1:${config.PORT}`;
      void fetch(new URL("/api/v1/responses", baseUrl), {
        method: "POST",
        headers: {
          Authorization: credential.authorization,
          "Content-Type": "application/json",
          "X-Project-Id": credential.projectId,
        },
        body: JSON.stringify({
          conversation: record.sessionId,
          model,
          stream: true,
          input: buildTaskExecutionPrompt(workflow),
        }),
      }).then(async (response) => {
        const reader = response.body?.getReader();
        if (reader) {
          while (!(await reader.read()).done) {
            // Drain the stream without retaining the model response in memory.
          }
        }
        if (!response.ok) {
          clearOperationalTaskLease(record.sessionId);
          logger.error({ sessionId: record.sessionId, status: response.status }, "background task execution failed");
        }
      }).catch((error) => {
        clearOperationalTaskLease(record.sessionId);
        logger.error({ err: error, sessionId: record.sessionId }, "background task execution request failed");
      });
    });
  };

  const prepareTaskExecution = async (
    record: SessionRecord,
    credential: NonNullable<ReturnType<typeof getTrustedProductCredential>>,
    input: z.infer<typeof taskExecutionPayloadSchema>,
  ) => {
    bindOperationalTaskCredential(record.sessionId, credential);
    try {
      const taskContext = await operationalTaskClient.getAgentContext(record.sessionId, record.taskId!);
      const task = taskContext.task && typeof taskContext.task === "object"
        ? taskContext.task as Record<string, unknown>
        : null;
      const activeRun = task?.active_run && typeof task.active_run === "object"
        ? task.active_run as Record<string, unknown>
        : null;
      if (
        !activeRun
        || activeRun.run_id !== input.run_id
        || activeRun.execution_request_id !== input.execution_request_id
        || typeof activeRun.revision !== "number"
        || typeof activeRun.status !== "string"
      ) {
        throw new Error("task context did not include the requested active run");
      }
      if (activeRun.status === "queued") {
        await operationalTaskClient.claimTask(record.sessionId, record.taskId!, input.run_id, {
          requestId: input.execution_request_id,
          expectedRevision: activeRun.revision,
          sourceSessionId: record.sessionId,
        });
        return "running";
      }
      if (
        ["claimed", "running"].includes(activeRun.status)
        && activeRun.agent_session_id === record.sessionId
        && activeRun.agent_instance_id === config.AGENT_INSTANCE_ID
      ) {
        return "running";
      }
      throw new Error(`task run is not executable from this session (${activeRun.status})`);
    } finally {
      releaseOperationalTaskCredential(record.sessionId);
    }
  };

  const resolveTaskExecutionHistory = async (
    projectId: string,
    credential: ProductCredential,
    taskId: string,
    runId: string,
    sessionId: string,
  ) => {
    const validationSessionId = `task-history-access-${crypto.randomUUID()}`;
    let taskTitle: string;
    bindOperationalTaskCredential(validationSessionId, credential);
    try {
      taskTitle = (await operationalTaskClient.getAgentContext(validationSessionId, taskId)).task.title;
    } finally {
      releaseOperationalTaskCredential(validationSessionId);
      clearOperationalTaskLease(validationSessionId);
    }

    const record = await sessionMetadataStore.getTaskExecutionBinding(
      projectId,
      taskId,
      runId,
      sessionId,
    );
    if (!record) return null;
    const state = await sessionUiStateStore.read(toSessionUiStateContext(record));
    if (!state) return null;
    return { record, state, taskTitle };
  };

  chatRouter.post("/session", async (req, res) => {
    const parsed = createSessionPayloadSchema.safeParse(req.body ?? {});
    if (!parsed.success) {
      res.status(400).json({
        message: "invalid request payload",
        detail: parsed.error.flatten(),
      });
      return;
    }
    const authContext = getLocalAgentContext(req);
    const projectId = authContext.projectId;
    const userId = authContext.userId;
    const actorKey = authContext.actorKey;
    const projectKey = authContext.projectKey;
    if (parsed.data.topic_id && !await sessionMetadataStore.topics.get(authContext, parsed.data.topic_id)) {
      res.status(404).json({ message: "主题不存在" }); return;
    }
    const trigger = parsed.data.trigger_context;
    if (trigger && !projectId) {
      res.status(403).json({ code: "PROJECT_CONTEXT_REQUIRED", message: "任务会话必须关联项目" });
      return;
    }
    const parentSessionId = parsed.data.parent_session_id?.trim();
    if (parentSessionId && !await sessionMetadataStore.get(authContext, parentSessionId)) {
      res.status(404).json({ message: "session not found" });
      return;
    }
    const purpose: SessionPurpose = trigger ? "task_execution" : "ordinary_chat";
    const taskId = trigger?.subject_id;
    if (trigger) {
      const existingTriggeredSession = await sessionMetadataStore.findByTrigger(
        { actorKey, projectId, projectKey, userId },
        trigger.trigger_id,
      );
      if (existingTriggeredSession) {
        res.status(200).json({
          session_id: existingTriggeredSession.sessionId,
          created_at: existingTriggeredSession.createdAt,
          updated_at: existingTriggeredSession.updatedAt,
          status: existingTriggeredSession.status,
          title: existingTriggeredSession.title,
          parent_session_id: existingTriggeredSession.parentSessionId,
          created: false,
          event_duplicate: true,
          trigger_id: trigger.trigger_id,
        });
        return;
      }
    }
    const requestedSessionId = parsed.data.session_id?.trim();
    if (requestedSessionId && !await sessionMetadataStore.get(authContext, requestedSessionId)) {
      res.status(404).json({ message: "session not found" });
      return;
    }
    let sessionId = requestedSessionId;
    if (!sessionId) {
      try {
        sessionId = (await runtime.createSession(undefined, { actorKey, projectKey })).id;
      } catch (error) {
        sessionId = sessionBridge.createClientSessionId();
        logger.warn(
          { err: error, sessionId },
          "runtime unavailable while creating chat session; using local degraded session",
        );
      }
    }

    const { record, created } = await sessionMetadataStore.ensure({
      actorKey,
      parentSessionId: parsed.data.parent_session_id,
      topicId: parsed.data.topic_id,
      projectId,
      projectKey,
      sessionId,
      purpose,
      taskId,
      trigger: trigger ? {
        triggerId: trigger.trigger_id,
        source: trigger.source,
        subjectType: trigger.subject_type,
        subjectId: trigger.subject_id,
        workflow: trigger.workflow,
        occurredAt: trigger.occurred_at,
      } : undefined,
      userId,
    });

    res.status(created ? 201 : 200).json({
      session_id: record.sessionId,
      created_at: record.createdAt,
      updated_at: record.updatedAt,
      status: record.status,
      title: record.title,
      parent_session_id: record.parentSessionId,
      topic_id: record.topicId ?? null,
      created,
      event_duplicate: false,
      trigger_id: record.triggerId,
      purpose: readPurpose(record),
      task_id: record.taskId,
    });
  });

  chatRouter.post("/task/:task_id/collaboration/session", async (req, res) => {
    const taskId = z.string().uuid().safeParse(req.params.task_id);
    if (!taskId.success) {
      res.status(400).json({ message: "valid task_id is required" });
      return;
    }
    const authContext = getLocalAgentContext(req);
    const credential = getTrustedProductCredential(req);
    if (!credential) {
      res.status(401).json({ message: "task collaboration requires product credentials" });
      return;
    }
    const metadataContext = {
      actorKey: authContext.actorKey,
      projectId: authContext.projectId,
      projectKey: authContext.projectKey,
      userId: authContext.userId,
    };
    const existing = await sessionMetadataStore.findByTask(
      metadataContext,
      taskId.data,
      "task_collaboration",
    );
    const validationSessionId = `task-access-${crypto.randomUUID()}`;
    let taskTitle: string;
    bindOperationalTaskCredential(validationSessionId, credential);
    try {
      taskTitle = (await operationalTaskClient.getAgentContext(validationSessionId, taskId.data)).task.title;
    } catch (error) {
      res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502).json({
        message: error instanceof Error ? error.message : "task access validation failed",
      });
      return;
    } finally {
      releaseOperationalTaskCredential(validationSessionId);
      clearOperationalTaskLease(validationSessionId);
    }
    if (existing) {
      const current = await sessionMetadataStore.get(metadataContext, existing.sessionId);
      if (!current) { res.status(404).json({ message: "session not found" }); return; }
      const titled = await refreshCollaborationTitle(current, credential, taskTitle);
      res.json({
        session_id: titled.sessionId,
        title: titled.title,
        purpose: readPurpose(titled),
        task_id: titled.taskId,
        created: false,
        status: titled.status,
      });
      return;
    }
    const runtimeSession = await runtime.createSession(undefined, metadataContext);
    const { record } = await sessionMetadataStore.ensure({
      ...metadataContext,
      sessionId: runtimeSession.id,
      purpose: "task_collaboration",
      taskId: taskId.data,
    });
    const titled = await sessionMetadataStore.touch(record, { title: taskTitle });
    res.status(201).json({
      session_id: titled.sessionId,
      title: titled.title,
      purpose: readPurpose(titled),
      task_id: titled.taskId,
      created: true,
      status: titled.status,
    });
  });

  chatRouter.get("/task/:task_id/execution-session", async (req, res) => {
    const taskId = z.string().uuid().safeParse(req.params.task_id);
    const query = taskExecutionHistoryQuerySchema.safeParse(req.query);
    if (!taskId.success || !query.success) {
      res.status(400).json({ message: "valid task_id, run_id and session_id are required" });
      return;
    }
    const authContext = getLocalAgentContext(req);
    const credential = getTrustedProductCredential(req);
    if (!authContext.projectId) {
      res.status(403).json({ message: "task execution history requires project context" });
      return;
    }
    if (!credential) {
      res.status(401).json({ message: "task execution history requires product credentials" });
      return;
    }

    let history: Awaited<ReturnType<typeof resolveTaskExecutionHistory>>;
    try {
      history = await resolveTaskExecutionHistory(
        authContext.projectId,
        credential,
        taskId.data,
        query.data.run_id,
        query.data.session_id,
      );
    } catch (error) {
      res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502).json({
        message: error instanceof Error ? error.message : "task access validation failed",
      });
      return;
    }
    if (!history) {
      res.status(404).json({ message: "task execution session not found" });
      return;
    }
    const { record, state, taskTitle } = history;
    const runStatus = getSessionRunStatus(record.sessionId);
    res.setHeader("Cache-Control", "no-store");
    res.json({
      id: record.sessionId,
      session_id: record.sessionId,
      title: `${taskTitle} · 执行记录`,
      created_at: record.createdAt,
      updated_at: record.updatedAt,
      status: record.status,
      messages: await withInteractionReplies(record, state.messages ?? []),
      execution_plans: state.executionPlans ?? [],
      is_streaming: runStatus === "running",
      run_status: runStatus,
      purpose: "task_execution",
      task_id: record.taskId,
      read_only: true,
    });
  });

  chatRouter.post("/task/:task_id/execution-session/debug", async (req, res) => {
    const taskId = z.string().uuid().safeParse(req.params.task_id);
    const payload = taskExecutionDebugPayloadSchema.safeParse(req.body);
    if (!taskId.success || !payload.success) {
      res.status(400).json({ message: "valid task_id, run_id and session_id are required" });
      return;
    }
    const authContext = getLocalAgentContext(req);
    const credential = getTrustedProductCredential(req);
    if (!authContext.projectId) {
      res.status(403).json({ message: "task execution debugging requires project context" });
      return;
    }
    if (!credential) {
      res.status(401).json({ message: "task execution debugging requires product credentials" });
      return;
    }

    let history: Awaited<ReturnType<typeof resolveTaskExecutionHistory>>;
    try {
      history = await resolveTaskExecutionHistory(
        authContext.projectId,
        credential,
        taskId.data,
        payload.data.run_id,
        payload.data.session_id,
      );
    } catch (error) {
      res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502).json({
        message: error instanceof Error ? error.message : "task access validation failed",
      });
      return;
    }
    if (!history) {
      res.status(404).json({ message: "task execution session not found" });
      return;
    }

    const metadataContext = {
      actorKey: authContext.actorKey,
      projectId: authContext.projectId,
      projectKey: authContext.projectKey,
      userId: authContext.userId,
    };
    const existing = (await sessionMetadataStore.list(metadataContext, {
      purpose: "task_collaboration",
      taskId: taskId.data,
    })).find((record) => record.parentSessionId === history.record.sessionId);
    if (existing) {
      res.json({
        session_id: existing.sessionId,
        title: existing.title,
        purpose: readPurpose(existing),
        task_id: existing.taskId,
        parent_session_id: existing.parentSessionId,
        created: false,
        status: existing.status,
      });
      return;
    }

    const runtimeSession = await runtime.createSession(undefined, metadataContext);
    const { record } = await sessionMetadataStore.ensure({
      ...metadataContext,
      sessionId: runtimeSession.id,
      parentSessionId: history.record.sessionId,
      purpose: "task_collaboration",
      taskId: taskId.data,
    });
    const titled = await sessionMetadataStore.touch(record, { title: `${history.taskTitle} · 调试` });
    const messages = await withInteractionReplies(history.record, history.state.messages ?? []);
    await sessionUiStateStore.write(
      toSessionUiStateContext(titled),
      buildForkedSessionUiState({ messages }, {
        keepMessageCount: messages.length,
        targetSessionId: titled.sessionId,
      }),
    );
    res.status(201).json({
      session_id: titled.sessionId,
      title: titled.title,
      purpose: readPurpose(titled),
      task_id: titled.taskId,
      parent_session_id: titled.parentSessionId,
      created: true,
      status: titled.status,
    });
  });

  chatRouter.post("/task-executions", async (req, res) => {
    const parsed = taskExecutionPayloadSchema.safeParse(req.body);
    if (!parsed.success) {
      res.status(400).json({ message: "invalid request payload", detail: parsed.error.flatten() });
      return;
    }
    if (parsed.data.agent_instance_id !== config.AGENT_INSTANCE_ID) {
      res.status(409).json({ message: "task execution targeted a different Agent instance" });
      return;
    }
    const credential = getTrustedProductCredential(req);
    if (!credential) {
      res.status(401).json({ message: "task execution requires product credentials" });
      return;
    }
    const authContext = getLocalAgentContext(req);
    const metadataContext = {
      actorKey: authContext.actorKey,
      projectId: authContext.projectId,
      projectKey: authContext.projectKey,
      userId: authContext.userId,
    };
    const existing = await sessionMetadataStore.findByRequest(metadataContext, parsed.data.execution_request_id);
    if (existing) {
      if (
        existing.taskId !== parsed.data.task_id
        || existing.runId !== parsed.data.run_id
        || readPurpose(existing) !== "task_execution"
      ) {
        res.status(409).json({ message: "request_id is already bound to another task" });
        return;
      }
      if (activeRuns.get(existing.sessionId)?.status === "running") {
        res.status(202).json({
          session_id: existing.sessionId,
          task_id: existing.taskId,
          request_id: existing.requestId,
          run_id: existing.runId,
          purpose: readPurpose(existing),
          created: false,
          status: "running",
          model: existing.modelId,
          agent_instance_id: config.AGENT_INSTANCE_ID,
        });
        return;
      }
      const taskModel = existing.modelId;
      if (!taskModel) {
        res.status(409).json({ message: "task execution session has no model binding" });
        return;
      }
      try {
        await prepareTaskExecution(existing, credential, parsed.data);
      } catch (error) {
        res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 409).json({
          message: error instanceof Error ? error.message : "task execution recovery failed",
        });
        return;
      }
      res.status(202).json({
        session_id: existing.sessionId,
        task_id: existing.taskId,
        request_id: existing.requestId,
        run_id: existing.runId,
        purpose: readPurpose(existing),
        created: false,
        status: "running",
        model: taskModel,
        agent_instance_id: config.AGENT_INSTANCE_ID,
      });
      startTaskExecution(existing, credential, taskModel, parsed.data.workflow);
      return;
    }
    let taskModel: string;
    try {
      taskModel = await runtime.resolveAvailableModel([
        config.OPERATIONAL_TASK_MODEL ?? config.OPENCODE_MODEL,
        ...agentModelOptions.map((option) => option.id),
      ]);
    } catch (error) {
      res.status(503).json({
        message: error instanceof Error ? error.message : "no operational task model is available",
      });
      return;
    }
    const runtimeSession = await runtime.createSession(undefined, metadataContext);
    const { record } = await sessionMetadataStore.ensure({
      ...metadataContext,
      sessionId: runtimeSession.id,
      purpose: "task_execution",
      taskId: parsed.data.task_id,
      runId: parsed.data.run_id,
      requestId: parsed.data.execution_request_id,
      modelId: taskModel,
    });
    try {
      await prepareTaskExecution(record, credential, parsed.data);
    } catch (error) {
      await sessionMetadataStore.remove(record);
      clearOperationalTaskLease(record.sessionId);
      res.status(error && typeof error === "object" && "status" in error && typeof error.status === "number" ? error.status : 502).json({
        message: error instanceof Error ? error.message : "task claim failed",
      });
      return;
    }
    res.status(202).json({
      session_id: record.sessionId,
      task_id: record.taskId,
      request_id: record.requestId,
      run_id: record.runId,
      purpose: readPurpose(record),
      created: true,
      status: "running",
      model: taskModel,
      agent_instance_id: config.AGENT_INSTANCE_ID,
    });
    startTaskExecution(record, credential, taskModel, parsed.data.workflow);
  });

  chatRouter.get("/sessions", async (req, res) => {
    const query = sessionListQuerySchema.safeParse(req.query);
    if (!query.success) {
      res.status(400).json({ message: "invalid session filter", detail: query.error.flatten() });
      return;
    }
    const authContext = getLocalAgentContext(req);
    const projectId = authContext.projectId;
    const userId = authContext.userId;
    const actorKey = authContext.actorKey;
    const projectKey = authContext.projectKey;
    const includeCollaboration = query.data.include_collaboration === "true";
    let records = await sessionMetadataStore.list({
      actorKey,
      projectId,
      projectKey,
      userId,
    }, {
      purpose: includeCollaboration ? undefined : query.data.purpose ?? "ordinary_chat",
      taskId: query.data.task_id,
    });
    if (includeCollaboration) records = records.filter((record) => ["ordinary_chat", "task_collaboration"].includes(readPurpose(record)));
    const topics = new Map((await sessionMetadataStore.topics.list(authContext)).map((topic) => [topic.id, topic]));
    const topicId = query.data.topic_id;
    if (topicId && topicId !== "unclassified" && !topics.has(topicId)) {
      res.status(404).json({ message: "主题不存在" }); return;
    }
    records = records.filter((record) => !topicId || (topicId === "unclassified"
      ? !record.topicId || !topics.has(record.topicId) : record.topicId === topicId));
    const searchable: Array<SessionRecord & { searchSnippet?: string }> = [];
    for (const record of records) {
      const titleMatches = (record.title ?? "新对话").toLocaleLowerCase().includes((query.data.q ?? "").toLocaleLowerCase());
      const state = query.data.q && !titleMatches ? await sessionUiStateStore.read(toSessionUiStateContext(record)) : null;
      const messages = activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? [];
      searchable.push({ ...record, searchSnippet: query.data.q && !titleMatches ? matchHistoryText(messages, query.data.q) : undefined });
    }
    let page;
    try {
      page = sessionHistoryPage(searchable, {
        scope: historyScope([actorKey, projectKey, userId, query.data.purpose ?? "ordinary_chat", query.data.task_id, topicId, includeCollaboration]),
        q: query.data.q, limit: query.data.limit ?? (query.data.q || query.data.cursor ? 30 : Math.max(records.length, 1)),
        cursor: query.data.cursor,
      });
    } catch (error) {
      if (!(error instanceof HistoryCursorError)) throw error;
      res.status(400).json({ message: error.message });
      return;
    }
    const credential = getTrustedProductCredential(req);
    const titleSignal = AbortSignal.timeout(2_000);
    const titleLimit = Math.min(page.sessions.length, 30);
    for (let offset = 0; offset < titleLimit && !titleSignal.aborted; offset += 4) {
      const batch = await Promise.all(page.sessions.slice(offset, Math.min(offset + 4, titleLimit)).map(async (record) => ({
        ...record, ...await refreshCollaborationTitle(record, credential, undefined, titleSignal),
      })));
      page.sessions.splice(offset, batch.length, ...batch);
    }
    res.json({
      has_more: page.has_more,
      next_cursor: page.next_cursor,
      sessions: page.sessions.map((record) => ({
        id: record.sessionId,
        title: record.title ?? "新对话",
        topic_id: record.topicId && topics.has(record.topicId) ? record.topicId : null,
        topic_name: record.topicId ? topics.get(record.topicId)?.name : undefined,
        search_snippet: record.searchSnippet,
        created_at: record.createdAt,
        updated_at: record.updatedAt,
        status: record.status,
        parent_session_id: record.parentSessionId,
        is_streaming: activeRuns.get(record.sessionId)?.status === "running",
        run_status: getSessionRunStatus(record.sessionId),
        purpose: readPurpose(record),
        task_id: record.taskId,
        trigger: record.triggerWorkflow === "water_quality_complaint" ? {
          subject_id: record.triggerSubjectId,
          workflow: record.triggerWorkflow,
          occurred_at: record.triggerOccurredAt,
        } : undefined,
      })),
    });
  });

  chatRouter.get("/session/:session_id/messages", async (req, res) => {
    const query = olderMessagesQuerySchema.safeParse(req.query);
    if (!query.success) { res.status(400).json({ message: "invalid message pagination" }); return; }
    const context = getLocalAgentContext(req);
    const record = await sessionMetadataStore.get(context, req.params.session_id);
    if (!record) { res.status(404).json({ message: "session not found" }); return; }
    const state = await sessionUiStateStore.read(toSessionUiStateContext(record));
    const messages = readPurpose(record) === "task_collaboration" ? state?.messages ?? []
      : activeRuns.get(record.sessionId)?.messages ?? state?.messages ?? [];
    try {
      const page = sessionMessagePage(messages, { scope: historyScope([record.actorKey, record.projectKey, record.sessionId]), ...query.data });
      res.json({ ...page, messages: await withInteractionReplies(record, page.messages) });
    } catch (error) {
      if (!(error instanceof HistoryCursorError)) throw error;
      res.status(400).json({ message: error.message });
    }
  });

  chatRouter.get("/session/:session_id", async (req, res) => {
    const query = messageQuerySchema.safeParse(req.query);
    if (!query.success) { res.status(400).json({ message: "invalid message pagination" }); return; }
    const sessionId = req.params.session_id?.trim();
    const authContext = getLocalAgentContext(req);
    const projectId = authContext.projectId;
    const userId = authContext.userId;
    const actorKey = authContext.actorKey;
    const projectKey = authContext.projectKey;
    if (!sessionId) {
      res.status(400).json({ message: "session_id is required" });
      return;
    }

    let sessionRecord = await sessionMetadataStore.get(
      {
        actorKey,
        projectId,
        projectKey,
        userId,
      },
      sessionId,
    );
    if (!sessionRecord) {
      res.status(404).json({ message: "session not found" });
      return;
    }

    sessionRecord = await refreshCollaborationTitle(sessionRecord, getTrustedProductCredential(req));

    const state = await sessionUiStateStore.read(
      toSessionUiStateContext(sessionRecord),
    );
    const messages = readPurpose(sessionRecord) === "task_collaboration" ? state?.messages ?? []
      : activeRuns.get(sessionRecord.sessionId)?.messages ?? state?.messages ?? [];
    const messagePayload = query.data.message_limit ? sessionMessagePage(messages, {
      scope: historyScope([sessionRecord.actorKey, sessionRecord.projectKey, sessionId]), limit: query.data.message_limit,
    }) : { messages };
    res.json({
      id: sessionRecord.sessionId,
      title: sessionRecord.title ?? "新对话",
      topic_id: sessionRecord.topicId && await sessionMetadataStore.topics.get(authContext, sessionRecord.topicId) ? sessionRecord.topicId : null,
      is_title_manually_edited: state?.isTitleManuallyEdited ?? false,
      created_at: sessionRecord.createdAt,
      updated_at: sessionRecord.updatedAt,
      status: sessionRecord.status,
      session_id: sessionRecord.sessionId,
      ...messagePayload,
      messages: await withInteractionReplies(sessionRecord, messagePayload.messages),
      execution_plans: state?.executionPlans ?? [],
      parent_session_id: sessionRecord.parentSessionId,
      is_streaming: activeRuns.get(sessionRecord.sessionId)?.status === "running",
      run_status: getSessionRunStatus(sessionRecord.sessionId),
      purpose: readPurpose(sessionRecord),
      task_id: sessionRecord.taskId,
    });
  });

  chatRouter.get("/session/:session_id/stream", async (req, res) => {
    const query = messageQuerySchema.safeParse(req.query);
    if (!query.success) { res.status(400).json({ message: "invalid message pagination" }); return; }
    const sessionId = req.params.session_id?.trim();
    const authContext = getLocalAgentContext(req);
    const projectId = authContext.projectId;
    const userId = authContext.userId;
    const actorKey = authContext.actorKey;
    const projectKey = authContext.projectKey;
    if (!sessionId) {
      res.status(400).json({ message: "session_id is required" });
      return;
    }

    const sessionRecord = await sessionMetadataStore.get(
      { actorKey, projectId, projectKey, userId },
      sessionId,
    );
    if (!sessionRecord) {
      res.status(404).json({ message: "session not found" });
      return;
    }

    res.status(200);
    res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
    res.setHeader("Cache-Control", "no-cache");
    res.setHeader("Connection", "keep-alive");
    res.setHeader("X-Accel-Buffering", "no");
    res.flushHeaders?.();

    const run = activeRuns.get(sessionRecord.sessionId);
    const state = await sessionUiStateStore.read(toSessionUiStateContext(sessionRecord));
    const messages = readPurpose(sessionRecord) === "task_collaboration" ? state?.messages ?? []
      : run?.messages ?? state?.messages ?? [];
    const messagePayload = query.data.message_limit ? sessionMessagePage(messages, {
      scope: historyScope([sessionRecord.actorKey, sessionRecord.projectKey, sessionId]), limit: query.data.message_limit,
    }) : { messages };
    res.write(
      toSse("state", {
        session_id: sessionRecord.sessionId,
        ...messagePayload,
        messages: await withInteractionReplies(sessionRecord, messagePayload.messages),
        execution_plans: run ? undefined : state?.executionPlans,
        is_streaming: run?.status === "running",
        run_status: getSessionRunStatus(sessionRecord.sessionId),
      }),
    );

    if (!run || run.status !== "running") {
      res.end();
      return;
    }

    const subscriber: StreamSubscriber = {
      write: (event, data) => {
        if (!res.writableEnded && !res.destroyed) {
          res.write(toSse(event, data));
        }
      },
      close: () => {
        if (!res.writableEnded && !res.destroyed) {
          res.end();
        }
      },
    };
    run.subscribers.add(subscriber);
    const pendingRefresh = credentialRefreshCoordinator.getPendingEvent(sessionRecord.sessionId);
    if (pendingRefresh) subscriber.write(pendingRefresh.type, {
      ...pendingRefresh, session_id: sessionRecord.sessionId, request_id: pendingRefresh.requestId,
    });

    const cleanup = () => {
      run.subscribers.delete(subscriber);
    };
    req.on("close", cleanup);
    res.on("close", cleanup);
  });

  chatRouter.patch("/session/:session_id/title", async (req, res) => {
    const sessionId = req.params.session_id?.trim();
    const title =
      typeof req.body?.title === "string" ? req.body.title.trim() : "";
    const isTitleManuallyEdited =
      typeof req.body?.is_title_manually_edited === "boolean"
        ? req.body.is_title_manually_edited
        : undefined;
    const authContext = getLocalAgentContext(req);
    const projectId = authContext.projectId;
    const userId = authContext.userId;
    const actorKey = authContext.actorKey;
    const projectKey = authContext.projectKey;
    if (!sessionId || !title) {
      res.status(400).json({ message: "session_id and title are required" });
      return;
    }
    const sessionRecord = await sessionMetadataStore.get(
      { actorKey, projectId, projectKey, userId },
      sessionId,
    );
    if (!sessionRecord) {
      res.status(404).json({ message: "session not found" });
      return;
    }
    const nextSessionRecord = await sessionMetadataStore.touch(sessionRecord, {
      title,
    });
    const state = await sessionUiStateStore.read(
      toSessionUiStateContext(nextSessionRecord),
    );
    if (state) {
      await sessionUiStateStore.write(
        toSessionUiStateContext(nextSessionRecord),
        {
          ...state,
          isTitleManuallyEdited:
            isTitleManuallyEdited ?? state.isTitleManuallyEdited,
        },
      );
    }
    res.json({
      id: nextSessionRecord.sessionId,
      title: nextSessionRecord.title,
      updated_at: nextSessionRecord.updatedAt,
    });
  });

  chatRouter.delete("/session/:session_id", async (req, res) => {
    try {
      const sessionId = req.params.session_id?.trim();
      const authContext = getLocalAgentContext(req);
      const projectId = authContext.projectId;
      const userId = authContext.userId;
      const actorKey = authContext.actorKey;
      const projectKey = authContext.projectKey;
      if (!sessionId) {
        res.status(400).json({ message: "session_id is required" });
        return;
      }
      const sessionRecord = await sessionMetadataStore.get(
        { actorKey, projectId, projectKey, userId },
        sessionId,
        { includeDeleted: true },
      );
      if (!sessionRecord) {
        res.status(204).end();
        return;
      }
      await sessionMetadataStore.remove(sessionRecord);
      frontendActionCoordinator.cancelSession(sessionRecord.sessionId);
      await learningOrchestrator.cancelSession(sessionRecord.sessionId);
      await sessionBridge.deleteSession({
        clientSessionId: sessionRecord.sessionId,
        sessionId: sessionRecord.sessionId,
        actorKey: sessionRecord.actorKey,
        projectKey: sessionRecord.projectKey,
      });
      await sessionMetadataStore.purge(sessionRecord);
      activeRuns.delete(sessionRecord.sessionId);
      lastRunStatuses.delete(sessionRecord.sessionId);
      res.status(204).end();
    } catch (error) {
      if (error instanceof SessionContextMismatchError) { res.status(404).json({ message: "session not found" }); return; }
      logger.warn({ err: error, sessionId: req.params.session_id }, "session cleanup incomplete");
      res.status(503).json({ code: "SESSION_CLEANUP_PENDING", message: "Session cleanup could not finish. Retry deleting this session." });
    }
  });

  chatRouter.post("/fork", async (req, res) => {
    const parsed = forkPayloadSchema.safeParse(req.body);
    if (!parsed.success) {
      res.status(400).json({
        message: "invalid request payload",
        detail: parsed.error.flatten(),
      });
      return;
    }

    try {
      const authContext = getLocalAgentContext(req);
      const projectId = authContext.projectId;
      const traceId = req.header("x-trace-id") ?? undefined;
      const userId = authContext.userId;

      const actorKey = authContext.actorKey;
      const projectKey = authContext.projectKey;
      const sourceSessionId = parsed.data.session_id?.trim();
      const sourceSessionRecord = sourceSessionId
        ? await sessionMetadataStore.get(
            {
              actorKey,
              projectId,
              projectKey,
              userId,
            },
            sourceSessionId,
          )
        : null;
      if (sourceSessionId && !sourceSessionRecord) {
        res.status(404).json({ message: "session not found" });
        return;
      }
      if (sourceSessionRecord && readPurpose(sourceSessionRecord) !== "ordinary_chat") {
        res.status(409).json({ message: "task sessions cannot be forked into ordinary chat history" });
        return;
      }
      const forkSession = await runtime.createSession(undefined, { actorKey, projectKey });
      const { record: targetSessionRecord } = await sessionMetadataStore.ensure({
        actorKey,
        parentSessionId: sourceSessionId,
        topicId: sourceSessionRecord?.topicId ?? undefined,
        projectId,
        projectKey,
        sessionId: forkSession.id,
        userId,
      });
      const nextSessionId = targetSessionRecord.sessionId;

      if (sourceSessionId) {
        await sessionTranscriptStore.cloneThread(
          {
            actorKey,
            clientSessionId: sourceSessionId,
            projectKey,
            sessionId: sourceSessionId,
          },
          {
            actorKey,
            clientSessionId: nextSessionId,
            projectKey,
            sessionId: nextSessionId,
          },
          parsed.data.keep_message_count,
        );
      }
      const sourceState = sourceSessionRecord
        ? await sessionUiStateStore.read(
            toSessionUiStateContext(sourceSessionRecord),
          )
        : null;
      const forkTitle = sourceSessionRecord?.title
        ? `${sourceSessionRecord.title} 副本`
        : "新对话副本";
      const titledTargetSessionRecord = await sessionMetadataStore.touch(
        targetSessionRecord,
        { title: forkTitle },
      );
      await sessionUiStateStore.write(
        toSessionUiStateContext(titledTargetSessionRecord),
        buildForkedSessionUiState(sourceState, {
          keepMessageCount: parsed.data.keep_message_count,
          targetSessionId: nextSessionId,
        }),
      );

      logger.info(
        {
          sourceSessionId: parsed.data.session_id,
          sessionId: nextSessionId,
          traceId,
          projectId,
          keepMessageCount: parsed.data.keep_message_count,
        },
        "forked chat session",
      );

      res.status(200).json({
        session_id: nextSessionId,
      });
    } catch (error) {
      const detail = error instanceof Error ? error.message : String(error);
      logger.error({ err: error }, "chat fork failed");
      res.status(500).json({
        message: "chat fork failed",
        detail,
      });
    }
  });
};