dsh-easy-desktop/plugins/agentrq/src/runtime.ts

/**
 * The per-agent AgentRQ runtime.
 *
 * One disposable projection per live root agent. AgentRQ decides *when* there
 * is work — it pushes a task the moment one is created for this agent, and
 * re-pushes the next unclaimed task every 60 seconds from
 * `WorkspaceServer.StartPoller` — so this runtime never asks. It keeps the
 * session open, drops repeats, and routes each push into the session.
 *
 * @module @agentrq/dsh-plugin-agentrq
 */

import type { Context } from '@deepseek-ai/cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { AgentRqClient, AgentRqTask, ChannelMessage } from './client.js'
import type { Config } from './config.js'
import { renderPushFraming, renderTaskFraming } from './prompt.js'

/** Source attribution carried by every message this plugin queues. */
const MESSAGE_SOURCE = { kind: 'plugin', plugin: 'agentrq' } as const

/**
 * How many `(task, content)` pairs to remember for repeat suppression.
 *
 * The workspace re-pushes an unclaimed task every 60 seconds with byte-identical
 * content, so without this the agent would be woken once a minute for work it
 * has already been handed. Bounded because a long-lived session sees many
 * distinct tasks and this is a cache, not a ledger.
 */
const SEEN_LIMIT = 200

/** What `agentrq_autopull` reports about the current runtime. */
export interface DeliveryStatus {
  /** Whether the workspace session is established right now. */
  readonly connected: boolean
  /** Whether pushes are configured to reach the session. */
  readonly configured: boolean
  /** Whether pushes are reaching the session (configured and not paused). */
  readonly active: boolean
  /** Task id most recently delivered to this agent, or null when none has been. */
  readonly lastDeliveredTaskId: string | null
}

/** Render an unknown thrown value for process-local diagnostics only. */
function renderThrown(value: unknown): string {
  return value instanceof Error ? value.message : String(value)
}

/** One live AgentRQ attachment bound to one exact root agent. */
export class AgentRqRuntime {
  private readonly abort = new AbortController()
  private readonly seen = new Set<string>()
  private lastDeliveredTaskId: string | undefined
  private paused = false
  private stopping = false

  constructor(
    private readonly ctx: Context,
    private readonly agent: Agent,
    private readonly client: AgentRqClient,
    private readonly config: Config,
  ) {}

  /** Open the workspace session and, optionally, claim any waiting task. */
  async start(): Promise<void> {
    await this.client.start()
    if (!this.config.catchUpOnStart || this.stopping) return
    try {
      const task = await this.client.fetchNextTask(this.abort.signal)
      if (task !== undefined) this.deliverTask(task)
    } catch (error: unknown) {
      // The workspace re-pushes an unclaimed task on its own schedule, so a
      // failed catch-up costs latency, not work.
      this.warn('startup task check failed', error)
    }
  }

  /** Stop delivering and close the workspace session. */
  async dispose(): Promise<void> {
    this.stopping = true
    this.abort.abort()
    await this.client.dispose()
  }

  /** Current runtime state, for the management tool. */
  status(): DeliveryStatus {
    return {
      connected: this.client.connected,
      configured: this.config.deliverPushes,
      active: this.config.deliverPushes && !this.paused && !this.stopping,
      lastDeliveredTaskId: this.lastDeliveredTaskId ?? null,
    }
  }

  /** Stop routing pushes into this session; the session itself stays open. */
  pause(): DeliveryStatus {
    this.paused = true
    return this.status()
  }

  /** Resume routing pushes into this session. */
  resume(): DeliveryStatus {
    this.paused = false
    return this.status()
  }

  /**
   * Dequeue the next task for an explicit request.
   *
   * The caller is a tool body, so the task travels back as the tool's own
   * result rather than as a queued turn.
   *
   * @param signal - tool-call cancellation.
   * @returns the task, or undefined when the queue is empty.
   */
  async pullNow(signal: AbortSignal): Promise<AgentRqTask | undefined> {
    const task = await this.client.fetchNextTask(signal)
    if (task === undefined) return undefined
    this.remember(task.id, task.text)
    this.lastDeliveredTaskId = task.id
    return task
  }

  /**
   * Route one workspace push into the live session.
   *
   * A new task, the periodic reminder, a status check, and a human's reply all
   * arrive on the same channel, and the plugin forwards each as written — the
   * content is the message, and deciding what kind it is would only add a way
   * to be wrong. A running agent takes it as injected context at its next step
   * boundary; an idle agent is woken with it, because nothing else would.
   *
   * @param message - the push AgentRQ delivered.
   */
  deliverPush(message: ChannelMessage): void {
    if (!this.deliverable()) return
    // The workspace repeats an unclaimed task verbatim every minute.
    if (this.remember(message.chatId, message.text)) return
    this.lastDeliveredTaskId = message.chatId
    this.queue(renderPushFraming(message, this.config.serverName))
  }

  /** Queue one task fetched by the plugin itself, framed as a task hand-off. */
  private deliverTask(task: AgentRqTask): void {
    if (!this.deliverable()) return
    if (this.remember(task.id, task.text)) return
    this.lastDeliveredTaskId = task.id
    this.queue(renderTaskFraming(task, this.config.serverName))
  }

  /** Hand framed text to the agent on the route its current state allows. */
  private queue(text: string): void {
    const framed = createUserMessage({
      content: [{ type: 'text', text }],
      source: MESSAGE_SOURCE,
    })
    try {
      this.ctx.agents.withoutInitiator(() => {
        if (this.agent.status === 'running') this.agent.inject(framed)
        else this.agent.followup(framed)
      })
    } catch (error: unknown) {
      this.warn('could not deliver a workspace push', error)
    }
  }

  /**
   * Record one `(task, content)` pair.
   *
   * @returns whether this exact content was already delivered for this task.
   */
  private remember(chatId: string, text: string): boolean {
    const key = JSON.stringify([chatId, text])
    if (this.seen.has(key)) return true
    this.seen.add(key)
    if (this.seen.size > SEEN_LIMIT) {
      // Insertion-ordered, so the first key is the oldest.
      const oldest = this.seen.values().next()
      if (!oldest.done) this.seen.delete(oldest.value)
    }
    return false
  }

  /** Whether a push may reach the agent right now. */
  private deliverable(): boolean {
    return !this.stopping && !this.paused && this.config.deliverPushes && this.isLive()
  }

  /** Whether this exact root lifecycle is still the authoritative one. */
  private isLive(): boolean {
    return this.ctx.agents.get(this.agent.id) === this.agent
  }

  private warn(what: string, error: unknown): void {
    if (this.stopping || !this.isLive()) return
    this.ctx.logger.warn(`agentrq: ${what} for agent "${this.agent.id}": ${renderThrown(error)}`)
  }
}