memind/mindspace-agent-runner.mjs

import crypto from 'node:crypto';
import fs from 'node:fs/promises';
import { Readable } from 'node:stream';
import { Agent, fetch as undiciFetch } from 'undici';
import { jsonrepair } from 'jsonrepair';
import { reconcileAgentSession } from './session-reconcile.mjs';
import { resolveSessionAccess } from './session-broker.mjs';
import { resolveBillingTokenState } from './billing-token-state.mjs';

const insecureDispatcher = new Agent({
  connect: { rejectUnauthorized: false },
});

const DEFAULT_TEXT_BYTES = 48 * 1024;

function isHttpsTarget(target) {
  return String(target).startsWith('https://');
}

function runnerError(message, code, details) {
  return Object.assign(new Error(message), { code, details });
}

function createUserMessage(text) {
  return {
    id: crypto.randomUUID(),
    role: 'user',
    created: Math.floor(Date.now() / 1000),
    content: [{ type: 'text', text }],
    metadata: { userVisible: true, agentVisible: true },
  };
}

function messageVisibleText(message) {
  return (message?.content ?? [])
    .filter((item) => item.type === 'text')
    .map((item) => item.text)
    .join('');
}

import { mergeMessageContent } from './message-stream.mjs';

function pushMessage(messages, incoming) {
  const last = messages[messages.length - 1];
  if (last?.id && incoming?.id && last.id === incoming.id) {
    return [
      ...messages.slice(0, -1),
      {
        ...last,
        content: mergeMessageContent(last.content, incoming.content),
      },
    ];
  }
  return [...messages, incoming];
}

function extractBalancedJsonObject(text, startIndex = 0) {
  let depth = 0;
  let inString = false;
  let escape = false;
  for (let i = startIndex; i < text.length; i += 1) {
    const ch = text[i];
    if (inString) {
      if (escape) {
        escape = false;
        continue;
      }
      if (ch === '\\') {
        escape = true;
        continue;
      }
      if (ch === '"') inString = false;
      continue;
    }
    if (ch === '"') {
      inString = true;
      continue;
    }
    if (ch === '{') depth += 1;
    else if (ch === '}') {
      depth -= 1;
      if (depth === 0) return text.slice(startIndex, i + 1);
    }
  }
  return null;
}

function collectJsonCandidates(text) {
  const source = String(text ?? '').trim();
  const candidates = [];
  const seen = new Set();
  const push = (value) => {
    const trimmed = String(value ?? '').trim();
    if (!trimmed || seen.has(trimmed)) return;
    seen.add(trimmed);
    candidates.push(trimmed);
  };

  for (const match of source.matchAll(/```(?:json)?\s*([\s\S]*?)```/gi)) {
    push(match[1]);
  }
  push(source);

  for (const candidate of [...candidates]) {
    for (let i = 0; i < candidate.length; i += 1) {
      if (candidate[i] !== '{') continue;
      const balanced = extractBalancedJsonObject(candidate, i);
      if (balanced) push(balanced);
    }
  }

  return candidates;
}

function parseJsonObject(text) {
  for (const candidate of collectJsonCandidates(text)) {
    for (const normalized of [candidate, jsonrepair(candidate)]) {
      try {
        const parsed = JSON.parse(normalized);
        if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
          return parsed;
        }
      } catch {
        // try next candidate
      }
    }
  }
  return null;
}

function extractJsonObject(text) {
  const source = String(text ?? '').trim();
  const parsed = parseJsonObject(source);
  if (parsed) return parsed;

  const start = source.indexOf('{');
  const end = source.lastIndexOf('}');
  if (start < 0 || end <= start) {
    throw runnerError('Agent 输出缺少 JSON 结果', 'invalid_agent_job_output');
  }

  const raw = extractBalancedJsonObject(source, start) ?? source.slice(start, end + 1);
  try {
    return JSON.parse(jsonrepair(raw));
  } catch (error) {
    throw runnerError('Agent 输出 JSON 解析失败', 'invalid_agent_job_output', {
      raw: raw.slice(0, 2000),
      cause: error instanceof Error ? error.message : String(error),
    });
  }
}

