508 lines
18 KiB
TypeScript
508 lines
18 KiB
TypeScript
|
import {
|
||
|
|
type Item,
|
||
|
|
type Request,
|
||
|
|
type Ref,
|
||
|
|
type ThreadState,
|
||
|
|
type ThreadController,
|
||
|
|
type Thread,
|
||
|
|
type Cursor,
|
||
|
|
itemKey,
|
||
|
|
terminal,
|
||
|
|
} from "./protocol.js";
|
||
|
|
|
||
|
|
export async function api<T = any>(path: string, input?: unknown, signal?: AbortSignal): Promise<T> {
|
||
|
|
const response = await fetch(`/api/agent${path}`, {
|
||
|
|
credentials: "same-origin",
|
||
|
|
signal,
|
||
|
|
...(input === undefined
|
||
|
|
? {}
|
||
|
|
: { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(input) }),
|
||
|
|
});
|
||
|
|
const value = await response.json();
|
||
|
|
if (!response.ok)
|
||
|
|
throw Object.assign(Error(value.error ?? `Agent service unavailable (${response.status})`), {
|
||
|
|
knownRejected: value.knownRejected === true,
|
||
|
|
});
|
||
|
|
if (value.codec !== "quixos-agent-http-v1") throw Error("Unsupported Agent service response");
|
||
|
|
return value;
|
||
|
|
}
|
||
|
|
export const control = async (input: Record<string, unknown>) => {
|
||
|
|
const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(JSON.stringify(input)));
|
||
|
|
const key = `quixos.agent.control:${location.origin}:${Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("")}`;
|
||
|
|
const previous = sessionStorage.getItem(key),
|
||
|
|
retained = previous
|
||
|
|
? JSON.parse(previous)
|
||
|
|
: {
|
||
|
|
operationId: crypto.randomUUID(),
|
||
|
|
createdAt: new Date().toISOString(),
|
||
|
|
...input,
|
||
|
|
};
|
||
|
|
sessionStorage.setItem(key, JSON.stringify(retained));
|
||
|
|
let result;
|
||
|
|
try {
|
||
|
|
result = await api("/control", retained);
|
||
|
|
} catch (error) {
|
||
|
|
if ((error as { knownRejected?: boolean }).knownRejected) sessionStorage.removeItem(key);
|
||
|
|
throw error;
|
||
|
|
}
|
||
|
|
sessionStorage.removeItem(key);
|
||
|
|
return result;
|
||
|
|
};
|
||
|
|
export const draftKey = (thread: string) => `quixos.agent.draft:${location.origin}:${thread}`;
|
||
|
|
|
||
|
|
/** Status and directory invalidation shared by shells, without a recurring
|
||
|
|
* snapshot timer. Reconnecting rechecks state to cover disconnected changes. */
|
||
|
|
export function subscribeAgentUpdates(listener: () => void, onError: (error: unknown) => void) {
|
||
|
|
const abort = new AbortController();
|
||
|
|
let timer: ReturnType<typeof setTimeout> | undefined;
|
||
|
|
const changed = () => {
|
||
|
|
timer ??= setTimeout(() => {
|
||
|
|
timer = undefined;
|
||
|
|
listener();
|
||
|
|
}, 100);
|
||
|
|
};
|
||
|
|
void (async () => {
|
||
|
|
while (!abort.signal.aborted) {
|
||
|
|
try {
|
||
|
|
const response = await fetch("/api/agent/updates", { credentials: "same-origin", signal: abort.signal });
|
||
|
|
if (!response.ok || !response.body) throw Error("Agent updates unavailable");
|
||
|
|
const reader = response.body.getReader(),
|
||
|
|
decoder = new TextDecoder();
|
||
|
|
let buffer = "";
|
||
|
|
try {
|
||
|
|
while (!abort.signal.aborted) {
|
||
|
|
const chunk = await reader.read();
|
||
|
|
if (chunk.done) break;
|
||
|
|
buffer += decoder.decode(chunk.value, { stream: true });
|
||
|
|
if (buffer.length > 65536) throw Error("Agent update frame exceeds limit");
|
||
|
|
let end: number;
|
||
|
|
while ((end = buffer.indexOf("\n")) !== -1) {
|
||
|
|
const line = buffer.slice(0, end);
|
||
|
|
buffer = buffer.slice(end + 1);
|
||
|
|
if (line && JSON.parse(line).type === "changed") changed();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} finally {
|
||
|
|
await reader.cancel().catch(() => {});
|
||
|
|
reader.releaseLock();
|
||
|
|
}
|
||
|
|
} catch (error) {
|
||
|
|
if (!abort.signal.aborted) onError(error);
|
||
|
|
}
|
||
|
|
if (!abort.signal.aborted)
|
||
|
|
await new Promise<void>((resolve) => {
|
||
|
|
const finish = () => {
|
||
|
|
clearTimeout(retry);
|
||
|
|
abort.signal.removeEventListener("abort", finish);
|
||
|
|
resolve();
|
||
|
|
};
|
||
|
|
const retry = setTimeout(finish, 1000);
|
||
|
|
abort.signal.addEventListener("abort", finish, { once: true });
|
||
|
|
if (abort.signal.aborted) finish();
|
||
|
|
});
|
||
|
|
}
|
||
|
|
})();
|
||
|
|
return () => {
|
||
|
|
abort.abort();
|
||
|
|
clearTimeout(timer);
|
||
|
|
};
|
||
|
|
}
|
||
|
|
|
||
|
|
/** A browser observer owns no executor. Closing it only stops reads. */
|
||
|
|
export class HttpThreadController implements ThreadController {
|
||
|
|
private state: ThreadState = {
|
||
|
|
items: [],
|
||
|
|
requests: [],
|
||
|
|
maintenance: false,
|
||
|
|
connected: false,
|
||
|
|
loading: true,
|
||
|
|
error: "",
|
||
|
|
};
|
||
|
|
private listeners = new Set<() => void>();
|
||
|
|
private items = new Map<string, Item>();
|
||
|
|
private requests = new Map<string, Request>();
|
||
|
|
private aliases = new Map<string, Ref>();
|
||
|
|
private abort = new AbortController();
|
||
|
|
private timer?: ReturnType<typeof setTimeout>;
|
||
|
|
private refreshNeeded = false;
|
||
|
|
private refreshing?: Promise<void>;
|
||
|
|
private closed = false;
|
||
|
|
constructor(
|
||
|
|
readonly thread: string,
|
||
|
|
initial?: Thread,
|
||
|
|
) {
|
||
|
|
if (initial) this.ingest({ data: { root: initial } });
|
||
|
|
void this.refresh()
|
||
|
|
.catch((error) => this.patch({ error: String(error), loading: false }))
|
||
|
|
.finally(() => this.stream());
|
||
|
|
}
|
||
|
|
getSnapshot = () => this.state;
|
||
|
|
subscribe = (callback: () => void) => {
|
||
|
|
this.listeners.add(callback);
|
||
|
|
return () => {
|
||
|
|
this.listeners.delete(callback);
|
||
|
|
};
|
||
|
|
};
|
||
|
|
private patch(value: Partial<ThreadState>) {
|
||
|
|
if (this.closed) return;
|
||
|
|
this.state = { ...this.state, ...value };
|
||
|
|
for (const listener of this.listeners) listener();
|
||
|
|
}
|
||
|
|
private request(value: Request) {
|
||
|
|
const object = value._camino.ref.object,
|
||
|
|
old = this.requests.get(object);
|
||
|
|
const progress =
|
||
|
|
old &&
|
||
|
|
((terminal.has(old.progress.status) && !terminal.has(value.progress.status)) ||
|
||
|
|
value.progress.generation < old.progress.generation)
|
||
|
|
? old.progress
|
||
|
|
: value.progress;
|
||
|
|
this.requests.set(object, { ...old, ...value, progress });
|
||
|
|
}
|
||
|
|
private item(value: Item) {
|
||
|
|
if (value.data.kind === "user")
|
||
|
|
value = {
|
||
|
|
...value,
|
||
|
|
request: {
|
||
|
|
...value.request,
|
||
|
|
admission: {
|
||
|
|
createdAt: value.data.createdAt,
|
||
|
|
sourceKind: "human",
|
||
|
|
sourceLabel: "User",
|
||
|
|
...value.request.admission,
|
||
|
|
text: value.data.text,
|
||
|
|
},
|
||
|
|
},
|
||
|
|
};
|
||
|
|
const replacement = this.aliases.get(value.request._camino.ref.object);
|
||
|
|
if (replacement) value = { ...value, request: { ...value.request, _camino: { ref: replacement } } };
|
||
|
|
this.request(value.request);
|
||
|
|
const key = itemKey(value),
|
||
|
|
previous = this.items.get(key);
|
||
|
|
if (!previous || BigInt(value.data.version) >= BigInt(previous.data.version))
|
||
|
|
this.items.set(key, { ...previous, ...value });
|
||
|
|
}
|
||
|
|
private publish() {
|
||
|
|
const items = [...this.items.values()].map((item) => ({
|
||
|
|
...item,
|
||
|
|
request: this.requests.get(item.request._camino.ref.object) ?? item.request,
|
||
|
|
}));
|
||
|
|
items.sort((a, b) => {
|
||
|
|
const ar = a.request.admission,
|
||
|
|
br = b.request.admission;
|
||
|
|
if (a.request._camino.ref.object !== b.request._camino.ref.object) {
|
||
|
|
if (ar?.sequence && br?.sequence && ar.sequence !== "0" && br.sequence !== "0")
|
||
|
|
return BigInt(ar.sequence) < BigInt(br.sequence) ? -1 : 1;
|
||
|
|
return (ar?.createdAt ?? a.data.createdAt).localeCompare(br?.createdAt ?? b.data.createdAt);
|
||
|
|
}
|
||
|
|
if (a.data.key === "input") return -1;
|
||
|
|
if (b.data.key === "input") return 1;
|
||
|
|
if (a.data.sequence !== "0" && b.data.sequence !== "0")
|
||
|
|
return BigInt(a.data.sequence) < BigInt(b.data.sequence) ? -1 : 1;
|
||
|
|
return a.data.createdAt.localeCompare(b.data.createdAt);
|
||
|
|
});
|
||
|
|
this.patch({ items, requests: [...this.requests.values()] });
|
||
|
|
}
|
||
|
|
private ingest(value: any, older = false) {
|
||
|
|
for (const repair of value.repair ?? []) {
|
||
|
|
if (repair.graphRequest) {
|
||
|
|
this.aliases.set(repair.request.object, repair.graphRequest);
|
||
|
|
const previous = this.requests.get(repair.request.object);
|
||
|
|
if (previous) {
|
||
|
|
this.requests.delete(repair.request.object);
|
||
|
|
this.request({ ...previous, _camino: { ref: repair.graphRequest }, pendingGraph: false });
|
||
|
|
}
|
||
|
|
for (const [key, item] of this.items)
|
||
|
|
if (item.request._camino.ref.object === repair.request.object) {
|
||
|
|
this.items.delete(key);
|
||
|
|
this.item({
|
||
|
|
...item,
|
||
|
|
request: { ...item.request, _camino: { ref: repair.graphRequest }, pendingGraph: false },
|
||
|
|
});
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
const request: Request = {
|
||
|
|
_camino: { ref: repair.request },
|
||
|
|
repair: repair.key,
|
||
|
|
pendingGraph: true,
|
||
|
|
admission: {
|
||
|
|
text: repair.text,
|
||
|
|
createdAt: repair.createdAt,
|
||
|
|
sourceKind: "human",
|
||
|
|
sourceLabel: "Source repair",
|
||
|
|
},
|
||
|
|
progress: {
|
||
|
|
status: "pending_graph",
|
||
|
|
attempt: "",
|
||
|
|
generation: 0,
|
||
|
|
reason: "Awaiting graph recording",
|
||
|
|
cancelRequested: false,
|
||
|
|
},
|
||
|
|
};
|
||
|
|
this.item({
|
||
|
|
request,
|
||
|
|
live: true,
|
||
|
|
data: {
|
||
|
|
key: "input",
|
||
|
|
kind: "user",
|
||
|
|
text: repair.text,
|
||
|
|
phase: "input",
|
||
|
|
activity: null,
|
||
|
|
presentation: null,
|
||
|
|
sequence: "0",
|
||
|
|
version: "0",
|
||
|
|
createdAt: repair.createdAt,
|
||
|
|
updatedAt: repair.createdAt,
|
||
|
|
},
|
||
|
|
});
|
||
|
|
}
|
||
|
|
}
|
||
|
|
const root: Thread | undefined = value.data?.root;
|
||
|
|
if (root) {
|
||
|
|
if (root.active) this.request(root.active);
|
||
|
|
for (const row of root.requests?.entries ?? []) this.request(row.node);
|
||
|
|
for (const row of root.items?.entries ?? []) this.item({ ...row.node, live: false });
|
||
|
|
this.patch({
|
||
|
|
thread: root,
|
||
|
|
pageInfo: older ? root.items?.pageInfo : (root.items?.pageInfo ?? this.state.pageInfo),
|
||
|
|
});
|
||
|
|
}
|
||
|
|
for (const row of value.requests ?? []) {
|
||
|
|
const request: Request = {
|
||
|
|
_camino: { ref: row.request },
|
||
|
|
admission: row.admission,
|
||
|
|
progress: {
|
||
|
|
...row.progress,
|
||
|
|
status: row.phase === "projected" ? row.progress.status : terminal.has(row.phase) ? "recording" : row.phase,
|
||
|
|
attempt: row.attempt ?? row.progress?.attempt ?? "",
|
||
|
|
generation: row.generation ?? row.progress?.generation ?? 0,
|
||
|
|
reason: row.reason ?? "",
|
||
|
|
cancelRequested: row.cancelRequested ?? false,
|
||
|
|
},
|
||
|
|
pendingGraph: row.pendingGraph,
|
||
|
|
repair: row.repair,
|
||
|
|
};
|
||
|
|
this.request(request);
|
||
|
|
if (request.admission?.text && !this.items.has(`${request._camino.ref.object}/input`))
|
||
|
|
this.item({
|
||
|
|
request,
|
||
|
|
live: true,
|
||
|
|
data: {
|
||
|
|
key: "input",
|
||
|
|
kind: "user",
|
||
|
|
text: request.admission.text,
|
||
|
|
phase: "input",
|
||
|
|
activity: null,
|
||
|
|
presentation: null,
|
||
|
|
sequence: "0",
|
||
|
|
version: "0",
|
||
|
|
createdAt: request.admission.createdAt,
|
||
|
|
updatedAt: request.admission.createdAt,
|
||
|
|
},
|
||
|
|
});
|
||
|
|
}
|
||
|
|
for (const live of value.live ?? []) {
|
||
|
|
const replacement = this.aliases.get(live.request.object) ?? live.request;
|
||
|
|
const request = this.requests.get(replacement.object) ?? {
|
||
|
|
_camino: { ref: replacement },
|
||
|
|
progress: {
|
||
|
|
status: "running",
|
||
|
|
attempt: live.attempt,
|
||
|
|
generation: live.generation,
|
||
|
|
reason: "",
|
||
|
|
cancelRequested: false,
|
||
|
|
},
|
||
|
|
};
|
||
|
|
if (
|
||
|
|
!this.items.has(`${replacement.object}/${live.data.key}`) ||
|
||
|
|
BigInt(live.data.version) > BigInt(this.items.get(`${replacement.object}/${live.data.key}`)!.data.version)
|
||
|
|
)
|
||
|
|
this.item({ request, data: live.data, live: true });
|
||
|
|
}
|
||
|
|
this.patch({
|
||
|
|
maintenance: value.maintenance === true,
|
||
|
|
loading: false,
|
||
|
|
error: "",
|
||
|
|
...(!older && value.cursor ? { cursor: value.cursor } : {}),
|
||
|
|
});
|
||
|
|
this.publish();
|
||
|
|
}
|
||
|
|
refresh = (older = false): Promise<void> => {
|
||
|
|
if (!older && this.refreshing) return this.refreshing;
|
||
|
|
const after = older ? this.state.pageInfo?.endCursor : null;
|
||
|
|
const task = api(
|
||
|
|
`/thread?thread=${encodeURIComponent(this.thread)}&first=30${after ? `&after=${encodeURIComponent(after)}` : ""}`,
|
||
|
|
undefined,
|
||
|
|
this.abort.signal,
|
||
|
|
)
|
||
|
|
.then((value) => this.ingest(value, older))
|
||
|
|
.catch((error) => {
|
||
|
|
if (!this.closed) this.patch({ error: String(error), loading: false });
|
||
|
|
throw error;
|
||
|
|
})
|
||
|
|
.finally(() => {
|
||
|
|
if (!older) this.refreshing = undefined;
|
||
|
|
});
|
||
|
|
if (!older) this.refreshing = task;
|
||
|
|
return task;
|
||
|
|
};
|
||
|
|
private scheduleRefresh() {
|
||
|
|
this.refreshNeeded = true;
|
||
|
|
if (this.closed || this.timer) return;
|
||
|
|
this.timer = setTimeout(() => {
|
||
|
|
this.timer = undefined;
|
||
|
|
this.refreshNeeded = false;
|
||
|
|
void (async () => {
|
||
|
|
// A notification arriving during a previous snapshot needs a subsequent
|
||
|
|
// read, not merely the same in-flight promise.
|
||
|
|
await this.refreshing?.catch(() => {});
|
||
|
|
await this.refresh().catch(() => {});
|
||
|
|
if (this.refreshNeeded) this.scheduleRefresh();
|
||
|
|
})();
|
||
|
|
}, 100);
|
||
|
|
}
|
||
|
|
private async stream() {
|
||
|
|
while (!this.closed) {
|
||
|
|
try {
|
||
|
|
const cursor = this.state.cursor;
|
||
|
|
const response = await fetch(
|
||
|
|
`/api/agent/events?thread=${encodeURIComponent(this.thread)}&after=${cursor?.sequence ?? 0}${cursor?.epoch ? `&epoch=${encodeURIComponent(cursor.epoch)}` : ""}`,
|
||
|
|
{ credentials: "same-origin", signal: this.abort.signal },
|
||
|
|
);
|
||
|
|
if (!response.ok || !response.body) throw Error("Live Agent stream unavailable");
|
||
|
|
this.patch({ connected: true });
|
||
|
|
this.scheduleRefresh();
|
||
|
|
const reader = response.body.getReader(),
|
||
|
|
decoder = new TextDecoder();
|
||
|
|
let buffer = "";
|
||
|
|
try {
|
||
|
|
while (!this.closed) {
|
||
|
|
const chunk = await reader.read();
|
||
|
|
if (chunk.done) break;
|
||
|
|
buffer += decoder.decode(chunk.value, { stream: true });
|
||
|
|
if (buffer.length > 1024 * 1024) throw Error("Agent stream frame exceeds limit");
|
||
|
|
let end: number;
|
||
|
|
while ((end = buffer.indexOf("\n")) !== -1) {
|
||
|
|
const line = buffer.slice(0, end);
|
||
|
|
buffer = buffer.slice(end + 1);
|
||
|
|
if (!line) continue;
|
||
|
|
const event = JSON.parse(line);
|
||
|
|
if (event.type === "graph") {
|
||
|
|
this.scheduleRefresh();
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
if (event.type === "reset") {
|
||
|
|
await this.refresh();
|
||
|
|
break;
|
||
|
|
}
|
||
|
|
if (event.sequence) this.patch({ cursor: { epoch: event.epoch, sequence: event.sequence } });
|
||
|
|
const ref: Ref | undefined = event.request;
|
||
|
|
if (event.kind === "item" && ref) {
|
||
|
|
const request = this.requests.get(ref.object) ?? {
|
||
|
|
_camino: { ref },
|
||
|
|
progress: {
|
||
|
|
status: "running",
|
||
|
|
attempt: event.attempt,
|
||
|
|
generation: event.generation,
|
||
|
|
reason: "",
|
||
|
|
cancelRequested: false,
|
||
|
|
},
|
||
|
|
};
|
||
|
|
this.item({ request, data: event.item, live: true });
|
||
|
|
this.publish();
|
||
|
|
} else if (ref && ["request", "execution"].includes(event.kind)) {
|
||
|
|
const previous = this.requests.get(ref.object);
|
||
|
|
if (previous)
|
||
|
|
this.request({
|
||
|
|
...previous,
|
||
|
|
progress: {
|
||
|
|
...previous.progress,
|
||
|
|
status: event.kind === "execution" && terminal.has(event.status) ? "recording" : event.status,
|
||
|
|
attempt: event.attempt,
|
||
|
|
generation: event.generation,
|
||
|
|
...(event.kind === "request" ? { result: event.result, reason: event.reason ?? "" } : {}),
|
||
|
|
},
|
||
|
|
});
|
||
|
|
this.publish();
|
||
|
|
void this.refresh().catch(() => {});
|
||
|
|
} else if (event.kind === "reconciled") {
|
||
|
|
await this.refresh();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
} finally {
|
||
|
|
await reader.cancel().catch(() => {});
|
||
|
|
reader.releaseLock();
|
||
|
|
this.patch({ connected: false });
|
||
|
|
}
|
||
|
|
} catch (error) {
|
||
|
|
if (!this.closed) this.patch({ connected: false });
|
||
|
|
}
|
||
|
|
if (!this.closed) await new Promise((resolve) => setTimeout(resolve, 750));
|
||
|
|
}
|
||
|
|
}
|
||
|
|
submit = async (
|
||
|
|
text: string,
|
||
|
|
options: { profile?: string; response?: unknown; repair?: boolean; retryOf?: string } = {},
|
||
|
|
) => {
|
||
|
|
const key = `${draftKey(this.thread)}:pending`,
|
||
|
|
previous = sessionStorage.getItem(key),
|
||
|
|
input = previous
|
||
|
|
? JSON.parse(previous)
|
||
|
|
: {
|
||
|
|
operationId: crypto.randomUUID(),
|
||
|
|
createdAt: new Date().toISOString(),
|
||
|
|
thread: this.thread,
|
||
|
|
text,
|
||
|
|
...options,
|
||
|
|
};
|
||
|
|
if (previous && input.text !== text)
|
||
|
|
throw Error("Recover the pending submission before sending a different message");
|
||
|
|
sessionStorage.setItem(key, JSON.stringify(input));
|
||
|
|
let value;
|
||
|
|
try {
|
||
|
|
value = await api(input.repair ? "/repair" : "/submit", input, this.abort.signal);
|
||
|
|
} catch (error) {
|
||
|
|
if ((error as { knownRejected?: boolean }).knownRejected) sessionStorage.removeItem(key);
|
||
|
|
throw error;
|
||
|
|
}
|
||
|
|
sessionStorage.removeItem(key);
|
||
|
|
if (value.repair) this.ingest({ repair: [value.repair] });
|
||
|
|
else if (value.admission.accepted && value.admission.request) {
|
||
|
|
const request: Request = {
|
||
|
|
_camino: { ref: value.admission.request },
|
||
|
|
admission: { text, createdAt: input.createdAt, sourceKind: "human", sourceLabel: "User" },
|
||
|
|
progress: { status: "queued", attempt: "", generation: 0, reason: "", cancelRequested: false },
|
||
|
|
};
|
||
|
|
this.item({
|
||
|
|
request,
|
||
|
|
data: {
|
||
|
|
key: "input",
|
||
|
|
kind: "user",
|
||
|
|
text,
|
||
|
|
phase: "input",
|
||
|
|
activity: null,
|
||
|
|
presentation: null,
|
||
|
|
sequence: "0",
|
||
|
|
version: "0",
|
||
|
|
createdAt: input.createdAt,
|
||
|
|
updatedAt: input.createdAt,
|
||
|
|
},
|
||
|
|
});
|
||
|
|
this.publish();
|
||
|
|
}
|
||
|
|
await this.refresh().catch(() => {});
|
||
|
|
};
|
||
|
|
control = async (input: Record<string, unknown>) => {
|
||
|
|
await control(input);
|
||
|
|
await this.refresh();
|
||
|
|
};
|
||
|
|
details = (request: string) => api(`/request?request=${encodeURIComponent(request)}`);
|
||
|
|
close = () => {
|
||
|
|
this.closed = true;
|
||
|
|
this.abort.abort();
|
||
|
|
clearTimeout(this.timer);
|
||
|
|
this.listeners.clear();
|
||
|
|
};
|
||
|
|
}
|