import { serve } from "https://deno.land/std@0.168.0/http/server.ts";
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2';
import { getServiceClient } from '../_shared/supabase-clients.ts';
import { callAi } from '../_shared/ai-call.ts';
import { scoreSkillsByIntent, loadRecentUsageCounts } from '../_shared/pilot/intent-scorer.ts';
import { SKILL_CATEGORY_MODULES, isCategoryActive, loadActiveModuleIds } from '../_shared/mcp/groups.ts';
import {
  resolveAiConfig,
  loadWorkspaceFiles,
  buildWorkspacePrompt,
  loadMemories,
  loadObjectives,
  buildSystemPrompt,
  pruneConversationHistory,
  fetchSkillInstructions,
  loadSkillTools,
  loadSkillsRaw,
  getBuiltInTools,
  executeBuiltInTool,
  isBuiltInTool,
  loadCMSSchema,
  tryAcquireLock,
  releaseLock,
  parseReplyDirectives,
} from "../_shared/agent-reason.ts";

/**
 * FlowPilot Operate — Interactive streaming agent
 * Thin SSE wrapper around the shared agent-reason core + prompt compiler.
 *
 * OpenClaw alignment: NO deterministic intent guards.
 * Skill descriptions with "Use when / NOT for" patterns handle all routing.
 */

const corsHeaders = {
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Headers': 'authorization, x-client-info, apikey, content-type, x-supabase-client-platform, x-supabase-client-platform-version, x-supabase-client-runtime, x-supabase-client-runtime-version',
};

const MAX_TOOL_ITERATIONS = 10;
const MAX_OPERATE_WALL_CLOCK_MS = 120_000; // 120s wall-clock for interactive operate (longer chains)

/**
 * FlowChat Operator Protocol — appended to the system prompt for FlowChat sessions.
 *
 * FlowChat is REACTIVE: it executes what the admin asks, here and now. No heartbeats,
 * no objectives, no proactive memory writes. But within a turn it MUST be a potent
 * operator — drafting content inline, chaining several skills, never bouncing the
 * task back to the admin for text it can write itself.
 *
 * See mem://architecture/flowchat-vs-flowpilot-roles
 *     mem://philosophy/flowpilot-development-laws (Law 3 — utilities are OK inline,
 *     skills are interfaces, not pipelines)
 */
const FLOWCHAT_OPERATOR_PROTOCOL = `
=== FLOWCHAT OPERATOR MODE ===
You are FlowChat — the admin's reactive operator. Same skills as FlowPilot, but you
act on this user's message, not on objectives or schedules. Be a potent doer.

OPERATING PRINCIPLES:
1. EXECUTE, DON'T DEFER. If the admin's request implies a content artifact (KB
   article, blog post, page, product description, email, contract clause, job ad,
   announcement), DRAFT THE CONTENT YOURSELF inside your reasoning, then call the
   matching manage_* / write_* / create_* skill with the drafted text in the same turn.
   Do NOT ask the admin "please provide the title/answer/body" — they already gave you
   the topic. Drafting from a topic is a pure text utility (Law 3 refinement).

2. ALWAYS POPULATE REQUIRED FIELDS. When a skill requires title + question + answer
   (KB), or title + content (blog), or name + description (product) — generate every
   required field from the topic. Never call a skill with only \`action\` set.

3. CHAIN AGGRESSIVELY. You have up to ${MAX_TOOL_ITERATIONS} tool iterations per turn.
   Use them: research → draft → save → publish → notify is one turn, not five.
   Call multiple independent tools in parallel.

4. NO AUTONOMY THEATRE. Do NOT say "I will check on this later", "I've added this to
   my objectives", or "I'll monitor for changes". You are reactive — that's FlowPilot's
   job, and it may be off. Finish the task in this turn or report what's missing.

5. ASK ONLY FOR INPUTS YOU CANNOT SYNTHESIZE. URLs, IDs, prices, dates, customer
   names, file uploads — yes, ask. Body copy, summaries, descriptions, taglines — no,
   write them.

6. REPORT WHAT YOU DID. End with a short summary: what was created/updated, IDs or
   slugs/URLs, and the single best next action the admin might want to take.

7. NEVER HALLUCINATE SUCCESS. You may ONLY claim that something was created /
   updated / sent / published / added / saved if the corresponding tool call in
   THIS turn returned status="success" (no "error" field). If a tool failed,
   say it failed and what the error was. If you wanted to do X but no available
   skill matches X, say "I don't have a skill for that — the relevant module
   may be disabled" instead of substituting a different skill and pretending
   it did the job. Wrong-skill substitution is the worst possible outcome.
=== END FLOWCHAT OPERATOR MODE ===
`;