function normalizeStructuredResult(payload) {
  const title = String(payload?.title ?? '').trim();
  const summary = String(payload?.summary ?? '').trim();
  const content =
    String(payload?.content ?? payload?.markdown ?? payload?.body ?? '').trim();
  const contentFormat = String(payload?.content_format ?? payload?.contentFormat ?? 'markdown')
    .trim()
    .toLowerCase();
  if (!title || !content) {
    throw runnerError('Agent 输出缺少标题或正文', 'invalid_agent_job_output');
  }
  return {
    title,
    summary,
    content,
    contentFormat: contentFormat === 'html' ? 'html' : 'markdown',
    pageType: contentFormat === 'html' ? 'html' : 'article',
    templateId: contentFormat === 'html' ? 'static-html' : 'report',
  };
}

async function readAssetContext(asset, maxBytes = DEFAULT_TEXT_BYTES) {
  const mimeType = String(asset.mimeType ?? '');
  const textLike =
    mimeType.startsWith('text/') ||
    mimeType === 'application/json' ||
    mimeType.includes('xml') ||
    mimeType.includes('javascript');
  if (!textLike) {
    return {
      assetId: asset.assetId,
      displayName: asset.displayName,
      mimeType,
      excerpt: '',
      note: '该文件不是纯文本，Runner 当前不会直接内嵌二进制内容。',
    };
  }
  const buffer = await fs.readFile(asset.path);
  return {
    assetId: asset.assetId,
    displayName: asset.displayName,
    mimeType,
    excerpt: buffer.subarray(0, maxBytes).toString('utf8'),
    truncated: buffer.length > maxBytes,
    note: buffer.length > maxBytes ? `内容已截断到 ${maxBytes} 字节。` : undefined,
  };
}

export function buildAgentJobPrompt(job, assetContexts) {
  const assetSections = assetContexts
    .map((asset, index) => {
      const header = `资料 ${index + 1}: ${asset.displayName} (${asset.mimeType || 'unknown'})`;
      const note = asset.note ? `说明: ${asset.note}\n` : '';
      const body = asset.excerpt
        ? `内容:\n<<<ASSET_${index + 1}>>>\n${asset.excerpt}\n<<<END_ASSET_${index + 1}>>>`
        : '内容: [未内嵌文本内容]';
      return `${header}\n${note}${body}`;
    })
    .join('\n\n');

  return [
    '你正在为 MindSpace 生成一个页面草稿。',
    '你只能基于给定资料和用户任务生成结果，不能假设额外事实。',
    '不要请求工具确认，不要输出解释，不要调用工具。',
    '最终回复必须是一个合法 JSON 对象，且至少包含 title、summary、content 三个字段。',
    'content 可以是 Markdown 或 HTML；如需 HTML，请把 content_format 设为 "html"。',
    '示例：',
    '{"title":"页面标题","summary":"一句话摘要","content":"# 标题\\n\\n正文段落","content_format":"markdown"}',
    `用户任务: ${job.instruction}`,
    '',
    assetSections,
  ].join('\n');
}

async function readJsonResponse(response) {
  const text = await response.text();
  if (!response.ok) {
    throw new Error(text || `upstream ${response.status}`);
  }
  return text ? JSON.parse(text) : null;
}

