spb/worthdoing Public

1/**2 * WorthDoing.ai3 * Author: Simon-Pierre Boucher4 * Contact: contact@spboucher.ai5 * File: src/lib/agent/engine.ts6 * Description: The explicit agent loop — Claude decides actions, the backend executes, state persists, repeat.7 */8import type Anthropic from "@anthropic-ai/sdk";9import { eq } from "drizzle-orm";10import { db } from "@/lib/db/client";11import { investigations, investigationSteps } from "@/lib/db/schema";12import { anthropic, agentModel, estimateCostUsd } from "@/lib/anthropic/client";13import { AGENT_SYSTEM_PROMPT } from "./prompts";14import { anthropicTools } from "./tools";15import { executeTool } from "./executors";16import { loadState, buildStateDigest, computePhase } from "./state";17import { emitEvent } from "./events";1819const MAX_TOKENS_PER_STEP = 16000;20const KEEP_RECENT_MESSAGES = 8; // messages kept verbatim; older tool results are compressed21const OLD_TOOL_RESULT_CAP = 400; // chars kept from old tool results after compression2223type MessageParam = Anthropic.MessageParam;2425/** Compress old tool results in-place so scraped pages are never resent in full. */26function compressHistory(messages: MessageParam[]): MessageParam[] {27  if (messages.length <= KEEP_RECENT_MESSAGES) return messages;28  const cutoff = messages.length - KEEP_RECENT_MESSAGES;29  return messages.map((m, i) => {30    if (i >= cutoff || m.role !== "user" || typeof m.content === "string") return m;31    const content = m.content.map((block) => {32      if (block.type === "tool_result" && typeof block.content === "string" && block.content.length > OLD_TOOL_RESULT_CAP) {33        return {34          ...block,35          content: `${block.content.slice(0, OLD_TOOL_RESULT_CAP)}\n…[older result compressed — durable facts live in the state digest]`,36        };37      }38      if (block.type === "text" && block.text.startsWith("CURRENT INVESTIGATION STATE")) {39        return { ...block, text: "[superseded state digest omitted]" };40      }41      return block;42    });43    return { ...m, content };44  });45}4647/**48 * Run an investigation to completion. Resumable: rebuilds context from49 * persisted state, so a restart continues rather than restarts.50 */51export async function runInvestigation(investigationId: string): Promise<void> {52  const model = agentModel();53  let state = await loadState(investigationId);54  const inv = state.investigation;5556  if (inv.status === "completed" || inv.status === "cancelled") return;5758  const startedFresh = inv.status === "pending";59  await db60    .update(investigations)61    .set({ status: "running", startedAt: inv.startedAt ?? new Date() })62    .where(eq(investigations.id, investigationId));6364  if (startedFresh) {65    await emitEvent(investigationId, "investigation.started", {66      objective: inv.objective,67      budget: inv.budget as unknown as Record<string, unknown>,68      model,69    });70  }7172  const wallDeadline = (inv.startedAt?.getTime() ?? Date.now()) + inv.budget.maxWallTimeMs;7374  let messages: MessageParam[] = [75    {76      role: "user",77      content: `CURRENT INVESTIGATION STATE\n${buildStateDigest(state)}\n\nBegin (or continue) the investigation. Narrate briefly, then act with tools.`,78    },79  ];8081  try {82    for (;;) {83      state = await loadState(investigationId);84      const used = state.investigation.budgetUsed;85      const budget = state.investigation.budget;8687      if (state.investigation.status === "cancelled") return;8889      if (used.agentSteps >= budget.maxAgentSteps || Date.now() > wallDeadline) {90        const reason = Date.now() > wallDeadline ? "budget_wall_time" : "budget_steps";91        await forceStop(investigationId, reason);92        return;93      }9495      // Phase bookkeeping — emitted only on real transitions.96      const phase = computePhase(state);97      if (phase !== state.investigation.phase) {98        await db.update(investigations).set({ phase }).where(eq(investigations.id, investigationId));99        await emitEvent(investigationId, "phase.changed", { from: state.investigation.phase, to: phase });100      }101102      const stepNumber = used.agentSteps + 1;103      const t0 = Date.now();104105      const stream = anthropic().messages.stream({106        model,107        max_tokens: MAX_TOKENS_PER_STEP,108        system: [{ type: "text", text: AGENT_SYSTEM_PROMPT, cache_control: { type: "ephemeral" } }],109        tools: anthropicTools,110        messages: compressHistory(messages),111      });112      const response = await stream.finalMessage();113      const latencyMs = Date.now() - t0;114115      // Count the step + tokens.116      const inputTokens = response.usage.input_tokens + (response.usage.cache_read_input_tokens ?? 0) + (response.usage.cache_creation_input_tokens ?? 0);117      const outputTokens = response.usage.output_tokens;118      const costUsd = estimateCostUsd(model, response.usage.input_tokens, outputTokens);119      await db120        .update(investigations)121        .set({122          budgetUsed: {123            ...used,124            agentSteps: used.agentSteps + 1,125            inputTokens: used.inputTokens + inputTokens,126            outputTokens: used.outputTokens + outputTokens,127            costUsd: Math.round((used.costUsd + costUsd) * 10000) / 10000,128          },129        })130        .where(eq(investigations.id, investigationId));131132      const textBlocks = response.content.filter((b): b is Anthropic.TextBlock => b.type === "text");133      const planText = textBlocks.map((b) => b.text).join("\n").trim();134      if (planText) {135        await emitEvent(investigationId, "agent.plan", { step: stepNumber, text: planText });136      }137138      await db.insert(investigationSteps).values({139        investigationId,140        stepNumber,141        model,142        inputTokens,143        outputTokens,144        costUsd,145        latencyMs,146        decisionSummary: planText.slice(0, 2000) || null,147      });148149      if (response.stop_reason === "refusal") {150        await failInvestigation(investigationId, "Model declined a step (safety classifiers).");151        return;152      }153154      // Keep assistant content verbatim (thinking blocks must round-trip unchanged).155      messages.push({ role: "assistant", content: response.content });156157      if (response.stop_reason === "pause_turn") continue;158159      const toolUses = response.content.filter((b): b is Anthropic.ToolUseBlock => b.type === "tool_use");160161      if (toolUses.length === 0) {162        // No tool call — nudge once toward action; the loop's only exit is finish_investigation.163        messages.push({164          role: "user",165          content:166            "You ended your turn without a tool call. Continue with tool calls, or end the investigation properly with finish_investigation.",167        });168        continue;169      }170171      const toolResults: Anthropic.ToolResultBlockParam[] = [];172      let finished = false;173      for (const call of toolUses) {174        const execT0 = Date.now();175        const outcome = await executeTool(investigationId, call.name, call.input);176        await db.insert(investigationSteps).values({177          investigationId,178          stepNumber,179          model,180          toolName: call.name,181          toolArgs: call.input,182          toolResultSummary: outcome.content.slice(0, 1500),183          latencyMs: Date.now() - execT0,184          error: outcome.isError ? outcome.content.slice(0, 500) : null,185        });186        toolResults.push({187          type: "tool_result",188          tool_use_id: call.id,189          content: outcome.content,190          is_error: outcome.isError,191        });192        if (outcome.finished) finished = true;193      }194195      if (finished) return;196197      // Fresh digest rides along with the tool results each turn.198      state = await loadState(investigationId);199      messages.push({200        role: "user",201        content: [202          ...toolResults,203          { type: "text", text: `CURRENT INVESTIGATION STATE\n${buildStateDigest(state)}` },204        ],205      });206    }207  } catch (err) {208    const message = err instanceof Error ? err.message : String(err);209    await failInvestigation(investigationId, message);210  }211}212213async function forceStop(investigationId: string, stopReason: "budget_steps" | "budget_wall_time"): Promise<void> {214  const state = await loadState(investigationId);215  const conclusion = `Investigation stopped: ${216    stopReason === "budget_steps" ? "agent step budget exhausted" : "wall-time budget exhausted"217  }. ${state.opportunities.length} opportunit${state.opportunities.length === 1 ? "y" : "ies"} and ${state.evidence.length} evidence items were gathered before the stop.`;218  await db219    .update(investigations)220    .set({221      status: "completed",222      phase: "done",223      stopReason,224      conclusion,225      outcome: state.opportunities.length > 0 ? "opportunities_found" : "insufficient_evidence",226      completedAt: new Date(),227    })228    .where(eq(investigations.id, investigationId));229  await emitEvent(investigationId, "investigation.completed", {230    conclusion,231    outcome: state.opportunities.length > 0 ? "opportunities_found" : "insufficient_evidence",232    stopReason,233    stats: {234      hypotheses: state.hypotheses.length,235      rejected: state.hypotheses.filter((h) => h.status === "rejected").length,236      evidence: state.evidence.length,237      searches: state.searches.length,238      opportunities: state.opportunities.length,239    },240  });241}242243async function failInvestigation(investigationId: string, message: string): Promise<void> {244  await db245    .update(investigations)246    .set({ status: "failed", stopReason: "error", error: message.slice(0, 1000), completedAt: new Date() })247    .where(eq(investigations.id, investigationId));248  await emitEvent(investigationId, "investigation.failed", { message: message.slice(0, 500) });249}250