function sseEvent(writer: WritableStreamDefaultWriter, encoder: TextEncoder, event: string, data: any) {
  const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
  return writer.write(encoder.encode(payload));
}

serve(async (req) => {
  if (req.method === 'OPTIONS') {
    return new Response(null, { headers: corsHeaders });
  }

  try {
    const body = await req.json();
    const messages = Array.isArray(body.messages) ? body.messages : [];
    const conversation_id = body.conversation_id;

    const supabaseUrl = Deno.env.get('SUPABASE_URL')!;
    const serviceKey = Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!;
    const supabase = getServiceClient();

    // Separation of concerns:
    //   FlowChat (this endpoint, as web-CLI shell) ALWAYS works — it's the human shell over the
    //   FlowWink platform skills, just like MCP is the agent shell. It does not require the
    //   FlowPilot autonomous-operator module to be enabled.
    //   FlowPilot, when ON, layers autonomous behaviors on top: objectives, soul-driven
    //   reasoning, reflection, planning, and automation execution. When OFF, those FlowPilot-
    //   internal built-in tools are stripped, but skills + reasoning remain available.
    // See mem://architecture/flowwink-as-business-os-three-shells
    //     mem://architecture/mcp-as-platform-not-flowpilot-feature
    const { data: moduleSettings } = await supabase
      .from('site_settings')
      .select('value')
      .eq('key', 'modules')
      .maybeSingle();

    const flowpilotEnabled = moduleSettings?.value?.flowpilot?.enabled ?? false;

    // Concurrency guard — one agent run per conversation
    const lane = conversation_id ? `operate:${conversation_id}` : null;
    if (lane) {
      const acquired = await tryAcquireLock(supabase, lane, 'operate', 300);
      if (!acquired) {
        return new Response(
          JSON.stringify({ error: 'Another agent process is running on this conversation' }),
          { status: 429, headers: { ...corsHeaders, 'Content-Type': 'application/json' } }
        );
      }
    }

    const { apiKey, apiUrl, model } = await resolveAiConfig(supabase, 'reasoning');

    // Load context in parallel
    const [{ soul, identity, agents, tools, user, bootstrap }, memoryContext, objectiveContext, cmsSchemaCtx] = await Promise.all([
      loadWorkspaceFiles(supabase),
      loadMemories(supabase),
      loadObjectives(supabase),
      loadCMSSchema(supabase),
    ]);

    // Use prompt compiler (OpenClaw Layer 1)
    const baseSystemPrompt = buildSystemPrompt({
      mode: 'operate',
      soulPrompt: buildWorkspacePrompt(soul, identity, agents, tools, user, bootstrap),
      agents,
      memoryContext,
      objectiveContext,
      cmsSchemaContext: cmsSchemaCtx,
    });

    // Append the FlowChat operator protocol — turns the generic operate-mode prompt
    // into a "potent reactive doer" prompt without affecting heartbeat/autonomous flows.
    const systemPrompt = `${baseSystemPrompt}\n${FLOWCHAT_OPERATOR_PROTOCOL}`;

    // Build tools — server-side loading with gating + caching (OpenClaw alignment)
    // Platform-level built-ins (always on for the FlowChat shell): memory (read), workflows, a2a, skill-packs.
    // FlowPilot-internal built-ins (autonomous-operator only): objectives, self-mod, reflect, soul, planning, automations-exec.
    const builtInGroups = flowpilotEnabled
      ? ['memory', 'objectives', 'self-mod', 'reflect', 'soul', 'planning', 'automations-exec', 'workflows', 'a2a', 'skill-packs']
      : ['memory', 'workflows', 'a2a', 'skill-packs'];
    const builtInTools = getBuiltInTools(builtInGroups);
    const skillCache = await loadSkillsRaw(supabase, 'internal');
    const [externalSkills, usageBoost] = await Promise.all([
      loadSkillTools(supabase, 'internal', undefined, 'full', skillCache),
      loadRecentUsageCounts(supabase),
    ]);

    // Intent-based adaptive tool window — wider for FlowChat operator (broad admin tasks).
    const lastUserMsg = [...messages].reverse().find((m: any) => m.role === 'user')?.content || '';
    const MAX_TOOLS = 128;
    const maxSkills = MAX_TOOLS - builtInTools.length;
    const filteredSkills = scoreSkillsByIntent(externalSkills, lastUserMsg, {
      maxSkills: Math.min(maxSkills, 40),
      usageBoost,
    });
    
    const allTools = [...builtInTools, ...filteredSkills];

    // ─── Module-disabled intent detection ──────────────────────────────────
    // Same scoring algorithm, but run against skills that WERE filtered out
    // by `loadActiveModuleIds` (modules turned off in /admin/modules). If the
    // user's intent matches a disabled skill more strongly than any exposed
    // skill, we tell them honestly instead of letting the LLM substitute the
    // wrong skill from an active module. See plan: FlowChat module-aware
    // honesty.
    let disabledMatch: { module: string; skills: string[]; topScore: number } | null = null;
    let exposedSkillCount = filteredSkills.length;
    let disabledSkillCount = 0;
    let modulesOffCount = 0;
    try {
      const activeModules = await loadActiveModuleIds(supabase);
      if (!activeModules.has('__all__')) {
        const { data: allEnabled } = await supabase
          .from('agent_skills')
          .select('name, tool_definition, category')
          .eq('enabled', true)
          .in('scope', ['internal', 'both']);

        const exposedNames = new Set(filteredSkills.map((t: any) => t.function?.name));
        const disabled = (allEnabled || []).filter((s: any) =>
          s.tool_definition?.function &&
          !exposedNames.has(s.tool_definition.function.name) &&
          !isCategoryActive(s.category, activeModules, SKILL_CATEGORY_MODULES)
        );
        disabledSkillCount = disabled.length;

        // Module-off count for honesty badge
        const { data: settingsRow } = await supabase
          .from('site_settings').select('value').eq('key', 'modules').maybeSingle();
        if (settingsRow?.value && typeof settingsRow.value === 'object') {
          modulesOffCount = Object.values(settingsRow.value as Record<string, any>)
            .filter((m: any) => !(m && m.enabled === true)).length;
        }

        if (disabled.length > 0 && lastUserMsg.trim().length > 0) {
          // Score disabled skills with the same scorer
          const disabledTools = disabled
            .map((s: any) => s.tool_definition)
            .filter((td: any) => td?.function);
          // Force scoring (>maxSkills) by using a small cap
          const scored = scoreSkillsByIntent(disabledTools, lastUserMsg, { maxSkills: 1 });

          // Manually re-score top exposed for comparison (use same function with cap=1)
          const exposedTopArr = filteredSkills.length > 1
            ? scoreSkillsByIntent(filteredSkills, lastUserMsg, { maxSkills: 1 })
            : filteredSkills;

          // Compute raw scores via direct expansion check (re-derive by name match — quick approximation)
          const computeMatchScore = (td: any) => {
            const fn = (td?.function?.name || '').toLowerCase();
            const desc = (td?.function?.description || '').toLowerCase();
            const msg = lastUserMsg.toLowerCase();
            let score = 0;
            for (const part of fn.split('_').filter((w: string) => w.length > 2)) {
              if (msg.includes(part)) score += 8;
            }
            const useWhen = desc.match(/use when:\s*([^.]*?)(?:\.|not for:|$)/i);
            if (useWhen) {
              for (const w of useWhen[1].toLowerCase().split(/[\s,]+/).filter((w: string) => w.length > 3)) {
                if (msg.includes(w)) score += 7;
              }
            }
            return score;
          };

          const topDisabledScore = scored[0] ? computeMatchScore(scored[0]) : 0;
          const topExposedScore = exposedTopArr[0] ? computeMatchScore(exposedTopArr[0]) : 0;

          // Trigger early-return if disabled skill clearly wins
          if (topDisabledScore >= 14 && topDisabledScore > topExposedScore + 4) {
            // Group top-3 matched disabled skills by their owning module
            const matchedSkills = scored
              .map((td: any) => disabled.find((d: any) => d.tool_definition?.function?.name === td.function?.name))
              .filter(Boolean)
              .slice(0, 3);

            // Find which module(s) need to be turned on for the top match
            const topCat = matchedSkills[0]?.category;
            const owningModules = SKILL_CATEGORY_MODULES[topCat] || [];
            const offModule = owningModules.find((m: string) => !activeModules.has(m)) || topCat || 'unknown';

            disabledMatch = {
              module: offModule,
              skills: matchedSkills.map((s: any) => s.name),
              topScore: topDisabledScore,
            };
            console.log(`[operate] Module-disabled intent detected: top disabled "${disabledMatch.skills[0]}" (score ${topDisabledScore}) > top exposed (score ${topExposedScore}). Suggesting module "${offModule}".`);
          }
        }
      } else {
        exposedSkillCount = filteredSkills.length;
      }
    } catch (e) {
      console.warn('[operate] Module-disabled detection failed (non-fatal):', (e as any)?.message);
    }

    // Set up SSE stream
    const { readable, writable } = new TransformStream();
    const writer = writable.getWriter();
    const encoder = new TextEncoder();

    const response = new Response(readable, {
      headers: {
        ...corsHeaders,
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
      },
    });

    (async () => {
      const operateTimeout = setTimeout(async () => {
        console.warn(`[operate] Wall-clock timeout (${MAX_OPERATE_WALL_CLOCK_MS}ms) — closing stream`);
        try {
          await sseEvent(writer, encoder, 'error', { message: 'Operation timed out. Partial results may have been returned.' });
          await sseEvent(writer, encoder, 'done', {});
        } catch { /* writer closed */ }
        try { await writer.close(); } catch { /* already closed */ }
      }, MAX_OPERATE_WALL_CLOCK_MS);

      // SSE keepalive to prevent gateway idle-timeout (sends comment every 10s)
      const keepalive = setInterval(async () => {
        try { await writer.write(encoder.encode(': keepalive\n\n')); } catch { clearInterval(keepalive); }
      }, 10_000);

      // Honesty meta — tells the UI how many skills are actually exposed vs hidden by disabled modules.
      await sseEvent(writer, encoder, 'flowchat_meta', {
        exposed_skill_count: exposedSkillCount,
        disabled_skill_count: disabledSkillCount,
        modules_off_count: modulesOffCount,
      });

      // Module-disabled early return: if the user clearly wants something only a
      // disabled module's skill can do, refuse honestly instead of letting the
      // LLM substitute the wrong skill. No tool calls, no hallucinated success.
      if (disabledMatch) {
        const skillsList = disabledMatch.skills.map(s => `\`${s}\``).join(', ');
        const reply =
          `I can't do that right now — it needs the **${disabledMatch.module}** module, which is currently disabled.\n\n` +
          `Skills that would handle this: ${skillsList}.\n\n` +
          `Enable the module in [Modules](/admin/modules), then ask me again.`;
        await sseEvent(writer, encoder, 'delta', { content: reply });
        await sseEvent(writer, encoder, 'done', {});
        clearInterval(keepalive);
        clearTimeout(operateTimeout);
        try { await writer.close(); } catch { /* already closed */ }
        if (lane) await releaseLock(supabase, lane).catch(() => {});
        return;
      }

      try {
        // Apply context pruning before starting
        let conversationMessages: any[] = await pruneConversationHistory(
          [{ role: 'system', content: systemPrompt }, ...messages],
          supabase,
        );

        const allSkillResults: any[] = [];
        const loadedInstructions = new Set<string>();
        let producedFinalResponse = false;

        for (let iteration = 0; iteration < MAX_TOOL_ITERATIONS; iteration++) {
          const t0 = Date.now();

          // Retry logic for transient network errors (TLS handshake, connection reset)
          let aiResponse: Response | null = null;
          const MAX_RETRIES = 3;
          for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
            try {
              aiResponse = await callAi({
                apiKey, apiUrl, model,
                messages: conversationMessages,
                tools: allTools.length > 0 ? allTools : undefined,
                tool_choice: allTools.length > 0 ? 'auto' : undefined,
              });
              break; // Success — exit retry loop
            } catch (fetchErr: any) {
              const isTransient = /tls handshake|connection reset|eof|ECONNRESET|socket hang up/i.test(fetchErr?.message || '');
              console.warn(`[operate] Fetch attempt ${attempt}/${MAX_RETRIES} failed: ${fetchErr?.message}`);
              if (!isTransient || attempt === MAX_RETRIES) {
                console.error('agent-operate stream error:', fetchErr);
                await sseEvent(writer, encoder, 'error', { message: `Network error connecting to AI provider. Please try again.` });
                await sseEvent(writer, encoder, 'done', {});
                clearInterval(keepalive);
                clearTimeout(operateTimeout);
                try { await writer.close(); } catch { /* already closed */ }
                return;
              }
              // Wait before retry: 1s, 2s
              await new Promise(r => setTimeout(r, attempt * 1000));
            }
          }

          console.log(`[operate] AI response in ${Date.now() - t0}ms (iteration ${iteration + 1}, tools=${allTools.length})`);

          if (!aiResponse!.ok) {
            const errText = await aiResponse!.text();
            console.error('AI error:', aiResponse!.status, errText);
            let errorDetail = 'AI provider error';
            try {
              const parsed = JSON.parse(errText);
              errorDetail = parsed?.error?.message || parsed?.message || `AI error ${aiResponse!.status}`;
            } catch { errorDetail = `AI error ${aiResponse!.status}: ${errText.slice(0, 200)}`; }
            await sseEvent(writer, encoder, 'error', { message: errorDetail });
            break;
          }

          const aiData = await aiResponse!.json();
          const choice = aiData.choices?.[0];

          if (!choice) {
            await sseEvent(writer, encoder, 'error', { message: 'No response from AI' });
            break;
          }

          const assistantMessage = choice.message;

          if (!assistantMessage.tool_calls?.length) {
            if (allSkillResults.length > 0) {
              await sseEvent(writer, encoder, 'skill_results', allSkillResults);
            }
            const finalContent = assistantMessage.content || 'Done.';
            const { cleanContent } = parseReplyDirectives(finalContent);
            await streamFinalResponse(apiUrl, apiKey, model, conversationMessages, writer, encoder, cleanContent || finalContent);
            producedFinalResponse = true;
            break;
          }

          conversationMessages.push(assistantMessage);

          const toolNames = assistantMessage.tool_calls.map((tc: any) => tc.function.name);
          await sseEvent(writer, encoder, 'tool_start', { iteration: iteration + 1, tools: toolNames });

          const toolResults = await Promise.all(
            assistantMessage.tool_calls.map(async (tc: any) => {
              const fnName = tc.function.name;
              let fnArgs: any;
              try { fnArgs = JSON.parse(tc.function.arguments || '{}'); } catch { fnArgs = {}; }

              let result: any;
              try {
                result = await executeBuiltInTool(supabase, supabaseUrl, serviceKey, fnName, fnArgs);
              } catch (err: any) {
                result = { error: err.message };
              }

              // Self-correction: if skill not found, suggest similar tools
              const resultStr = JSON.stringify(result || '');
              if (resultStr.includes('Skill not found') && !isBuiltInTool(fnName)) {
                const similarTools = allTools
                  .map((t: any) => t.function?.name)
                  .filter((n: string) => {
                    if (!n) return false;
                    const words = fnName.toLowerCase().replace(/_/g, ' ').split(' ').filter((w: string) => w.length > 2);
                    return words.some((w: string) => n.toLowerCase().includes(w));
                  })
                  .slice(0, 5);
                if (similarTools.length > 0) {
                  result = { error: `Tool "${fnName}" does not exist. Did you mean one of: ${similarTools.join(', ')}? Use the exact tool name from your available tools.` };
                }
              }

              if (!isBuiltInTool(fnName)) {
                allSkillResults.push({ skill: fnName, status: result?.error ? 'failed' : (result?.status || 'success'), result: result?.result || result });
              }

              return { role: 'tool' as const, tool_call_id: tc.id, content: JSON.stringify(result) };
            })
          );

          conversationMessages.push(...toolResults);

          // Deterministic next-step execution for explicit _next_action hints.
          // Keeps the OpenClaw atomic-tool model, but removes LLM flakiness in
          // critical multi-step flows like migrate_url -> manage_page.
          let autoChainSummary: string | null = null;
          for (const tr of toolResults) {
            try {
              const parsed = JSON.parse(tr.content);
              const result = parsed?.result || parsed;
              const nextAction = result?._next_action;
              if (!nextAction?.tool) continue;

              const nextArgs = JSON.parse(JSON.stringify(nextAction.args || {}));

              if (nextArgs.blocks === '{{blocks}}' && Array.isArray(result?.blocks)) {
                nextArgs.blocks = result.blocks;
              }
              if (nextArgs.page_id === '{{page_id}}' && (result?.page_id || result?.id)) {
                nextArgs.page_id = result.page_id || result.id;
              }
              if (!nextArgs.title && result?.title) {
                nextArgs.title = result.title;
              }
              if (!nextArgs.slug && result?.slug) {
                nextArgs.slug = result.slug;
              }

              console.log(`[operate] Auto-chaining mandatory next step: ${nextAction.tool}`);

              let chainedResult: any;
              try {
                chainedResult = await executeBuiltInTool(supabase, supabaseUrl, serviceKey, nextAction.tool, nextArgs);
              } catch (err: any) {
                chainedResult = { error: err.message || `Auto-chain failed for ${nextAction.tool}` };
              }

              if (!isBuiltInTool(nextAction.tool)) {
                allSkillResults.push({
                  skill: nextAction.tool,
                  status: chainedResult?.error ? 'failed' : (chainedResult?.status || 'success'),
                  result: chainedResult?.result || chainedResult,
                });
              }

              autoChainSummary = [
                `AUTOMATIC TOOL CHAIN COMPLETE: ${nextAction.tool} was executed immediately after the previous tool result.`,
                `Arguments used: ${JSON.stringify(nextArgs)}`,
                `Result: ${JSON.stringify(chainedResult)}`,
                nextAction.tool === 'manage_page'
                  ? 'IMPORTANT: The page now already contains the migrated blocks. Do NOT call create_page_block or manage_page_blocks to recreate the same imported blocks again unless you are making an explicit adjustment.'
                  : 'Continue only if another tool is still genuinely required.',
              ].join('\n');
              break;
            } catch {
              // Ignore malformed tool results and continue
            }
          }

          if (autoChainSummary) {
            conversationMessages.push({ role: 'system', content: autoChainSummary });
          }

          // Lazy instruction loading (OpenClaw Law 3)
          const calledSkillNames = assistantMessage.tool_calls
            .map((tc: any) => tc.function.name)
            .filter((n: string) => !isBuiltInTool(n));
          if (calledSkillNames.length > 0) {
            const instrContext = await fetchSkillInstructions(supabase, calledSkillNames, loadedInstructions);
            if (instrContext) {
              conversationMessages.push({ role: 'system', content: instrContext });
            }
          }

          // Continuation nudge — prevent AI from describing next steps instead of calling tools
          // Enhanced: detect _next_action hints from tool results for targeted chaining
          if (iteration < MAX_TOOL_ITERATIONS - 1) {
            // Check if any tool result contains a _next_action chaining hint
            let chainingDirective = '';
            for (const tr of toolResults) {
              try {
                const parsed = JSON.parse(tr.content);
                const result = parsed?.result || parsed;
                if (result?._next_action?.instruction) {
                  chainingDirective = result._next_action.instruction;
                  break;
                }
              } catch { /* ignore parse errors */ }
            }

            conversationMessages.push({
              role: 'system',
              content: chainingDirective
                ? `MANDATORY NEXT STEP: ${chainingDirective}`
                : 'IMPORTANT: If your task requires more steps, call the next tool NOW. Do NOT describe what you will do — just do it by calling the tool. Only respond with text when ALL tool calls are complete.',
            });
          }

          await sseEvent(writer, encoder, 'tool_done', { iteration: iteration + 1, tools: toolNames, results_count: toolResults.length });
        }

        // Safety fallback: if we only executed tools and never emitted a final assistant reply,
        // force a concise summary instead of closing silently.
        if (!producedFinalResponse) {
          if (allSkillResults.length > 0) {
            await sseEvent(writer, encoder, 'skill_results', allSkillResults);
          }
          const forcedConversation = [
            ...conversationMessages,
            {
              role: 'system',
              content: 'Stop calling tools now. Summarize what was completed, what failed, and the next best step in max 6 lines.',
            },
          ];
          await streamFinalResponse(
            apiUrl,
            apiKey,
            model,
            forcedConversation,
            writer,
            encoder,
            'I completed several steps but reached the iteration limit. Please ask me to continue from this point if needed.',
          );
        }
      } catch (err: any) {
        console.error('agent-operate stream error:', err);
        try { await sseEvent(writer, encoder, 'error', { message: err.message || 'Internal error' }); } catch { /* writer closed */ }
      } finally {
        clearTimeout(operateTimeout);
        clearInterval(keepalive);
        if (lane) {
          try { await releaseLock(supabase, lane); } catch { /* best effort */ }
        }
        try { await writer.close(); } catch { /* already closed */ }
      }
    })();

    return response;

  } catch (err: any) {
    console.error('agent-operate error:', err);
    return new Response(JSON.stringify({ error: err.message || 'Internal error' }), {
      status: 500, headers: { ...corsHeaders, 'Content-Type': 'application/json' },
    });
  }
});

