feat(ai): add remote coding agent sessions for SSH hosts (#1545)

* feat(ai): add host coding agent sessions with configured providers * Wait for agent initialization and shutdown before resuming sessions * Keep Pi transcripts outside its configuration migration directory * Cover Pi session directory separation * Update core permission catalog expectations for remote agents * Install remote agent runtimes from verified official sources * Wait for the identity panel credentials request in its test * Add scoped SSH forwarding setup for remote agents * Verify forwarding setup through a fresh SSH connection * Bound the SSH forwarding verification timeout * Return the forwarding verification callback result * fix(ai): support agent send and interrupt keyboard shortcuts * feat(ai): add agent queues attachments worktrees and session recovery * fix(ai): handle deleted workspace paths and duplicate status events * fix(ai): accept file references in root workspaces * fix(ai): refresh paused queues after agent errors * fix(ai): store agent sessions in plugin tables and lock down OpenCode Agent sessions lived in a single ctx.kv value holding up to 2000 events, which passed the 256 KB kv limit within minutes of a normal run. Every save after that failed, so history, nativeId, queue and archive changes were lost and every event wrote a kv audit row. Sessions, events and queued prompts now have their own tables. A running agent appends events in batches at most once a second with a lazy save; session and queue changes save immediately. Prompts, drafts and stored event text are capped in UTF-8 bytes to fit MySQL TEXT. kv:own is no longer requested. The OpenCode server now requires a per-session password, so other local accounts on the host cannot drive the agent through its loopback port. * test(upgrade): include agent session tables in the reviewed upgrade SQL
2026-10-30 15:15:24 +00:00 · 2026-10-03 16:01:47 -05:00
1 parent 77bcbdb3d6
commit bf817d8bd1
@@ -0,0 +1,110 @@
+import { afterEach, expect, it, vi } from "vitest";
+import { Duplex, PassThrough } from "node:stream";
+import { EventEmitter } from "node:events";
+import { startServer, type TestServer } from "../helpers.js";
+let server: TestServer;
+afterEach(async () => {
+  await server?.close();
+});
+it("dispatches FIFO once per completed turn and pauses pending work when cancelled", async () => {
+  server = await startServer({
+    permissions: ["ai.agents", "ai.manage_providers"],
+  });
+  await server.enableFor("user-1");
+  const prompts: { text: string }[] = [];
+  class Channel extends Duplex {
+    stderr = new PassThrough();
+    _read() {}
+    _write(chunk: Buffer, _encoding: string, done: () => void) {
+      const m = JSON.parse(chunk.toString());
+      if (m.type === "start") this.complete();
+      if (m.type === "prompt") prompts.push(m);
+      if (m.type === "stop") {
+        this.push(null);
+        queueMicrotask(() => this.emit("close"));
+      }
+      done();
+    }
+    complete() {
+      this.push(JSON.stringify({ kind: "status", text: "ready" }) + "\n");
+    }
+  }
+  const channel = new Channel();
+  const client = Object.assign(new EventEmitter(), {
+    forwardIn: (
+      _host: string,
+      _port: number,
+      callback: (error: null, port: number) => void,
+    ) => callback(null, 30000),
+    exec: (_cmd: string, callback: (error: null, ch: Channel) => void) =>
+      callback(null, channel),
+  });
+  vi.spyOn(server.mock.ctx.ssh, "connect").mockResolvedValue({
+    client,
+    dispose: () => channel.destroy(),
+  } as never);
+  const p = await server.request("POST", "/providers", {
+    body: {
+      providerType: "openai",
+      label: "Test",
+      apiKey: "unused",
+      defaultModel: "test",
+    },
+  });
+  const started = await server.request("POST", "/agents", {
+    body: {
+      agent: "pi",
+      hostId: 1,
+      providerId: p.body.provider.id,
+      model: "test",
+      cwd: "/tmp",
+    },
+  });
+  expect(started.status).toBe(201);
+  const id = started.body.id;
+  try {
+    expect(
+      (
+        await server.request("POST", `/agents/${id}/input`, {
+          body: { type: "prompt", text: "first" },
+        })
+      ).status,
+    ).toBe(200);
+    for (const text of ["second", "third"])
+      expect(
+        (
+          await server.request("POST", `/agents/${id}/queue`, {
+            body: { operation: "add", text },
+          })
+        ).status,
+      ).toBe(200);
+    expect(prompts.map((p) => p.text)).toEqual(["first"]);
+    channel.complete();
+    await vi.waitFor(() =>
+      expect(prompts.map((p) => p.text)).toEqual(["first", "second"]),
+    );
+    await server.request("POST", `/agents/${id}/input`, {
+      body: { type: "cancel" },
+    });
+    channel.complete();
+    await vi.waitFor(async () =>
+      expect((await server.request("GET", `/agents/${id}`)).body.status).toBe(
+        "ready",
+      ),
+    );
+    expect(prompts).toHaveLength(2);
+    const paused = (await server.request("GET", `/agents/${id}`)).body;
+    expect(paused.queuePaused).toBe(true);
+    expect(paused.queue.map((p: { text: string }) => p.text)).toEqual([
+      "third",
+    ]);
+    await server.request("POST", `/agents/${id}/queue`, {
+      body: { operation: "continue" },
+    });
+    await vi.waitFor(() =>
+      expect(prompts.map((p) => p.text)).toEqual(["first", "second", "third"]),
+    );
+  } finally {
+    await server.request("POST", `/agents/${id}/stop`);
+  }
+});