开发 dsh-plugin-deepseek-browser-use：会话、作业、文件与图片

import { randomInt } from 'node:crypto';
import { BrowserClient, requireSuccess, type Envelope, type Params } from './client.js';

export interface Owner { ctx: { effect(callback: () => () => Promise<void>, label?: string): () => Promise<void> }; }
interface Entry { id: number; tail: Promise<unknown>; started: boolean; closed: boolean; abort: AbortController; cleanup: () => Promise<void>; jobs: Set<string>; activeJob?: string; batchUnknown?: boolean; }
export interface SessionOptions { browser: string; headless: boolean; }

/** Live owner identity, not a model-supplied task ID, controls resource access. */
export class BrowserSessions {
  private entries = new Map<Owner, Entry>();
  private disposed = false;
  constructor(readonly client: BrowserClient, readonly options: SessionOptions) {}

  private entry(owner: Owner): Entry {
    if (this.disposed) throw new Error('Browser plugin is disposed');
    const previous = this.entries.get(owner);
    if (previous) return previous;
    // 48 random bits: exact JSON/JavaScript integer, unlike server Snowflake IDs.
    const entry: Entry = { id: randomInt(1, 2 ** 48 - 1), tail: Promise.resolve(), started: false, closed: false,
      abort: new AbortController(), cleanup: async () => {}, jobs: new Set() };
    this.entries.set(owner, entry);
    entry.cleanup = owner.ctx.effect(() => async () => {
      entry.closed = true;
      entry.abort.abort(new Error('Browser session disposed'));
      await entry.tail.catch(() => {});
      if (entry.batchUnknown) throw new Error(`Task ${entry.id}: batch submission outcome unknown. Inspect service list_jobs and finish/cancel the batch before manually closing this task. Cleanup did not close potentially active work.`);
      if (entry.started) {
        const signal = AbortSignal.timeout(this.client.options.timeoutMs);
        if (entry.activeJob) {
          requireSuccess(await this.client.command(entry.id, 'cancel_job', { jobId: entry.activeJob }, signal));
          for (;;) {
            const status = requireSuccess(await this.client.command(entry.id, 'get_job', { jobId: entry.activeJob, includeResult: false }, signal));
            if ((status.data as Params)?.status !== 'running') break;
            await new Promise(resolve => setTimeout(resolve, 100));
            signal.throwIfAborted();
          }
        }
        requireSuccess(await this.client.command(entry.id, 'close', {}, signal));
      }
      this.entries.delete(owner);
    }, 'deepseek-browser-use.session');
    return entry;
  }

  run(owner: Owner, method: string, params: Params, signal: AbortSignal): Promise<Envelope> {
    signal.throwIfAborted();
    const entry = this.entry(owner);
    const combined = AbortSignal.any([signal, entry.abort.signal]);
    const operation = entry.tail.then(async (): Promise<Envelope> => {
      combined.throwIfAborted();
      if (entry.closed) throw new Error('Browser session is closed');
      if (entry.batchUnknown) throw new Error(`Task ${entry.id}: batch submission outcome unknown; inspect service list_jobs before manually recovering. This Session will not submit more browser actions.`);
      if (method === 'get_job' || method === 'cancel_job') {
        if (typeof params.jobId !== 'string' || !entry.jobs.has(params.jobId)) throw new Error('Job does not belong to this Session');
      } else if (entry.activeJob) {
        throw new Error(`Browser batch ${entry.activeJob} is pending. Use dsb_job until it finishes before issuing other commands.`);
      }
      if (method === 'close' && !entry.started) return { ok: true, data: { id: entry.id, closed: true } };
      if (!entry.started) {
        // Remember ownership before HTTP, so timeout during start is still cleaned up.
        entry.started = true;
        const started = await this.client.command(entry.id, 'start', { ...this.options }, combined);
        if (!started.ok) entry.started = false;
        requireSuccess(started);
        if (method === 'start') return started;
      } else if (method === 'start') {
        return { ok: true, data: { id: entry.id, reused: true } };
      }
      let result: Envelope;
      try { result = await this.client.command(entry.id, method, params, combined); }
      catch (error) {
        if (method === 'commands') entry.batchUnknown = true;
        throw error;
      }
      const data = result.data as Params | undefined;
      if (method === 'commands' && result.ok) {
        if (typeof data?.jobId !== 'string') {
          entry.batchUnknown = true;
          throw new Error(`Task ${entry.id}: async batch response missing jobId; inspect service before resubmitting`);
        }
        entry.jobs.add(data.jobId);
        entry.activeJob = data.jobId;
      }
      if (method === 'get_job' && result.ok && data && ['done', 'failed', 'cancelled'].includes(String(data.status)) && params.jobId === entry.activeJob) {
        entry.activeJob = undefined;
      }
      if (method === 'close' && result.ok) entry.started = false;
      return result;
    });
    entry.tail = operation.catch(() => {});
    return operation;
  }

