Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 25 additions & 20 deletions backend/src/modules/assistant-workspace/page-review.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,44 +87,49 @@ export async function readAssistantPageReview(
AND comment.body ~* '(^|[[:space:]])@orbyn\\y'
AND (comment.parent_id IS NULL OR EXISTS(SELECT 1 FROM doc_comments parent WHERE parent.id=comment.parent_id
AND parent.doc_id=comment.doc_id AND parent.resolved_at IS NULL AND NOT parent.detached))))`;
const counts = (
await db.query<AssistantProfileCounts>(
`WITH eligible AS (${eligible}) SELECT
count(*) FILTER(WHERE state='running' AND lease_expires_at>$5)::int AS working,
count(*) FILTER(WHERE state='waiting')::int AS waiting,
count(*) FILTER(WHERE state='queued')::int AS queued,
count(*) FILTER(WHERE state='running' AND (lease_expires_at IS NULL OR lease_expires_at<=$5))::int AS recovering FROM eligible`,
params.values,
)
).rows[0];
type Row = {
id: string;
state: AssistantPageReviewRun["state"];
lease_expires_at: Date | null;
updated_at: Date;
lease_expires_at: string | null;
updated_at: string;
doc_id: string;
title: string;
work_kind: AssistantPageReviewRun["work_kind"];
origin: AssistantPageReviewRun["origin"];
};
const rows = (
await db.query<Row>(
`WITH eligible AS (${eligible}),active AS (
// Authorize the complete set once, before either counts or display limits.
const result = (
await db.query<{
counts: AssistantProfileCounts;
runs: Row[] | null;
}>(
`WITH eligible AS MATERIALIZED (${eligible}),active AS (
SELECT * FROM eligible WHERE state IN ('queued','running','waiting')
ORDER BY CASE WHEN state='waiting' THEN 0 WHEN state='running' THEN 1 ELSE 2 END,updated_at DESC,id LIMIT 21
),terminal AS (SELECT * FROM eligible WHERE state IN ('done','failed','cancelled') ORDER BY updated_at DESC,id LIMIT 6)
SELECT * FROM active UNION ALL SELECT * FROM terminal`,
SELECT
(SELECT jsonb_build_object(
'working',count(*) FILTER(WHERE state='running' AND lease_expires_at>$5)::int,
'waiting',count(*) FILTER(WHERE state='waiting')::int,
'queued',count(*) FILTER(WHERE state='queued')::int,
'recovering',count(*) FILTER(WHERE state='running' AND (lease_expires_at IS NULL OR lease_expires_at<=$5))::int
) FROM eligible) AS counts,
(SELECT jsonb_agg(to_jsonb(selected) ORDER BY
CASE selected.state WHEN 'waiting' THEN 0 WHEN 'running' THEN 1 WHEN 'queued' THEN 2 ELSE 3 END,
selected.updated_at DESC,selected.id)
FROM (SELECT * FROM active UNION ALL SELECT * FROM terminal) selected) AS runs`,
params.values,
)
).rows;
).rows[0];
const rows = result.runs ?? [];
const active = rows.filter((r) =>
["queued", "running", "waiting"].includes(r.state),
),
terminal = rows.filter(
(r) => !["queued", "running", "waiting"].includes(r.state),
);
return {
counts,
counts: result.counts,
truncated: active.length > 20 || terminal.length > 5,
runs: [...active.slice(0, 20), ...terminal.slice(0, 5)].map((r) => ({
id: r.id,
Expand All @@ -134,8 +139,8 @@ export async function readAssistantPageReview(
state: r.state,
recovering:
r.state === "running" &&
(!r.lease_expires_at || r.lease_expires_at <= now),
updated_at: r.updated_at.toISOString(),
(!r.lease_expires_at || new Date(r.lease_expires_at) <= now),
updated_at: new Date(r.updated_at).toISOString(),
work_kind: r.work_kind,
})),
};
Expand Down
64 changes: 64 additions & 0 deletions backend/tests/assistant-page-review.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -254,6 +254,70 @@ test("twenty active plus five terminal page rows do not truncate each other and
assert.equal(page.profiles[0].page_runs_truncated, true);
assert.ok(!page.profiles[0].page_runs!.some((r) => r.id === hidden.run.id));
});
test("one eligibility statement retains complete counts and deterministic tied row ordering", async () => {
const p = await person();
const expected: { id: string; state: string }[] = [];
for (const state of ["waiting", "running", "queued", "done"] as const) {
const count = state === "done" ? 7 : 5;
for (let index = 0; index < count; index++) {
const value = await pageRun(p, {
state,
expired: state === "running" && index < 2,
});
expected.push({ id: value.run.id, state });
}
}
await pool.query(
"UPDATE assistant_page_runs SET updated_at='2026-10-30T12:00:00.123456Z' WHERE user_id=$1",
[p.id],
);
const statements: string[] = [];
const result = await readTransaction(
(db) =>
readAssistantPageReview(
{
query: ((...args: unknown[]) => {
statements.push(String(args[0]));
return Reflect.apply(db.query, db, args);
}) as typeof db.query,
},
p.id,
"background",
new Date(),
),
{ primary: true, isolation: "repeatable read", jit: false },
);
assert.equal(
statements.filter((sql) => sql.includes("FROM assistant_page_runs r"))
.length,
1,
"counts and selected rows share one eligibility evaluation",
);
assert.deepEqual(result.counts, {
working: 3,
waiting: 5,
queued: 5,
recovering: 2,
});
const ids = (state: string) =>
expected
.filter((row) => row.state === state)
.map((row) => row.id)
.sort();
assert.deepEqual(
result.runs.map((row) => row.id),
[
...ids("waiting"),
...ids("running"),
...ids("queued"),
...ids("done").slice(0, 5),
],
);
assert.equal(result.truncated, true);
assert.ok(
result.runs.every((row) => row.updated_at === "2026-10-30T12:00:00.123Z"),
);
});
test("current project and grant/rule/privacy authority control titles and counts without effect-guard locks", async () => {
const p = await person(),
pr = (
Expand Down
34 changes: 34 additions & 0 deletions backend/tests/client-refresh.unit.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -448,3 +448,37 @@ test("shared 204 replies retain undefined and settle cleanly", async () => {
assert.deepEqual(await replies, [undefined, undefined]);
assert.equal(count, 1);
});

// React Native supplies the abort-controller signal, without the newer browser
// convenience method or a reason. Exercise that shape without replacing globals.
test("legacy native signals can read and cancel without throwIfAborted or reason", async () => {
const hold = gate();
const stop = new AbortController();
Object.defineProperty(stop.signal, "throwIfAborted", { value: undefined });
Object.defineProperty(stop.signal, "reason", { value: undefined });
let count = 0;
const api = client(async () => {
count++;
if (count === 1) return Response.json({ read: true });
await hold.waiting;
return Response.json({ read: true });
});
assert.deepEqual(await api.request("/me", { signal: stop.signal }), {
read: true,
});
const cancelled = api.request("/me", { signal: stop.signal });
const failure = assert.rejects(cancelled, { name: "AbortError" });
await tick();
stop.abort();
await failure;
await assert.rejects(api.request("/me", { signal: stop.signal }), {
name: "AbortError",
});
assert.equal(
count,
2,
"an already cancelled native reader must not dispatch",
);
hold.release();
await tick();
});
9 changes: 5 additions & 4 deletions packages/api-client/src/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@ import {
AssistantRunProgress,
DocNews,
LiveNews,
abortError,
MAX_CACHED,
OrbynClientOptions,
pause,
Expand Down Expand Up @@ -317,7 +318,7 @@ export class OrbynClient {
// A read begun after a mutation must not join an older in-flight read.
const epoch = method === "GET" ? this.readEpoch : ++this.readEpoch;
const token = options.anonymous ? null : await this.getToken();
options.signal?.throwIfAborted();
if (options.signal?.aborted) throw options.signal.reason ?? abortError();
const share =
method === "GET" &&
!options.headers &&
Expand Down Expand Up @@ -371,7 +372,7 @@ export class OrbynClient {
return true;
};
const cancel = () => {
if (release()) reject(options.signal!.reason);
if (release()) reject(options.signal!.reason ?? abortError());
};
options.signal?.addEventListener("abort", cancel, { once: true });
if (options.signal?.aborted) cancel();
Expand Down Expand Up @@ -481,7 +482,7 @@ export class OrbynClient {
response = await send();
}
}
options.signal?.throwIfAborted();
if (options.signal?.aborted) throw options.signal.reason ?? abortError();
const noStore = response.headers
.get("Cache-Control")
?.split(",")
Expand Down Expand Up @@ -535,7 +536,7 @@ export class OrbynClient {
}
if (response.status === 204) return undefined as T;
const text = await response.text();
options.signal?.throwIfAborted();
if (options.signal?.aborted) throw options.signal.reason ?? abortError();
const etag = response.headers.get("ETag");
if (retain && etag) {
this.cache.delete(key);
Expand Down
Loading