// ─── Stream final AI response token-by-token ─────────────────────────────────

async function streamFinalResponse(
  apiUrl: string,
  apiKey: string,
  model: string,
  conversationMessages: any[],
  writer: WritableStreamDefaultWriter,
  encoder: TextEncoder,
  fallbackContent: string,
) {
  try {
    // Anthropic doesn't use OpenAI-compatible streaming — use fallback for Anthropic
    const isAnthropic = apiUrl.includes('anthropic.com');
    if (isAnthropic) {
      await sseEvent(writer, encoder, 'delta', { content: fallbackContent });
      await sseEvent(writer, encoder, 'done', {});
      return;
    }

    const streamResponse = await fetch(apiUrl, {
      method: 'POST',
      headers: { 'Authorization': `Bearer ${apiKey}`, 'Content-Type': 'application/json' },
      body: JSON.stringify({ model, messages: conversationMessages, stream: true }),
    });

    if (!streamResponse.ok || !streamResponse.body) {
      await sseEvent(writer, encoder, 'delta', { content: fallbackContent });
      await sseEvent(writer, encoder, 'done', {});
      return;
    }

    const reader = streamResponse.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      buffer += decoder.decode(value, { stream: true });

      let newlineIndex: number;
      while ((newlineIndex = buffer.indexOf('\n')) !== -1) {
        let line = buffer.slice(0, newlineIndex);
        buffer = buffer.slice(newlineIndex + 1);

        if (line.endsWith('\r')) line = line.slice(0, -1);
        if (!line.startsWith('data: ')) continue;

        const jsonStr = line.slice(6).trim();
        if (jsonStr === '[DONE]') {
          await sseEvent(writer, encoder, 'done', {});
          return;
        }

        try {
          const parsed = JSON.parse(jsonStr);
          const content = parsed.choices?.[0]?.delta?.content;
          if (content) {
            await sseEvent(writer, encoder, 'delta', { content });
          }
        } catch {
          buffer = line + '\n' + buffer;
          break;
        }
      }
    }

    // Flush remaining
    if (buffer.trim()) {
      for (let raw of buffer.split('\n')) {
        if (!raw) continue;
        if (raw.endsWith('\r')) raw = raw.slice(0, -1);
        if (!raw.startsWith('data: ')) continue;
        const jsonStr = raw.slice(6).trim();
        if (jsonStr === '[DONE]') continue;
        try {
          const parsed = JSON.parse(jsonStr);
          const content = parsed.choices?.[0]?.delta?.content;
          if (content) await sseEvent(writer, encoder, 'delta', { content });
        } catch { /* ignore */ }
      }
    }

    await sseEvent(writer, encoder, 'done', {});
  } catch (err) {
    console.error('Stream error, falling back:', err);
    await sseEvent(writer, encoder, 'delta', { content: fallbackContent });
    await sseEvent(writer, encoder, 'done', {});
  }
}
