Export template source from ca5b05584b89e90cb259b4596b1bc01d3d692136:quixos-instance/packages/agent-presentation
This commit is contained in:
Quixos Template Publisher
+507
@@ -0,0 +1,507 @@
|
||||
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();
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user