perf(inbox): combine startup reads and reduce refetch pressure (#18790)

## Thinking Path

> - Paperclip helps people manage AI agents and their work.
> - Mine shows the work and failures that need the viewing user's
attention.
> - A reload waits for account access, task rows, parent captions and
failure state.
> - Those reads form several browser request stages before useful rows
appear.
> - Live invalidations can also restart the same task-list request more
than once.
> - This change supplies startup data together and removes redundant
refresh work.
> - It preserves current authority, failure presentation, archive state
and row stability.

## Linked Issues or Issue Description

Refs #18645. Related: #18642. This change implements the read-receipt
behavior independently on current master. It does not use that PR's
branch or patch.

**What happened?**

Mine starts its primary request after the session, company and
board-access reads. It then requests parent captions and rich failure
context. Four observed loads spent another 0.45–0.70 seconds after the
primary rows arrived. One slower load waited 1.71 seconds for parent
context. The same run observed two HTTP 429 responses. These timings
identify request waits; they do not attribute the delay to SQL or Node
CPU.

Read receipts also invalidate general task collections and dashboards.
Overlapping parent and child invalidations can start two requests and
abort the first.

**Expected behavior**

The initial page should use one account-safe startup projection and one
complete Mine response. A read receipt should refresh the affected
viewer's inbox state. Each invalidation batch should refresh a covered
query once. Required captions and current failure state must be ready
before first reveal. Existing rows should remain usable during a
transient refresh error.

**Steps to reproduce**

1. Open a populated Mine inbox with parent tasks outside the visible
index and a current failed run.
2. Reload with browser HTTP cache retained and application query state
empty.
3. Inspect the session, company, board access, task-index and context
request sequence.
4. Open or mark a task read and inspect the unrelated collection
refetches.

**Paperclip version or commit**

Baseline: `44a9a043b80c18314d8dd89556f444dae696d3d6`.

**Deployment mode**

Authenticated board UI. Local proof uses the real APIs and database with
a synthetic session resolver. No hosted candidate deployment is claimed.

## What Changed

- Add optional, same-user `boardBootstrap` data to the existing session
response. Seed account-keyed company and board-access reads from it.
Keep ordinary endpoint reads for fallback, retry and live
reconciliation. A returning account cannot reuse revoked cached access.
Cache collection and later real refetches remain effective.
- Add the board-only Mine snapshot endpoint. Reuse the canonical issue
selector, private-task filtering, run-read decisions, secret redaction
and interruption/recovery presentation. Assemble the fixed 100-task
page, parent captions and bounded 200-run failure window on the server.
Start independent reads together. Optional source errors keep successful
task rows usable.
- Keep array-shaped source caches for archive and undo. Key Mine source,
context and reveal state by account. Validate the response's user stamp
before committing it. Remove array-only cross-tab Mine broadcasts, which
cannot carry its paired failure context. The server retains
authenticated in-flight coalescing.
- Scope read-receipt invalidation to the affected viewer's inbox, badges
and chat unread state. Keep company and task audit views fresh. Preserve
genuine task mutations, run updates and subscription-gap reconciliation.
- Remove covered scalar-prefix invalidations within a batch. Preserve
predicates, exact matches and different query/refetch constraints. A
later event can still refresh again.
- Retire Mine in-flight reuse after local company activity or run
lifecycle changes. A post-archive read starts fresh. Old work keeps its
admission slot until it settles, even if every HTTP client aborts. The
eight-compute limit stays in place.
- Add API, component and browser regressions. Extend the production
timing harness with matching API/UI artifact receipts, retained-cache
reloads, request status counts, layout checks and optional controlled
network latency. Update API and verification documentation.

## Verification

- Full `pnpm -r typecheck` passes.
- Full `pnpm build` passes.
- Token gates, map checks and diff whitespace checks pass.
- 72 Inbox component tests pass. Startup-specific checks pass, including
account switching, revoked cached access, malformed/absent bootstrap
data and cache collection. The canonical authenticated bootstrap route
also passes against real PostgreSQL.
- Six real PostgreSQL Mine route cases pass, including full equivalence
with legacy authorized projections, privacy, fixed bounds, membership
revocation, coalescing, archive freshness and aborted-client admission.
Four focused legacy coordinator controls pass. Six assembly tests pass.
- Refetch and invalidation regressions pass. Meaningful pre-fix tests
observed two request starts and one abort for overlapping invalidations;
the fix produces one start and no abort. The archive race test observed
the post-archive GET joining the old body before the fix.
- All 11 existing startup browser checks pass, including Tasks/Mine code
loading, desktop/mobile bounds, non-streamlined Coordinator, task
navigation, failed preloading and Back position. The nine focused
startup, fallback, partial-source error, rate-limit recovery,
archive/undo and geometry cases also pass. The 429 cases use the
inherited retry window. Archive and keyboard undo are corroborated
through the real API and reload. Authentication and failure fault
injection in these cases are explicitly simulated.
- Matched production before/after timings are complete. Full aggregate
Vitest coverage is left to CI. At 15:15 UTC, the final-head runner gate
and `review` workflow pass. The 45 build/test/policy jobs are queued;
the Build job has no `ubuntu-latest` runner assigned and has executed no
steps, so remote aggregate results remain unverified. Paperclip Review
and Superagent Security Scan are in progress. Superseded runs for this
PR were cancelled. No current-head check has reported a failure; the CI
checklist remains open. The remaining step is runner allocation and
completion of those external checks; persistent queueing needs the
repository CI owner's attention.
- Greptile completed on final head
`ecde295565309275c4f2fae903caf6892f1eaafd` at 5/5. The map-scope finding
is fixed and resolved; there are no open review threads.

### Matched browser timings

Five alternating baseline/candidate pairs per network scenario, measured
serially
in headful Chrome 156.0.8078.80 on Apple silicon with Node 24.22.2 and a
1440×900
viewport. Both production UIs use their matching real API revision
against one
fixed disposable PostgreSQL fixture: 34 tasks, 31 visible task rows, one
paused
agent and one terminal failure. No active agent/provider work. HTTP
cache is
retained; each initial sample reloads the document and clears
application query
state. Warm navigation and Back retain query state. CPU and throughput
are native.
The 100ms scenario uses CDP minimum request-to-response-header latency
without
request interception. It is a controlled request-dependency experiment,
not a
staging reproduction.

Values are median milliseconds with min–max in parentheses. Paired
saving is
baseline minus candidate for each alternating pair; its median differs
from the
difference of arm medians. Positive saving favors the candidate.

| Network / journey | Baseline | Candidate | Median paired saving |
| --- | ---: | ---: | ---: |
| Native / initial rows | 187.6 (128.9–208.5) | 191.7 (130.3–210.9) |
−23.3 |
| Native / Recent → Mine | 100.17 (99.90–101.87) | 100.07 (99.88–101.37)
| −0.16 |
| Native / task → Back | 70.45 (56.63–72.02) | 66.64 (56.52–67.71) |
+4.94 |
| 100ms / initial rows | 930.3 (914.1–1015.7) | 799.0 (697.1–826.0) |
+211.8 |
| 100ms / Recent → Mine | 100.09 (97.73–100.16) | 100.18 (96.65–100.98)
| −0.02 |
| 100ms / task → Back | 74.97 (62.77–75.82) | 75.47 (61.51–76.19) |
+0.35 |

The controlled-latency initial pairs all improve, saving 108.3–217.0ms.
Native
initial pairs have mixed signs (−67.5 to +78.1ms), so they establish no
reliable
native page-speed improvement. Warm and Back timings are effectively
unchanged.
Median primary-request start is 110.4→135.2ms natively and 497.8→480.9ms
with
100ms latency; these runs do not establish a large startup-only timing
gain.

API requests through the two-second post-reveal window fall from median
32
(32–37) to 27 (25–28) natively, and 37 (33–38) to 30 (25–32) with 100ms
latency.
All 20 samples preserve the same ordered 31 task links, first-row DOM
identity
and geometry. All recorded API responses are HTTP 200; zero unfinished
requests,
429s or JavaScript errors. Lower 429 incidence on staging remains
unproven.

Measured source: baseline `44a9a043b80c18314d8dd89556f444dae696d3d6`,
candidate
`4c12b2ef0ecf0754676e99bd6b9399e84408646c`. Final head
`ecde295565309275c4f2fae903caf6892f1eaafd` changes only two feature-map
files
and existing browser test expectations since that measured candidate;
production executable inputs are identical to the measured candidate.
Asset hashes, API
health revision receipts and every raw sample were retained. These
results do
not claim hosted-page improvement or deployment of this PR.

## Risks

- Startup enrichment adds work to the session read. Failure of this
optional projection falls back to the ordinary company and access
endpoints. It does not invalidate a valid session or replace per-request
server authorization.
- Account changes, stale cache entries and same-company navigation can
expose old presentation data. Account keys, principal stamps, reveal
scopes and canonical bootstrap replacement have dedicated negative
cases. Permission failures hide cached task rows; transient network, 429
and 5xx failures preserve them.
- Current failure state can differ from the historical failed run. The
snapshot uses the existing canonical rich projection and verified
missing-ID coverage. Partial context failures retain an explicit retry
path.
- Mutation retirement follows events delivered to the current server
process. This adds no cross-replica invalidation protocol or distributed
read-after-write guarantee.
- New UI assets require the matching API. Older UI/session consumers
retain their existing response and endpoint behavior. There is no
database migration, provider call or deployment change.

## Model Used

OpenAI Codex, based on GPT-6, with reasoning, code execution, browser
automation and delegated source review. The exact runtime model ID and
context-window size are not exposed by this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-11-03 09:28:57 -06:00
1 parent 5dee4993f1
commit b40e6edaea
40 files changed
+2457 -350

No files matched your search

+9
View File
@@ -98,6 +98,15 @@ describe("auth routes", () => {
});
});
it("preserves the session when optional navigation bootstrap cannot load", async () => {
const app = createApp({ type: "board", userId: "user-1", source: "session", companyIds: [] }, baseUser);
// This profile-only database cannot execute the bootstrap's company join.
const response = await request(app).get("/api/auth/get-session");
expect(response.status).toBe(200);
expect(response.body.user.id).toBe("user-1");
expect(response.body.boardBootstrap).toBeUndefined();
});
it("sends sentryDsn for a board actor when SENTRY_DSN is set", async () => {
process.env.SENTRY_DSN = "https://public@o0.ingest.sentry.io/1";
const app = await createApp(
@@ -0,0 +1,49 @@
import { randomUUID } from "node:crypto";
import express from "express";
import request from "supertest";
import { eq } from "drizzle-orm";
import { authUsers, companies, companyMemberships, instanceUserRoles } from "@paperclipai/db";
import { readBoardBootstrap } from "@paperclipai/shared";
import { expect, it } from "vitest";
import { actorMiddleware } from "../middleware/auth.js";
import { errorHandler } from "../middleware/error-handler.js";
import { authRoutes } from "../routes/auth.js";
import { describeEmbeddedPostgres, useEmbeddedPostgres } from "./helpers/route-test-harness.js";
describeEmbeddedPostgres("authenticated board bootstrap", () => {
const ctx = useEmbeddedPostgres("board-bootstrap-");
it("uses the authenticated account and current memberships, including for instance admins", async () => {
const userId = `bootstrap-${randomUUID()}`, companyId = randomUUID(), foreignId = randomUUID();
const now = new Date();
await ctx.db.insert(authUsers).values({ id: userId, name: "Bootstrap user", email: `${userId}@example.com`, emailVerified: true, createdAt: now, updatedAt: now });
await ctx.db.insert(companies).values([
{ id: companyId, name: "Accessible", issuePrefix: `A${companyId.slice(0, 7)}` },
{ id: foreignId, name: "Directory only", issuePrefix: `B${foreignId.slice(0, 7)}` },
]);
const [membership] = await ctx.db.insert(companyMemberships).values({
companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active",
}).returning();
await ctx.db.insert(instanceUserRoles).values({ userId, role: "instance_admin" });
const app = express();
app.use(actorMiddleware(ctx.db, { deploymentMode: "authenticated", resolveSession: async () => ({
session: { id: "bootstrap-session", userId }, user: { id: userId, name: "Bootstrap user", email: `${userId}@example.com` },
}) }));
app.use("/api/auth", authRoutes(ctx.db));
app.use(errorHandler);
const first = await request(app).get(`/api/auth/get-session?userId=foreign-user&companyId=${foreignId}`).expect(200);
expect(first.body.user.id).toBe(userId);
expect(first.body.boardBootstrap.userId).toBe(userId);
expect(first.body.boardBootstrap.companies.map((value: { id: string }) => value.id)).toEqual([companyId]);
expect(first.body.boardBootstrap.boardAccess).toEqual({ userId, isInstanceAdmin: true, companyIds: [companyId] });
expect(readBoardBootstrap(first.body.boardBootstrap, userId)?.companies[0]?.id).toBe(companyId);
expect(first.headers["cache-control"]).toBe("private, no-store");
await ctx.db.update(companyMemberships).set({ status: "archived" }).where(eq(companyMemberships.id, membership!.id));
await ctx.db.delete(instanceUserRoles).where(eq(instanceUserRoles.userId, userId));
const revoked = await request(app).get("/api/auth/get-session").expect(200);
expect(revoked.body.boardBootstrap.companies).toEqual([]);
expect(revoked.body.boardBootstrap.boardAccess).toEqual({ userId, isInstanceAdmin: false, companyIds: [] });
});
});
@@ -0,0 +1,296 @@
import { randomUUID } from "node:crypto";
import express from "express";
import request from "supertest";
import { eq } from "drizzle-orm";
import { expect, it, vi } from "vitest";
import { agents, authUsers, companyMemberships, heartbeatRuns, issueInboxArchives, issues } from "@paperclipai/db";
import { INBOX_MINE_ISSUE_STATUS_FILTER, type CompactIssue, type InboxMineSnapshot } from "@paperclipai/shared";
import { actorMiddleware } from "../middleware/auth.js";
import { errorHandler } from "../middleware/error-handler.js";
import { agentRoutes } from "../routes/agents.js";
import { issueRoutes } from "../routes/issues.js";
import * as issueServices from "../services/issues.js";
import * as heartbeatServices from "../services/heartbeat.js";
import { describeEmbeddedPostgres, seedCompanyWithBoardAccess, useEmbeddedPostgres } from "./helpers/route-test-harness.js";
describeEmbeddedPostgres("Mine inbox snapshot routes", () => {
const ctx = useEmbeddedPostgres("inbox-mine-routes-");
async function seed(name: string) {
const fixture = await seedCompanyWithBoardAccess(ctx.db, name);
await ctx.db.insert(authUsers).values({
id: fixture.userId, name, email: `${fixture.userId}@example.test`,
emailVerified: true, createdAt: new Date(), updatedAt: new Date(),
});
return fixture;
}
function appFor(userId: string | null, options: Parameters<typeof issueRoutes>[2] = {}, actor?: Express.Request["actor"]) {
const app = express();
app.use(express.json());
if (actor) app.use((req, _res, next) => { req.actor = actor; next(); });
else app.use(actorMiddleware(ctx.db, {
deploymentMode: "authenticated",
resolveSession: async () => userId ? {
session: { id: `session-${userId}`, userId },
user: { id: userId, name: "Inbox viewer", email: `${userId}@example.test` },
} : null,
}));
app.use("/api", issueRoutes(ctx.db, {} as never, options));
app.use("/api", agentRoutes(ctx.db));
app.use(errorHandler);
return app;
}
async function addAgent(companyId: string) {
const [agent] = await ctx.db.insert(agents).values({
companyId, name: `Inbox agent ${randomUUID()}`, adapterType: "process", status: "active",
}).returning();
return agent!;
}
async function addTask(companyId: string, userId: string, values: Partial<typeof issues.$inferInsert> = {}) {
const [task] = await ctx.db.insert(issues).values({
companyId, title: "Inbox task", status: "todo", createdByUserId: userId, ...values,
}).returning();
return task!;
}
const sortedContext = (rows: CompactIssue[]) => [...rows].sort((a, b) => a.id.localeCompare(b.id));
it("returns the legacy authorized index, archived parent captions and canonical rich failure state in one snapshot", async () => {
const f = await seed("Snapshot equivalence");
const app = appFor(f.userId);
const [failedAgent, resolvedAgent, supersededAgent] = await Promise.all([
addAgent(f.companyId), addAgent(f.companyId), addAgent(f.companyId),
]);
const indexedParent = await addTask(f.companyId, f.userId, { title: "Indexed parent" });
const archivedParent = await addTask(f.companyId, f.userId, { title: "Archived parent", status: "done" });
const failedTask = await addTask(f.companyId, f.userId, { title: "Failed task", status: "blocked", assigneeAgentId: failedAgent.id });
const resolvedTask = await addTask(f.companyId, f.userId, { title: "Resolved task", status: "done", assigneeAgentId: resolvedAgent.id });
const supersededTask = await addTask(f.companyId, f.userId, { title: "Superseded failure", assigneeAgentId: supersededAgent.id });
await ctx.db.insert(issueInboxArchives).values({ companyId: f.companyId, issueId: archivedParent.id, userId: f.userId });
await addTask(f.companyId, f.userId, { parentId: indexedParent.id });
await addTask(f.companyId, f.userId, { parentId: archivedParent.id });
await ctx.db.insert(heartbeatRuns).values([
{ companyId: f.companyId, agentId: failedAgent.id, issueId: failedTask.id, status: "failed", responsibleUserId: f.userId, contextSnapshot: { issueId: failedTask.id }, createdAt: new Date("2026-11-01T00:00:01Z") },
{ companyId: f.companyId, agentId: resolvedAgent.id, issueId: resolvedTask.id, status: "failed", responsibleUserId: f.userId, contextSnapshot: { issueId: resolvedTask.id }, createdAt: new Date("2026-11-01T00:00:02Z") },
{ companyId: f.companyId, agentId: supersededAgent.id, issueId: supersededTask.id, status: "failed", responsibleUserId: f.userId, contextSnapshot: { issueId: supersededTask.id }, createdAt: new Date("2026-11-01T00:00:03Z") },
{ companyId: f.companyId, agentId: supersededAgent.id, issueId: supersededTask.id, status: "succeeded", responsibleUserId: f.userId, contextSnapshot: { issueId: supersededTask.id }, createdAt: new Date("2026-11-01T00:00:04Z") },
]);
const snapshotResponse = await request(app).get(`/api/companies/${f.companyId}/inbox/mine`).expect(200);
const snapshot = snapshotResponse.body as InboxMineSnapshot;
const index = await request(app).get(`/api/companies/${f.companyId}/issues`).query({
touchedByUserId: "me", inboxArchivedByUserId: "me", status: INBOX_MINE_ISSUE_STATUS_FILTER,
view: "index", sortField: "updated", sortDir: "desc", limit: 100,
includeRoutineExecutions: true, includeLiveDescendantSummary: true,
}).expect(200);
const failures = await request(app).get(`/api/companies/${f.companyId}/heartbeat-runs`)
.query({ summary: true, latestFailuresOnly: true, limit: 200 }).expect(200);
const parent = await request(app).get(`/api/companies/${f.companyId}/issues`)
.query({ view: "index", ids: archivedParent.id, includeRoutineExecutions: true, limit: 100 }).expect(200);
const failureIds = [failedTask.id, resolvedTask.id].sort();
const rich = await request(app).get(`/api/companies/${f.companyId}/issues`)
.query({ view: "rows", ids: failureIds.join(","), includeRoutineExecutions: true, limit: 100 }).expect(200);
expect(snapshot.userId).toBe(f.userId);
expect(snapshot.issues).toEqual(index.body);
expect(snapshot.failureRuns).toEqual(failures.body);
expect(sortedContext(snapshot.contextIssues)).toEqual(sortedContext([...parent.body, ...rich.body]));
expect(snapshot.failureIssueIds).toEqual(failureIds);
expect(snapshot.contextIssues).not.toEqual(expect.arrayContaining([expect.objectContaining({ id: indexedParent.id })]));
expect(snapshot.contextIssues.find(row => row.id === archivedParent.id)).toMatchObject({ title: "Archived parent", listProjection: "index" });
expect(snapshot.contextIssues.find(row => row.id === failedTask.id)?.interruption).toMatchObject({ state: "failed", canRetry: true });
expect(snapshot.contextIssues.find(row => row.id === resolvedTask.id)?.interruption).toMatchObject({ state: "resolved", canRetry: false });
expect(snapshot.failureRuns.map(run => run.agentId)).not.toContain(supersededAgent.id);
expect(snapshotResponse.headers["cache-control"]).toBe("private, no-store");
expect(snapshot).not.toHaveProperty("failureError");
expect(snapshot).not.toHaveProperty("contextError");
});
it("keeps private and missing task runs locked and excludes another company's data", async () => {
const f = await seed("Snapshot privacy");
const foreign = await seed("Foreign snapshot");
const privateUser = `private-${randomUUID()}`;
const agent = await addAgent(f.companyId), missingAgent = await addAgent(f.companyId);
const privateId = randomUUID(), missingId = randomUUID();
const publicTask = await addTask(f.companyId, f.userId, { title: "Visible task" });
await addTask(f.companyId, privateUser, {
id: privateId, title: "PRIVATE_TASK_TITLE", visibility: "private", privacyRootIssueId: privateId,
responsibleUserId: privateUser, assigneeAgentId: agent.id,
});
await addTask(foreign.companyId, f.userId, { title: "FOREIGN_TASK_TITLE" });
const privateRunId = randomUUID(), missingRunId = randomUUID();
await ctx.db.insert(heartbeatRuns).values([
{ id: privateRunId, companyId: f.companyId, agentId: agent.id, issueId: privateId, status: "failed", contextSnapshot: { issueId: privateId, secret: "PRIVATE_RUN_CONTEXT" }, error: "PRIVATE_RUN_ERROR" },
{ id: missingRunId, companyId: f.companyId, agentId: missingAgent.id, status: "failed", contextSnapshot: { issueId: missingId, secret: "MISSING_RUN_CONTEXT" } },
]);
const app = appFor(f.userId);
const snapshot = (await request(app).get(`/api/companies/${f.companyId}/inbox/mine`).expect(200)).body as InboxMineSnapshot;
expect(snapshot.issues.map(row => row.id)).toEqual([publicTask.id]);
expect(snapshot.contextIssues).toEqual([]);
expect(snapshot.failureIssueIds).toEqual([]);
expect(snapshot.failureRuns).toEqual(expect.arrayContaining([
expect.objectContaining({ id: privateRunId, redacted: true }),
expect.objectContaining({ id: missingRunId, redacted: true }),
]));
const serialized = JSON.stringify(snapshot);
for (const secret of [privateId, missingId, "PRIVATE_TASK_TITLE", "FOREIGN_TASK_TITLE", "PRIVATE_RUN_CONTEXT", "PRIVATE_RUN_ERROR", "MISSING_RUN_CONTEXT"]) {
expect(serialized).not.toContain(secret);
}
const legacy = await request(app).get(`/api/companies/${f.companyId}/heartbeat-runs`)
.query({ summary: true, latestFailuresOnly: true, limit: 200 }).expect(200);
expect(snapshot.failureRuns).toEqual(legacy.body);
});
it("owns the 100-task and 200-run bounds and rejects caller filters or unauthorized identities", async () => {
const f = await seed("Snapshot bounds");
const ids = Array.from({ length: 105 }, () => randomUUID());
await ctx.db.insert(issues).values(ids.map((id, index) => ({
id, companyId: f.companyId, title: `Bounded task ${index}`, status: "todo", createdByUserId: f.userId,
createdAt: new Date("2026-10-01T00:00:00Z"),
updatedAt: new Date(Date.UTC(2026, 10, 1, 0, index)),
})));
const runAgents = await ctx.db.insert(agents).values(Array.from({ length: 205 }, (_, index) => ({
companyId: f.companyId, name: `Bounded run agent ${index}`, adapterType: "process",
}))).returning();
await ctx.db.insert(heartbeatRuns).values(runAgents.map((agent, index) => ({
companyId: f.companyId, agentId: agent.id, status: "failed",
createdAt: new Date(Date.UTC(2026, 10, 1, 0, index)),
})));
const app = appFor(f.userId);
const snapshot = (await request(app).get(`/api/companies/${f.companyId}/inbox/mine`).expect(200)).body as InboxMineSnapshot;
expect(snapshot.issues.map(row => row.id)).toEqual([...ids].reverse().slice(0, 100));
expect(snapshot.failureRuns).toHaveLength(200);
expect(snapshot.failureRuns.map(run => run.agentId)).not.toContain(runAgents[0]!.id);
for (const query of [{ limit: 1000 }, { ids: ids[0] }, { userId: "other-user" }, { view: "rows" }]) {
await request(app).get(`/api/companies/${f.companyId}/inbox/mine`).query(query).expect(400);
}
await request(appFor(null)).get(`/api/companies/${f.companyId}/inbox/mine`).expect(403);
await request(appFor(null, {}, { type: "agent", agentId: runAgents[0]!.id, companyId: f.companyId, source: "agent_key" }))
.get(`/api/companies/${f.companyId}/inbox/mine`).expect(403);
await request(appFor(null, {}, { ...f.actor, userId: undefined }))
.get(`/api/companies/${f.companyId}/inbox/mine`).expect(401);
await ctx.db.update(companyMemberships).set({ status: "archived" }).where(eq(companyMemberships.principalId, f.userId));
await request(app).get(`/api/companies/${f.companyId}/inbox/mine`).expect(403);
});
it("starts a fresh snapshot after archive while an older snapshot is still loading failure context", async () => {
const f = await seed("Snapshot mutation freshness");
const task = await addTask(f.companyId, f.userId);
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
let releaseAdmission!: () => void;
const admissionGate = new Promise<void>(resolve => { releaseAdmission = resolve; });
let holdNewComputes = false;
const primaryRead = vi.fn();
const heldFailureRead = vi.fn();
const issueFactory = issueServices.issueService;
const issueSpy = vi.spyOn(issueServices, "issueService").mockImplementation((...args) => {
const service = issueFactory(...args);
return { ...service, list: async (...listArgs) => {
const rows = await service.list(...listArgs);
if (listArgs[0] === f.companyId && listArgs[1]?.touchedByUserId === f.userId) primaryRead();
return rows;
} };
});
const heartbeatFactory = heartbeatServices.heartbeatService;
let firstFailureRead = true;
const heartbeatSpy = vi.spyOn(heartbeatServices, "heartbeatService").mockImplementation((...args) => {
const service = heartbeatFactory(...args);
return { ...service, list: async (...listArgs) => {
const runs = await service.list(...listArgs);
if (listArgs[0] === f.companyId && firstFailureRead) {
firstFailureRead = false;
heldFailureRead();
await gate;
}
return runs;
} };
});
const started = vi.fn(async () => { if (holdNewComputes) await admissionGate; });
const pending: Array<Promise<request.Response>> = [];
try {
const app = appFor(f.userId, { issueListDiagnostics: { onComputeStart: started } });
const path = `/api/companies/${f.companyId}/inbox/mine`;
const original = request(app).get(path).then(response => response);
pending.push(original);
await vi.waitFor(() => {
expect(primaryRead).toHaveBeenCalledTimes(1);
expect(heldFailureRead).toHaveBeenCalledTimes(1);
}, { timeout: 5000 });
await request(app).post(`/api/issues/${task.id}/inbox-archive`).send({}).expect(200);
const fresh = request(app).get(path).then(response => response);
pending.push(fresh);
await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(2), { timeout: 2000 });
const afterArchive = await fresh;
expect(afterArchive.status).toBe(200);
expect(afterArchive.headers["x-paperclip-request-cache"]).toBe("miss");
expect(afterArchive.body.issues).toEqual([]);
// Retiring reuse does not retire the old compute's admission. It still
// occupies one of the same eight slots until its failure read settles.
holdNewComputes = true;
for (let offset = 0; offset < 7; offset += 1) {
pending.push(request(app).get(`/api/companies/${f.companyId}/issues`)
.query({ view: "index", offset }).then(response => response));
}
await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(9), { timeout: 5000 });
await request(app).get(`/api/companies/${f.companyId}/issues`).query({ view: "index", offset: 99 }).expect(429);
expect(started).toHaveBeenCalledTimes(9);
release();
const beforeArchive = await original;
expect(beforeArchive.body.issues.map((row: CompactIssue) => row.id)).toEqual([task.id]);
pending.push(request(app).get(`/api/companies/${f.companyId}/issues`)
.query({ view: "index", offset: 100 }).then(response => response));
await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(10), { timeout: 5000 });
releaseAdmission();
expect((await Promise.all(pending)).every(response => response.status === 200)).toBe(true);
} finally {
release();
releaseAdmission();
await Promise.allSettled(pending);
issueSpy.mockRestore();
heartbeatSpy.mockRestore();
}
});
it.each([false, true])("retains admission until aborted snapshot work settles (coalesced reader: %s)", async withCoalescedReader => {
const f = await seed("Snapshot admission");
await addTask(f.companyId, f.userId);
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
const started = vi.fn(async () => gate);
const app = appFor(f.userId, { issueListDiagnostics: { onComputeStart: started } });
const requests: Array<Promise<{ status: number }>> = [];
const snapshotPath = `/api/companies/${f.companyId}/inbox/mine`;
const original = request(app).get(snapshotPath);
// Retain a rejection handler before aborting the socket.
const aborted = original.then(() => undefined, () => undefined);
try {
await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(1), { timeout: 5000 });
const coalesced = withCoalescedReader ? request(app).get(snapshotPath).then(response => response) : undefined;
if (coalesced) requests.push(coalesced);
for (let offset = 0; offset < 7; offset += 1) {
requests.push(request(app).get(`/api/companies/${f.companyId}/issues`)
.query({ view: "index", offset }).then(response => response));
}
await vi.waitFor(() => expect(started).toHaveBeenCalledTimes(8), { timeout: 5000 });
original.abort();
await aborted;
const denied = await request(app).get(`/api/companies/${f.companyId}/issues`).query({ view: "index", offset: 99 }).expect(429);
expect(denied.headers["retry-after"]).toBe("1");
expect(started).toHaveBeenCalledTimes(8);
release();
const responses = await Promise.all(requests);
expect(responses.every(response => response.status === 200)).toBe(true);
if (coalesced) expect((await coalesced).headers["x-paperclip-request-cache"]).toBe("coalesced");
await request(app).get(snapshotPath).expect(200);
expect(started).toHaveBeenCalledTimes(9);
} finally {
release();
await aborted;
await Promise.allSettled(requests);
}
});
});
+14 -3
View File
@@ -10,6 +10,8 @@ import {
import { unauthorized } from "../errors.js";
import { validate } from "../middleware/validate.js";
import { resolveSentryDsns } from "../sentry-dsn.js";
import { logger } from "../middleware/logger.js";
import { loadBoardBootstrap } from "../services/board-bootstrap.js";
async function loadCurrentUserProfile(db: Db, userId: string) {
const user = await db
@@ -43,8 +45,16 @@ export function authRoutes(db: Db) {
throw unauthorized("Board authentication required");
}
const user = await loadCurrentUserProfile(db, req.actor.userId);
res.json(authSessionSchema.parse({
const bootstrap = (req.actor.source === "session" || req.actor.source === "cloud_tenant")
&& Array.isArray(req.actor.companyIds)
? loadBoardBootstrap(db, { userId: req.actor.userId, companyIds: req.actor.companyIds }).catch((err) => {
logger.warn({ err }, "failed to load optional board bootstrap");
return undefined;
})
: Promise.resolve(undefined);
const [user, boardBootstrap] = await Promise.all([loadCurrentUserProfile(db, req.actor.userId), bootstrap]);
res.setHeader("Cache-Control", "private, no-store");
const session = authSessionSchema.parse({
session: {
id: `paperclip:${req.actor.source ?? "none"}:${req.actor.userId}`,
userId: req.actor.userId,
@@ -57,7 +67,8 @@ export function authRoutes(db: Db) {
sentryDsn: resolveSentryDsns().frontend,
// Match the server SDK's runtime environment, including in reused images.
sentryEnvironment: process.env.SENTRY_ENVIRONMENT || null,
}));
});
res.json({ ...session, ...(boardBootstrap ? { boardBootstrap } : {}) });
});
router.get("/profile", async (req, res) => {
+164 -78
View File
@@ -4,6 +4,10 @@ import { notifyNativeStatusReassessment } from "../services/native-status-reasse
import { notifyUnsafeWorkspaceExportRecovery } from "../services/unsafe-workspace-export-notifications.js";
import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js";
import { completeIssueCommentReceipt } from "../services/issue-comment-receipt.js";
import { readInboxMine } from "../services/inbox-mine-read.js";
import { subscribeCompanyLiveEvents } from "../services/live-events.js";
import { INBOX_MINE_ISSUE_STATUS_FILTER, latestFailedRunsByAgent } from "@paperclipai/shared";
import { getHeartbeatRunReadDecisions, redactHeartbeatRunListRow } from "../services/heartbeat-run-privacy.js";
import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js";
import { parseObject } from "../adapters/utils.js";
import { prepareRestrictedWorkspaceRetry } from "../services/restricted-workspace-retry.js";
@@ -2983,6 +2987,7 @@ const ISSUE_LIST_STORM_THRESHOLD = 4;
const ISSUE_LIST_MAX_ACTOR_CLIENT_INFLIGHT = 8;
type IssueListPreparedResponse =
| { kind: "inbox"; body: unknown }
| {
kind: "compact";
body: CompactIssue[];
@@ -3145,8 +3150,9 @@ function issueListRequestKey(input: {
req: Request;
companyId: string;
normalizedQuery: Record<string, unknown>;
route?: string;
}) {
const route = "GET /api/companies/:companyId/issues";
const route = input.route ?? "GET /api/companies/:companyId/issues";
const actor = issueListActorIdentity(input.req, input.companyId);
const client = issueListClientIdentity(input.req);
const normalizedQuery = normalizeIssueListCacheValue(
@@ -3247,6 +3253,7 @@ async function coordinateIssueListGet(input: {
companyId: string;
requestKey: ReturnType<typeof issueListRequestKey>;
allowTtlCache: boolean;
retireOnCompanyChange?: boolean;
diagnostics?: IssueListDiagnostics;
compute: () => Promise<IssueListPreparedResponse>;
}): Promise<{
@@ -3336,6 +3343,17 @@ async function coordinateIssueListGet(input: {
stormLogged: false,
};
issueListInflight.set(input.requestKey.key, inflightEntry);
// Mine reads its primary rows before its optional context has necessarily
// settled. A mutation must make subsequent GETs read again, while this work
// and its existing readers keep their original admission until completion.
const unsubscribe = input.retireOnCompanyChange
? subscribeCompanyLiveEvents(input.companyId, (event) => {
if (event.type !== "activity.logged" && event.type !== "heartbeat.run.queued" && event.type !== "heartbeat.run.status") return;
if (issueListInflight.get(input.requestKey.key) === inflightEntry) {
issueListInflight.delete(input.requestKey.key);
}
})
: undefined;
try {
const response = await promise;
@@ -3348,6 +3366,7 @@ async function coordinateIssueListGet(input: {
}
return { response, cacheStatus: "miss", identicalInFlightCount: 1 };
} finally {
unsubscribe?.();
if (issueListInflight.get(input.requestKey.key) === inflightEntry) {
issueListInflight.delete(input.requestKey.key);
}
@@ -8387,6 +8406,149 @@ export function issueRoutes(
res.json(summaries);
});
async function prepareIssueListForActor(req: Request, companyId: string, listFilters: IssueFilters, view: string | undefined): Promise<IssueListPreparedResponse> {
const compactView = view === "compact" || view === "index" || view === "rows";
const rawResult = await svc.list(companyId, listFilters);
const visibleRows = await actorCanReadCompanyScope(req, companyId)
? rawResult
: await filterIssuesForActor(req, rawResult);
const result = await redactIssueRelationEdgesOnRows(req, companyId, visibleRows);
if (view === "index") {
const body = result.map((issue) => ({ ...toCompactIssue(issue), listProjection: "index" as const }));
return { kind: "compact", body, etag: compactIssueListEtag(body), cacheControl: "private, must-revalidate" };
}
const issueIds = result.map((issue) => issue.id);
const interruptions = await taskInterruptionPresentations(db, companyId, result);
if (compactView) {
const [handoffStates, recoveryActionByIssue] = await Promise.all([
listSuccessfulRunHandoffStates(db, companyId, issueIds),
recoveryActionsSvc.listActiveForIssues(companyId, issueIds),
]);
const actor = getActorInfo(req);
await Promise.all(
result.map(async (issue) => {
const activeRecoveryAction =
recoveryActionByIssue.get(issue.id) ?? null;
if (!activeRecoveryAction) return;
const revalidated = await revalidateActiveSourceRecoveryForRead({
issue,
trigger: "read_projection",
actor,
activeRecoveryAction,
});
if (revalidated) recoveryActionByIssue.set(issue.id, revalidated);
else recoveryActionByIssue.delete(issue.id);
}),
);
const compactResult = result.map((issue) =>
({ ...toCompactIssue({
...issue,
activeRecoveryAction: recoveryActionByIssue.get(issue.id) ?? null,
interruption: interruptions.get(issue.id) ?? null,
successfulRunHandoff: handoffStates.get(issue.id) ?? null,
}), ...(view === "rows" ? { listProjection: "rows" as const } : {}) }),
);
return {
kind: "compact",
body: compactResult,
etag: compactIssueListEtag(compactResult),
cacheControl: "private, must-revalidate",
};
}
const [handoffStates, recoveryActionByIssue] = await Promise.all([
listSuccessfulRunHandoffStates(db, companyId, issueIds),
recoveryActionsSvc.listActiveForIssues(companyId, issueIds),
]);
const actor = getActorInfo(req);
await Promise.all(
result.map(async (issue) => {
const activeRecoveryAction =
recoveryActionByIssue.get(issue.id) ?? null;
if (!activeRecoveryAction) return;
const revalidated = await revalidateActiveSourceRecoveryForRead({
issue,
trigger: "read_projection",
actor,
activeRecoveryAction,
});
if (revalidated) recoveryActionByIssue.set(issue.id, revalidated);
else recoveryActionByIssue.delete(issue.id);
}),
);
return {
kind: "full",
body: result.map((issue) => ({
...issue,
successfulRunHandoff: handoffStates.get(issue.id) ?? null,
activeRecoveryAction: recoveryActionByIssue.get(issue.id) ?? null,
interruption: interruptions.get(issue.id) ?? null,
})),
};
}
router.get("/companies/:companyId/inbox/mine", async (req, res) => {
const startedAt = Date.now();
const companyId = req.params.companyId as string;
assertBoard(req);
assertCompanyAccess(req, companyId);
if (Object.keys(req.query).length > 0) {
res.status(400).json({ error: "The Mine snapshot does not accept query parameters" });
return;
}
const userId = req.actor.userId;
if (!userId) throw new HttpError(401, "A signed-in board user is required");
const readCondition = await issueReadSqlCondition(db, req.actor);
const listFilters: IssueFilters = {
readCondition, touchedByUserId: userId, inboxArchivedByUserId: userId,
status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index",
sortField: "updated", sortDir: "desc", limit: 100,
includeRoutineExecutions: true, includeLiveDescendantSummary: true,
};
const readCompact = async (filters: IssueFilters, projection: "index" | "rows") => {
const prepared = await prepareIssueListForActor(req, companyId, filters, projection);
if (prepared.kind !== "compact") throw new Error("Expected compact inbox rows");
return prepared.body;
};
const requestKey = issueListRequestKey({ req, companyId, normalizedQuery: { view: "mine-snapshot" }, route: "GET /api/companies/:companyId/inbox/mine" });
const coordinated = await coordinateIssueListGet({
req, companyId, requestKey, allowTtlCache: false, retireOnCompanyChange: true, diagnostics: opts.issueListDiagnostics,
compute: async () => ({
kind: "inbox",
body: { userId, ...await readInboxMine({
loadIssues: () => readCompact(listFilters, "index"),
loadFailures: async () => {
const permission = await access.decide({ actor: req.actor, action: "company_scope:read", resource: { type: "company", companyId } });
if (!permission.allowed) throw new HttpError(403, "Run telemetry is outside this actor's authorization boundary");
const runs = latestFailedRunsByAgent(await heartbeat.list(companyId, undefined, 200, { summary: true }));
const decisions = await getHeartbeatRunReadDecisions(db, access, req.actor, runs);
return runRedactions.redactForRuns(companyId, runs.map((run, index) => decisions[index] ? run : redactHeartbeatRunListRow(run)), getActorInfo(req).actorId);
},
loadContext: (ids, projection) => readCompact({
readCondition, issueIds: ids, listProjection: projection, limit: 100, includeRoutineExecutions: true,
}, projection),
sourceError: (error, source) => {
logger.warn({ err: error, companyId, source }, "inbox source read failed");
return {
status: error instanceof HttpError ? error.status : 503,
message: error instanceof HttpError && error.status < 500 ? error.message
: source === "failures" ? "Could not load failed runs" : "Could not load task context",
};
},
}) },
}),
});
res.setHeader("Cache-Control", "private, no-store");
res.setHeader("X-Paperclip-Request-Cache", coordinated.cacheStatus);
if (!coordinated.response) {
res.setHeader("Retry-After", String(coordinated.retryAfterSeconds ?? 1));
res.status(429).json({ error: "Too many concurrent issue-list requests for this actor/client", retryAfterSeconds: coordinated.retryAfterSeconds ?? 1 });
return;
}
logIssueListRequest({ req, res, companyId, requestKey, startedAt, cacheStatus: coordinated.cacheStatus,
bodyBytes: estimatedJsonBytes(coordinated.response.body), etagOutcome: "none", identicalInFlightCount: coordinated.identicalInFlightCount });
res.json(coordinated.response.body);
});
router.get("/companies/:companyId/issues", async (req, res) => {
const startedAt = Date.now();
const companyId = req.params.companyId as string;
@@ -8439,7 +8601,6 @@ export function issueRoutes(
const sortField = req.query.sortField as string | undefined;
const sortDir = req.query.sortDir as string | undefined;
const view = req.query.view as string | undefined;
const compactView = view === "compact" || view === "index" || view === "rows";
const issueIds = typeof req.query.ids === "string" ? [...new Set(req.query.ids.split(","))] : undefined;
const parentIds = typeof req.query.parentIds === "string" ? [...new Set(req.query.parentIds.split(","))] : undefined;
const oneChildPerParent = parseOptionalBooleanQuery(req.query.oneChildPerParent);
@@ -8685,82 +8846,7 @@ export function issueRoutes(
allowTtlCache: view === "compact",
diagnostics: opts.issueListDiagnostics,
compute: async () => {
const rawResult = await svc.list(companyId, listFilters);
const visibleRows = await actorCanReadCompanyScope(req, companyId)
? rawResult
: await filterIssuesForActor(req, rawResult);
const result = await redactIssueRelationEdgesOnRows(req, companyId, visibleRows);
if (view === "index") {
const body = result.map((issue) => ({ ...toCompactIssue(issue), listProjection: "index" as const }));
return { kind: "compact", body, etag: compactIssueListEtag(body), cacheControl: "private, must-revalidate" };
}
const issueIds = result.map((issue) => issue.id);
const interruptions = await taskInterruptionPresentations(db, companyId, result);
if (compactView) {
const [handoffStates, recoveryActionByIssue] = await Promise.all([
listSuccessfulRunHandoffStates(db, companyId, issueIds),
recoveryActionsSvc.listActiveForIssues(companyId, issueIds),
]);
const actor = getActorInfo(req);
await Promise.all(
result.map(async (issue) => {
const activeRecoveryAction =
recoveryActionByIssue.get(issue.id) ?? null;
if (!activeRecoveryAction) return;
const revalidated = await revalidateActiveSourceRecoveryForRead({
issue,
trigger: "read_projection",
actor,
activeRecoveryAction,
});
if (revalidated) recoveryActionByIssue.set(issue.id, revalidated);
else recoveryActionByIssue.delete(issue.id);
}),
);
const compactResult = result.map((issue) =>
({ ...toCompactIssue({
...issue,
activeRecoveryAction: recoveryActionByIssue.get(issue.id) ?? null,
interruption: interruptions.get(issue.id) ?? null,
successfulRunHandoff: handoffStates.get(issue.id) ?? null,
}), ...(view === "rows" ? { listProjection: "rows" as const } : {}) }),
);
return {
kind: "compact",
body: compactResult,
etag: compactIssueListEtag(compactResult),
cacheControl: "private, must-revalidate",
};
}
const [handoffStates, recoveryActionByIssue] = await Promise.all([
listSuccessfulRunHandoffStates(db, companyId, issueIds),
recoveryActionsSvc.listActiveForIssues(companyId, issueIds),
]);
const actor = getActorInfo(req);
await Promise.all(
result.map(async (issue) => {
const activeRecoveryAction =
recoveryActionByIssue.get(issue.id) ?? null;
if (!activeRecoveryAction) return;
const revalidated = await revalidateActiveSourceRecoveryForRead({
issue,
trigger: "read_projection",
actor,
activeRecoveryAction,
});
if (revalidated) recoveryActionByIssue.set(issue.id, revalidated);
else recoveryActionByIssue.delete(issue.id);
}),
);
return {
kind: "full",
body: result.map((issue) => ({
...issue,
successfulRunHandoff: handoffStates.get(issue.id) ?? null,
activeRecoveryAction: recoveryActionByIssue.get(issue.id) ?? null,
interruption: interruptions.get(issue.id) ?? null,
})),
};
return prepareIssueListForActor(req, companyId, listFilters, view);
},
});
+11
View File
@@ -4231,6 +4231,16 @@ registry.registerPath({
responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden, 404: r.notFound },
});
registry.registerPath({
method: "get",
path: "/api/companies/{companyId}/inbox/mine",
tags: ["issues"],
summary: "Get the board user's initial Mine inbox and required task context",
description: "Board-only presentation read. Returns userId, issues, contextIssues, failureRuns and verified failureIssueIds. Uses the existing company, private-task, run-read and secret-redaction policies. Optional failureError or contextError retains successful task rows. The initial task page is bounded to 100 and failed-run history to 200. No query parameters are accepted. This GET shares issue-list coalescing and admission limits; it creates no work.",
request: { params: z.object({ companyId: z.string() }) },
responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 429: r.tooManyRequests },
});
registry.registerPath({
method: "get",
path: "/api/companies/{companyId}/issues",
@@ -7434,6 +7444,7 @@ registry.registerPath({
path: "/api/auth/get-session",
tags: ["auth"],
summary: "Get current session",
description: "Authenticated board sessions can include an optional same-user boardBootstrap with accessible companies and boardAccess. Existing session fields are unchanged; clients retain ordinary company and access reads when the optional projection is absent.",
responses: { 200: r.ok(), 401: r.unauthorized },
});
+27
View File
@@ -0,0 +1,27 @@
import type { Db } from "@paperclipai/db";
import type { BoardBootstrap } from "@paperclipai/shared";
import { boardAuthService } from "./board-auth.js";
import { companyService } from "./companies.js";
/** The actor has already authenticated and resolved current company authority. */
export async function loadBoardBootstrap(db: Db, principal: {
userId: string;
companyIds: readonly string[];
}): Promise<BoardBootstrap> {
const [companies, access] = await Promise.all([
companyService(db).list(),
boardAuthService(db).resolveBoardAccess(principal.userId),
]);
const accessible = new Set(principal.companyIds);
return {
userId: principal.userId,
// Match /companies?scope=accessible: directory visibility through instance
// admin alone does not grant entry to a company's inbox or tasks.
companies: companies.filter((company) => accessible.has(company.id)),
boardAccess: {
userId: principal.userId,
isInstanceAdmin: access.isInstanceAdmin,
companyIds: access.companyIds,
},
};
}
+6 -1
View File
@@ -37,6 +37,7 @@ import {
import { environmentService } from "./environments.js";
import { logActivity, type ActivityPublication } from "./activity-log.js";
import { builtInAgentService } from "./built-in-agents.js";
import type { Company } from "@paperclipai/shared";
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
@@ -117,6 +118,8 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
name: companies.name,
description: companies.description,
status: companies.status,
pauseReason: companies.pauseReason,
pausedAt: companies.pausedAt,
issuePrefix: companies.issuePrefix,
issueCounter: companies.issueCounter,
budgetMonthlyCents: companies.budgetMonthlyCents,
@@ -133,9 +136,11 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
updatedAt: companies.updatedAt,
};
function enrichCompany<T extends { logoAssetId: string | null }>(company: T) {
function enrichCompany<T extends { logoAssetId: string | null; status: string; pauseReason: string | null }>(company: T) {
return {
...company,
status: company.status as Company["status"],
pauseReason: company.pauseReason as Company["pauseReason"],
logoUrl: company.logoAssetId ? `/api/assets/${company.logoAssetId}/content` : null,
};
}
+102
View File
@@ -0,0 +1,102 @@
import { describe, expect, it, vi } from "vitest";
import { readInboxMine } from "./inbox-mine-read.js";
function deferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((done) => { resolve = done; });
return { promise, resolve };
}
type Row = { id: string; parentId?: string; interruption?: string };
const sourceError = (_error: unknown, source: string) => ({ status: 503, message: `${source} unavailable` });
describe("Mine read composition", () => {
it("loads failure state before the primary list finishes and reveals one complete snapshot", async () => {
const primary = deferred<Row[]>(), rich = deferred<Row[]>();
const loadContext = vi.fn((ids: string[], projection: string) => projection === "rows"
? rich.promise : Promise.resolve(ids.map((id) => ({ id }))));
const result = readInboxMine({
loadIssues: () => primary.promise,
loadFailures: async () => [{ id: "run", issueId: "failure" }],
loadContext, sourceError,
});
await vi.waitFor(() => expect(loadContext).toHaveBeenCalledWith(["failure"], "rows"));
primary.resolve([{ id: "child", parentId: "parent" }]);
await vi.waitFor(() => expect(loadContext).toHaveBeenCalledWith(["parent"], "index"));
const settled = vi.fn();
void result.then(settled);
await Promise.resolve();
expect(settled).not.toHaveBeenCalled();
rich.resolve([{ id: "failure", interruption: "current" }]);
expect(await result).toMatchObject({
issues: [{ id: "child", parentId: "parent" }],
contextIssues: [{ id: "parent" }, { id: "failure", interruption: "current" }],
failureIssueIds: ["failure"],
});
});
it("reuses indexed parents, preserves rich overlap, and records known missing failures", async () => {
const loadContext = vi.fn(async (ids: string[], projection: string) => ids.filter((id) => id !== "missing")
.map((id) => ({ id, ...(projection === "rows" ? { interruption: "resolved" } : {}) })));
const result = await readInboxMine({
loadIssues: async () => [{ id: "indexed" }, { id: "a", parentId: "indexed" }, { id: "b", parentId: "overlap" }],
loadFailures: async () => [{ id: "run-a", issueId: "overlap" }, { id: "run-b", contextSnapshot: { taskId: "missing" } }],
loadContext, sourceError,
});
expect(loadContext.mock.calls.every(([ids]) => !ids.includes("indexed"))).toBe(true);
expect(result.contextIssues).toEqual([{ id: "overlap", interruption: "resolved" }]);
expect(result.failureIssueIds).toEqual(["missing", "overlap"]);
});
it("keeps tasks and captions when the optional run read fails", async () => {
const result = await readInboxMine({
loadIssues: async () => [{ id: "child", parentId: "parent" }],
loadFailures: async (): Promise<{ id: string }[]> => { throw new Error("private database detail"); },
loadContext: async (ids) => ids.map((id) => ({ id })), sourceError,
});
expect(result.contextIssues).toEqual([{ id: "parent" }]);
expect(result.failureRuns).toEqual([]);
expect(result.failureError).toEqual({ status: 503, message: "failures unavailable" });
});
it("keeps parent captions when rich context fails without claiming verified failure state", async () => {
const result = await readInboxMine({
loadIssues: async () => [{ id: "child", parentId: "parent" }],
loadFailures: async () => [{ id: "run", issueId: "failure" }],
loadContext: async (ids, projection) => {
if (projection === "rows") throw new Error("read failed");
return ids.map((id) => ({ id }));
}, sourceError,
});
expect(result.contextIssues).toEqual([{ id: "parent" }]);
expect(result.failureIssueIds).toEqual([]);
expect(result.contextError).toEqual({ status: 503, message: "context unavailable" });
});
it("bounds context batches even with a full failed-run window", async () => {
const loadContext = vi.fn(async (ids: string[]) => ids.map((id) => ({ id })));
const result = await readInboxMine({
loadIssues: async (): Promise<Row[]> => [],
loadFailures: async () => Array.from({ length: 200 }, (_, i) => ({ id: `run-${i}`, issueId: `issue-${i}` })),
loadContext, sourceError,
});
expect(loadContext.mock.calls.map(([ids]) => ids.length)).toEqual([100, 100]);
expect(result.failureIssueIds).toHaveLength(200);
});
it("does not release the read while another started branch still runs after a primary error", async () => {
const rich = deferred<Row[]>();
const loadContext = vi.fn(() => rich.promise);
const result = readInboxMine({
loadIssues: async (): Promise<Row[]> => { throw new Error("primary unavailable"); },
loadFailures: async () => [{ id: "run", issueId: "failure" }],
loadContext, sourceError,
});
const rejected = vi.fn();
const observed = result.catch(rejected);
await vi.waitFor(() => expect(loadContext).toHaveBeenCalled());
expect(rejected).not.toHaveBeenCalled();
rich.resolve([]);
await observed;
expect(rejected).toHaveBeenCalledWith(expect.objectContaining({ message: "primary unavailable" }));
});
});
+77
View File
@@ -0,0 +1,77 @@
import type { InboxSourceError } from "@paperclipai/shared";
type IssueRow = { id: string; parentId?: string | null };
type RunRow = { id: string; issueId?: unknown; contextSnapshot?: unknown };
type ContextResult<I> = { issues: I[]; error?: InboxSourceError };
const CONTEXT_BATCH_SIZE = 100;
export function inboxFailureIssueId(run: RunRow): string | null {
if (typeof run.issueId === "string" && run.issueId) return run.issueId;
const context = run.contextSnapshot;
if (!context || typeof context !== "object" || Array.isArray(context)) return null;
for (const key of ["issueId", "taskId"] as const) {
const value = (context as Record<string, unknown>)[key];
if (typeof value === "string" && value) return value;
}
return null;
}
/** Compose canonical, authorized reads without a browser round trip between them. */
export async function readInboxMine<I extends IssueRow, R extends RunRow>(deps: {
loadIssues(): Promise<I[]>;
loadFailures(): Promise<R[]>;
loadContext(ids: string[], projection: "index" | "rows"): Promise<I[]>;
sourceError(error: unknown, source: "failures" | "context"): InboxSourceError;
}) {
const loadContext = async (ids: string[], projection: "index" | "rows"): Promise<ContextResult<I>> => {
try {
const issues: I[] = [];
for (let offset = 0; offset < ids.length; offset += CONTEXT_BATCH_SIZE) {
issues.push(...await deps.loadContext(ids.slice(offset, offset + CONTEXT_BATCH_SIZE), projection));
}
return { issues };
} catch (error) {
return { issues: [], error: deps.sourceError(error, "context") };
}
};
const issuesPromise = deps.loadIssues();
const failuresPromise = deps.loadFailures().then(
(runs) => ({ runs, error: undefined as InboxSourceError | undefined }),
(error: unknown) => ({ runs: [] as R[], error: deps.sourceError(error, "failures") }),
);
// Start each branch as soon as its own input arrives. In particular, rich
// failure reads do not wait for the primary task list or its parent captions.
const parentsPromise = issuesPromise.then((issues) => {
const loaded = new Set(issues.map((issue) => issue.id));
const ids = [...new Set(issues.map((issue) => issue.parentId)
.filter((id): id is string => typeof id === "string" && id.length > 0 && !loaded.has(id)))].sort();
return loadContext(ids, "index");
});
const failureContextPromise = failuresPromise.then(async ({ runs }) => {
const ids = [...new Set(runs.map(inboxFailureIssueId).filter((id): id is string => Boolean(id)))].sort();
return { ...await loadContext(ids, "rows"), ids };
});
// Admission remains occupied until every started read settles, including when
// the critical primary query fails while a failure read is still running.
const results = await Promise.allSettled([
issuesPromise, failuresPromise, parentsPromise, failureContextPromise,
]);
const [issueResult, failuresResult, parentsResult, contextResult] = results;
if (issueResult.status === "rejected") throw issueResult.reason;
if (failuresResult.status === "rejected") throw failuresResult.reason;
if (parentsResult.status === "rejected") throw parentsResult.reason;
if (contextResult.status === "rejected") throw contextResult.reason;
const issues = issueResult.value, failures = failuresResult.value;
const parents = parentsResult.value, failureContext = contextResult.value;
// A task that serves both roles keeps its canonical rich failure state.
const context = new Map(parents.issues.map((issue) => [issue.id, issue]));
for (const issue of failureContext.issues) context.set(issue.id, issue);
return {
issues,
contextIssues: [...context.values()],
failureRuns: failures.runs,
failureIssueIds: failureContext.error ? [] : failureContext.ids,
...(failures.error ? { failureError: failures.error } : {}),
...(parents.error || failureContext.error ? { contextError: parents.error ?? failureContext.error } : {}),
};
}