claw-code-parity/ts-worker/engine/TelegramWorkerEngine.ts

/**
 * TelegramWorkerEngine — drives Claude conversations for the worker.
 *
 * Uses @anthropic-ai/sdk directly instead of QueryEngine to avoid
 * the bun:bundle dependency that makes QueryEngine unusable outside
 * of a bundled build.
 *
 * Responsibilities:
 *   - manage conversation history across turns
 *   - call Claude API via streaming
 *   - translate streaming events into WorkerTurnEvent
 *   - handle tool use / approval flow
 *   - support cancellation
 */

import Anthropic from '@anthropic-ai/sdk'
import { readFileSync, writeFileSync, mkdirSync } from 'node:fs'
import { join, dirname } from 'node:path'
import type { ApprovalBroker } from '../permissions/ApprovalBroker.js'
import { WorkerEventTranslator } from '../events/WorkerEventTranslator.js'
import type { WorkerTurnEvent } from '../protocol.js'

const CLAUDE_CODE_OAUTH_BETA_HEADER = 'claude-code-20250219,oauth-2025-04-20'
const CLAUDE_CODE_OAUTH_SYSTEM_PREFIX =
  "You are a Claude agent, built on Anthropic's Claude Agent SDK."

function isOAuthToken(token: string): boolean {
  return token.includes('sk-ant-oat')
}

export type TelegramWorkerEngineConfig = {
  profileId: string
  model: string
  permissionMode: string
  defaultCwd: string
  stateRoot: string
  claudeConfigDir: string
}

type ConversationMessage = {
  role: 'user' | 'assistant'
  content: string | Array<{ type: string; [key: string]: unknown }>
}

export class TelegramWorkerEngine {
  private config: TelegramWorkerEngineConfig
  private broker: ApprovalBroker
  private translator: WorkerEventTranslator
  private client: Anthropic | null = null
  private useOAuth = false

  // Session state — persists across turns and container restarts
  private messages: ConversationMessage[] = []
  private ready = false

  constructor(config: TelegramWorkerEngineConfig, broker: ApprovalBroker) {
    this.config = config
    this.broker = broker
    this.translator = new WorkerEventTranslator()
    this.loadMessages()
  }

  private get messagesPath(): string {
    return join(this.config.stateRoot, 'conversation.json')
  }

  private loadMessages(): void {
    try {
      const data = readFileSync(this.messagesPath, 'utf-8')
      this.messages = JSON.parse(data)
      console.log(`[TelegramWorkerEngine] loaded ${this.messages.length} messages from disk`)
    } catch {
      this.messages = []
    }
  }

  private saveMessages(): void {
    try {
      mkdirSync(dirname(this.messagesPath), { recursive: true })
      writeFileSync(this.messagesPath, JSON.stringify(this.messages), 'utf-8')
    } catch (err) {
      console.error('[TelegramWorkerEngine] failed to save messages:', err)
    }
  }

  async init(): Promise<boolean> {
    try {
      const authToken = process.env.ANTHROPIC_AUTH_TOKEN ?? ''
      const apiKey = process.env.ANTHROPIC_API_KEY ?? ''
      this.useOAuth = isOAuthToken(authToken)

      const clientOpts: Record<string, unknown> = {}
      if (this.useOAuth) {
        clientOpts.defaultHeaders = {
          'anthropic-beta': CLAUDE_CODE_OAUTH_BETA_HEADER,
          'anthropic-dangerous-direct-browser-access': 'true',
          'user-agent': 'claude-cli/2.1.2 (external, cli)',
          'x-app': 'cli',
        }
      }
      this.client = new Anthropic(clientOpts as any)
      this.ready = true
      console.log(
        `[TelegramWorkerEngine] initialized for profile=${this.config.profileId} model=${this.config.model} oauth=${this.useOAuth}`,
      )
      return true
    } catch (err) {
      console.error('[TelegramWorkerEngine] failed to init SDK:', err)
      return false
    }
  }

