mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
## Thinking Path > - Paperclip helps people manage AI agents and their work. > - The inbox collects requests that need a human decision. > - Teams need to know who owns each request and who may act on it. > - A shared close operation must differ from hiding a card for one person. > - This pull request adds personal routing and durable outcomes to the existing attention service. > - People can claim or transfer work and inspect what another person did for them. > - Domain permissions still decide who may resolve each request. ## Linked Issues or Issue Description **Subsystem affected** Server, shared contracts, and database. This is the routing foundation for a stacked inbox UI change. **Problem or motivation** A company-wide attention feed does not give teams a clear personal queue. Archive, withdrawal, transfer, and decisions by another person need distinct scopes and records. **Proposed solution** Resolve a request audience from its addressee, task participants, and configured owners. Provide personal, unassigned, company, all-assignee, and archived views. Add claim, transfer, personal archive and snooze, shared withdrawal, receipts, and personal history. **Alternatives considered** A single shared inbox leaves ownership and personal hiding ambiguous. The service keeps one canonical request and projects its audience instead of copying requests for each person. **Roadmap alignment** This extends the existing Multiple Human Users and Agent Reviews and Approvals capabilities in ROADMAP.md. **Additional context** Refs #19363 and #19520 for the current inbox navigation and confirmation replacement rules. Refs #20268 for related actionable-request work. That PR filters actionability; this stack adds audience routing and personal outcomes. This is part 1 of a two-PR stack. The inbox UI PR #20961 targets this branch. ## What Changed - Add nullable company and project decision owners and a company budget owner. - Add audience views and personal badge counts to the canonical attention projections. - Add claim, transfer, withdrawal, personal archive, snooze, and restore operations with permission checks and activity records. Lock addressed transfers to the current addressee or routing administrators, and validate the recipient against canonical resolution policies. - Add receipts and scoped history for work handled by another person. - Keep closed-task requests and replaced confirmations out of current cards. - Add an idempotent migration and a concurrent receipt-history index. - Update typed fixtures so the existing UI remains compatible with the new contracts. ## Verification - Server layered typechecking passes, including all three test-tree partitions. - Seven routing and interaction-resolution suites pass all 249 tests against a dedicated external PostgreSQL test database on `4caecc21d8`. Six new regressions failed before the repair; they cover takeover followed by approval, legitimate delegation, excluded creators/review requesters, and unauthorized claims. - Prior focused verification covers private approval history and receipts, transfer visibility, audience escalation, withdrawal governance, hidden-card counts, per-recipient confirmation replacement, and SDK source contracts. - A synthetic three-person company passed 37 routing and lifecycle acceptance checks. The UI stack adds browser acceptance for the shared controls. - The complete [GitHub Actions CI matrix](https://github.com/paperclipai/paperclip/actions/runs/43134997331) passes on `4caecc21d891677323a7335f23b1a015cce24c13` (Linux/Node 24: typecheck, build, general and serialized tests, Runner support, browser shards, and aggregate gates). Greptile is 5/5 on that exact head with no open findings. The fresh security scan passes after explicitly rechecking the addressee-transfer repair. - `pnpm verify` reports exit 2 because a CI-global script changed; full CI supplies that broader coverage. Embedded PostgreSQL could not start on this Mac because shared-memory slots were exhausted, so local database checks use a separate external test database. Linux CI covers the embedded setup. ## Risks - Personal badge counts change to the viewer's assigned requests. Accessible feed totals and task membership keep their existing scope. - The migration adds nullable owner fields and a dismissal fingerprint. Its index runs outside a transaction. The migration is classified and rerunnable. - Routing changes the addressee used by resolution authorization. The current addressee or a routing administrator must authorize an addressed transfer; the recipient must satisfy writer access, canonical resolver exclusions, and task review policy under row locks. - The shared validator export barrel crosses 1,000 lines because it exports the new attention schemas. Schema implementation stays in the dedicated attention validator module. ## Model Used OpenAI GPT-6 through Codex. Used repository tools, code execution, and browser checks. This session does not expose a more specific runtime model ID or its context-window size. ## 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 described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket id - [x] I have run focused tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes (OpenAPI contracts and docs/api/attention.md) - [x] I have considered and documented any risks above - [x] 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: Scott Tong <scott@scottsmbpm5max.lan> Co-authored-by: Paperclip <noreply@paperclip.ing>
803 lines
54 KiB
TypeScript
803 lines
54 KiB
TypeScript
import { assertAgentInteraction } from "./agent-interaction-policy.js";
|
|
import { decisionTargetService, notifyDecisionTargetWork } from "./decision-targets.js";
|
|
import { randomUUID } from "node:crypto";
|
|
import { and, asc, count, desc, eq, gte, inArray, sql } from "drizzle-orm";
|
|
import type { Db } from "@paperclipai/db";
|
|
import { databaseWorkScopeKey, companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db";
|
|
import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared";
|
|
import type { AttentionArchiveManifestEntry, DecisionEffect, DecisionInput, DecisionOption, DecisionStatsCounts, DecisionStatsResponse } from "@paperclipai/shared";
|
|
import { conflict, forbidden, notFound, tooManyRequests, unprocessable } from "../errors.js";
|
|
import { authorizationService, type AuthorizationActor } from "./authorization.js";
|
|
import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js";
|
|
import { signDecisionSpec, verifyDecisionSpec } from "./decision-signing.js";
|
|
import {
|
|
issueService,
|
|
type IssuePostCommitAction,
|
|
} from "./issues.js";
|
|
import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js";
|
|
|
|
import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js";
|
|
import { decisionEffectRecoveryService } from "./decision-effect-recovery.js";
|
|
import { decisionExpiryService } from "./decision-expiry.js";
|
|
import { notifyDecisionCreated, writeDecisionChange } from "./decision-work-notifications.js";
|
|
|
|
type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; updatedAt: string;
|
|
descendantCount?: number; descendantIds?: string[]; childCount?: number; attentionArchive?: unknown };
|
|
|
|
export type DecisionServiceOptions = { wakeOriginAgent: DecisionContinuationWake };
|
|
const DAY = 86_400_000;
|
|
type DecisionOutcome = typeof decisions.$inferSelect & { executions: Array<typeof decisionEffectExecutions.$inferSelect> };
|
|
type EffectAttempt = { outcome: DecisionOutcome; resumed: boolean };
|
|
const effectFlights = new WeakMap<object, Map<string, Promise<EffectAttempt | null>>>();
|
|
|
|
function targetIds(options: DecisionOption[]) {
|
|
const result = new Set<string>();
|
|
for (const option of options) for (const effect of option.effects) for (const id of decisionEffectTargetIssueIds(effect)) result.add(id);
|
|
return [...result];
|
|
}
|
|
|
|
function targetActions(options: DecisionOption[]) {
|
|
const result = new Map<string, Set<"issue:comment" | "issue:mutate">>();
|
|
for (const option of options) for (const effect of option.effects) {
|
|
const action = effect.type === "comment_on_issue" ? "issue:comment" as const : "issue:mutate" as const;
|
|
for (const id of decisionEffectTargetIssueIds(effect)) {
|
|
const actions = result.get(id) ?? new Set();
|
|
actions.add(action);
|
|
result.set(id, actions);
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function sameIds(left: readonly string[], right: readonly string[]) {
|
|
if (left.length !== right.length) return false;
|
|
const rightIds = new Set(right);
|
|
return rightIds.size === right.length && left.every((id) => rightIds.has(id));
|
|
}
|
|
|
|
function sameInputValues(left: Record<string, string>, right: Record<string, string>) {
|
|
const leftKeys = Object.keys(left);
|
|
const rightKeys = Object.keys(right);
|
|
return leftKeys.length === rightKeys.length && leftKeys.every((key) => left[key] === right[key]);
|
|
}
|
|
|
|
function spec(decision: { id: string; options: DecisionOption[]; targetSnapshots: Record<string, Snapshot> }) {
|
|
return { decisionId: decision.id, options: decision.options, targetSnapshots: decision.targetSnapshots };
|
|
}
|
|
|
|
function resource(issue: typeof issues.$inferSelect) {
|
|
return { type: "issue" as const, companyId: issue.companyId, issueId: issue.id, projectId: issue.projectId,
|
|
parentIssueId: issue.parentId, assigneeAgentId: issue.assigneeAgentId, assigneeUserId: issue.assigneeUserId, status: issue.status };
|
|
}
|
|
|
|
function interpolate(text: string, values: Record<string, string>) {
|
|
return text.replace(/\{\{input\.([A-Za-z0-9_-]+)\}\}/g, (_all, id: string) => values[id] ?? "");
|
|
}
|
|
|
|
function canonicalJson(value: unknown): string {
|
|
if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`;
|
|
if (value && typeof value === "object") {
|
|
return `{${Object.entries(value as Record<string, unknown>).sort(([left], [right]) => left.localeCompare(right))
|
|
.map(([key, item]) => `${JSON.stringify(key)}:${canonicalJson(item)}`).join(",")}}`;
|
|
}
|
|
return JSON.stringify(value);
|
|
}
|
|
|
|
function parseAttentionArchiveManifest(
|
|
companyId: string,
|
|
targetSnapshots: Record<string, Snapshot>,
|
|
): AttentionArchiveManifestEntry[] | null {
|
|
const entries: AttentionArchiveManifestEntry[] = [];
|
|
const seen = new Set<string>();
|
|
for (const [key, snapshot] of Object.entries(targetSnapshots)) {
|
|
if (!key.startsWith("attention:")) return null;
|
|
const entry = snapshot.attentionArchive;
|
|
if (!entry || typeof entry !== "object" || Array.isArray(entry)) return null;
|
|
const candidate = entry as Record<string, unknown>;
|
|
if (candidate.companyId !== companyId
|
|
|| typeof candidate.sourceKind !== "string"
|
|
|| !ATTENTION_SOURCE_KINDS.includes(candidate.sourceKind as (typeof ATTENTION_SOURCE_KINDS)[number])
|
|
|| typeof candidate.sourceId !== "string"
|
|
|| !candidate.sourceId
|
|
|| !Number.isInteger(candidate.expectedVersion)
|
|
|| Number(candidate.expectedVersion) < 1
|
|
|| typeof candidate.activityAt !== "string"
|
|
|| !Number.isFinite(Date.parse(candidate.activityAt))
|
|
|| typeof candidate.reason !== "string"
|
|
|| !candidate.reason.trim()) return null;
|
|
const expectedKey = `attention:${candidate.sourceKind}:${candidate.sourceId}`;
|
|
if (key !== expectedKey || seen.has(expectedKey)) return null;
|
|
seen.add(expectedKey);
|
|
entries.push({
|
|
companyId,
|
|
sourceKind: candidate.sourceKind,
|
|
sourceId: candidate.sourceId,
|
|
expectedVersion: Number(candidate.expectedVersion),
|
|
activityAt: candidate.activityAt,
|
|
reason: candidate.reason,
|
|
});
|
|
}
|
|
return entries.length > 0 ? entries : null;
|
|
}
|
|
|
|
function boardCanActDirectly(actor: AuthorizationActor, companyId: string) {
|
|
if (actor.type !== "board") return false;
|
|
if (actor.source === "local_implicit" || actor.isInstanceAdmin) return true;
|
|
return actor.companyIds?.includes(companyId) === true ||
|
|
actor.memberships?.some((membership) => membership.companyId === companyId && membership.status === "active") === true;
|
|
}
|
|
|
|
export function decisionService(db: Db, options: DecisionServiceOptions) {
|
|
const continuations = decisionContinuationService(db, options.wakeOriginAgent);
|
|
const expiry = decisionExpiryService(db, continuations.deliver);
|
|
const targets = decisionTargetService(db, expiry.expire);
|
|
const effectRecovery = decisionEffectRecoveryService(db, recoverDecisionEffects);
|
|
const scope = databaseWorkScopeKey(db);
|
|
let flights = effectFlights.get(scope);
|
|
if (!flights) effectFlights.set(scope, flights = new Map());
|
|
const executing = flights;
|
|
const authz = authorizationService(db);
|
|
type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null;
|
|
ruleKey?: string | null; title: string; body: string; options: DecisionOption[]; inputs?: DecisionInput[] | null; expiresAt?: Date | null;
|
|
idempotencyKey?: string | null; continuationPolicy?: "none" | "wake_origin_agent"; metadata?: Record<string, unknown> };
|
|
type CreateInputWithSnapshots = CreateInput & { additionalTargetSnapshots?: Record<string, Snapshot> };
|
|
|
|
async function origin(companyId: string, agentId: string, runId: string) {
|
|
const run = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId)))
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!run) throw forbidden("Decision provenance requires the origin run");
|
|
const issueId = typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null;
|
|
if (!issueId) throw unprocessable("Origin run is not issue-scoped");
|
|
return { run, issueId };
|
|
}
|
|
|
|
async function recoveryActor(companyId: string, userId: string): Promise<AuthorizationActor> {
|
|
const membership = await db.select({ companyId: companyMemberships.companyId, membershipRole: companyMemberships.membershipRole,
|
|
status: companyMemberships.status }).from(companyMemberships).where(and(eq(companyMemberships.companyId, companyId),
|
|
eq(companyMemberships.principalType, "user"), eq(companyMemberships.principalId, userId), eq(companyMemberships.status, "active")))
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!membership) return { type: "none", source: "none" };
|
|
return { type: "board", userId, companyIds: [companyId], memberships: [membership], source: "session" };
|
|
}
|
|
|
|
async function collectDescendantIds(companyId: string, rootId: string, dbOrTx: Db) {
|
|
const queue = [rootId]; const visited = new Set([rootId]); const result: string[] = [];
|
|
while (queue.length) {
|
|
const parentId = queue.shift()!;
|
|
const children = await dbOrTx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, companyId), eq(issues.parentId, parentId)));
|
|
for (const child of children) {
|
|
if (visited.has(child.id)) continue;
|
|
visited.add(child.id);
|
|
result.push(child.id);
|
|
queue.push(child.id);
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
async function snapshots(companyId: string, ids: string[], actor: AuthorizationActor, requiredActions: ReturnType<typeof targetActions>,
|
|
cancellationTargetIds: ReadonlySet<string>, dbOrTx: Db) {
|
|
const rows = ids.length ? await dbOrTx.select().from(issues).where(and(eq(issues.companyId, companyId), inArray(issues.id, ids))) : [];
|
|
if (rows.length !== ids.length) throw unprocessable("All referenced issues must exist in the company");
|
|
const result: Record<string, Snapshot> = {};
|
|
for (const issue of rows) {
|
|
const access = await authz.decide({ actor, action: "issue:read", resource: resource(issue) });
|
|
if (!access.allowed) throw forbidden("Decision target is outside the origin visibility boundary");
|
|
for (const action of requiredActions.get(issue.id) ?? []) {
|
|
const effectAccess = await authz.decide({ actor, action, resource: resource(issue) });
|
|
if (!effectAccess.allowed) throw forbidden("Decision effect exceeds the origin authority boundary");
|
|
}
|
|
const cancellationTarget = cancellationTargetIds.has(issue.id);
|
|
const descendantIds = cancellationTarget ? await collectDescendantIds(companyId, issue.id, dbOrTx) : [];
|
|
if (cancellationTarget && descendantIds.length) {
|
|
const descendants = await dbOrTx.select().from(issues)
|
|
.where(and(eq(issues.companyId, companyId), inArray(issues.id, descendantIds)));
|
|
const descendantAccess = await Promise.all(descendants.map((descendant) =>
|
|
authz.decide({ actor, action: "issue:mutate", resource: resource(descendant) })));
|
|
if (descendants.length !== descendantIds.length || descendantAccess.some((access) => !access.allowed)) {
|
|
throw forbidden("Decision effect exceeds the origin authority boundary");
|
|
}
|
|
}
|
|
result[issue.id] = { status: issue.status, assigneeAgentId: issue.assigneeAgentId, assigneeUserId: issue.assigneeUserId,
|
|
updatedAt: issue.updatedAt.toISOString(),
|
|
...(cancellationTarget ? { descendantCount: descendantIds.length, descendantIds } : {}) };
|
|
}
|
|
return result;
|
|
}
|
|
|
|
async function createInStore(input: CreateInputWithSnapshots, dbOrTx: Db) {
|
|
if (input.idempotencyKey) {
|
|
const lockKey = `decision-create:${input.companyId}:${input.idempotencyKey}`;
|
|
await dbOrTx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`);
|
|
}
|
|
const provenance = await origin(input.companyId, input.agentId, input.runId);
|
|
if (input.idempotencyKey) {
|
|
const existing = await dbOrTx.select().from(decisions).where(and(eq(decisions.companyId, input.companyId), eq(decisions.idempotencyKey, input.idempotencyKey)))
|
|
.then((rows) => rows[0] ?? null);
|
|
if (existing) {
|
|
const archiveEquivalent = input.additionalTargetSnapshots === undefined || (
|
|
canonicalJson(existing.metadata ?? {}) === canonicalJson(input.metadata ?? {}) &&
|
|
canonicalJson(Object.fromEntries(Object.entries(existing.targetSnapshots as Record<string, Snapshot>)
|
|
.filter(([key]) => key.startsWith("attention:")))) === canonicalJson(input.additionalTargetSnapshots)
|
|
);
|
|
const equivalent = existing.title === input.title && existing.body === input.body &&
|
|
canonicalJson(existing.options) === canonicalJson(input.options) && canonicalJson(existing.inputs ?? null) === canonicalJson(input.inputs ?? null) &&
|
|
archiveEquivalent;
|
|
if (!equivalent) throw conflict("Decision idempotency key already used with a different payload");
|
|
return existing;
|
|
}
|
|
}
|
|
const open = await dbOrTx.select({ value: count() }).from(decisions).where(and(eq(decisions.companyId, input.companyId), eq(decisions.originAgentId, input.agentId), eq(decisions.status, "open")));
|
|
const cap = Number(process.env.PAPERCLIP_DECISIONS_OPEN_CAP ?? 50);
|
|
if (Number(open[0]?.value ?? 0) >= cap) throw tooManyRequests("Open decision cap reached");
|
|
const expiresAt = input.expiresAt ?? new Date(Date.now() + 7 * DAY);
|
|
if (expiresAt.getTime() <= Date.now() || expiresAt.getTime() > Date.now() + 30 * DAY) throw unprocessable("expiresAt must be within 30 days");
|
|
const ids = targetIds(input.options);
|
|
const cancellationTargetIds = new Set(input.options.flatMap((option) => option.effects
|
|
.filter((effect) => effect.type === "cancel_issue_tree").map((effect) => effect.targetIssueId)));
|
|
const targetSnapshots = {
|
|
...await snapshots(input.companyId, ids, input.actor, targetActions(input.options), cancellationTargetIds, dbOrTx),
|
|
...(input.additionalTargetSnapshots ?? {}),
|
|
};
|
|
const id = randomUUID();
|
|
await notifyDecisionCreated(dbOrTx, { companyId: input.companyId, sourceId: id });
|
|
await notifyDecisionTargetWork(dbOrTx, `decision:${id}`);
|
|
const [created] = await dbOrTx.insert(decisions).values({ id, companyId: input.companyId, bundleId: input.bundleId ?? null,
|
|
originAgentId: input.agentId, originIssueId: provenance.issueId, originRunId: input.runId, ruleKey: input.ruleKey ?? null,
|
|
title: input.title, body: input.body, options: input.options, inputs: input.inputs ?? null, expiresAt,
|
|
idempotencyKey: input.idempotencyKey ?? null, signedSpec: signDecisionSpec(spec({ id, options: input.options, targetSnapshots })),
|
|
targetSnapshots, continuationPolicy: input.continuationPolicy ?? "none", metadata: input.metadata ?? {} }).onConflictDoNothing().returning();
|
|
if (!created) {
|
|
const existing = input.idempotencyKey
|
|
? await dbOrTx.select().from(decisions).where(and(eq(decisions.companyId, input.companyId), eq(decisions.idempotencyKey, input.idempotencyKey))).then((rows) => rows[0] ?? null)
|
|
: null;
|
|
const archiveEquivalent = !existing || input.additionalTargetSnapshots === undefined || (
|
|
canonicalJson(existing.metadata ?? {}) === canonicalJson(input.metadata ?? {}) &&
|
|
canonicalJson(Object.fromEntries(Object.entries(existing.targetSnapshots as Record<string, Snapshot>)
|
|
.filter(([key]) => key.startsWith("attention:")))) === canonicalJson(input.additionalTargetSnapshots)
|
|
);
|
|
const equivalent = existing && existing.title === input.title && existing.body === input.body &&
|
|
canonicalJson(existing.options) === canonicalJson(input.options) && canonicalJson(existing.inputs ?? null) === canonicalJson(input.inputs ?? null) &&
|
|
archiveEquivalent;
|
|
if (equivalent) return existing;
|
|
throw conflict("Decision idempotency key already used with a different payload");
|
|
}
|
|
if (ids.length) await dbOrTx.insert(decisionTargetIssues).values(ids.map((issueId) => ({ decisionId: id, issueId, companyId: input.companyId })));
|
|
await logActivity(dbOrTx, { companyId: input.companyId, actorType: "agent", actorId: input.agentId, agentId: input.agentId,
|
|
runId: input.runId, action: "decision.created", entityType: "decision", entityId: id,
|
|
details: { originIssueId: provenance.issueId, originAgentId: input.agentId, originResponsibleUserId: provenance.run.responsibleUserId } });
|
|
return created;
|
|
}
|
|
|
|
async function create(input: CreateInputWithSnapshots, dbOrTx?: Db) {
|
|
if (dbOrTx) return createInStore(input, dbOrTx);
|
|
return db.transaction((tx) => createInStore(input, tx as unknown as Db));
|
|
}
|
|
|
|
const get = (id: string) => db.select().from(decisions).where(eq(decisions.id, id)).then((rows) => rows[0] ?? null);
|
|
async function outcome(id: string) {
|
|
const decision = await get(id);
|
|
const executions = await db.select().from(decisionEffectExecutions).where(eq(decisionEffectExecutions.decisionId, id)).orderBy(asc(decisionEffectExecutions.effectIndex));
|
|
return { ...decision, executions };
|
|
}
|
|
|
|
async function list(companyId: string, filter: { status?: string; bundleId?: string; targetIssueId?: string; originAgentId?: string; limit?: number } = {}) {
|
|
const conditions = [eq(decisions.companyId, companyId)];
|
|
if (filter.status) conditions.push(eq(decisions.status, filter.status));
|
|
if (filter.bundleId) conditions.push(eq(decisions.bundleId, filter.bundleId));
|
|
if (filter.originAgentId) conditions.push(eq(decisions.originAgentId, filter.originAgentId));
|
|
if (filter.targetIssueId) {
|
|
const links = await db.select({ id: decisionTargetIssues.decisionId }).from(decisionTargetIssues).where(and(eq(decisionTargetIssues.companyId, companyId), eq(decisionTargetIssues.issueId, filter.targetIssueId)));
|
|
if (!links.length) return [];
|
|
conditions.push(inArray(decisions.id, links.map((row) => row.id)));
|
|
}
|
|
const rows = await db.select().from(decisions).where(and(...conditions)).orderBy(desc(decisions.createdAt)).limit(Math.min(filter.limit ?? 50, 100));
|
|
const openDecisionIds = rows.filter((decision) => decision.status === "open").map((decision) => decision.id);
|
|
const currentIssueTargetIds = rows.flatMap((decision) => Object.keys(decision.targetSnapshots as Record<string, Snapshot>))
|
|
.filter((id) => /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(id));
|
|
const currentTargets = openDecisionIds.length && currentIssueTargetIds.length
|
|
? await db.select({ id: issues.id, updatedAt: issues.updatedAt })
|
|
.from(decisionTargetIssues)
|
|
.innerJoin(issues, and(eq(issues.companyId, companyId), eq(issues.id, decisionTargetIssues.issueId)))
|
|
.where(and(eq(decisionTargetIssues.companyId, companyId), inArray(decisionTargetIssues.decisionId, openDecisionIds), inArray(issues.id, currentIssueTargetIds)))
|
|
: [];
|
|
const currentTargetsById = new Map(currentTargets.map((target) => [target.id, target.updatedAt]));
|
|
const attentionSnapshotIds = rows.flatMap((decision) => Object.values(decision.targetSnapshots as Record<string, Snapshot>))
|
|
.map((snapshot) => snapshot.attentionArchive as AttentionArchiveManifestEntry | undefined)
|
|
.filter((entry): entry is AttentionArchiveManifestEntry => Boolean(entry))
|
|
.map((entry) => entry.sourceId);
|
|
const currentAttentionStates = attentionSnapshotIds.length
|
|
? await db.select().from(decisionRetention).where(and(
|
|
eq(decisionRetention.companyId, companyId),
|
|
inArray(decisionRetention.sourceId, attentionSnapshotIds),
|
|
))
|
|
: [];
|
|
const currentAttentionByKey = new Map(currentAttentionStates.map((state) => [
|
|
`attention:${state.sourceKind}:${state.sourceId}`,
|
|
state,
|
|
]));
|
|
const terminalDecisionIds = rows.filter((decision) => decision.status !== "open").map((decision) => decision.id);
|
|
const terminalExecutions = terminalDecisionIds.length
|
|
? await db.select().from(decisionEffectExecutions)
|
|
.where(inArray(decisionEffectExecutions.decisionId, terminalDecisionIds))
|
|
.orderBy(asc(decisionEffectExecutions.decisionId), asc(decisionEffectExecutions.effectIndex))
|
|
: [];
|
|
const executionsByDecision = new Map<string, typeof terminalExecutions>();
|
|
for (const execution of terminalExecutions) {
|
|
const grouped = executionsByDecision.get(execution.decisionId) ?? [];
|
|
grouped.push(execution);
|
|
executionsByDecision.set(execution.decisionId, grouped);
|
|
}
|
|
return rows.map((decision) => {
|
|
const changed: Record<string, boolean> = {};
|
|
if (decision.status === "open") for (const [id, snapshot] of Object.entries(decision.targetSnapshots as Record<string, Snapshot>)) {
|
|
if (id.startsWith("attention:")) {
|
|
const entry = snapshot.attentionArchive as AttentionArchiveManifestEntry | undefined;
|
|
const current = currentAttentionByKey.get(id);
|
|
changed[id] = !entry || !current || current.version !== entry.expectedVersion
|
|
|| current.sourceActivityAt.toISOString() !== entry.activityAt || Boolean(current.archivedAt);
|
|
continue;
|
|
}
|
|
const currentUpdatedAt = currentTargetsById.get(id);
|
|
changed[id] = !currentUpdatedAt || currentUpdatedAt.toISOString() !== snapshot.updatedAt;
|
|
}
|
|
return { ...decision, targetChanged: changed,
|
|
...(decision.status === "open" ? {} : { executions: executionsByDecision.get(decision.id) ?? [] }) };
|
|
});
|
|
}
|
|
|
|
async function stats(companyId: string, filter: { originAgentId?: string; since?: Date } = {}): Promise<DecisionStatsResponse> {
|
|
const conditions = [eq(decisions.companyId, companyId)];
|
|
if (filter.originAgentId) conditions.push(eq(decisions.originAgentId, filter.originAgentId));
|
|
if (filter.since) conditions.push(gte(decisions.createdAt, filter.since));
|
|
const dismissed = sql<boolean>`coalesce(${decisions.metadata}->'dismissed' = 'true'::jsonb, false)`;
|
|
const rows = await db.select({
|
|
ruleKey: decisions.ruleKey,
|
|
status: decisions.status,
|
|
chosenOptionId: decisions.chosenOptionId,
|
|
dismissed,
|
|
value: count(),
|
|
}).from(decisions).where(and(...conditions))
|
|
.groupBy(decisions.ruleKey, decisions.status, decisions.chosenOptionId, dismissed);
|
|
const emptyCounts = (): DecisionStatsCounts => ({ proposed: 0, accepted: 0, rejected: 0, expired: 0 });
|
|
const totals = emptyCounts();
|
|
const grouped = new Map<string | null, { counts: DecisionStatsCounts; chosenOptions: Map<string, number> }>();
|
|
for (const row of rows) {
|
|
const value = Number(row.value);
|
|
const group = grouped.get(row.ruleKey) ?? { counts: emptyCounts(), chosenOptions: new Map<string, number>() };
|
|
grouped.set(row.ruleKey, group);
|
|
totals.proposed += value;
|
|
group.counts.proposed += value;
|
|
if (row.status === "expired") {
|
|
totals.expired += value;
|
|
group.counts.expired += value;
|
|
continue;
|
|
}
|
|
if (row.status !== "decided") continue;
|
|
const rejected = row.chosenOptionId === "dismissed" || row.dismissed;
|
|
if (rejected) {
|
|
totals.rejected += value;
|
|
group.counts.rejected += value;
|
|
continue;
|
|
}
|
|
totals.accepted += value;
|
|
group.counts.accepted += value;
|
|
if (row.chosenOptionId) group.chosenOptions.set(row.chosenOptionId, (group.chosenOptions.get(row.chosenOptionId) ?? 0) + value);
|
|
}
|
|
return {
|
|
groupBy: "ruleKey",
|
|
filters: { originAgentId: filter.originAgentId ?? null, since: filter.since?.toISOString() ?? null },
|
|
totals,
|
|
groups: [...grouped.entries()]
|
|
.sort(([left], [right]) => left === null ? 1 : right === null ? -1 : left.localeCompare(right))
|
|
.map(([ruleKey, group]) => ({
|
|
ruleKey,
|
|
...group.counts,
|
|
chosenOptions: [...group.chosenOptions.entries()]
|
|
.sort(([leftId, leftCount], [rightId, rightCount]) => rightCount - leftCount || leftId.localeCompare(rightId))
|
|
.map(([optionId, optionCount]) => ({ optionId, count: optionCount })),
|
|
})),
|
|
};
|
|
}
|
|
|
|
async function effectAudit(tx: Db, decision: typeof decisions.$inferSelect, executionId: string, effect: DecisionEffect,
|
|
status: "executed" | "failed" | "skipped", decidedByUserId: string, originResponsibleUserId: string | null, details: Record<string, unknown>) {
|
|
return logActivity(tx, { companyId: decision.companyId, actorType: "system", actorId: "decision-executor", agentId: decision.originAgentId,
|
|
runId: decision.originRunId, responsibleUserIdOverride: decidedByUserId, action: `decision.effect_${status}`, entityType: "decision", entityId: decision.id,
|
|
details: { effectType: effect.type, targetIssueId: effect.targetIssueId, originIssueId: decision.originIssueId, originAgentId: decision.originAgentId,
|
|
chosenOptionId: decision.chosenOptionId, executionId, decidedByUserId, originResponsibleUserId, ...details } });
|
|
}
|
|
|
|
async function executeEffect(decision: typeof decisions.$inferSelect, effect: DecisionEffect, effectIndex: number,
|
|
userActor: AuthorizationActor, decidedByUserId: string, originResponsibleUserId: string | null) {
|
|
const lockKey = `decision-effect:${decision.id}:${effectIndex}`;
|
|
const recordFailure = async (reason: string, details: Record<string, unknown>) => db.transaction(async (tx) => {
|
|
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`);
|
|
let execution = await tx.select().from(decisionEffectExecutions).where(and(eq(decisionEffectExecutions.decisionId, decision.id), eq(decisionEffectExecutions.effectIndex, effectIndex)))
|
|
.then((rows) => rows[0] ?? null);
|
|
if (execution && execution.status !== "claimed") return execution;
|
|
if (!execution) {
|
|
[execution] = await tx.insert(decisionEffectExecutions).values({ decisionId: decision.id, effectIndex, effectType: effect.type, targetIssueId: effect.targetIssueId }).onConflictDoNothing().returning();
|
|
if (!execution) execution = await tx.select().from(decisionEffectExecutions).where(and(eq(decisionEffectExecutions.decisionId, decision.id), eq(decisionEffectExecutions.effectIndex, effectIndex))).then((rows) => rows[0] ?? null);
|
|
}
|
|
if (!execution || execution.status !== "claimed") return execution;
|
|
const activity = await effectAudit(tx as unknown as Db, decision, execution.id, effect, "failed", decidedByUserId, originResponsibleUserId, details);
|
|
const [row] = await tx.update(decisionEffectExecutions).set({ status: "failed", error: reason, result: details, activityLogId: activity?.id ?? null, executedAt: new Date() }).where(eq(decisionEffectExecutions.id, execution.id)).returning();
|
|
return row;
|
|
});
|
|
|
|
try {
|
|
const postCommitActivityPublications: ActivityPublication[] = [];
|
|
const postCommitIssueActions: IssuePostCommitAction[] = [];
|
|
const executionResult = await db.transaction(async (tx) => {
|
|
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`);
|
|
let execution = await tx.select().from(decisionEffectExecutions).where(and(eq(decisionEffectExecutions.decisionId, decision.id), eq(decisionEffectExecutions.effectIndex, effectIndex)))
|
|
.then((rows) => rows[0] ?? null);
|
|
if (execution && execution.status !== "claimed") return execution;
|
|
if (!execution) {
|
|
[execution] = await tx.insert(decisionEffectExecutions).values({ decisionId: decision.id, effectIndex, effectType: effect.type, targetIssueId: effect.targetIssueId }).onConflictDoNothing().returning();
|
|
if (!execution) execution = await tx.select().from(decisionEffectExecutions).where(and(eq(decisionEffectExecutions.decisionId, decision.id), eq(decisionEffectExecutions.effectIndex, effectIndex))).then((rows) => rows[0] ?? null);
|
|
if (!execution || execution.status !== "claimed") return execution;
|
|
}
|
|
const finish = async (status: "failed" | "skipped", reason: string, details: Record<string, unknown>) => {
|
|
const activity = await effectAudit(tx as unknown as Db, decision, execution!.id, effect, status, decidedByUserId, originResponsibleUserId, details);
|
|
const [row] = await tx.update(decisionEffectExecutions).set({ status, error: reason, result: details, activityLogId: activity?.id ?? null, executedAt: new Date() }).where(eq(decisionEffectExecutions.id, execution!.id)).returning();
|
|
return row;
|
|
};
|
|
const directReferencedIds = new Set(decisionEffectTargetIssueIds(effect));
|
|
const snapshots = decision.targetSnapshots as Record<string, Snapshot>;
|
|
const cancellationDescendantIds = effect.type === "cancel_issue_tree"
|
|
? snapshots[effect.targetIssueId]?.descendantIds
|
|
: undefined;
|
|
if (effect.type === "cancel_issue_tree" && !Array.isArray(cancellationDescendantIds)) {
|
|
return finish("failed", "invalid_signed_cancellation_scope", { reason: "invalid_signed_cancellation_scope" });
|
|
}
|
|
const referencedIds = new Set([...directReferencedIds, ...(cancellationDescendantIds ?? [])]);
|
|
const referencedIssues = await tx.select().from(issues).where(and(eq(issues.companyId, decision.companyId), inArray(issues.id, [...referencedIds])));
|
|
if (referencedIssues.length !== referencedIds.size) return finish("failed", "invalid_effect_reference", { reason: "invalid_effect_reference" });
|
|
const target = referencedIssues.find((item) => item.id === effect.targetIssueId)!;
|
|
const originActor: AuthorizationActor = { type: "agent", agentId: decision.originAgentId, companyId: decision.companyId,
|
|
runId: decision.originRunId, onBehalfOfUserId: originResponsibleUserId, source: "agent_jwt" };
|
|
const originAction = effect.type === "comment_on_issue" ? "issue:comment" : "issue:mutate";
|
|
const originAccess = await Promise.all(referencedIssues.map((item) => authz.decide({ actor: originActor, action: originAction, resource: resource(item) })));
|
|
let userAccess: { allowed: boolean; reason: string };
|
|
if (effect.type === "assign_issue" || (effect.type === "create_issue" && (effect.draft.assigneeAgentId || effect.draft.assigneeUserId))) {
|
|
const assigneeAgentId = effect.type === "assign_issue" ? effect.assigneeAgentId : effect.draft.assigneeAgentId;
|
|
const assigneeUserId = effect.type === "assign_issue" ? effect.assigneeUserId : effect.draft.assigneeUserId;
|
|
const parentIssueId = effect.type === "create_issue" ? effect.draft.parentId ?? target.id : target.parentId;
|
|
const projectId = effect.type === "create_issue" ? effect.draft.projectId ?? target.projectId : target.projectId;
|
|
userAccess = await authz.decide({ actor: userActor, action: "tasks:assign", resource: { ...resource(target), parentIssueId, projectId },
|
|
scope: { issueId: target.id, parentIssueId, projectId, assigneeAgentId, assigneeUserId } });
|
|
} else {
|
|
userAccess = boardCanActDirectly(userActor, decision.companyId)
|
|
? { allowed: true, reason: "allow_board_direct_route" }
|
|
: { allowed: false, reason: "deny_company_boundary" };
|
|
}
|
|
const deniedOrigin = originAccess.find((access) => !access.allowed);
|
|
if (!userAccess.allowed || deniedOrigin) return finish("failed", "deny_decision_intersection",
|
|
{ reason: "deny_decision_intersection", userReason: userAccess.reason, originReason: deniedOrigin?.reason ?? null });
|
|
const assignment = effect.type === "assign_issue" ? effect : effect.type === "create_issue" ? effect.draft : null;
|
|
await assertAgentInteraction(tx, {
|
|
companyId: decision.companyId, actorAgentId: decision.originAgentId,
|
|
operation: assignment ? "assign" : effect.type === "comment_on_issue" ? "comment" : "edit",
|
|
issueId: effect.type === "create_issue" ? undefined : target.id,
|
|
targetAgentId: assignment?.assigneeAgentId, assigneeUserId: assignment?.assigneeUserId,
|
|
parentIssueId: effect.type === "create_issue" ? effect.draft.parentId ?? target.id : undefined,
|
|
texts: [interpolate(JSON.stringify(effect), decision.inputValues ?? {})], lock: true,
|
|
});
|
|
const effectComment = effect.type === "cancel_issue_tree" ? effect.reasonComment
|
|
: "comment" in effect ? effect.comment : undefined;
|
|
if (effectComment) await assertAgentInteraction(tx, {
|
|
companyId: decision.companyId, actorAgentId: decision.originAgentId,
|
|
operation: "comment", issueId: target.id, texts: [interpolate(effectComment, decision.inputValues ?? {})], lock: true,
|
|
});
|
|
if (effect.type === "cancel_issue_tree") {
|
|
for (const child of referencedIssues.filter(item => item.id !== target.id)) {
|
|
await assertAgentInteraction(tx, { companyId: decision.companyId, actorAgentId: decision.originAgentId,
|
|
operation: "edit", issueId: child.id, lock: true });
|
|
}
|
|
}
|
|
if (effect.staleness === "strict") {
|
|
const staleReference = referencedIssues.some((item) => directReferencedIds.has(item.id) &&
|
|
snapshots[item.id]?.updatedAt !== item.updatedAt.toISOString());
|
|
const changedTree = effect.type === "cancel_issue_tree" && !sameIds(cancellationDescendantIds!,
|
|
await collectDescendantIds(decision.companyId, target.id, tx as unknown as Db));
|
|
if (staleReference || changedTree) return finish("skipped", "target_changed", { reason: "target_changed" });
|
|
}
|
|
const svc = issueService(db);
|
|
const values = decision.inputValues ?? {};
|
|
let result: Record<string, unknown>;
|
|
if (effect.type === "comment_on_issue") {
|
|
const comment = await svc.addComment(target.id, interpolate(effect.bodyMarkdown, values), { userId: decidedByUserId, runId: decision.originRunId }, { clientRequestId: lockKey }, tx);
|
|
result = { commentId: comment.id };
|
|
} else if (effect.type === "update_issue_status") {
|
|
const updated = await svc.update(
|
|
target.id,
|
|
{ status: effect.status, actorUserId: decidedByUserId },
|
|
tx,
|
|
postCommitActivityPublications,
|
|
postCommitIssueActions,
|
|
);
|
|
if (effect.comment) await svc.addComment(target.id, interpolate(effect.comment, values), { userId: decidedByUserId, runId: decision.originRunId }, { clientRequestId: lockKey }, tx);
|
|
result = { issueId: updated?.id, status: updated?.status };
|
|
} else if (effect.type === "assign_issue") {
|
|
const updated = await svc.update(
|
|
target.id,
|
|
{
|
|
assigneeAgentId: effect.assigneeAgentId ?? null,
|
|
assigneeUserId: effect.assigneeUserId ?? null,
|
|
actorUserId: decidedByUserId,
|
|
},
|
|
tx,
|
|
postCommitActivityPublications,
|
|
postCommitIssueActions,
|
|
);
|
|
if (effect.comment) await svc.addComment(target.id, interpolate(effect.comment, values), { userId: decidedByUserId, runId: decision.originRunId }, { clientRequestId: lockKey }, tx);
|
|
result = { issueId: updated?.id };
|
|
} else if (effect.type === "resolve_blocker") {
|
|
const current = await tx.select({ id: issueRelations.issueId }).from(issueRelations).where(and(eq(issueRelations.companyId, decision.companyId), eq(issueRelations.relatedIssueId, target.id), eq(issueRelations.type, "blocks")));
|
|
await svc.update(
|
|
target.id,
|
|
{
|
|
blockedByIssueIds: current.map((row) => row.id).filter((id) => !effect.removeBlockedByIssueIds.includes(id)),
|
|
actorUserId: decidedByUserId,
|
|
},
|
|
tx,
|
|
postCommitActivityPublications,
|
|
postCommitIssueActions,
|
|
);
|
|
result = { removedBlockedByIssueIds: effect.removeBlockedByIssueIds };
|
|
} else if (effect.type === "create_issue") {
|
|
const draft = effect.draft;
|
|
const created = await svc.create(decision.companyId, { title: draft.title, description: draft.description ?? null, parentId: draft.parentId ?? target.id,
|
|
assigneeAgentId: draft.assigneeAgentId ?? null, assigneeUserId: draft.assigneeUserId ?? null, projectId: draft.projectId ?? target.projectId,
|
|
goalId: draft.goalId ?? null, blockedByIssueIds: draft.blockedByIssueIds ?? [], createdByUserId: decidedByUserId, actorRunId: decision.originRunId,
|
|
idempotencyKey: `decision-effect:${decision.id}:${effectIndex}` });
|
|
result = { issueId: created.id };
|
|
} else {
|
|
const cancelled = [target.id, ...cancellationDescendantIds!].reverse();
|
|
for (const id of cancelled) {
|
|
await svc.update(
|
|
id,
|
|
{ status: "cancelled", actorUserId: decidedByUserId },
|
|
tx,
|
|
postCommitActivityPublications,
|
|
postCommitIssueActions,
|
|
);
|
|
}
|
|
await svc.addComment(target.id, interpolate(effect.reasonComment, values), { userId: decidedByUserId, runId: decision.originRunId }, { clientRequestId: lockKey }, tx);
|
|
result = { cancelledIssueIds: cancelled };
|
|
}
|
|
const activity = await effectAudit(tx as unknown as Db, decision, execution.id, effect, "executed", decidedByUserId, originResponsibleUserId, result);
|
|
const [row] = await tx.update(decisionEffectExecutions).set({ status: "executed", result, error: null, activityLogId: activity?.id ?? null, executedAt: new Date() }).where(eq(decisionEffectExecutions.id, execution.id)).returning();
|
|
return row;
|
|
});
|
|
for (const publication of postCommitActivityPublications) publishActivity(publication);
|
|
if (postCommitIssueActions.length > 0) {
|
|
const { executeIssuePostCommitActions } = await import("./issues.js");
|
|
await executeIssuePostCommitActions(db, postCommitIssueActions);
|
|
}
|
|
return executionResult;
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : "Decision effect execution failed";
|
|
return recordFailure("effect_execution_failed", { reason: "effect_execution_failed", message });
|
|
}
|
|
}
|
|
|
|
function runEffects(id: string, userActor?: AuthorizationActor): Promise<EffectAttempt | null> {
|
|
const existing = executing.get(id);
|
|
if (existing) return existing;
|
|
const attempt = (async () => {
|
|
const current = await get(id);
|
|
if (!current) return null;
|
|
if (current.status !== "decided" || current.executionStatus !== "running") {
|
|
return { outcome: await outcome(id), resumed: false };
|
|
}
|
|
// Recovery derives authority from the current claim inside the shared flight.
|
|
const actor = userActor ?? (current.decidedByUserId
|
|
? await recoveryActor(current.companyId, current.decidedByUserId)
|
|
: { type: "none" as const, source: "none" as const });
|
|
return { outcome: await executeDecisionEffects(current, actor), resumed: true };
|
|
})().finally(() => { executing.delete(id); });
|
|
executing.set(id, attempt);
|
|
return attempt;
|
|
}
|
|
|
|
async function settleInvalidEffectOutcome(decision: typeof decisions.$inferSelect) {
|
|
const id = decision.id;
|
|
const publications: ActivityPublication[] = [];
|
|
await writeDecisionChange(db, async tx => {
|
|
const [current] = await tx.select().from(decisions).where(eq(decisions.id, id)).for("update");
|
|
if (!current || current.status !== "decided" || current.executionStatus !== "running") return;
|
|
const reason = !current.decidedByUserId ? "missing_decider"
|
|
: !current.options.some(option => option.id === current.chosenOptionId) ? "missing_option" : null;
|
|
if (!reason) return;
|
|
const [executed] = await tx.select({ id: decisionEffectExecutions.id }).from(decisionEffectExecutions)
|
|
.where(and(eq(decisionEffectExecutions.decisionId, id), eq(decisionEffectExecutions.status, "executed"))).limit(1);
|
|
const executionStatus = executed ? "partial" : "failed";
|
|
await tx.update(decisions).set({ executionStatus, updatedAt: new Date(), metadata: { ...current.metadata,
|
|
effectRecoveryError: reason,
|
|
...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}),
|
|
} }).where(eq(decisions.id, id));
|
|
await logActivity(tx, { companyId: current.companyId, actorType: "system", actorId: "decision-executor",
|
|
agentId: current.originAgentId, runId: current.originRunId, responsibleUserIdOverride: current.decidedByUserId,
|
|
action: "decision.recovery_failed", entityType: "decision", entityId: id,
|
|
details: { reason, executionStatus, chosenOptionId: current.chosenOptionId, decidedByUserId: current.decidedByUserId },
|
|
}, publications);
|
|
}, { companyId: decision.companyId, sourceId: id });
|
|
for (const publication of publications) publishActivity(publication);
|
|
return outcome(id);
|
|
}
|
|
|
|
async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
|
|
const option = decision.options.find((item) => item.id === decision.chosenOptionId);
|
|
if (!option || !decision.decidedByUserId) return settleInvalidEffectOutcome(decision);
|
|
const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null);
|
|
const metadata = decision.metadata as Record<string, unknown>;
|
|
if (metadata.kind === "attention_archive_proposal") {
|
|
if (!verifyDecisionSpec(spec({
|
|
id: decision.id,
|
|
options: decision.options,
|
|
targetSnapshots: decision.targetSnapshots as Record<string, Snapshot>,
|
|
}), decision.signedSpec)) {
|
|
await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
|
|
.where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
|
|
return outcome(decision.id);
|
|
}
|
|
if (option.id === "archive") {
|
|
const manifest = parseAttentionArchiveManifest(
|
|
decision.companyId,
|
|
decision.targetSnapshots as Record<string, Snapshot>,
|
|
);
|
|
const manifestHash = manifest ? hashAttentionArchiveManifest(manifest) : null;
|
|
if (!manifest || manifestHash !== metadata.manifestHash) {
|
|
await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
|
|
.where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
|
|
return outcome(decision.id);
|
|
}
|
|
const originActor: AuthorizationActor = {
|
|
type: "agent",
|
|
agentId: decision.originAgentId,
|
|
companyId: decision.companyId,
|
|
runId: decision.originRunId,
|
|
onBehalfOfUserId: run?.responsibleUserId ?? null,
|
|
source: "agent_jwt",
|
|
};
|
|
try {
|
|
await decisionRetentionService(db).archiveReviewedManifest({
|
|
companyId: decision.companyId,
|
|
manifest,
|
|
originActor,
|
|
decidingActor: userActor,
|
|
decidedByUserId: decision.decidedByUserId,
|
|
});
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : "archive_proposal_failed";
|
|
await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
|
|
.where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
|
|
return outcome(decision.id);
|
|
}
|
|
}
|
|
await writeDecisionChange(db, async tx => tx.update(decisions).set({
|
|
executionStatus: "succeeded",
|
|
updatedAt: new Date(),
|
|
metadata: { ...metadata, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) },
|
|
}).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
|
|
return outcome(decision.id);
|
|
}
|
|
for (let index = 0; index < option.effects.length; index += 1) await executeEffect(decision, option.effects[index]!, index, userActor, decision.decidedByUserId, run?.responsibleUserId ?? null);
|
|
const rows = await db.select().from(decisionEffectExecutions).where(eq(decisionEffectExecutions.decisionId, decision.id));
|
|
const successful = rows.filter((row) => row.status === "executed").length;
|
|
const status = rows.every((row) => row.status === "executed") ? "succeeded" : successful ? "partial" : "failed";
|
|
await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata,
|
|
...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
|
|
return outcome(decision.id);
|
|
}
|
|
|
|
async function resumeDecision(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
|
|
const attempt = decision.executionStatus === "running" ? await runEffects(decision.id, userActor) : undefined;
|
|
if (attempt === null) throw notFound("Decision not found");
|
|
const result = attempt?.outcome ?? await outcome(decision.id);
|
|
await continuations.deliver(result.id);
|
|
return outcome(decision.id);
|
|
}
|
|
|
|
async function decide(input: { id: string; optionId: string; inputValues?: Record<string, string>; idempotencyKey?: string | null; decidedByUserId: string;
|
|
userActor: AuthorizationActor; dismissed?: boolean; dismissReason?: string | null;
|
|
/** Extra activity details, e.g. the receipt for the person the card was for (PAP-749 plan §6). */
|
|
activityDetails?: Record<string, unknown> }) {
|
|
const current = await get(input.id); if (!current) throw notFound("Decision not found");
|
|
const metadata = current.metadata as Record<string, unknown>;
|
|
if (!verifyDecisionSpec(spec({ id: current.id, options: current.options, targetSnapshots: current.targetSnapshots as Record<string, Snapshot> }), current.signedSpec)) throw forbidden("Decision signature verification failed");
|
|
if (current.status === "decided" && input.idempotencyKey && metadata.decideIdempotencyKey === input.idempotencyKey) {
|
|
if (current.decidedByUserId !== input.decidedByUserId) throw forbidden("Decision replay belongs to a different user");
|
|
return resumeDecision(current, input.userActor);
|
|
}
|
|
if (current.status === "decided" && current.chosenOptionId === input.optionId &&
|
|
sameInputValues(current.inputValues ?? {}, input.inputValues ?? {})) {
|
|
if (current.decidedByUserId !== input.decidedByUserId) throw forbidden("Decision replay belongs to a different user");
|
|
return resumeDecision(current, input.userActor);
|
|
}
|
|
if (current.status !== "open") throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
|
|
if (current.expiresAt <= new Date()) {
|
|
await expiry.expire(current, "ttl", new Date());
|
|
throw conflict("decision_expired", { code: "decision_expired" });
|
|
}
|
|
if (!current.options.some((option) => option.id === input.optionId)) throw unprocessable("Unknown optionId");
|
|
const values = input.inputValues ?? {};
|
|
for (const field of current.inputs ?? []) { const value = values[field.id] ?? ""; if (field.required && !value.trim()) throw unprocessable(`Input ${field.id} is required`); if (field.maxLength && value.length > field.maxLength) throw unprocessable(`Input ${field.id} is too long`); }
|
|
const [claimed] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values,
|
|
decidedByUserId: input.decidedByUserId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...metadata,
|
|
decideIdempotencyKey: input.idempotencyKey ?? null,
|
|
...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}),
|
|
...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}) } })
|
|
.where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: current.id });
|
|
if (!claimed) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
|
|
const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, claimed.originRunId)).then((rows) => rows[0] ?? null);
|
|
await logActivity(db, { companyId: claimed.companyId, actorType: "system", actorId: "decision-executor", agentId: claimed.originAgentId, runId: claimed.originRunId,
|
|
responsibleUserIdOverride: input.decidedByUserId, action: input.dismissed ? "decision.dismissed" : "decision.decided", entityType: "decision", entityId: claimed.id,
|
|
details: { chosenOptionId: input.optionId, decidedByUserId: input.decidedByUserId, originResponsibleUserId: run?.responsibleUserId ?? null,
|
|
...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}), ...input.activityDetails } });
|
|
return resumeDecision(claimed, input.userActor);
|
|
}
|
|
|
|
async function cancel(id: string, actor: { actorType: "agent" | "user"; actorId: string; runId?: string | null },
|
|
options: { reason?: string | null; activityDetails?: Record<string, unknown> } = {}) {
|
|
const current = await get(id); if (!current) throw notFound("Decision not found");
|
|
if (actor.actorType === "agent" && actor.actorId !== current.originAgentId) throw forbidden("Only the origin agent may cancel");
|
|
if (current.status === "cancelled") {
|
|
await continuations.deliver(current.id);
|
|
return (await get(id))!;
|
|
}
|
|
// A person who withdraws an out-of-date decision (PAP-749) is recorded as
|
|
// its closer, with the reason, so history can say who withdrew it.
|
|
const reason = options.reason?.trim() || null;
|
|
const byUser = actor.actorType === "user" ? { decidedByUserId: actor.actorId, decidedAt: new Date() } : {};
|
|
const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "cancelled", ...byUser, updatedAt: new Date(), metadata: { ...current.metadata,
|
|
...(reason ? { cancelReason: reason } : {}),
|
|
...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: id });
|
|
if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
|
|
const details = { ...(reason ? { reason } : {}), ...options.activityDetails };
|
|
await logActivity(db, { companyId: updated.companyId, actorType: actor.actorType, actorId: actor.actorId, runId: actor.runId, action: "decision.cancelled", entityType: "decision", entityId: id,
|
|
...(Object.keys(details).length > 0 ? { details } : {}) });
|
|
await continuations.deliver(updated.id);
|
|
return (await get(id))!;
|
|
}
|
|
|
|
async function dismiss(id: string, userId: string, userActor: AuthorizationActor, reason?: string | null) {
|
|
const current = await get(id); if (!current) throw notFound("Decision not found");
|
|
if (!verifyDecisionSpec(spec({ id: current.id, options: current.options, targetSnapshots: current.targetSnapshots as Record<string, Snapshot> }), current.signedSpec)) throw forbidden("Decision signature verification failed");
|
|
const empty = current.options.find((option) => option.effects.length === 0);
|
|
if (empty) return decide({ id, optionId: empty.id, decidedByUserId: userId, userActor, dismissed: true, dismissReason: reason });
|
|
const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId,
|
|
decidedAt: new Date(), updatedAt: new Date(), metadata: { ...current.metadata, dismissed: true, dismissReason: reason ?? null,
|
|
...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: id });
|
|
if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
|
|
await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-executor", agentId: updated.originAgentId,
|
|
runId: updated.originRunId, responsibleUserIdOverride: userId, action: "decision.dismissed", entityType: "decision", entityId: updated.id,
|
|
details: { chosenOptionId: "dismissed", decidedByUserId: userId, dismissed: true } });
|
|
await continuations.deliver(updated.id);
|
|
return outcome(id);
|
|
}
|
|
|
|
async function createBundle(input: { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; title: string; summary: string;
|
|
decisions: Array<Omit<Parameters<typeof create>[0], "companyId" | "actor" | "agentId" | "runId" | "bundleId">> }) {
|
|
return db.transaction(async (tx) => { const provenance = await origin(input.companyId, input.agentId, input.runId);
|
|
const [bundle] = await tx.insert(decisionBundles).values({ companyId: input.companyId, title: input.title, summary: input.summary, originAgentId: input.agentId, originIssueId: provenance.issueId, originRunId: input.runId }).returning();
|
|
const created = []; for (const item of input.decisions) created.push(await create({ ...item, companyId: input.companyId, actor: input.actor, agentId: input.agentId, runId: input.runId, bundleId: bundle.id }, tx as unknown as Db));
|
|
return { ...bundle, decisions: created }; });
|
|
}
|
|
|
|
async function recoverDecisionEffects(id: string) {
|
|
const attempt = await runEffects(id);
|
|
if (!attempt?.resumed || attempt.outcome.executionStatus === "running") return false;
|
|
await continuations.deliver(attempt.outcome.id);
|
|
return true;
|
|
}
|
|
|
|
return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepContinuations: continuations.sweep,
|
|
sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next,
|
|
recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next,
|
|
sweepInvalidTargets: targets.sweep, decisionTargetsChanged: targets.changed, hasDecisionTargetWork: targets.hasPending };
|
|
}
|