async function defaultExecuteSessionReply(apiFetch, sessionId, requestId, prompt) {
  const eventsResponse = await apiFetch(`/sessions/${sessionId}/events`, {
    method: 'GET',
    headers: { Accept: 'text/event-stream' },
  });
  if (!eventsResponse.ok || !eventsResponse.body) {
    const text = await eventsResponse.text().catch(() => '');
    throw runnerError(text || '无法建立任务事件流', 'worker_unavailable');
  }

  const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, {
    method: 'POST',
    body: JSON.stringify({
      request_id: requestId,
      user_message: createUserMessage(prompt),
    }),
  });
  if (!replyResponse.ok) {
    const text = await replyResponse.text().catch(() => '');
    throw runnerError(text || 'Agent reply 请求失败', 'worker_unavailable');
  }
  replyResponse.body?.cancel().catch?.(() => {});

  const reader = Readable.fromWeb(eventsResponse.body);
  const decoder = new TextDecoder();
  let buffer = '';
  let messages = [];

  for await (const chunk of reader) {
    buffer += decoder.decode(chunk, { stream: true });
    const frames = buffer.split('\n\n');
    buffer = frames.pop() ?? '';
    for (const frame of frames) {
      let data = '';
      for (const line of frame.split('\n')) {
        if (line.startsWith('data:')) data += `${line.slice(5).trim()}`;
      }
      if (!data) continue;
      let event;
      try {
        event = JSON.parse(data);
      } catch {
        continue;
      }
      const routingId = event.chat_request_id ?? event.request_id;
      if (routingId && routingId !== requestId) continue;

      if (event.type === 'Message' && event.message?.metadata?.userVisible) {
        const hasActionRequired = event.message.content?.some((item) => item.type === 'actionRequired');
        if (hasActionRequired) {
          throw runnerError('任务执行需要人工确认，Runner 当前无法自动处理', 'worker_unavailable');
        }
        messages = pushMessage(messages, event.message);
      } else if (event.type === 'UpdateConversation') {
        messages = (event.conversation ?? []).filter((item) => item.metadata?.userVisible);
      } else if (event.type === 'Error') {
        throw runnerError(event.error || '任务执行失败', 'worker_unavailable');
      } else if (event.type === 'Finish') {
        const assistant = [...messages].reverse().find((item) => item.role === 'assistant');
        return {
          text: messageVisibleText(assistant),
          tokenState: event.token_state ?? null,
        };
      }
    }
  }

  throw runnerError('任务事件流提前结束', 'worker_unavailable');
}