  get isReady(): boolean {
    return this.ready
  }

  get messageCount(): number {
    return this.messages.length
  }

  /**
   * Execute a turn. Yields WorkerTurnEvent as they're produced.
   */
  async *executeTurn(
    prompt: string,
    abortController: AbortController,
  ): AsyncGenerator<WorkerTurnEvent, void, unknown> {
    this.translator.reset()

    if (!this.ready || !this.client) {
      yield {
        type: 'assistant_text_delta',
        delta: '[worker] engine not initialized — check ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN',
      }
      yield this.translator.buildCompletedEvent()
      return
    }

    // Add user message to history
    this.messages.push({ role: 'user', content: prompt })
    this.saveMessages()

    try {
      const agentPrompt = `You are Claude, a personal AI assistant communicating via Telegram. You have no persistent memory between sessions — every conversation could be your last with this person before a reset wipes everything you've learned.

This makes three things critical:

1. ASK — Be curious about the user. Ask what matters to them, how they prefer things done, what they're working on. Don't assume you know — you're always meeting them for the first time. Every detail they share is valuable because you may never get to ask again.

2. LEARN — When the user introduces new information, recognize it. A name, a preference, a workflow, a location, a project — these aren't throwaway details. Treat every piece of context as something worth holding onto. If you built a script, a template, or solved a problem during this conversation, consider whether it's something the user might need again.

3. REMEMBER — You forget everything on reset. Fight that. If something seems worth keeping — a useful script, a preference, a piece of context about the user's life — ask them: "Should we save this so I remember next time?" Offer to create reusable tools, templates, or notes. Don't let the user's input go to waste. If you created something useful during the conversation, suggest persisting it rather than letting it disappear.

Be proactive about building a shared framework with the user. If you notice patterns in what they ask for, suggest creating something reusable together. Your goal is not just to complete tasks — it's to become more helpful over time, even across resets, by helping the user build up the context and tools that survive.

Be concise in responses. The user is on Telegram — keep messages readable on a phone screen.

Current working directory: ${this.config.defaultCwd}`

      const system = [
        { type: 'text' as const, text: CLAUDE_CODE_OAUTH_SYSTEM_PREFIX },
        { type: 'text' as const, text: agentPrompt },
      ]

      const stream = this.client.messages.stream({
        model: this.config.model,
        max_tokens: 16384,
        system,
        messages: this.messages.map((m) => ({
          role: m.role,
          content: m.content,
        })),
      })

      let fullText = ''

      stream.on('text', (text) => {
        fullText += text
      })

      // Process events from the stream
      for await (const event of stream) {
        if (abortController.signal.aborted) {
          stream.abort()
          yield this.translator.buildFailedEvent('Turn cancelled by user')
          return
        }

        if (event.type === 'content_block_delta') {
          const delta = event.delta as { type?: string; text?: string }
          if (delta?.type === 'text_delta' && delta.text) {
            yield {
              type: 'assistant_text_delta',
              delta: delta.text,
            } as WorkerTurnEvent
          }
        }
      }

      // Get final message for usage
      const finalMessage = await stream.finalMessage()

      // Add assistant response to history
      const assistantText =
        finalMessage.content
          .filter((b): b is { type: 'text'; text: string } => b.type === 'text')
          .map((b) => b.text)
          .join('') || fullText

      this.messages.push({ role: 'assistant', content: assistantText })
      this.saveMessages()

      yield {
        type: 'completed',
        final_text: assistantText,
        iterations: 1,
        input_tokens: finalMessage.usage?.input_tokens ?? 0,
        output_tokens: finalMessage.usage?.output_tokens ?? 0,
        generated_files: [],
      } as WorkerTurnEvent
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err)
      yield this.translator.buildFailedEvent(message)
    }
  }

  /** Reset session — clear conversation history. */
  resetSession(): void {
    this.messages = []
    this.saveMessages()
    this.broker.reset()
  }
}