  async dispose(): Promise<void> {
    this.disposed = true;
    const results = await Promise.allSettled([...this.entries.values()].map(entry => entry.cleanup()));
    const errors = results.filter(r => r.status === 'rejected').map(r => r.reason);
    if (errors.length) throw new AggregateError(errors, 'Browser session cleanup failed');
  }
}

const operation = entry.tail.then(async (): Promise<Envelope> => {
  // 在这里检查取消、启动任务、发送本次命令。
});
entry.tail = operation.catch(() => {});
return operation;

const forbidden = new Set(['start', 'close', 'shutdown', 'cleanup', 'get_config', 'list_tasks', 'list_jobs', 'get_job', 'cancel_job', 'run_recipe', 'commands']);
export function validateCommand(method: string): void {
  if (!commandNames.includes(method) || forbidden.has(method)) {
    throw new Error(`Command ${method} is not an exposed task command. Use dedicated lifecycle/job tools; service-wide commands and recipes are not exposed.`);
  }
}

  add('command', 'Execute an exact supported task-scoped command. Use dsb_methods and the skill for names/parameters. No lifecycle, global management, nested batches or recipes.',
    z.strictObject({ method: z.string().min(1), params: object.optional() }), (a, e) => { validateCommand(a.method); return command(e, a.method, a.params ?? {}); });
  add('batch', 'Submit sequential task commands as a background job. Poll dsb_job before issuing other actions. Do not batch indices whose validity depends on earlier page changes.', z.strictObject({
    commands: z.array(z.strictObject({ method: z.string(), params: object.optional() })).min(1).max(200),
    stopOnError: z.boolean().optional(),
  }), (a, e) => {
    for (const step of a.commands) validateCommand(step.method);
    return command(e, 'commands', { commands: a.commands.map(step => ({ [step.method]: step.params ?? {} })), async: true, stopOnError: a.stopOnError ?? true });
  });
  add('job', 'Read or request cooperative cancellation of a job created by this Session. Cancellation does not undo an already delivered action.',
    z.strictObject({ jobId: z.string().min(1), cancel: z.boolean().optional() }), (a, e) => command(e, a.cancel ? 'cancel_job' : 'get_job', { jobId: a.jobId, ...(a.cancel ? {} : { includeResult: true }) }));

  add('upload', 'Upload a file from the Harness host to a file input. localPath is on the Harness host; bytes are transferred over HTTP even if browser service is remote.',
    z.strictObject({ ...target, localPath: z.string().min(1) }), async (a, e) => {
      checkTarget(a);
      const info = await stat(a.localPath);
      if (!info.isFile() || info.size > options.maxUploadBytes) throw new Error('Upload must be a regular file within maxUploadBytes');
      const bytes = await readFile(a.localPath, { signal: e.signal });
      if (bytes.length > options.maxUploadBytes) throw new Error('Upload exceeds maxUploadBytes');
      const { localPath, ...locator } = a;
      return command(e, 'upload_file', { ...locator, filename: basename(localPath), contentBase64: bytes.toString('base64') });
    });

import { readFile, stat } from 'node:fs/promises';
import { basename } from 'node:path';

  add('screenshot', 'Capture the page or a selector. Set view=true to attach image for a verified vision-capable model; otherwise return a saved screenshot URL.',
    z.strictObject({ fullPage: z.boolean().optional(), selector: z.string().optional(), view: z.boolean().optional() }), async (a, e) => {
      const attachments = ctx.get('attachments');
      if (a.view) {
        if (!attachments) throw new Error('view=true requires the Harness attachment service');
        const route = e.agent?.session.requestHeader()?.config;
        const provider = route?.provider ?? e.agent?.options.provider;
        const model = route?.model ?? e.agent?.options.model;
        const llm = ctx.get('llm');
        if (!provider || !model || !llm) throw new Error('Cannot verify the active model image capability; use view=false');
        const info = await llm.resolveModelInfo(provider, model, e.signal);
        if (!info.inputModalities?.includes('image')) throw new Error('Current model does not declare image input; use view=false and DOM text tools');
      }
      const { view, ...params } = a;
      const result = await command(e, 'screenshot', { ...params, inline: view ?? false });
      if (!result.ok || !view) return result;
      const data = result.data as Params;
      if (typeof data?.base64 !== 'string') throw new Error('Screenshot response missing base64');
      e.signal.throwIfAborted();
      const attachment = await attachments!.saveImage({ data: Buffer.from(data.base64, 'base64'), mediaType: 'image/png', name: 'browser-screenshot.png' });
      const { base64: _bytes, ...metadata } = data;
      return { ...result, data: metadata, attachment: attachment as unknown as Params };
    });