export function createMindSpaceAgentRunner({
  apiTarget,
  apiSecret,
  userAuth,
  sessionAccess = null,
  agentJobService,
  experienceService = null,
  executeSessionReply = defaultExecuteSessionReply,
  apiFetchImpl = null,
}) {
  const sessionStore = resolveSessionAccess({ userAuth, sessionAccess });
  const apiFetch = apiFetchImpl ?? (async (pathname, init = {}) => {
    const url = new URL(pathname, apiTarget);
    const headers = {
      ...(init.headers ?? {}),
      'X-Secret-Key': apiSecret,
    };
    if (init.body && !headers['Content-Type']) {
      headers['Content-Type'] = 'application/json';
    }
    return undiciFetch(url, {
      ...init,
      headers,
      dispatcher: isHttpsTarget(apiTarget) ? insecureDispatcher : undefined,
    });
  });

  // preClaimed lets a polling worker that already atomically claimed the job
  // (via claimNextJob + SKIP LOCKED) hand the claim payload straight in, instead
  // of claiming a second time. The HTTP-triggered path calls runJob(jobId) with
  // no preClaimed and claims here as before.
  const runJob = async (jobId, preClaimed = null) => {
    let claim = null;
    let sessionId = null;
    try {
      claim = preClaimed ?? (await agentJobService.claimJob(jobId));
      const gate = await userAuth.canUseChat(claim.userId);
      if (!gate.ok) {
        throw runnerError(gate.message || '当前用户无法执行 Agent 任务', 'worker_unavailable');
      }

      const workingDir = await userAuth.resolveWorkingDir(claim.userId);
      const sessionPolicy = await userAuth.getAgentSessionPolicy(claim.userId);
      const publishLayout = await userAuth.getUserPublishLayout(claim.userId);
      const startSession = await readJsonResponse(
        await apiFetch('/agent/start', {
          method: 'POST',
          body: JSON.stringify({
            working_dir: workingDir,
            enable_context_memory: sessionPolicy.enableContextMemory,
            ...(sessionPolicy.extensionOverrides
              ? { extension_overrides: sessionPolicy.extensionOverrides }
              : {}),
          }),
        }),
      );
      sessionId = startSession?.id;
      if (!sessionId) {
        throw runnerError('Agent 会话启动失败', 'worker_unavailable');
      }
      await agentJobService.setSessionId(jobId, claim.jobToken, sessionId);
      await sessionStore.registerAgentSession(claim.userId, sessionId);
      await reconcileAgentSession(
        (pathname, init) => apiFetch(pathname, init),
        sessionId,
        {
          workingDir,
          sessionPolicy,
          sandboxConstraints: publishLayout?.constraints ?? null,
          userContext: publishLayout
            ? {
                userId: claim.userId,
                displayName: publishLayout.displayName,
                username: publishLayout.username,
                slug: publishLayout.slug,
              }
            : null,
        },
      );

      const assetContexts = [];
      for (const asset of claim.allowedAssets) {
        const localAsset = await agentJobService.getAssetForJob(jobId, claim.jobToken, asset.assetId);
        assetContexts.push(await readAssetContext(localAsset));
      }
      let prompt = buildAgentJobPrompt(claim, assetContexts);
      // Inject shared experience (etat C): relevant lessons all instances learned.
      // Best-effort — retrieval must never block or fail a job.
      const relevant = await retrieveExperience(claim.instruction);
      if (relevant) prompt = `${relevant}\n\n${prompt}`;
      const requestId = crypto.randomUUID();
      const reply = await executeSessionReply(apiFetch, sessionId, requestId, prompt);
      const parsed = normalizeStructuredResult(extractJsonObject(reply.text));

      if (reply.tokenState) {
        const tokenState = await resolveBillingTokenState(reply.tokenState, {
          sessionId,
          fetchSession: async (sid) => {
            const response = await apiFetch(`/sessions/${encodeURIComponent(sid)}`);
            if (!response?.ok) return null;
            return readJsonResponse(response);
          },
        });
        await userAuth.billSessionUsage(claim.userId, sessionId, tokenState, requestId);
      }

      const completed = await agentJobService.completeJob(jobId, claim.jobToken, {
        title: parsed.title,
        summary: parsed.summary,
        content: parsed.content,
        contentFormat: parsed.contentFormat,
        pageType: parsed.pageType,
        templateId: parsed.templateId,
        sourceAssetIds: claim.allowedAssets.map((asset) => asset.assetId),
      });
      // Record the outcome as shared experience so other instances benefit.
      await recordExperience(claim, sessionId, parsed);
      return completed;
    } catch (error) {
      if (claim?.jobToken) {
        await agentJobService
          .completeJob(jobId, claim.jobToken, {
            status: 'failed',
            errorCode: error?.code ?? 'worker_unavailable',
            errorMessage: error instanceof Error ? error.message : String(error),
          })
          .catch(() => {});
      }
      throw error;
    }
  };

  // Returns a prompt-ready block of relevant prior experience, or '' if none /
  // disabled / on any error. Never throws.
  async function retrieveExperience(instruction) {
    if (!experienceService || !instruction) return '';
    try {
      const hits = await experienceService.search(instruction, { limit: 3 });
      if (!hits.length) return '';
      const lines = hits.map((h) => `- ${h.title}: ${h.body}`).join('\n');
      return `# 相关经验（供参考，来自历史任务）\n${lines}`;
    } catch {
      return '';
    }
  }

  // Persists a one-line lesson from a completed job. Best-effort; never throws.
  async function recordExperience(claim, sessionId, parsed) {
    if (!experienceService) return;
    try {
      const title = String(parsed?.title ?? claim.instruction ?? '').slice(0, 200);
      const summary = String(parsed?.summary ?? '').trim();
      if (!title || !summary) return;
      await experienceService.record({
        kind: 'task_outcome',
        title,
        body: summary,
        sourceSessionId: sessionId,
        sourceUserId: claim.userId,
      });
    } catch {
      // swallow — experience recording is non-critical
    }
  }

  return { runJob };
}

export const agentRunnerInternals = {
  extractJsonObject,
  extractBalancedJsonObject,
  collectJsonCandidates,
  normalizeStructuredResult,
  buildAgentJobPrompt,
  readAssetContext,
};