feat(decisions): add queue and triage APIs

Add company-scoped decision queue membership, decide-by triage, lazy data-backed starter queues, source-mediated authorization, and immutable provenance.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-01 17:01:20 +00:00
1 parent d4f57eabfe
commit 38b05a9eda
20 files changed
+39554 -17

No files matched your search

+6
View File
@@ -177,6 +177,12 @@ Both tables use a unique key on `(company_id, user_id, resource_id)` and keep `s
This policy makes training exports self-describing while keeping the decision record usable after a comment deletion without retaining content the author removed.
## Decision queues and triage provenance
The decisions desk stores queue membership and decide-by/snooze state in `decision_queues`, `decision_queue_items`, and `decision_triage`. These sidecars use the stable attention identity `(source_kind, source_id)` so all attention source kinds can participate without copying source titles, bodies, projects, or other visibility-sensitive data.
`decision_triage_events` is append-only history for queue and triage changes. Current rows and history both carry server-derived user/agent, heartbeat run, API-key, and responsible-user attribution where applicable. Queue reads must resolve and authorize their source rows at read time; a sidecar row is never a visibility grant.
## Plugin database namespaces
The plugin runtime tracks plugin-owned database namespaces and migrations in `plugin_database_namespaces` and `plugin_migrations`. Hosted deployments that separate runtime and migration connections should set `DATABASE_MIGRATION_URL`; plugin namespace migration work uses the migration connection when present.
+17
View File
@@ -455,6 +455,14 @@ The current implementation includes additional V1-control-plane tables beyond th
- Plugins and routines: `plugins`, plugin config/state/entities/jobs/logs/webhooks, plugin database namespaces/migrations, plugin company settings, `routines`, `routine_revisions`, `routine_triggers`, and `routine_runs`.
- Access and operations: company memberships, instance roles, principal permission grants, invites, join requests, board API keys, CLI auth challenges, budget policies/incidents, feedback exports/votes, company skills, sidebar preferences, and company logos.
Decision-desk triage uses company-scoped sidecars rather than adding queue fields to every attention source:
- `decision_queues` stores durable named queues, optional retention overrides, server-derived creator/run provenance, and data-backed seed rules.
- `decision_queue_items` keys membership by `(queue_id, source_kind, source_id)` and repeats `company_id` for company-consistent joins.
- `decision_triage` keys current decide-by/snooze state by `(company_id, source_kind, source_id)` and preserves the latest setter attribution.
- `decision_triage_events` is the immutable mutation history for queue membership and triage overrides, including actor, run, API-key, and responsible-user provenance.
- Queue membership never grants source visibility. Item writes re-authorize the referenced source, and queue reads re-authorize every member before returning rows or counts.
## 8. State Machines
## 8.1 Agent Status
@@ -974,6 +982,15 @@ The current app also exposes V1-supporting surfaces for:
- plugin installation, configuration, state, jobs, logs, webhooks, and plugin database namespace migration
- company import/export preview/apply, feedback export/vote routes, instance backup/config routes, invites, join requests, memberships, and permission grants
- company skill policy read/replace/reset/simulation, enforced by the same core evaluator used by skill mutation routes
- decision queues and per-attention-item triage:
- `GET|POST /companies/:companyId/decision-queues`
- `PATCH /companies/:companyId/decision-queues/:key`
- `GET|POST /companies/:companyId/decision-queues/:key/items`
- `DELETE /companies/:companyId/decision-queues/:key/items/:sourceKind/:sourceId`
- `GET /companies/:companyId/decision-queue-seed-rules`
- `GET|PUT /companies/:companyId/decision-triage/:sourceKind/:sourceId`
Queue and triage mutations accept board non-viewers and active standard-scope agents, apply responsible-user intersection for run JWTs, and reject low-trust, `task_bridge`, and `skill_test` contexts. Missing, cross-company, and unauthorized attention sources share the same not-found response.
## 11. Heartbeat and Adapter Contract
@@ -0,0 +1,121 @@
CREATE TABLE "decision_queue_items" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"company_id" uuid NOT NULL,
"queue_id" uuid NOT NULL,
"source_kind" text NOT NULL,
"source_id" text NOT NULL,
"added_by_type" text NOT NULL,
"added_by_agent_id" uuid,
"added_by_user_id" text,
"added_by_run_id" uuid,
"added_by_agent_api_key_id" uuid,
"responsible_user_id" text,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT "decision_queue_items_actor_check" CHECK ((
("decision_queue_items"."added_by_type" = 'agent' AND "decision_queue_items"."added_by_agent_id" IS NOT NULL AND "decision_queue_items"."added_by_user_id" IS NULL)
OR ("decision_queue_items"."added_by_type" = 'user' AND "decision_queue_items"."added_by_agent_id" IS NULL AND "decision_queue_items"."added_by_user_id" IS NOT NULL)
OR ("decision_queue_items"."added_by_type" = 'system' AND "decision_queue_items"."added_by_agent_id" IS NULL AND "decision_queue_items"."added_by_user_id" IS NULL)
))
);
--> statement-breakpoint
CREATE TABLE "decision_queues" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"company_id" uuid NOT NULL,
"key" text NOT NULL,
"title" text NOT NULL,
"description" text,
"created_by_type" text NOT NULL,
"created_by_agent_id" uuid,
"created_by_user_id" text,
"created_by_run_id" uuid,
"created_by_agent_api_key_id" uuid,
"retention_days" integer,
"seed_rules" jsonb DEFAULT '[]'::jsonb NOT NULL,
"seed_rules_enabled" boolean DEFAULT false NOT NULL,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"updated_at" timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT "decision_queues_creator_check" CHECK ((
("decision_queues"."created_by_type" = 'agent' AND "decision_queues"."created_by_agent_id" IS NOT NULL AND "decision_queues"."created_by_user_id" IS NULL)
OR ("decision_queues"."created_by_type" = 'user' AND "decision_queues"."created_by_agent_id" IS NULL AND "decision_queues"."created_by_user_id" IS NOT NULL)
OR ("decision_queues"."created_by_type" = 'system' AND "decision_queues"."created_by_agent_id" IS NULL AND "decision_queues"."created_by_user_id" IS NULL)
)),
CONSTRAINT "decision_queues_retention_days_check" CHECK ("decision_queues"."retention_days" IS NULL OR ("decision_queues"."retention_days" >= 1 AND "decision_queues"."retention_days" <= 3650))
);
--> statement-breakpoint
CREATE TABLE "decision_triage" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"company_id" uuid NOT NULL,
"source_kind" text NOT NULL,
"source_id" text NOT NULL,
"decide_by" text,
"decide_by_date" date,
"snoozed_until" timestamp with time zone,
"set_by_type" text NOT NULL,
"set_by_agent_id" uuid,
"set_by_user_id" text,
"set_by_run_id" uuid,
"set_by_agent_api_key_id" uuid,
"responsible_user_id" text,
"version" integer DEFAULT 1 NOT NULL,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"updated_at" timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT "decision_triage_actor_check" CHECK ((
("decision_triage"."set_by_type" = 'agent' AND "decision_triage"."set_by_agent_id" IS NOT NULL AND "decision_triage"."set_by_user_id" IS NULL)
OR ("decision_triage"."set_by_type" = 'user' AND "decision_triage"."set_by_agent_id" IS NULL AND "decision_triage"."set_by_user_id" IS NOT NULL)
)),
CONSTRAINT "decision_triage_decide_by_check" CHECK ((
("decision_triage"."decide_by" IS NULL AND "decision_triage"."decide_by_date" IS NULL)
OR ("decision_triage"."decide_by" IN ('today', 'this_week', 'whenever') AND "decision_triage"."decide_by_date" IS NULL)
OR ("decision_triage"."decide_by" = 'date' AND "decision_triage"."decide_by_date" IS NOT NULL)
))
);
--> statement-breakpoint
CREATE TABLE "decision_triage_events" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"company_id" uuid NOT NULL,
"queue_id" uuid,
"source_kind" text,
"source_id" text,
"action" text NOT NULL,
"actor_type" text NOT NULL,
"actor_agent_id" uuid,
"actor_user_id" text,
"actor_run_id" uuid,
"agent_api_key_id" uuid,
"responsible_user_id" text,
"details" jsonb DEFAULT '{}'::jsonb NOT NULL,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT "decision_triage_events_actor_check" CHECK ((
("decision_triage_events"."actor_type" = 'agent' AND "decision_triage_events"."actor_agent_id" IS NOT NULL AND "decision_triage_events"."actor_user_id" IS NULL)
OR ("decision_triage_events"."actor_type" = 'user' AND "decision_triage_events"."actor_agent_id" IS NULL AND "decision_triage_events"."actor_user_id" IS NOT NULL)
OR ("decision_triage_events"."actor_type" = 'system' AND "decision_triage_events"."actor_agent_id" IS NULL AND "decision_triage_events"."actor_user_id" IS NULL)
))
);
--> statement-breakpoint
ALTER TABLE "decision_queue_items" ADD CONSTRAINT "decision_queue_items_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queue_items" ADD CONSTRAINT "decision_queue_items_added_by_agent_id_agents_id_fk" FOREIGN KEY ("added_by_agent_id") REFERENCES "public"."agents"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queue_items" ADD CONSTRAINT "decision_queue_items_added_by_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("added_by_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queue_items" ADD CONSTRAINT "decision_queue_items_added_by_agent_api_key_id_agent_api_keys_id_fk" FOREIGN KEY ("added_by_agent_api_key_id") REFERENCES "public"."agent_api_keys"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
CREATE UNIQUE INDEX "decision_queues_id_company_uq" ON "decision_queues" USING btree ("id","company_id");--> statement-breakpoint
ALTER TABLE "decision_queue_items" ADD CONSTRAINT "decision_queue_items_queue_company_fk" FOREIGN KEY ("queue_id","company_id") REFERENCES "public"."decision_queues"("id","company_id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queues" ADD CONSTRAINT "decision_queues_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queues" ADD CONSTRAINT "decision_queues_created_by_agent_id_agents_id_fk" FOREIGN KEY ("created_by_agent_id") REFERENCES "public"."agents"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queues" ADD CONSTRAINT "decision_queues_created_by_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("created_by_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_queues" ADD CONSTRAINT "decision_queues_created_by_agent_api_key_id_agent_api_keys_id_fk" FOREIGN KEY ("created_by_agent_api_key_id") REFERENCES "public"."agent_api_keys"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage" ADD CONSTRAINT "decision_triage_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage" ADD CONSTRAINT "decision_triage_set_by_agent_id_agents_id_fk" FOREIGN KEY ("set_by_agent_id") REFERENCES "public"."agents"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage" ADD CONSTRAINT "decision_triage_set_by_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("set_by_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage" ADD CONSTRAINT "decision_triage_set_by_agent_api_key_id_agent_api_keys_id_fk" FOREIGN KEY ("set_by_agent_api_key_id") REFERENCES "public"."agent_api_keys"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage_events" ADD CONSTRAINT "decision_triage_events_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage_events" ADD CONSTRAINT "decision_triage_events_queue_id_decision_queues_id_fk" FOREIGN KEY ("queue_id") REFERENCES "public"."decision_queues"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage_events" ADD CONSTRAINT "decision_triage_events_actor_agent_id_agents_id_fk" FOREIGN KEY ("actor_agent_id") REFERENCES "public"."agents"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage_events" ADD CONSTRAINT "decision_triage_events_actor_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("actor_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "decision_triage_events" ADD CONSTRAINT "decision_triage_events_agent_api_key_id_agent_api_keys_id_fk" FOREIGN KEY ("agent_api_key_id") REFERENCES "public"."agent_api_keys"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
CREATE UNIQUE INDEX "decision_queue_items_queue_source_uq" ON "decision_queue_items" USING btree ("queue_id","source_kind","source_id");--> statement-breakpoint
CREATE INDEX "decision_queue_items_company_source_idx" ON "decision_queue_items" USING btree ("company_id","source_kind","source_id");--> statement-breakpoint
CREATE UNIQUE INDEX "decision_queues_company_key_uq" ON "decision_queues" USING btree ("company_id","key");--> statement-breakpoint
CREATE INDEX "decision_queues_company_updated_idx" ON "decision_queues" USING btree ("company_id","updated_at");--> statement-breakpoint
CREATE UNIQUE INDEX "decision_triage_company_source_uq" ON "decision_triage" USING btree ("company_id","source_kind","source_id");--> statement-breakpoint
CREATE INDEX "decision_triage_company_decide_by_idx" ON "decision_triage" USING btree ("company_id","decide_by");--> statement-breakpoint
CREATE INDEX "decision_triage_events_company_source_created_idx" ON "decision_triage_events" USING btree ("company_id","source_kind","source_id","created_at");--> statement-breakpoint
CREATE INDEX "decision_triage_events_queue_created_idx" ON "decision_triage_events" USING btree ("queue_id","created_at");
File diff suppressed because it is too large. Load diff
@@ -1373,6 +1373,13 @@
"when": 1785175200000,
"tag": "0197_decisions_v1",
"breakpoints": true
},
{
"idx": 198,
"version": "7",
"when": 1785603158310,
"tag": "0198_decision_queues_and_triage",
"breakpoints": true
}
]
}
+183
View File
@@ -0,0 +1,183 @@
import { sql } from "drizzle-orm";
import {
boolean,
check,
date,
foreignKey,
index,
integer,
jsonb,
pgTable,
text,
timestamp,
uniqueIndex,
uuid,
} from "drizzle-orm/pg-core";
import type { DecisionQueueSeedRule } from "@paperclipai/shared";
import { agentApiKeys } from "./agent_api_keys.js";
import { agents } from "./agents.js";
import { companies } from "./companies.js";
import { heartbeatRuns } from "./heartbeat_runs.js";
export const decisionQueues = pgTable(
"decision_queues",
{
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
key: text("key").notNull(),
title: text("title").notNull(),
description: text("description"),
createdByType: text("created_by_type").notNull(),
createdByAgentId: uuid("created_by_agent_id").references(() => agents.id),
createdByUserId: text("created_by_user_id"),
createdByRunId: uuid("created_by_run_id").references(() => heartbeatRuns.id),
createdByAgentApiKeyId: uuid("created_by_agent_api_key_id").references(() => agentApiKeys.id),
retentionDays: integer("retention_days"),
seedRules: jsonb("seed_rules").$type<DecisionQueueSeedRule[]>().notNull().default([]),
seedRulesEnabled: boolean("seed_rules_enabled").notNull().default(false),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
},
(table) => ({
companyKeyUq: uniqueIndex("decision_queues_company_key_uq").on(table.companyId, table.key),
companyUpdatedIdx: index("decision_queues_company_updated_idx").on(table.companyId, table.updatedAt),
idCompanyUq: uniqueIndex("decision_queues_id_company_uq").on(table.id, table.companyId),
creatorCheck: check(
"decision_queues_creator_check",
sql`(
(${table.createdByType} = 'agent' AND ${table.createdByAgentId} IS NOT NULL AND ${table.createdByUserId} IS NULL)
OR (${table.createdByType} = 'user' AND ${table.createdByAgentId} IS NULL AND ${table.createdByUserId} IS NOT NULL)
OR (${table.createdByType} = 'system' AND ${table.createdByAgentId} IS NULL AND ${table.createdByUserId} IS NULL)
)`,
),
retentionDaysCheck: check(
"decision_queues_retention_days_check",
sql`${table.retentionDays} IS NULL OR (${table.retentionDays} >= 1 AND ${table.retentionDays} <= 3650)`,
),
}),
);
export const decisionQueueItems = pgTable(
"decision_queue_items",
{
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
queueId: uuid("queue_id").notNull(),
sourceKind: text("source_kind").notNull(),
sourceId: text("source_id").notNull(),
addedByType: text("added_by_type").notNull(),
addedByAgentId: uuid("added_by_agent_id").references(() => agents.id),
addedByUserId: text("added_by_user_id"),
addedByRunId: uuid("added_by_run_id").references(() => heartbeatRuns.id),
addedByAgentApiKeyId: uuid("added_by_agent_api_key_id").references(() => agentApiKeys.id),
responsibleUserId: text("responsible_user_id"),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
},
(table) => ({
queueSourceUq: uniqueIndex("decision_queue_items_queue_source_uq").on(
table.queueId,
table.sourceKind,
table.sourceId,
),
companySourceIdx: index("decision_queue_items_company_source_idx").on(
table.companyId,
table.sourceKind,
table.sourceId,
),
queueCompanyFk: foreignKey({
columns: [table.queueId, table.companyId],
foreignColumns: [decisionQueues.id, decisionQueues.companyId],
name: "decision_queue_items_queue_company_fk",
}).onDelete("cascade"),
actorCheck: check(
"decision_queue_items_actor_check",
sql`(
(${table.addedByType} = 'agent' AND ${table.addedByAgentId} IS NOT NULL AND ${table.addedByUserId} IS NULL)
OR (${table.addedByType} = 'user' AND ${table.addedByAgentId} IS NULL AND ${table.addedByUserId} IS NOT NULL)
OR (${table.addedByType} = 'system' AND ${table.addedByAgentId} IS NULL AND ${table.addedByUserId} IS NULL)
)`,
),
}),
);
export const decisionTriage = pgTable(
"decision_triage",
{
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
sourceKind: text("source_kind").notNull(),
sourceId: text("source_id").notNull(),
decideBy: text("decide_by"),
decideByDate: date("decide_by_date"),
snoozedUntil: timestamp("snoozed_until", { withTimezone: true }),
setByType: text("set_by_type").notNull(),
setByAgentId: uuid("set_by_agent_id").references(() => agents.id),
setByUserId: text("set_by_user_id"),
setByRunId: uuid("set_by_run_id").references(() => heartbeatRuns.id),
setByAgentApiKeyId: uuid("set_by_agent_api_key_id").references(() => agentApiKeys.id),
responsibleUserId: text("responsible_user_id"),
version: integer("version").notNull().default(1),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
},
(table) => ({
companySourceUq: uniqueIndex("decision_triage_company_source_uq").on(
table.companyId,
table.sourceKind,
table.sourceId,
),
companyDecideByIdx: index("decision_triage_company_decide_by_idx").on(table.companyId, table.decideBy),
actorCheck: check(
"decision_triage_actor_check",
sql`(
(${table.setByType} = 'agent' AND ${table.setByAgentId} IS NOT NULL AND ${table.setByUserId} IS NULL)
OR (${table.setByType} = 'user' AND ${table.setByAgentId} IS NULL AND ${table.setByUserId} IS NOT NULL)
)`,
),
decideByCheck: check(
"decision_triage_decide_by_check",
sql`(
(${table.decideBy} IS NULL AND ${table.decideByDate} IS NULL)
OR (${table.decideBy} IN ('today', 'this_week', 'whenever') AND ${table.decideByDate} IS NULL)
OR (${table.decideBy} = 'date' AND ${table.decideByDate} IS NOT NULL)
)`,
),
}),
);
export const decisionTriageEvents = pgTable(
"decision_triage_events",
{
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
queueId: uuid("queue_id").references(() => decisionQueues.id),
sourceKind: text("source_kind"),
sourceId: text("source_id"),
action: text("action").notNull(),
actorType: text("actor_type").notNull(),
actorAgentId: uuid("actor_agent_id").references(() => agents.id),
actorUserId: text("actor_user_id"),
actorRunId: uuid("actor_run_id").references(() => heartbeatRuns.id),
agentApiKeyId: uuid("agent_api_key_id").references(() => agentApiKeys.id),
responsibleUserId: text("responsible_user_id"),
details: jsonb("details").$type<Record<string, unknown>>().notNull().default({}),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
},
(table) => ({
companySourceCreatedIdx: index("decision_triage_events_company_source_created_idx").on(
table.companyId,
table.sourceKind,
table.sourceId,
table.createdAt,
),
queueCreatedIdx: index("decision_triage_events_queue_created_idx").on(table.queueId, table.createdAt),
actorCheck: check(
"decision_triage_events_actor_check",
sql`(
(${table.actorType} = 'agent' AND ${table.actorAgentId} IS NOT NULL AND ${table.actorUserId} IS NULL)
OR (${table.actorType} = 'user' AND ${table.actorAgentId} IS NULL AND ${table.actorUserId} IS NOT NULL)
OR (${table.actorType} = 'system' AND ${table.actorAgentId} IS NULL AND ${table.actorUserId} IS NULL)
)`,
),
}),
);
+6
View File
@@ -84,6 +84,12 @@ export { userInboxAgentPolicies } from "./user_inbox_agent_policies.js";
export { inboxDismissals } from "./inbox_dismissals.js";
export { feedbackVotes } from "./feedback_votes.js";
export { decisionTrainingExamples } from "./decision_training_examples.js";
export {
decisionQueues,
decisionQueueItems,
decisionTriage,
decisionTriageEvents,
} from "./decision_queues.js";
export { feedbackExports } from "./feedback_exports.js";
export { issueReadStates } from "./issue_read_states.js";
export { assets } from "./assets.js";
+22
View File
@@ -102,6 +102,28 @@ export type {
AttentionSubjectKind,
AttentionWorkspaceRef,
} from "./types/attention.js";
export { ATTENTION_SOURCE_KINDS } from "./types/attention.js";
export type {
DecisionQueue,
DecisionQueueItem,
DecisionQueueSeedRule,
DecisionQueueSeedRuleSignal,
DecisionTriage,
DecisionTriageDecideBy,
} from "./types/decision-queue.js";
export {
decisionAttentionSourceKindSchema,
decisionQueueKeySchema,
createDecisionQueueSchema,
updateDecisionQueueSchema,
addDecisionQueueItemSchema,
decisionTriageDecideBySchema,
updateDecisionTriageSchema,
type CreateDecisionQueueInput,
type UpdateDecisionQueueInput,
type AddDecisionQueueItemInput,
type UpdateDecisionTriageInput,
} from "./validators/decision-queue.js";
export type {
DecisionTrainingExample,
DecisionTrainingNotesHistoryEntry,
+15 -12
View File
@@ -1,17 +1,20 @@
import type { InboxDismissalKind } from "./inbox-dismissal.js";
export type AttentionSourceKind =
| "approval"
| "decision"
| "issue_thread_interaction"
| "join_request"
| "recovery_action"
| "productivity_review"
| "blocker_attention"
| "review"
| "failed_run"
| "budget_alert"
| "agent_error_alert";
export const ATTENTION_SOURCE_KINDS = [
"approval",
"decision",
"issue_thread_interaction",
"join_request",
"recovery_action",
"productivity_review",
"blocker_attention",
"review",
"failed_run",
"budget_alert",
"agent_error_alert",
] as const;
export type AttentionSourceKind = (typeof ATTENTION_SOURCE_KINDS)[number];
export type AttentionSubjectKind =
| "approval"
@@ -0,0 +1,63 @@
import type { AttentionSourceKind } from "./attention.js";
export type DecisionQueueSeedRuleSignal =
| "issue_has_pull_request_work_product"
| "plan_document_confirmation"
| "ask_user_questions";
export interface DecisionQueueSeedRule {
key: string;
description: string;
signal: DecisionQueueSeedRuleSignal;
}
export interface DecisionQueueItem {
id: string;
companyId: string;
queueId: string;
sourceKind: AttentionSourceKind;
sourceId: string;
addedByType: "agent" | "user" | "system";
addedByAgentId: string | null;
addedByUserId: string | null;
addedByRunId: string | null;
responsibleUserId: string | null;
createdAt: Date;
}
export interface DecisionQueue {
id: string;
companyId: string;
key: string;
title: string;
description: string | null;
createdByType: "agent" | "user" | "system";
createdByAgentId: string | null;
createdByUserId: string | null;
createdByRunId: string | null;
retentionDays: number | null;
seedRules: DecisionQueueSeedRule[];
seedRulesEnabled: boolean;
itemCount: number;
createdAt: Date;
updatedAt: Date;
}
export type DecisionTriageDecideBy = "today" | "this_week" | "whenever" | string;
export interface DecisionTriage {
id: string;
companyId: string;
sourceKind: AttentionSourceKind;
sourceId: string;
decideBy: DecisionTriageDecideBy | null;
snoozedUntil: Date | null;
setByType: "agent" | "user";
setByAgentId: string | null;
setByUserId: string | null;
setByRunId: string | null;
responsibleUserId: string | null;
version: number;
createdAt: Date;
updatedAt: Date;
}
+9
View File
@@ -26,6 +26,15 @@ export type {
AttentionSubjectKind,
AttentionWorkspaceRef,
} from "./attention.js";
export { ATTENTION_SOURCE_KINDS } from "./attention.js";
export type {
DecisionQueue,
DecisionQueueItem,
DecisionQueueSeedRule,
DecisionQueueSeedRuleSignal,
DecisionTriage,
DecisionTriageDecideBy,
} from "./decision-queue.js";
export type {
DecisionTrainingExample,
DecisionTrainingNotesHistoryEntry,
@@ -0,0 +1,49 @@
import { z } from "zod";
import { ATTENTION_SOURCE_KINDS } from "../types/attention.js";
export const decisionAttentionSourceKindSchema = z.enum(ATTENTION_SOURCE_KINDS);
export const decisionQueueKeySchema = z.string()
.trim()
.min(1)
.max(80)
.regex(/^[a-z0-9]+(?:-[a-z0-9]+)*$/, "Queue key must be URL-safe lowercase kebab-case");
export const createDecisionQueueSchema = z.object({
key: decisionQueueKeySchema,
title: z.string().trim().min(1).max(120),
description: z.string().trim().max(2_000).nullable().optional(),
retentionDays: z.number().int().min(1).max(3_650).nullable().optional(),
}).strict();
export const updateDecisionQueueSchema = z.object({
title: z.string().trim().min(1).max(120).optional(),
description: z.string().trim().max(2_000).nullable().optional(),
retentionDays: z.number().int().min(1).max(3_650).nullable().optional(),
seedRulesEnabled: z.boolean().optional(),
}).strict().refine((value) => Object.keys(value).length > 0, "At least one queue field is required");
export const addDecisionQueueItemSchema = z.object({
sourceKind: decisionAttentionSourceKindSchema,
sourceId: z.string().trim().min(1).max(500),
}).strict();
const calendarDateSchema = z.string().regex(/^\d{4}-\d{2}-\d{2}$/).refine((value) => {
const parsed = new Date(`${value}T00:00:00.000Z`);
return Number.isFinite(parsed.getTime()) && parsed.toISOString().slice(0, 10) === value;
}, "Invalid calendar date");
export const decisionTriageDecideBySchema = z.union([
z.enum(["today", "this_week", "whenever"]),
calendarDateSchema,
]);
export const updateDecisionTriageSchema = z.object({
decideBy: decisionTriageDecideBySchema.nullable().optional(),
snoozedUntil: z.string().datetime({ offset: true }).nullable().optional(),
}).strict().refine((value) => Object.keys(value).length > 0, "At least one triage field is required");
export type CreateDecisionQueueInput = z.infer<typeof createDecisionQueueSchema>;
export type UpdateDecisionQueueInput = z.infer<typeof updateDecisionQueueSchema>;
export type AddDecisionQueueItemInput = z.infer<typeof addDecisionQueueItemSchema>;
export type UpdateDecisionTriageInput = z.infer<typeof updateDecisionTriageSchema>;
@@ -1734,7 +1734,7 @@ describeEmbeddedPostgres("authorization service", () => {
});
});
it("scopes task bridge keys away from company-wide reads and unrelated issue writes", async () => {
it("scopes task bridge JWTs away from company-wide reads and unrelated issue writes", async () => {
const company = await createCompany(db, "TaskBridge");
const bridgeAgent = await createAgent(db, company.id);
const targetAgent = await createAgent(db, company.id);
@@ -1751,7 +1751,7 @@ describeEmbeddedPostgres("authorization service", () => {
type: "agent" as const,
agentId: bridgeAgent.id,
companyId: company.id,
source: "agent_key" as const,
source: "agent_jwt" as const,
keyId,
keyScope: {
kind: "task_bridge" as const,
@@ -0,0 +1,416 @@
import { randomUUID } from "node:crypto";
import express from "express";
import request from "supertest";
import { and, eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
activityLog,
agents,
approvals,
companies,
createDb,
decisionQueueItems,
decisionQueues,
decisionTriage,
decisionTriageEvents,
issueThreadInteractions,
issueWorkProducts,
issues,
} from "@paperclipai/db";
import type { AttentionItem } from "@paperclipai/shared";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import { errorHandler } from "../middleware/index.js";
import { decisionQueueRoutes } from "../routes/decision-queues.js";
import { decisionQueueService } from "../services/decision-queues.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres decision queue tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
describeEmbeddedPostgres("decision queue routes", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-");
db = createDb(tempDb.connectionString);
}, 30_000);
afterEach(async () => {
await db.delete(decisionTriageEvents);
await db.delete(decisionQueueItems);
await db.delete(decisionTriage);
await db.delete(decisionQueues);
await db.delete(activityLog);
await db.delete(issueThreadInteractions);
await db.delete(issueWorkProducts);
await db.delete(approvals);
await db.delete(issues);
await db.delete(agents);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
async function seed() {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const interactionId = randomUUID();
const approvalId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Decision Queue Co",
issuePrefix: "DQC",
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "Prioritizer",
role: "engineer",
status: "idle",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
identifier: "DQC-1",
title: "Review the rollout",
status: "in_review",
assigneeAgentId: agentId,
});
await db.insert(issueThreadInteractions).values({
id: interactionId,
companyId,
issueId,
kind: "ask_user_questions",
status: "pending",
payload: { version: 1, questions: [] } as never,
});
await db.insert(approvals).values({
id: approvalId,
companyId,
type: "request_board_approval",
status: "pending",
payload: {},
});
return { companyId, agentId, issueId, interactionId, approvalId };
}
function app(actor: Record<string, unknown>) {
const testApp = express();
testApp.use(express.json());
testApp.use((req, _res, next) => {
(req as any).actor = actor;
next();
});
testApp.use("/api", decisionQueueRoutes(db));
testApp.use(errorHandler);
return testApp;
}
function boardActor(companyId: string, userId = "board-user") {
return {
type: "board",
source: "local_implicit",
userId,
companyIds: [companyId],
isInstanceAdmin: false,
};
}
function agentActor(companyId: string, agentId: string) {
return {
type: "agent",
source: "agent_key",
companyId,
agentId,
keyId: null,
keyScope: { kind: "standard" },
runId: null,
};
}
it("creates idempotently, patches, lists by updated time, and audits queue mutations", async () => {
const { companyId } = await seed();
const board = boardActor(companyId);
const first = await request(app(board)).post(`/api/companies/${companyId}/decision-queues`).send({
key: "launches",
title: "Launches",
description: "Ship decisions",
}).expect(201);
const repeated = await request(app(board)).post(`/api/companies/${companyId}/decision-queues`).send({
key: "launches",
title: "Ignored duplicate title",
}).expect(200);
expect(repeated.body.id).toBe(first.body.id);
expect(repeated.body.title).toBe("Launches");
const patched = await request(app(board))
.patch(`/api/companies/${companyId}/decision-queues/launches`)
.send({ title: "Launch desk", retentionDays: 45 })
.expect(200);
expect(patched.body).toMatchObject({ title: "Launch desk", retentionDays: 45 });
await request(app(board)).post(`/api/companies/${companyId}/decision-queues`).send({
key: "older",
title: "Older",
}).expect(201);
const listed = await request(app(board)).get(`/api/companies/${companyId}/decision-queues`).expect(200);
expect(listed.body.map((queue: { key: string }) => queue.key)).toEqual(["older", "launches"]);
const seedRules = await request(app(board))
.get(`/api/companies/${companyId}/decision-queue-seed-rules`)
.expect(200);
expect(seedRules.body.map((seed: { key: string }) => seed.key)).toEqual(["prs", "plans", "questions"]);
const events = await db.select().from(decisionTriageEvents)
.where(eq(decisionTriageEvents.queueId, first.body.id));
expect(events.map((event) => event.action)).toEqual(["queue.created", "queue.updated"]);
expect(events.every((event) => event.actorUserId === "board-user")).toBe(true);
});
it("adds and removes three attention source kinds and hides board-only membership from agents", async () => {
const { companyId, agentId, issueId, interactionId, approvalId } = await seed();
const board = boardActor(companyId);
await request(app(board)).post(`/api/companies/${companyId}/decision-queues`).send({
key: "triage",
title: "Triage",
}).expect(201);
for (const source of [
{ sourceKind: "approval", sourceId: approvalId },
{ sourceKind: "issue_thread_interaction", sourceId: interactionId },
{ sourceKind: "review", sourceId: issueId },
]) {
await request(app(board))
.post(`/api/companies/${companyId}/decision-queues/triage/items`)
.send(source)
.expect(201);
}
const items = await request(app(board))
.get(`/api/companies/${companyId}/decision-queues/triage/items`)
.expect(200);
expect(items.body).toHaveLength(3);
const agentList = await request(app(agentActor(companyId, agentId)))
.get(`/api/companies/${companyId}/decision-queues`)
.expect(200);
expect(agentList.body[0].itemCount).toBe(2);
const agentItems = await request(app(agentActor(companyId, agentId)))
.get(`/api/companies/${companyId}/decision-queues/triage/items`)
.expect(200);
expect(agentItems.body.map((item: { sourceKind: string }) => item.sourceKind).sort()).toEqual([
"issue_thread_interaction",
"review",
]);
await request(app(board))
.delete(`/api/companies/${companyId}/decision-queues/triage/items/review/${issueId}`)
.expect(200);
expect(await db.select().from(decisionQueueItems).where(and(
eq(decisionQueueItems.companyId, companyId),
eq(decisionQueueItems.sourceKind, "review"),
))).toHaveLength(0);
});
it("materializes data-backed starter queues from plan, question, and pull-request signals", async () => {
const { companyId, issueId, interactionId } = await seed();
await db.insert(issueWorkProducts).values({
companyId,
issueId,
type: "pull_request",
provider: "github",
title: "PR 42",
status: "open",
});
function attentionItem(input: {
sourceKind: AttentionItem["sourceKind"];
sourceId: string;
subjectKind: AttentionItem["subject"]["kind"];
metadata?: Record<string, unknown>;
issueId?: string;
}): AttentionItem {
return {
id: `${input.sourceKind}:${input.sourceId}`,
companyId,
sourceKind: input.sourceKind,
subject: {
kind: input.subjectKind,
id: input.sourceId,
companyId,
title: "Seed candidate",
identifier: null,
status: "pending",
href: null,
metadata: input.metadata,
},
whyNow: "test",
decisionVerbs: [],
inlineResolvable: true,
entryRule: "test",
exitRule: "test",
dedupKey: `${input.sourceKind}:${input.sourceId}`,
dismissalKey: `${input.sourceKind}:${input.sourceId}`,
dismissal: null,
severity: "medium",
rank: 1,
activityAt: new Date().toISOString(),
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
relatedIssue: input.issueId ? {
kind: "issue",
id: input.issueId,
companyId,
title: "Issue",
identifier: "DQC-1",
status: "in_review",
href: null,
} : null,
project: null,
workspace: null,
detail: null,
trainingExampleId: null,
};
}
const planId = randomUUID();
const candidates = [
attentionItem({ sourceKind: "review", sourceId: issueId, subjectKind: "issue", issueId }),
attentionItem({
sourceKind: "issue_thread_interaction",
sourceId: planId,
subjectKind: "interaction",
metadata: { kind: "request_confirmation", isPlanTarget: true, issueId },
issueId,
}),
attentionItem({
sourceKind: "issue_thread_interaction",
sourceId: interactionId,
subjectKind: "interaction",
metadata: { kind: "ask_user_questions", issueId },
issueId,
}),
];
await decisionQueueService(db).materializeSeededQueues(companyId, candidates);
const queues = await db.select().from(decisionQueues).where(eq(decisionQueues.companyId, companyId));
expect(queues.map((queue) => queue.key).sort()).toEqual(["plans", "prs", "questions"]);
expect(queues.every((queue) => queue.seedRulesEnabled && queue.seedRules.length === 1)).toBe(true);
const seededItems = await db.select({ queueId: decisionQueueItems.queueId })
.from(decisionQueueItems)
.where(eq(decisionQueueItems.companyId, companyId));
expect(seededItems).toHaveLength(5);
const prQueue = queues.find((queue) => queue.key === "prs")!;
expect(seededItems.filter((item) => item.queueId === prQueue.id)).toHaveLength(3);
const firstEvents = await db.select().from(decisionTriageEvents)
.where(eq(decisionTriageEvents.companyId, companyId));
const firstActivity = await db.select().from(activityLog).where(eq(activityLog.companyId, companyId));
const firstUpdatedAt = new Map(queues.map((queue) => [queue.key, queue.updatedAt.toISOString()]));
await decisionQueueService(db).materializeSeededQueues(companyId, candidates);
const unchangedQueues = await db.select().from(decisionQueues).where(eq(decisionQueues.companyId, companyId));
expect(unchangedQueues.every((queue) => queue.updatedAt.toISOString() === firstUpdatedAt.get(queue.key))).toBe(true);
expect(await db.select().from(decisionTriageEvents).where(eq(decisionTriageEvents.companyId, companyId)))
.toHaveLength(firstEvents.length);
expect(await db.select().from(activityLog).where(eq(activityLog.companyId, companyId)))
.toHaveLength(firstActivity.length);
const questionsQueue = queues.find((queue) => queue.key === "questions")!;
await db.update(decisionQueues).set({ seedRulesEnabled: false }).where(eq(decisionQueues.id, questionsQueue.id));
await decisionQueueService(db).materializeSeededQueues(companyId, [attentionItem({
sourceKind: "issue_thread_interaction",
sourceId: randomUUID(),
subjectKind: "interaction",
metadata: { kind: "ask_user_questions", issueId },
issueId,
})]);
expect(await db.select().from(decisionQueueItems).where(eq(decisionQueueItems.queueId, questionsQueue.id)))
.toHaveLength(1);
});
it("records agent decide-by, preserves override history, and exposes the board override attribution", async () => {
const { companyId, agentId, issueId } = await seed();
const agent = agentActor(companyId, agentId);
const board = boardActor(companyId, "override-user");
const agentSet = await request(app(agent))
.put(`/api/companies/${companyId}/decision-triage/review/${issueId}`)
.send({ decideBy: "today", snoozedUntil: "2026-08-03T12:00:00.000Z" })
.expect(200);
expect(agentSet.body).toMatchObject({
decideBy: "today",
setByType: "agent",
setByAgentId: agentId,
version: 1,
});
const overridden = await request(app(board))
.put(`/api/companies/${companyId}/decision-triage/review/${issueId}`)
.send({ decideBy: "2026-08-08", snoozedUntil: null })
.expect(200);
expect(overridden.body).toMatchObject({
decideBy: "2026-08-08",
setByType: "user",
setByUserId: "override-user",
version: 2,
});
const events = await db.select().from(decisionTriageEvents).where(and(
eq(decisionTriageEvents.companyId, companyId),
eq(decisionTriageEvents.sourceKind, "review"),
eq(decisionTriageEvents.sourceId, issueId),
));
expect(events).toHaveLength(2);
expect(events[0]?.actorAgentId).toBe(agentId);
expect(events[1]?.actorUserId).toBe("override-user");
expect(events[1]?.details).toMatchObject({ previousDecideBy: "today", decideBy: "2026-08-08" });
const current = await db.select().from(decisionTriage).where(eq(decisionTriage.sourceId, issueId));
expect(current[0]?.setByUserId).toBe("override-user");
});
it("rejects task-bridge JWTs and returns the same 404 for missing and unauthorized sources", async () => {
const { companyId, agentId, approvalId } = await seed();
const bridge = {
...agentActor(companyId, agentId),
source: "agent_jwt",
keyId: randomUUID(),
keyScope: { kind: "task_bridge", parentIssueId: randomUUID() },
};
await request(app(bridge)).post(`/api/companies/${companyId}/decision-queues`).send({
key: "blocked",
title: "Blocked",
}).expect(403);
await request(app(boardActor(companyId))).post(`/api/companies/${companyId}/decision-queues`).send({
key: "secure",
title: "Secure",
}).expect(201);
const standardAgent = app(agentActor(companyId, agentId));
const unauthorized = await request(standardAgent)
.post(`/api/companies/${companyId}/decision-queues/secure/items`)
.send({ sourceKind: "approval", sourceId: approvalId })
.expect(404);
const missing = await request(standardAgent)
.post(`/api/companies/${companyId}/decision-queues/secure/items`)
.send({ sourceKind: "approval", sourceId: randomUUID() })
.expect(404);
expect(unauthorized.body).toEqual(missing.body);
});
});
+2
View File
@@ -43,6 +43,7 @@ import { dashboardRoutes } from "./routes/dashboard.js";
import { attentionRoutes } from "./routes/attention.js";
import { decisionTrainingRoutes } from "./routes/decision-training.js";
import { decisionRoutes } from "./routes/decisions.js";
import { decisionQueueRoutes } from "./routes/decision-queues.js";
import type { DecisionServiceOptions } from "./services/decisions.js";
import { userProfileRoutes } from "./routes/user-profiles.js";
import { sidebarBadgeRoutes } from "./routes/sidebar-badges.js";
@@ -407,6 +408,7 @@ export async function createApp(
api.use(attentionRoutes(db));
api.use(decisionTrainingRoutes(db));
api.use(decisionRoutes(db, opts.decisionServiceOptions));
api.use(decisionQueueRoutes(db));
api.use(userProfileRoutes(db));
api.use(sidebarBadgeRoutes(db));
api.use(sidebarPreferenceRoutes(db));
+172
View File
@@ -0,0 +1,172 @@
import { Router } from "express";
import type { Db } from "@paperclipai/db";
import {
addDecisionQueueItemSchema,
createDecisionQueueSchema,
decisionAttentionSourceKindSchema,
updateDecisionQueueSchema,
updateDecisionTriageSchema,
} from "@paperclipai/shared";
import { forbidden } from "../errors.js";
import { validate } from "../middleware/validate.js";
import {
authorizationDeniedDetails,
authorizationService,
type AuthorizationAction,
} from "../services/authorization.js";
import {
DECISION_QUEUE_SEEDS,
decisionQueueService,
type DecisionMutationActor,
} from "../services/decision-queues.js";
import { assertBoardOrAgent, assertCompanyAccess, getActorInfo } from "./authz.js";
function mutationActor(req: Parameters<typeof getActorInfo>[0]): DecisionMutationActor {
const actor = getActorInfo(req);
return {
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
userId: actor.actorType === "user" ? actor.actorId : null,
runId: actor.runId,
agentApiKeyId: actor.agentApiKeyId,
responsibleUserId: req.actor.onBehalfOfUserId ?? (actor.actorType === "user" ? actor.actorId : null),
};
}
async function assertDecisionAccess(
db: Db,
req: Parameters<typeof getActorInfo>[0],
companyId: string,
action: Extract<AuthorizationAction, "decision_queue:read" | "decision_queue:manage" | "decision_triage:manage">,
) {
assertBoardOrAgent(req);
assertCompanyAccess(req, companyId);
const decision = await authorizationService(db).decide({
actor: req.actor,
action,
resource: { type: "company", companyId },
});
if (!decision.allowed) {
throw forbidden(decision.explanation, authorizationDeniedDetails(decision));
}
}
function sourceParams(req: Parameters<typeof getActorInfo>[0]) {
const parsed = decisionAttentionSourceKindSchema.safeParse(req.params.sourceKind);
if (!parsed.success) return null;
const sourceId = (req.params.sourceId as string | undefined)?.trim();
if (!sourceId || sourceId.length > 500) return null;
return { sourceKind: parsed.data, sourceId };
}
export function decisionQueueRoutes(db: Db) {
const router = Router();
const svc = decisionQueueService(db);
router.get("/companies/:companyId/decision-queue-seed-rules", async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:read");
res.json(DECISION_QUEUE_SEEDS);
});
router.get("/companies/:companyId/decision-queues", async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:read");
res.json(await svc.list(companyId, req.actor));
});
router.post("/companies/:companyId/decision-queues", validate(createDecisionQueueSchema), async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:manage");
const result = await svc.create({ companyId, actor: mutationActor(req), ...req.body });
res.status(result.created ? 201 : 200).json(result.queue);
});
router.patch("/companies/:companyId/decision-queues/:key", validate(updateDecisionQueueSchema), async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:manage");
res.json(await svc.update({
companyId,
key: req.params.key as string,
patch: req.body,
authActor: req.actor,
actor: mutationActor(req),
}));
});
router.get("/companies/:companyId/decision-queues/:key/items", async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:read");
res.json(await svc.listItems(companyId, req.params.key as string, req.actor));
});
router.post(
"/companies/:companyId/decision-queues/:key/items",
validate(addDecisionQueueItemSchema),
async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:manage");
const result = await svc.addItem({
companyId,
key: req.params.key as string,
sourceKind: req.body.sourceKind,
sourceId: req.body.sourceId,
authActor: req.actor,
actor: mutationActor(req),
});
res.status(result.created ? 201 : 200).json(result.item);
},
);
router.delete("/companies/:companyId/decision-queues/:key/items/:sourceKind/:sourceId", async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:manage");
const source = sourceParams(req);
if (!source) {
res.status(400).json({ error: "Invalid attention source identity" });
return;
}
res.json(await svc.removeItem({
companyId,
key: req.params.key as string,
...source,
authActor: req.actor,
actor: mutationActor(req),
}));
});
router.get("/companies/:companyId/decision-triage/:sourceKind/:sourceId", async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_queue:read");
const source = sourceParams(req);
if (!source) {
res.status(400).json({ error: "Invalid attention source identity" });
return;
}
res.json(await svc.getTriage(companyId, source.sourceKind, source.sourceId, req.actor));
});
router.put(
"/companies/:companyId/decision-triage/:sourceKind/:sourceId",
validate(updateDecisionTriageSchema),
async (req, res) => {
const companyId = req.params.companyId as string;
await assertDecisionAccess(db, req, companyId, "decision_triage:manage");
const source = sourceParams(req);
if (!source) {
res.status(400).json({ error: "Invalid attention source identity" });
return;
}
res.json(await svc.updateTriage({
companyId,
...source,
...req.body,
authActor: req.actor,
actor: mutationActor(req),
}));
},
);
return router;
}
+1
View File
@@ -24,6 +24,7 @@ export { activityRoutes } from "./activity.js";
export { dashboardRoutes } from "./dashboard.js";
export { attentionRoutes } from "./attention.js";
export { decisionRoutes } from "./decisions.js";
export { decisionQueueRoutes } from "./decision-queues.js";
export { sidebarBadgeRoutes } from "./sidebar-badges.js";
export { sidebarPreferenceRoutes } from "./sidebar-preferences.js";
export { resourceMembershipRoutes } from "./resource-memberships.js";
+2
View File
@@ -42,6 +42,7 @@ import { budgetService } from "./budgets.js";
import { issueService } from "./issues.js";
import { parseIssueExecutionState } from "./issue-execution-policy.js";
import { isProspectiveBlockedTransition } from "./routable-blocked.js";
import { decisionQueueService } from "./decision-queues.js";
const ATTENTION_SOURCE_KINDS: AttentionSourceKind[] = [
"approval",
@@ -1386,6 +1387,7 @@ export function attentionService(db: Db, options: AttentionServiceOptions = {})
const items = [...deduped.values()]
.sort(compareAttentionItems)
.map((item, index) => ({ ...item, rank: index + 1 }));
await decisionQueueService(db).materializeSeededQueues(companyId, items);
if (options.userId) {
const trainable: Array<{ sourceKind: "approval" | "interaction"; sourceId: string }> = [];
for (const item of items) {
+47 -3
View File
@@ -63,6 +63,9 @@ export type AuthorizationAction =
| "agent:read"
| "agent:wake"
| "company_scope:read"
| "decision_queue:manage"
| "decision_queue:read"
| "decision_triage:manage"
| "issue:comment"
| "issue:mutate"
| "issue:read"
@@ -145,6 +148,9 @@ function permissionForAction(action: AuthorizationAction): PermissionKey | null
action === "agent:read" ||
action === "agent:wake" ||
action === "company_scope:read" ||
action === "decision_queue:manage" ||
action === "decision_queue:read" ||
action === "decision_triage:manage" ||
action === "issue:read" ||
action === "project:read" ||
action === "runtime:manage" ||
@@ -969,6 +975,9 @@ export function authorizationService(db: Db) {
if (
input.action === "company_scope:read" ||
input.action === "decision_queue:manage" ||
input.action === "decision_queue:read" ||
input.action === "decision_triage:manage" ||
input.action === "agent_config:read" ||
input.action === "agent_config:update" ||
input.action === "skill_config:update" ||
@@ -1140,6 +1149,9 @@ export function authorizationService(db: Db) {
if (
input.action === "company_scope:read" ||
input.action === "decision_queue:manage" ||
input.action === "decision_queue:read" ||
input.action === "decision_triage:manage" ||
input.action === "agent:read" ||
input.action === "agent:wake" ||
input.action === "project:read" ||
@@ -1206,6 +1218,9 @@ export function authorizationService(db: Db) {
if (
input.action === "company_scope:read" ||
input.action === "decision_queue:manage" ||
input.action === "decision_queue:read" ||
input.action === "decision_triage:manage" ||
input.action === "agent:read" ||
input.action === "agent:wake" ||
input.action === "project:read" ||
@@ -1570,6 +1585,7 @@ export function authorizationService(db: Db) {
if (
input.action === "agent:read" ||
input.action === "company_scope:read" ||
input.action === "decision_queue:read" ||
input.action === "issue:read" ||
input.action === "project:read"
) {
@@ -1580,7 +1596,12 @@ export function authorizationService(db: Db) {
});
}
if (
(input.action === "issue:comment" || input.action === "issue:mutate") &&
(
input.action === "issue:comment" ||
input.action === "issue:mutate" ||
input.action === "decision_queue:manage" ||
input.action === "decision_triage:manage"
) &&
membership.membershipRole !== "viewer"
) {
return allow({
@@ -1638,6 +1659,9 @@ export function authorizationService(db: Db) {
if (
input.action === "agent:read" ||
input.action === "company_scope:read" ||
input.action === "decision_queue:manage" ||
input.action === "decision_queue:read" ||
input.action === "decision_triage:manage" ||
input.action === "issue:read" ||
input.action === "project:read" ||
input.action === "runtime:manage" ||
@@ -1647,7 +1671,10 @@ export function authorizationService(db: Db) {
// Mirroring the tasks:assign carve-out above, viewers keep the
// read-only visibility actions but not the privileged ones.
const requiresNonViewer =
input.action === "runtime:manage" || input.action === "secrets:read";
input.action === "runtime:manage" ||
input.action === "secrets:read" ||
input.action === "decision_queue:manage" ||
input.action === "decision_triage:manage";
if (membership && (!requiresNonViewer || membership.membershipRole !== "viewer")) {
return allow({
action: input.action,
@@ -1725,7 +1752,7 @@ export function authorizationService(db: Db) {
if (skillTestDecision) return skillTestDecision;
}
if (input.actor.source === "agent_key" && input.actor.keyScope?.kind === "task_bridge") {
if (input.actor.keyScope?.kind === "task_bridge") {
const keyId = input.actor.keyId ?? null;
if (!keyId) {
return deny({
@@ -1771,6 +1798,7 @@ export function authorizationService(db: Db) {
input.action === "agent:read" ||
input.action === "agent:wake" ||
input.action === "company_scope:read" ||
input.action === "decision_queue:read" ||
input.action === "issue:comment" ||
input.action === "issue:read" ||
input.action === "project:read" ||
@@ -1907,6 +1935,7 @@ export function authorizationService(db: Db) {
if (
input.action === "agent:read" ||
input.action === "company_scope:read" ||
input.action === "decision_queue:read" ||
input.action === "issue:read" ||
input.action === "project:read" ||
input.action === "runtime:manage" ||
@@ -1919,6 +1948,21 @@ export function authorizationService(db: Db) {
});
}
if (input.action === "decision_queue:manage" || input.action === "decision_triage:manage") {
if (!isSimpleAssignableAgentStatus(actorAgent.status)) {
return deny({
action: input.action,
reason: "deny_missing_membership",
explanation: "Actor agent is not active in the target company.",
});
}
return allow({
action: input.action,
reason: "allow_company_agent",
explanation: "Allowed for an active standard-scope company agent.",
});
}
if (input.action === "agent:wake" && input.resource.type === "agent" && input.resource.agentId === actorAgentId) {
return allow({
action: input.action,
+749
View File
@@ -0,0 +1,749 @@
import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
agents,
approvals,
budgetIncidents,
decisionQueueItems,
decisionQueues,
decisions,
decisionTriage,
decisionTriageEvents,
heartbeatRuns,
issueApprovals,
issueRecoveryActions,
issueThreadInteractions,
issueWorkProducts,
issues,
joinRequests,
} from "@paperclipai/db";
import type {
AttentionItem,
AttentionSourceKind,
DecisionQueue,
DecisionQueueItem,
DecisionQueueSeedRule,
DecisionTriage,
} from "@paperclipai/shared";
import { notFound, unprocessable } from "../errors.js";
import { logActivity } from "./activity-log.js";
import {
authorizationService,
type AuthorizationActor,
type AuthorizationResource,
} from "./authorization.js";
export type DecisionMutationActor = {
actorType: "agent" | "user" | "system";
actorId: string;
agentId: string | null;
userId: string | null;
runId: string | null;
agentApiKeyId: string | null;
responsibleUserId: string | null;
};
type SeedDefinition = {
key: string;
title: string;
description: string;
rules: DecisionQueueSeedRule[];
};
export const DECISION_QUEUE_SEEDS: readonly SeedDefinition[] = [
{
key: "prs",
title: "PRs",
description: "Pull-request and merge decisions detected from issue work products.",
rules: [{
key: "issue-pull-request-work-product",
signal: "issue_has_pull_request_work_product",
description: "Attention items whose issue has a pull_request work product.",
}],
},
{
key: "plans",
title: "Plans",
description: "Plan revisions waiting for confirmation.",
rules: [{
key: "plan-document-confirmation",
signal: "plan_document_confirmation",
description: "Pending request_confirmation interactions bound to the issue's plan document.",
}],
},
{
key: "questions",
title: "Questions",
description: "Structured questions waiting for a board response.",
rules: [{
key: "ask-user-questions",
signal: "ask_user_questions",
description: "Pending ask_user_questions interactions.",
}],
},
] as const;
const SYSTEM_ACTOR: DecisionMutationActor = {
actorType: "system",
actorId: "decision_queue_seed",
agentId: null,
userId: null,
runId: null,
agentApiKeyId: null,
responsibleUserId: null,
};
function actorColumns(actor: DecisionMutationActor, prefix: "added" | "set" | "created") {
const values = {
[`${prefix}ByType`]: actor.actorType,
[`${prefix}ByAgentId`]: actor.agentId,
[`${prefix}ByUserId`]: actor.userId,
[`${prefix}ByRunId`]: actor.runId,
[`${prefix}ByAgentApiKeyId`]: actor.agentApiKeyId,
};
return values as Record<string, string | null>;
}
function eventActorColumns(actor: DecisionMutationActor) {
return {
actorType: actor.actorType,
actorAgentId: actor.agentId,
actorUserId: actor.userId,
actorRunId: actor.runId,
agentApiKeyId: actor.agentApiKeyId,
responsibleUserId: actor.responsibleUserId,
};
}
function toQueue(row: typeof decisionQueues.$inferSelect, itemCount: number): DecisionQueue {
return {
id: row.id,
companyId: row.companyId,
key: row.key,
title: row.title,
description: row.description ?? null,
createdByType: row.createdByType as DecisionQueue["createdByType"],
createdByAgentId: row.createdByAgentId ?? null,
createdByUserId: row.createdByUserId ?? null,
createdByRunId: row.createdByRunId ?? null,
retentionDays: row.retentionDays ?? null,
seedRules: row.seedRules ?? [],
seedRulesEnabled: row.seedRulesEnabled,
itemCount,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
function toQueueItem(row: typeof decisionQueueItems.$inferSelect): DecisionQueueItem {
return {
id: row.id,
companyId: row.companyId,
queueId: row.queueId,
sourceKind: row.sourceKind as AttentionSourceKind,
sourceId: row.sourceId,
addedByType: row.addedByType as DecisionQueueItem["addedByType"],
addedByAgentId: row.addedByAgentId ?? null,
addedByUserId: row.addedByUserId ?? null,
addedByRunId: row.addedByRunId ?? null,
responsibleUserId: row.responsibleUserId ?? null,
createdAt: row.createdAt,
};
}
function toTriage(row: typeof decisionTriage.$inferSelect): DecisionTriage {
return {
id: row.id,
companyId: row.companyId,
sourceKind: row.sourceKind as AttentionSourceKind,
sourceId: row.sourceId,
decideBy: row.decideBy === "date" ? row.decideByDate : row.decideBy,
snoozedUntil: row.snoozedUntil ?? null,
setByType: row.setByType as DecisionTriage["setByType"],
setByAgentId: row.setByAgentId ?? null,
setByUserId: row.setByUserId ?? null,
setByRunId: row.setByRunId ?? null,
responsibleUserId: row.responsibleUserId ?? null,
version: row.version,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
async function loadIssueResource(db: Db, companyId: string, issueId: string): Promise<AuthorizationResource | null> {
return db
.select({
issueId: issues.id,
companyId: issues.companyId,
projectId: issues.projectId,
parentIssueId: issues.parentId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
originKind: issues.originKind,
originId: issues.originId,
status: issues.status,
})
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId), isNull(issues.hiddenAt)))
.then((rows) => rows[0] ? ({ type: "issue", ...rows[0] } as const) : null);
}
async function sourceIssueId(
db: Db,
companyId: string,
sourceKind: AttentionSourceKind,
sourceId: string,
): Promise<{ exists: boolean; issueId: string | null; ownerAgentId?: string | null; agentId?: string | null }> {
switch (sourceKind) {
case "approval": {
const row = await db.select({ id: approvals.id, issueId: issueApprovals.issueId })
.from(approvals)
.leftJoin(issueApprovals, and(
eq(issueApprovals.companyId, companyId),
eq(issueApprovals.approvalId, approvals.id),
))
.where(and(eq(approvals.companyId, companyId), eq(approvals.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: row?.issueId ?? null };
}
case "decision": {
const row = await db.select({ issueId: decisions.originIssueId })
.from(decisions)
.where(and(eq(decisions.companyId, companyId), eq(decisions.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: row?.issueId ?? null };
}
case "issue_thread_interaction": {
const row = await db.select({ issueId: issueThreadInteractions.issueId })
.from(issueThreadInteractions)
.where(and(eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: row?.issueId ?? null };
}
case "recovery_action": {
const row = await db.select({ issueId: issueRecoveryActions.sourceIssueId })
.from(issueRecoveryActions)
.where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: row?.issueId ?? null };
}
case "productivity_review":
case "blocker_attention":
case "review": {
const row = await db.select({ id: issues.id })
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.id, sourceId), isNull(issues.hiddenAt)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: row?.id ?? null };
}
case "failed_run": {
const row = await db.select({ agentId: heartbeatRuns.agentId, contextSnapshot: heartbeatRuns.contextSnapshot })
.from(heartbeatRuns)
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, sourceId)))
.then((rows) => rows[0] ?? null);
const snapshot = row?.contextSnapshot && typeof row.contextSnapshot === "object"
? row.contextSnapshot as Record<string, unknown>
: {};
const issueId = typeof snapshot.issueId === "string"
? snapshot.issueId
: typeof snapshot.taskId === "string" ? snapshot.taskId : null;
return { exists: Boolean(row), issueId, ownerAgentId: row?.agentId ?? null };
}
case "agent_error_alert": {
const row = await db.select({ id: agents.id })
.from(agents)
.where(and(eq(agents.companyId, companyId), eq(agents.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: null, agentId: row?.id ?? null };
}
case "join_request": {
const row = await db.select({ id: joinRequests.id })
.from(joinRequests)
.where(and(eq(joinRequests.companyId, companyId), eq(joinRequests.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: null };
}
case "budget_alert": {
const row = await db.select({ id: budgetIncidents.id })
.from(budgetIncidents)
.where(and(eq(budgetIncidents.companyId, companyId), eq(budgetIncidents.id, sourceId)))
.then((rows) => rows[0] ?? null);
return { exists: Boolean(row), issueId: null };
}
}
}
export async function canReadDecisionSource(
db: Db,
actor: AuthorizationActor,
companyId: string,
sourceKind: AttentionSourceKind,
sourceId: string,
) {
const source = await sourceIssueId(db, companyId, sourceKind, sourceId);
if (!source.exists) return false;
const authz = authorizationService(db);
if (source.issueId) {
const resource = await loadIssueResource(db, companyId, source.issueId);
if (!resource) return false;
return (await authz.decide({ actor, action: "issue:read", resource })).allowed;
}
if (sourceKind === "agent_error_alert" && source.agentId) {
return (await authz.decide({
actor,
action: "agent:read",
resource: { type: "agent", companyId, agentId: source.agentId },
})).allowed;
}
if (sourceKind === "failed_run" && actor.type === "agent") {
return source.ownerAgentId === actor.agentId;
}
// Join requests, unlinked approvals, and budget incidents are board-only
// governance data. Same-company existence is deliberately not authority.
if (actor.type !== "board") return false;
return (await authz.decide({
actor,
action: "company_scope:read",
resource: { type: "company", companyId },
})).allowed;
}
async function requireSourceRead(
db: Db,
actor: AuthorizationActor,
companyId: string,
sourceKind: AttentionSourceKind,
sourceId: string,
) {
if (!(await canReadDecisionSource(db, actor, companyId, sourceKind, sourceId))) {
throw notFound("Attention source not found");
}
}
async function recordActivity(
db: Db,
actor: DecisionMutationActor,
input: { companyId: string; action: string; entityType: string; entityId: string; details?: Record<string, unknown> },
) {
await logActivity(db, {
companyId: input.companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
runId: actor.runId,
agentApiKeyId: actor.agentApiKeyId,
responsibleUserIdOverride: actor.responsibleUserId,
action: input.action,
entityType: input.entityType,
entityId: input.entityId,
details: input.details ?? null,
});
}
function itemIssueId(item: AttentionItem) {
if (item.subject.kind === "issue") return item.subject.id;
const related = item.relatedIssue?.kind === "issue" ? item.relatedIssue.id : null;
if (related) return related;
const metadataIssueId = item.subject.metadata?.issueId ?? item.subject.metadata?.originIssueId;
return typeof metadataIssueId === "string" ? metadataIssueId : null;
}
export function decisionQueueService(db: Db) {
async function getQueue(companyId: string, key: string) {
return db.select().from(decisionQueues)
.where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.key, key)))
.then((rows) => rows[0] ?? null);
}
async function visibleItems(companyId: string, queueId: string, authActor: AuthorizationActor) {
const rows = await db.select().from(decisionQueueItems)
.where(and(eq(decisionQueueItems.companyId, companyId), eq(decisionQueueItems.queueId, queueId)))
.orderBy(desc(decisionQueueItems.createdAt), desc(decisionQueueItems.id));
const visible: DecisionQueueItem[] = [];
for (const row of rows) {
if (await canReadDecisionSource(db, authActor, companyId, row.sourceKind as AttentionSourceKind, row.sourceId)) {
visible.push(toQueueItem(row));
}
}
return visible;
}
return {
create: async (input: {
companyId: string;
key: string;
title: string;
description?: string | null;
retentionDays?: number | null;
actor: DecisionMutationActor;
}) => {
return db.transaction(async (tx) => {
const txDb = tx as unknown as Db;
const inserted = await txDb.insert(decisionQueues).values({
companyId: input.companyId,
key: input.key,
title: input.title,
description: input.description ?? null,
retentionDays: input.retentionDays ?? null,
...(actorColumns(input.actor, "created") as typeof decisionQueues.$inferInsert),
}).onConflictDoNothing({ target: [decisionQueues.companyId, decisionQueues.key] }).returning();
const row = inserted[0] ?? await txDb.select().from(decisionQueues)
.where(and(eq(decisionQueues.companyId, input.companyId), eq(decisionQueues.key, input.key)))
.then((rows) => rows[0] ?? null);
if (!row) throw new Error("Decision queue create did not return a row");
if (inserted[0]) {
await txDb.insert(decisionTriageEvents).values({
companyId: input.companyId,
queueId: row.id,
action: "queue.created",
...eventActorColumns(input.actor),
details: { key: row.key },
});
await recordActivity(txDb, input.actor, {
companyId: input.companyId,
action: "decision_queue.created",
entityType: "decision_queue",
entityId: row.id,
details: { key: row.key },
});
}
return { queue: toQueue(row, 0), created: Boolean(inserted[0]) };
});
},
list: async (companyId: string, authActor: AuthorizationActor) => {
const rows = await db.select().from(decisionQueues)
.where(eq(decisionQueues.companyId, companyId))
.orderBy(desc(decisionQueues.updatedAt), desc(decisionQueues.id));
const result: DecisionQueue[] = [];
for (const row of rows) {
const items = await visibleItems(companyId, row.id, authActor);
result.push(toQueue(row, items.length));
}
return result;
},
update: async (input: {
companyId: string;
key: string;
patch: { title?: string; description?: string | null; retentionDays?: number | null; seedRulesEnabled?: boolean };
authActor: AuthorizationActor;
actor: DecisionMutationActor;
}) => {
const row = await db.transaction(async (tx) => {
const txDb = tx as unknown as Db;
const updated = await txDb.update(decisionQueues).set({ ...input.patch, updatedAt: new Date() })
.where(and(eq(decisionQueues.companyId, input.companyId), eq(decisionQueues.key, input.key)))
.returning().then((rows) => rows[0] ?? null);
if (!updated) throw notFound("Decision queue not found");
await txDb.insert(decisionTriageEvents).values({
companyId: input.companyId,
queueId: updated.id,
action: "queue.updated",
...eventActorColumns(input.actor),
details: { fields: Object.keys(input.patch) },
});
await recordActivity(txDb, input.actor, {
companyId: input.companyId,
action: "decision_queue.updated",
entityType: "decision_queue",
entityId: updated.id,
details: { key: updated.key, fields: Object.keys(input.patch) },
});
return updated;
});
return toQueue(row, (await visibleItems(input.companyId, row.id, input.authActor)).length);
},
listItems: async (companyId: string, key: string, authActor: AuthorizationActor) => {
const queue = await getQueue(companyId, key);
if (!queue) throw notFound("Decision queue not found");
return visibleItems(companyId, queue.id, authActor);
},
addItem: async (input: {
companyId: string;
key: string;
sourceKind: AttentionSourceKind;
sourceId: string;
authActor: AuthorizationActor;
actor: DecisionMutationActor;
}) => {
const queue = await getQueue(input.companyId, input.key);
if (!queue) throw notFound("Decision queue not found");
await requireSourceRead(db, input.authActor, input.companyId, input.sourceKind, input.sourceId);
return db.transaction(async (tx) => {
const txDb = tx as unknown as Db;
const inserted = await txDb.insert(decisionQueueItems).values({
companyId: input.companyId,
queueId: queue.id,
sourceKind: input.sourceKind,
sourceId: input.sourceId,
responsibleUserId: input.actor.responsibleUserId,
...(actorColumns(input.actor, "added") as typeof decisionQueueItems.$inferInsert),
}).onConflictDoNothing({
target: [decisionQueueItems.queueId, decisionQueueItems.sourceKind, decisionQueueItems.sourceId],
}).returning();
const row = inserted[0] ?? await txDb.select().from(decisionQueueItems).where(and(
eq(decisionQueueItems.queueId, queue.id),
eq(decisionQueueItems.sourceKind, input.sourceKind),
eq(decisionQueueItems.sourceId, input.sourceId),
)).then((rows) => rows[0] ?? null);
if (!row) throw new Error("Decision queue item create did not return a row");
if (inserted[0]) {
await txDb.update(decisionQueues).set({ updatedAt: new Date() })
.where(and(eq(decisionQueues.id, queue.id), eq(decisionQueues.companyId, input.companyId)));
await txDb.insert(decisionTriageEvents).values({
companyId: input.companyId,
queueId: queue.id,
sourceKind: input.sourceKind,
sourceId: input.sourceId,
action: "queue_item.added",
...eventActorColumns(input.actor),
});
await recordActivity(txDb, input.actor, {
companyId: input.companyId,
action: "decision_queue_item.added",
entityType: "decision_queue",
entityId: queue.id,
details: { sourceKind: input.sourceKind, sourceId: input.sourceId },
});
}
return { item: toQueueItem(row), created: Boolean(inserted[0]) };
});
},
removeItem: async (input: {
companyId: string;
key: string;
sourceKind: AttentionSourceKind;
sourceId: string;
authActor: AuthorizationActor;
actor: DecisionMutationActor;
}) => {
const queue = await getQueue(input.companyId, input.key);
if (!queue) throw notFound("Decision queue not found");
if (input.authActor.type !== "board") {
await requireSourceRead(db, input.authActor, input.companyId, input.sourceKind, input.sourceId);
}
return db.transaction(async (tx) => {
const txDb = tx as unknown as Db;
const removed = await txDb.delete(decisionQueueItems).where(and(
eq(decisionQueueItems.companyId, input.companyId),
eq(decisionQueueItems.queueId, queue.id),
eq(decisionQueueItems.sourceKind, input.sourceKind),
eq(decisionQueueItems.sourceId, input.sourceId),
)).returning().then((rows) => rows[0] ?? null);
if (!removed) throw notFound("Decision queue item not found");
await txDb.update(decisionQueues).set({ updatedAt: new Date() })
.where(and(eq(decisionQueues.id, queue.id), eq(decisionQueues.companyId, input.companyId)));
await txDb.insert(decisionTriageEvents).values({
companyId: input.companyId,
queueId: queue.id,
sourceKind: input.sourceKind,
sourceId: input.sourceId,
action: "queue_item.removed",
...eventActorColumns(input.actor),
});
await recordActivity(txDb, input.actor, {
companyId: input.companyId,
action: "decision_queue_item.removed",
entityType: "decision_queue",
entityId: queue.id,
details: { sourceKind: input.sourceKind, sourceId: input.sourceId },
});
return toQueueItem(removed);
});
},
getTriage: async (
companyId: string,
sourceKind: AttentionSourceKind,
sourceId: string,
authActor: AuthorizationActor,
) => {
await requireSourceRead(db, authActor, companyId, sourceKind, sourceId);
const row = await db.select().from(decisionTriage).where(and(
eq(decisionTriage.companyId, companyId),
eq(decisionTriage.sourceKind, sourceKind),
eq(decisionTriage.sourceId, sourceId),
)).then((rows) => rows[0] ?? null);
return row ? toTriage(row) : null;
},
updateTriage: async (input: {
companyId: string;
sourceKind: AttentionSourceKind;
sourceId: string;
decideBy?: string | null;
snoozedUntil?: string | null;
authActor: AuthorizationActor;
actor: DecisionMutationActor;
}) => {
await requireSourceRead(db, input.authActor, input.companyId, input.sourceKind, input.sourceId);
return db.transaction(async (tx) => {
const txDb = tx as unknown as Db;
const current = await txDb.select().from(decisionTriage).where(and(
eq(decisionTriage.companyId, input.companyId),
eq(decisionTriage.sourceKind, input.sourceKind),
eq(decisionTriage.sourceId, input.sourceId),
)).then((rows) => rows[0] ?? null);
const decideByDate = input.decideBy && /^\d{4}-\d{2}-\d{2}$/.test(input.decideBy) ? input.decideBy : null;
const decideBy = input.decideBy === undefined
? current?.decideBy ?? null
: decideByDate ? "date" : input.decideBy;
const finalDecideByDate = input.decideBy === undefined ? current?.decideByDate ?? null : decideByDate;
const snoozedUntil = input.snoozedUntil === undefined
? current?.snoozedUntil ?? null
: input.snoozedUntil === null ? null : new Date(input.snoozedUntil);
if (snoozedUntil && snoozedUntil.getTime() > Date.now() + 5 * 366 * 24 * 60 * 60 * 1_000) {
throw unprocessable("snoozedUntil must be within five years");
}
const now = new Date();
const values = {
decideBy,
decideByDate: finalDecideByDate,
snoozedUntil,
setByType: input.actor.actorType as "agent" | "user",
setByAgentId: input.actor.agentId,
setByUserId: input.actor.userId,
setByRunId: input.actor.runId,
setByAgentApiKeyId: input.actor.agentApiKeyId,
responsibleUserId: input.actor.responsibleUserId,
version: (current?.version ?? 0) + 1,
updatedAt: now,
};
const row = await txDb.insert(decisionTriage).values({
companyId: input.companyId,
sourceKind: input.sourceKind,
sourceId: input.sourceId,
...values,
}).onConflictDoUpdate({
target: [decisionTriage.companyId, decisionTriage.sourceKind, decisionTriage.sourceId],
set: values,
}).returning().then((rows) => rows[0]!);
await txDb.insert(decisionTriageEvents).values({
companyId: input.companyId,
sourceKind: input.sourceKind,
sourceId: input.sourceId,
action: "triage.updated",
...eventActorColumns(input.actor),
details: {
previousDecideBy: current ? toTriage(current).decideBy : null,
decideBy: toTriage(row).decideBy,
previousSnoozedUntil: current?.snoozedUntil?.toISOString() ?? null,
snoozedUntil: row.snoozedUntil?.toISOString() ?? null,
version: row.version,
},
});
await recordActivity(txDb, input.actor, {
companyId: input.companyId,
action: "decision_triage.updated",
entityType: "attention_source",
entityId: `${input.sourceKind}:${input.sourceId}`,
details: { sourceKind: input.sourceKind, sourceId: input.sourceId, version: row.version },
});
return toTriage(row);
});
},
materializeSeededQueues: async (companyId: string, items: AttentionItem[]) => {
if (items.length === 0) return;
const issueIds = [...new Set(items.map(itemIssueId).filter((id): id is string => Boolean(id)))];
const prIssueIds = new Set(issueIds.length === 0 ? [] : await db
.select({ issueId: issueWorkProducts.issueId })
.from(issueWorkProducts)
.where(and(
eq(issueWorkProducts.companyId, companyId),
eq(issueWorkProducts.type, "pull_request"),
inArray(issueWorkProducts.issueId, issueIds),
)).then((rows) => rows.map((row) => row.issueId)));
const matches = new Map<string, AttentionItem[]>();
for (const item of items) {
if (prIssueIds.has(itemIssueId(item) ?? "")) {
matches.set("prs", [...(matches.get("prs") ?? []), item]);
}
if (item.sourceKind === "issue_thread_interaction" && item.subject.metadata?.isPlanTarget === true) {
matches.set("plans", [...(matches.get("plans") ?? []), item]);
}
if (item.sourceKind === "issue_thread_interaction" && item.subject.metadata?.kind === "ask_user_questions") {
matches.set("questions", [...(matches.get("questions") ?? []), item]);
}
}
for (const seed of DECISION_QUEUE_SEEDS) {
const matchingItems = matches.get(seed.key) ?? [];
if (matchingItems.length === 0) continue;
await db.transaction(async (tx) => {
const txDb = tx as unknown as Db;
const insertedQueue = await txDb.insert(decisionQueues).values({
companyId,
key: seed.key,
title: seed.title,
description: seed.description,
createdByType: "system",
seedRules: seed.rules,
seedRulesEnabled: true,
}).onConflictDoNothing({ target: [decisionQueues.companyId, decisionQueues.key] }).returning();
const queue = insertedQueue[0] ?? await txDb.select().from(decisionQueues)
.where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.key, seed.key)))
.then((rows) => rows[0] ?? null);
if (!queue || !queue.seedRulesEnabled) return;
if (insertedQueue[0]) {
await txDb.insert(decisionTriageEvents).values({
companyId,
queueId: queue.id,
action: "queue.seeded",
...eventActorColumns(SYSTEM_ACTOR),
details: { key: seed.key, rules: seed.rules.map((rule) => rule.key) },
});
await recordActivity(txDb, SYSTEM_ACTOR, {
companyId,
action: "decision_queue.seeded",
entityType: "decision_queue",
entityId: queue.id,
details: { key: seed.key },
});
}
let insertedAnyItem = false;
for (const item of matchingItems) {
const inserted = await txDb.insert(decisionQueueItems).values({
companyId,
queueId: queue.id,
sourceKind: item.sourceKind,
sourceId: item.subject.id,
addedByType: "system",
}).onConflictDoNothing({
target: [decisionQueueItems.queueId, decisionQueueItems.sourceKind, decisionQueueItems.sourceId],
}).returning();
if (!inserted[0]) continue;
insertedAnyItem = true;
await txDb.insert(decisionTriageEvents).values({
companyId,
queueId: queue.id,
sourceKind: item.sourceKind,
sourceId: item.subject.id,
action: "queue_item.seeded",
...eventActorColumns(SYSTEM_ACTOR),
details: { seedKey: seed.key },
});
await recordActivity(txDb, SYSTEM_ACTOR, {
companyId,
action: "decision_queue_item.seeded",
entityType: "decision_queue",
entityId: queue.id,
details: { sourceKind: item.sourceKind, sourceId: item.subject.id, seedKey: seed.key },
});
}
if (insertedAnyItem) {
await txDb.update(decisionQueues).set({ updatedAt: new Date() })
.where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.id, queue.id)));
}
});
}
},
};
}