Official
chat
Built-in chat UI (JavaScript UI plugin).
The app opens the listing; nothing installs until an agent in your Plugins workspace has read the files and you enable the plugin. In a terminal: cvg install convergence/chat@0.2.0
Permissions in 0.2.0
Files
testing.ts13.3 KB
// Stateful input-store fixture: CAS, idempotency, structured bytes, admission
// and settlement are separate operations. Tests cannot get a run by enqueueing
// or observe settled success merely by emitting run_finished.
import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";
import * as z from "zod";
import { Host, Kernel, KernelCallFailed, promise } from "convergence/effect";
import { testLayer } from "convergence/effect/testing";
import type {
Api,
ContentBlock,
HostMethods,
KernelMethods,
InputSnapshot,
PendingInput,
SettledRun,
SettingValue,
UsageLimits,
} from "convergence";
import { Controller } from "./controller.ts";
import { chat } from "./ui-for-tests.test.ts";
import { contentBlocks } from "./input-schema.ts";
const paramsSchema = z.object({
chatId: z.string().optional(),
id: z.string().optional(),
inputId: z.string().optional(),
revision: z.number().optional(),
blocks: contentBlocks.optional(),
intent: z.enum(["queue", "send", "steer"]).optional(),
origin: z.string().optional(),
append: z.boolean().optional(),
policyState: z.unknown().optional(),
pauseReason: z.string().nullable().optional(),
remove: z.boolean().optional(),
scope: z.string().optional(),
policy: z.string().optional(),
agentId: z.string().optional(),
key: z.string().optional(),
value: z.unknown().optional(),
});
export class InputLedger {
snapshots: Record<string, InputSnapshot> = {};
calls: { method: string; params: unknown }[] = [];
deliveries: { chatId: string; inputId: string; scope?: string; policy?: string; blocks: ContentBlock[] }[] = [];
notices: string[] = [];
forgotten: string[] = [];
settings: Record<string, SettingValue> = {};
storage: Record<string, unknown> = {};
token: string | null = "policy:stock:1";
rejectEnqueue = false;
rejectDispatch = false;
missingImage = false;
availability: (agentId: string) => Promise<{ limits: UsageLimits | null }> = async () => ({ limits: null });
beforeEnqueue: (() => Promise<void>) | undefined;
afterClaim: ((snapshot: InputSnapshot) => Promise<void>) | undefined;
beforePolicy: (() => Promise<void>) | undefined;
constructor(ids = ["c1"]) {
for (const id of ids)
this.snapshots[id] = {
chatId: id,
agentId: "agent",
sessionId: `session:${id}`,
generation: 1,
revision: 1,
inputs: [],
running: false,
settling: false,
policyState: null,
};
}
mutate(id: string, update: (snapshot: InputSnapshot) => void) {
const snapshot = structuredClone(this.snapshots[id]);
update(snapshot);
snapshot.revision++;
this.snapshots[id] = snapshot;
return structuredClone(snapshot);
}
finalizing(id: string) {
return this.mutate(id, (snapshot) => {
snapshot.running = false;
snapshot.settling = true;
});
}
settled(id: string, run: SettledRun) {
return this.mutate(id, (snapshot) => {
snapshot.running = false;
snapshot.settling = false;
snapshot.lastRun = run;
snapshot.runId = undefined;
});
}
consumed(id: string, inputId: string) {
// Only explicitly injected correlated pickup evidence changes this state;
// run settlement alone cannot acknowledge an admitted native inbox entry.
return this.mutate(id, (snapshot) => {
const input = snapshot.inputs.find((input) => input.id === inputId);
if (input) input.state = "observed_consumed";
});
}
async call(method: string, raw: unknown): Promise<unknown> {
this.calls.push({ method, params: structuredClone(raw) });
const params = paramsSchema.parse(raw),
id = params.chatId ?? "c1";
if (method === "host/storage.get") return { value: this.storage[params.key ?? ""] };
if (method === "host/storage.set") {
this.storage[params.key ?? ""] = params.value;
return {};
}
if (method === "host/chats.list")
return {
chats: Object.values(this.snapshots).map((snapshot) =>
chat({ id: snapshot.chatId, agentId: snapshot.agentId, sessionId: snapshot.sessionId ?? undefined }),
),
};
if (method === "host/chats.options") return { options: [] };
if (method === "host/chats.commands") return { commands: [] };
if (method === "host/chats.skills") return { skills: [] };
if (method === "host/chats.cancel") {
this.finalizing(id);
return {};
}
if (method === "host/agents.refresh_usage") return this.availability(params.agentId ?? "agent");
if (method === "host/inputs.policy") {
await this.beforePolicy?.();
return { token: this.token };
}
if (method === "host/inputs.list") return structuredClone(this.snapshots[id]);
if (method === "host/inputs.enqueue") {
await this.beforeEnqueue?.();
if (this.rejectEnqueue) throw new Error("disk unavailable");
const snapshot = this.snapshots[id];
if (snapshot.inputs.some((input) => input.id === params.id)) return structuredClone(snapshot);
return this.mutate(id, (snapshot) => {
if ((params.origin ?? "user") === "user")
snapshot.inputs = snapshot.inputs.filter(
(input) => !(input.origin === "generated_continue" && ["held", "rejected"].includes(input.state)),
);
if (
params.origin === "generated_continue" &&
snapshot.inputs.some((input) => input.state !== "observed_consumed")
)
return;
const last = snapshot.inputs.filter((input) => input.intent === "steer" && input.state === "held").at(-1);
if (params.append && last) {
for (const block of params.blocks ?? []) {
const tail = last.blocks.at(-1);
if (block.type === "text" && tail?.type === "text") tail.text += `\n${block.text}`;
else last.blocks.push(block);
}
return;
}
snapshot.inputs.push({
id: params.id ?? "invalid",
chatId: id,
agentId: snapshot.agentId,
generation: snapshot.generation,
revision: 1,
blocks: params.blocks ?? [],
intent: params.intent ?? "send",
originalIntent: params.intent ?? "send",
state: "held",
origin: params.origin ?? "user",
createdAt: new Date().toISOString(),
});
});
}
if (method === "host/inputs.update" || method === "host/inputs.dispatch") {
const current = this.snapshots[id];
if (current.revision !== params.revision) throw new Error("Input revision changed");
const input = current.inputs.find((input) => input.id === params.inputId);
if (input && !["held", "rejected"].includes(input.state)) throw new Error("Input is not editable");
if (method === "host/inputs.update")
return this.mutate(id, (snapshot) => {
if (Object.hasOwn(raw ?? {}, "policyState")) snapshot.policyState = params.policyState;
const entry = snapshot.inputs.find((input) => input.id === params.inputId);
if (entry) {
if (params.blocks) entry.blocks = params.blocks;
if (params.intent) entry.intent = params.intent;
if (params.pauseReason !== undefined) entry.pauseReason = params.pauseReason ?? undefined;
if (params.remove) snapshot.inputs = snapshot.inputs.filter((item) => item.id !== entry.id);
}
});
if (!input || (current.running && input.intent === "queue") || current.settling)
throw new Error("Input cannot dispatch");
if (params.policy && params.policy !== this.token) throw new Error("Policy replaced");
if (
params.scope &&
this.deliveries.some((delivery) => delivery.scope === params.scope && this.snapshots[delivery.chatId]?.running)
)
throw new Error("Scope already claimed");
if (this.rejectDispatch) {
this.mutate(id, (snapshot) => {
const entry = snapshot.inputs.find((entry) => entry.id === input.id);
if (entry) {
entry.state = "delivery_unknown";
entry.error = "transport lost";
}
});
throw new Error("transport lost");
}
this.deliveries.push({
chatId: id,
inputId: input.id,
scope: params.scope,
policy: params.policy,
blocks: structuredClone(input.blocks),
});
const nonPrompt = input.blocks.some((block) => block.type === "text" && block.text === "/no-prompt");
const claimed = this.mutate(id, (snapshot) => {
const entry = snapshot.inputs.find((entry) => entry.id === input.id);
if (entry) {
entry.state = "dispatching";
entry.attemptId = `attempt:${snapshot.revision}`;
entry.runId = snapshot.running ? snapshot.runId : `run:${input.id}`;
}
if (!nonPrompt) {
snapshot.running = true;
snapshot.runId = entry?.runId;
}
});
// A provider may still be awaiting admission while enqueue/cancel mutate
// the host. Tests inject receipts or positive rejection explicitly here.
await this.afterClaim?.(claimed);
return this.mutate(id, (snapshot) => {
const entry = snapshot.inputs.find((entry) => entry.id === input.id);
if (entry) entry.state = nonPrompt ? "observed_consumed" : "native_admitted";
});
}
throw new Error(`Unexpected fixture method ${method}`);
}
}
export const settledRun = (id: string, overrides: Partial<SettledRun> = {}): SettledRun => ({
id,
generation: 1,
outcome: { status: "completed" },
stoppedAt: "2026-10-04T12:00:00Z",
steering: false,
...overrides,
});
export const held = (id: string, overrides: Partial<PendingInput> = {}): PendingInput => ({
id,
chatId: "c1",
agentId: "agent",
generation: 1,
revision: 1,
createdAt: "2026-10-04T10:00:00Z",
blocks: [{ type: "text", text: id }],
intent: "queue",
originalIntent: "queue",
origin: "user",
state: "held",
...overrides,
});
export function withController<A>(
ledger: InputLedger,
fn: (controller: Controller) => Effect.Effect<A, unknown>,
): Promise<A> {
const host: Host["Service"] = {
call: <M extends keyof HostMethods>(method: M, params: HostMethods[M]["params"]) =>
promise(method, async () => {
// This assertion describes a local, stateful host fixture. Production code
// uses the generated typed host API, never an unchecked external cast.
return (await ledger.call(method, params)) as HostMethods[M]["result"];
}),
};
const fixture = testLayer({ host });
const api = {
...fixture.api,
app: { state: () => ({ route: "chat", chatId: "c1" }) },
mentions: { list: () => [] },
slashCommands: () => [],
settings: {
...fixture.api.settings,
get: (key?: string) => (key ? (ledger.settings[key] ?? null) : ledger.settings),
set: async (key: string, value: SettingValue) => {
ledger.settings[key] = value;
return value;
},
},
} as unknown as Api;
const kernel: Kernel["Service"] = {
call: <M extends keyof KernelMethods>(method: M, params: KernelMethods[M]["params"]) =>
Effect.try({
try: () => {
let value: unknown;
if (method === "image_data") {
if (ledger.missingImage) throw new Error("image missing");
value = { mimeType: "image/png", data: "persisted-image-bytes" };
} else if (method === "forget_image") {
ledger.forgotten.push(z.object({ id: z.string() }).parse(params).id);
value = {};
} else throw new Error(`Unexpected kernel ${method}`);
// Local fixture replies are confined to this typed adapter.
return value as KernelMethods[M]["result"];
},
catch: (error) => new KernelCallFailed({ method, message: String(error) }),
}),
};
return Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const controller = new Controller({
api,
host: yield* Host,
kernel,
app: {
changes: Stream.empty,
state: Effect.succeed({ route: "chat", chatId: "c1" }),
select: () => Effect.void,
route: () => Effect.void,
update: () => Effect.void,
},
notify: {
error: (message) =>
Effect.sync(() => {
ledger.notices.push(message);
}),
warning: (message) =>
Effect.sync(() => {
ledger.notices.push(message);
}),
info: (message) =>
Effect.sync(() => {
ledger.notices.push(message);
}),
},
views: { slot: () => Effect.succeed({ id: 1, update() {}, remove() {} }), focus: () => Effect.void },
scope: yield* Effect.scope,
});
controller.state.agents.agent = { id: "agent", name: "Agent", capabilities: { steer: true } };
for (const snapshot of Object.values(ledger.snapshots)) {
controller.state.chats[snapshot.chatId] = chat({ id: snapshot.chatId, agentId: snapshot.agentId, sessionId: snapshot.sessionId ?? undefined });
controller.composer(snapshot.chatId).loading = false;
controller.scheduler.receive(structuredClone(snapshot));
}
controller.scheduler.available = true;
controller.scheduler.ready = true;
return yield* fn(controller);
}),
).pipe(Effect.provide(fixture)),
);
}Versions
| Version | Published | Plugin API | Size | Permissions | Status |
|---|---|---|---|---|---|
| 0.2.0latest | Oct 5, 2026 | >=2 <3 | 105.1 KB | 5 permissions | Listed |
No comments yet.