diff --git a/server/src/__tests__/inbox-archive-routes.test.ts b/server/src/__tests__/inbox-archive-routes.test.ts index 4e6bc3063a..089882b724 100644 --- a/server/src/__tests__/inbox-archive-routes.test.ts +++ b/server/src/__tests__/inbox-archive-routes.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { activityLog, @@ -20,6 +20,7 @@ import { import { LOW_TRUST_REVIEW_PRESET } from "@paperclipai/shared"; import { errorHandler } from "../middleware/index.js"; import { issueRoutes } from "../routes/issues.js"; +import { subscribeCompanyLiveEvents } from "../services/live-events.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -192,6 +193,112 @@ describeEmbeddedPostgres("inbox archive routes", () => { .expect(({ body }) => expect(body).toEqual({ ok: true, userId: seeded.responsibleUserId })); }); + it("silently archives an issue for the board user who moves it to done", async () => { + const seeded = await seed(); + const app = appFor({ + type: "board", + source: "session", + userId: seeded.responsibleUserId, + companyIds: [seeded.companyId], + memberships: [{ companyId: seeded.companyId, membershipRole: "operator", status: "active" }], + isInstanceAdmin: false, + }); + + await request(app) + .patch(`/api/issues/${seeded.issueId}`) + .send({ status: "done" }) + .expect(200) + .expect(({ body }) => expect(body).toMatchObject({ id: seeded.issueId, status: "done" })); + + const [archive] = await db + .select() + .from(issueInboxArchives) + .where(eq(issueInboxArchives.issueId, seeded.issueId)); + expect(archive).toMatchObject({ + issueId: seeded.issueId, + userId: seeded.responsibleUserId, + archivedByActorType: "user", + archivedByAgentId: null, + archivedByRunId: null, + }); + + const [archiveAudit] = await db + .select() + .from(activityLog) + .where(eq(activityLog.action, "issue.inbox_archived")); + expect(archiveAudit).toMatchObject({ + actorType: "user", + actorId: seeded.responsibleUserId, + entityId: seeded.issueId, + details: { + userId: seeded.responsibleUserId, + targetResolvedFrom: "responsible_user", + source: "issue_status_done", + }, + }); + + await request(app) + .get(`/api/companies/${seeded.companyId}/issues`) + .query({ + touchedByUserId: seeded.responsibleUserId, + inboxArchivedByUserId: seeded.responsibleUserId, + status: "backlog,todo,in_progress,in_review,blocked,done", + }) + .expect(200) + .expect(({ body }) => expect(body.map((issue: { id: string }) => issue.id)).not.toContain(seeded.issueId)); + }); + + it("rolls back completion when the inbox archive audit cannot be written", async () => { + const seeded = await seed(); + const liveEvents: Array<{ type: string; payload: Record }> = []; + const unsubscribe = subscribeCompanyLiveEvents(seeded.companyId, (event) => liveEvents.push(event)); + const app = appFor({ + type: "board", + source: "session", + userId: seeded.responsibleUserId, + companyIds: [seeded.companyId], + memberships: [{ companyId: seeded.companyId, membershipRole: "operator", status: "active" }], + isInstanceAdmin: false, + }); + + await db.execute(sql` + alter table activity_log + add constraint reject_done_inbox_archive_audit + check (action <> 'issue.inbox_archived') + `); + try { + await request(app) + .patch(`/api/issues/${seeded.issueId}`) + .send({ status: "done" }) + .expect(500); + expect(liveEvents).not.toContainEqual(expect.objectContaining({ + type: "activity.logged", + payload: expect.objectContaining({ action: "issue.inbox_archived" }), + })); + } finally { + unsubscribe(); + await db.execute(sql` + alter table activity_log + drop constraint reject_done_inbox_archive_audit + `); + } + + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(issue.status).toBe("todo"); + expect(await db.select().from(issueInboxArchives)).toHaveLength(0); + }); + + it("does not archive a responsible user's inbox when an agent moves an issue to done", async () => { + const seeded = await seed(); + + await request(appFor(agentActor(seeded))) + .patch(`/api/issues/${seeded.issueId}`) + .send({ status: "done" }) + .expect(200); + + expect(await db.select().from(issueInboxArchives)).toHaveLength(0); + }); + it("archives for the responsible user with agent/run attribution and resurfaces after new activity", async () => { const seeded = await seed(); const app = appFor(agentActor(seeded)); diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index 85bd66d907..2cfaf96775 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -1890,6 +1890,7 @@ describe.sequential("issue comment reopen routes", () => { }), }), mockTx, + expect.any(Array), ); const updatePatch = mockIssueService.update.mock.calls[0]?.[1] as Record; const decisionId = updatePatch.executionState.lastDecisionId; diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 911137dc79..236f884c34 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -13,6 +13,7 @@ import { environments, heartbeatRuns, issueComments, + issueInboxArchives, issueRecoveryActions, issueRelations, issues, @@ -140,6 +141,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); await db.delete(environments); + await db.delete(issueInboxArchives); await db.delete(issues); await db.delete(agents); await db.delete(companies); @@ -1389,6 +1391,12 @@ describeEmbeddedPostgres("issue recovery actions", () => { }); expect(resolved.body.recoveryAction.resolvedAt).toBeTruthy(); expect(await recoveryActionSvc.getActiveForIssue(companyId, sourceIssueId)).toBeNull(); + expect( + await db + .select() + .from(issueInboxArchives) + .where(eq(issueInboxArchives.issueId, sourceIssueId)), + ).toHaveLength(1); const detail = await request(app).get(`/api/issues/${sourceIssueId}`).expect(200); expect(detail.body.activeRecoveryAction).toBeNull(); diff --git a/server/src/__tests__/low-trust-red-team-routes.test.ts b/server/src/__tests__/low-trust-red-team-routes.test.ts index 10fe7f96c7..1464ab6d32 100644 --- a/server/src/__tests__/low-trust-red-team-routes.test.ts +++ b/server/src/__tests__/low-trust-red-team-routes.test.ts @@ -28,6 +28,7 @@ import { issueApprovals, issueComments, issueDocuments, + issueInboxArchives, issueRelations, issues, issueThreadInteractions, @@ -723,6 +724,7 @@ describeEmbeddedPostgres("low-trust red-team HTTP route regression suite", () => await db.delete(documents); await db.delete(issueComments); await db.delete(issueRelations); + await db.delete(issueInboxArchives); await db.delete(activityLog); await db.delete(heartbeatRunEvents); await deleteHeartbeatRunsAndWakeupsAfterActivityLogDrains(db); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index d60866e64d..6515491516 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -117,11 +117,13 @@ import { ISSUE_LIST_MAX_LIMIT, issueReferenceService, issueService, + type ActivityPublication, type IssueFilters, clampIssueListLimit, documentService, documentAnnotationService, logActivity, + publishActivity, projectService, routineService, workProductService, @@ -5737,6 +5739,7 @@ export function issueRoutes( }); const actionStatus = outcome === "cancelled" ? "cancelled" : "resolved"; + const postCommitActivityPublications: ActivityPublication[] = []; const result = await db.transaction(async (tx) => { let issue = existing; if (outcome === "blocked") { @@ -5768,6 +5771,7 @@ export function issueRoutes( actorUserId: actor.actorType === "user" ? actor.actorId : null, }, tx, + postCommitActivityPublications, ); if (!updatedIssue) throw notFound("Issue not found"); issue = updatedIssue; @@ -5788,6 +5792,7 @@ export function issueRoutes( return { issue, recoveryAction }; }); + for (const publication of postCommitActivityPublications) publishActivity(publication); await routinesSvc.syncRunStatusForIssue(result.issue.id); @@ -8202,20 +8207,30 @@ export function issueRoutes( const stopRelayResult: { value: Awaited>; } = { value: null }; + const postCommitActivityPublications: ActivityPublication[] = []; + const issueUpdateData = { + ...updateFields, + actorAgentId: actor.agentId ?? null, + actorUserId: actor.actorType === "user" ? actor.actorId : null, + }; + const shouldCollectCompletionPublication = + actor.actorType === "user" && existing.status !== "done" && updateFields.status === "done"; + const updateIssue = (tx?: Parameters[2]) => { + if (tx) { + return shouldCollectCompletionPublication + ? svc.update(id, issueUpdateData, tx, postCommitActivityPublications) + : svc.update(id, issueUpdateData, tx); + } + return shouldCollectCompletionPublication + ? svc.update(id, issueUpdateData, db, postCommitActivityPublications) + : svc.update(id, issueUpdateData); + }; let issue: Awaited>; try { if (transition.decision && decisionId) { const decision = transition.decision; issue = await db.transaction(async (tx) => { - const updated = await svc.update( - id, - { - ...updateFields, - actorAgentId: actor.agentId ?? null, - actorUserId: actor.actorType === "user" ? actor.actorId : null, - }, - tx, - ); + const updated = await updateIssue(tx); if (!updated) return null; await tx.insert(issueExecutionDecisions).values({ @@ -8239,21 +8254,13 @@ export function issueRoutes( }); } else if (shouldRelayStop) { issue = await db.transaction(async (tx) => { - const updated = await svc.update(id, { - ...updateFields, - actorAgentId: actor.agentId ?? null, - actorUserId: actor.actorType === "user" ? actor.actorId : null, - }, tx); + const updated = await updateIssue(tx); if (!updated) return null; stopRelayResult.value = await svc.addStopRelayCommentIfNeeded(updated, tx); return updated; }); } else { - issue = await svc.update(id, { - ...updateFields, - actorAgentId: actor.agentId ?? null, - actorUserId: actor.actorType === "user" ? actor.actorId : null, - }); + issue = await updateIssue(); } } catch (err) { if (err instanceof HttpError && err.status === 422) { @@ -8282,6 +8289,7 @@ export function issueRoutes( res.status(404).json({ error: "Issue not found" }); return; } + for (const publication of postCommitActivityPublications) publishActivity(publication); if (enteringBlocked) { const blockedIssue = issue; @@ -10331,6 +10339,7 @@ export function issueRoutes( sourceTrust, }; let txResult: { comment: Awaited>; issue: NonNullable>> }; + const postCommitActivityPublications: ActivityPublication[] = []; try { txResult = await db.transaction(async (tx) => { const insertedComment = await svc.addComment( @@ -10344,7 +10353,9 @@ export function issueRoutes( commentOptions, tx, ); - const updated = await svc.update(id, updatePatch, tx); + const updated = actor.actorType === "user" && currentIssue.status !== "done" + ? await svc.update(id, updatePatch, tx, postCommitActivityPublications) + : await svc.update(id, updatePatch, tx); // Throw (not return null) so drizzle rolls back the inserted comment when the issue // has been concurrently deleted between the initial fetch and the in-transaction update. if (!updated) throw new AutoApprovalIssueMissingError(); @@ -10373,6 +10384,7 @@ export function issueRoutes( } throw err; } + for (const publication of postCommitActivityPublications) publishActivity(publication); comment = txResult.comment; currentIssue = txResult.issue; // Mirror the normal status-change audit trail: every other in_review -> done path diff --git a/server/src/services/activity-log.ts b/server/src/services/activity-log.ts index 5ff0a6fd78..2ba70359ce 100644 --- a/server/src/services/activity-log.ts +++ b/server/src/services/activity-log.ts @@ -66,6 +66,12 @@ export interface LogActivityInput { responsibleUserIdOverride?: string | null; } +export interface ActivityPublication { + companyId: string; + payload: Record; + pluginEvent: PluginEvent | null; +} + export async function createActivityDetailsRedactor(db: Db) { const currentUserRedactionOptions = { enabled: (await instanceSettingsService(db).getGeneral()).censorUsernameInLogs, @@ -142,7 +148,16 @@ export async function resolveResponsibleUserIdForActivity(db: Db, input: LogActi return readNonEmptyString(company?.defaultResponsibleUserId); } -export async function logActivity(db: Db, input: LogActivityInput) { +export function publishActivity(publication: ActivityPublication) { + publishLiveEvent({ + companyId: publication.companyId, + type: "activity.logged", + payload: publication.payload, + }); + if (publication.pluginEvent) publishPluginDomainEvent(publication.pluginEvent); +} + +export async function persistActivity(db: Db, input: LogActivityInput) { const redactedDetails = await redactActivityDetails(db, input.details ?? null); const responsibleUserId = await resolveResponsibleUserIdForActivity(db, input); const [activity] = await db.insert(activityLog).values({ @@ -158,42 +173,49 @@ export async function logActivity(db: Db, input: LogActivityInput) { details: redactedDetails, }).returning({ id: activityLog.id }); - publishLiveEvent({ - companyId: input.companyId, - type: "activity.logged", - payload: { - actorType: input.actorType, - actorId: input.actorId, - action: input.action, - entityType: input.entityType, - entityId: input.entityId, - agentId: input.agentId ?? null, - runId: input.runId ?? null, - responsibleUserId, - details: redactedDetails, - }, - }); - + const payload = { + actorType: input.actorType, + actorId: input.actorId, + action: input.action, + entityType: input.entityType, + entityId: input.entityId, + agentId: input.agentId ?? null, + runId: input.runId ?? null, + responsibleUserId, + details: redactedDetails, + }; const pluginEventType = eventTypeForActivityAction(input.action); - if (pluginEventType) { - const event: PluginEvent = { - eventId: randomUUID(), - eventType: pluginEventType, - occurredAt: new Date().toISOString(), - actorId: input.actorId, - actorType: input.actorType, - entityId: input.entityId, - entityType: input.entityType, - companyId: input.companyId, - payload: { - ...redactedDetails, - agentId: input.agentId ?? null, - runId: input.runId ?? null, - responsibleUserId, - }, - }; - publishPluginDomainEvent(event); - } + const pluginEvent: PluginEvent | null = pluginEventType + ? { + eventId: randomUUID(), + eventType: pluginEventType, + occurredAt: new Date().toISOString(), + actorId: input.actorId, + actorType: input.actorType, + entityId: input.entityId, + entityType: input.entityType, + companyId: input.companyId, + payload: { + ...redactedDetails, + agentId: input.agentId ?? null, + runId: input.runId ?? null, + responsibleUserId, + }, + } + : null; + return { + activity, + publication: { + companyId: input.companyId, + payload, + pluginEvent, + } satisfies ActivityPublication, + }; +} + +export async function logActivity(db: Db, input: LogActivityInput) { + const { activity, publication } = await persistActivity(db, input); + publishActivity(publication); return activity; } diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index 22435dd1fe..64fe0a91fd 100644 --- a/server/src/services/decisions.ts +++ b/server/src/services/decisions.ts @@ -5,7 +5,7 @@ import { companyMemberships, decisionBundles, decisionEffectExecutions, decision import type { 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 } from "./activity-log.js"; +import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; import { signDecisionSpec, verifyDecisionSpec } from "./decision-signing.js"; import { issueService } from "./issues.js"; @@ -337,7 +337,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }); try { - return await db.transaction(async (tx) => { + const postCommitActivityPublications: ActivityPublication[] = []; + 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); @@ -391,23 +392,45 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { 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(tx as unknown as Db); + const svc = issueService(db); const values = decision.inputValues ?? {}; let result: Record; if (effect.type === "comment_on_issue") { const comment = await svc.addComment(target.id, interpolate(effect.bodyMarkdown, values), { userId: decidedByUserId }, undefined, 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); + const updated = await svc.update( + target.id, + { status: effect.status, actorUserId: decidedByUserId }, + tx, + postCommitActivityPublications, + ); if (effect.comment) await svc.addComment(target.id, interpolate(effect.comment, values), { userId: decidedByUserId }, undefined, 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); + const updated = await svc.update( + target.id, + { + assigneeAgentId: effect.assigneeAgentId ?? null, + assigneeUserId: effect.assigneeUserId ?? null, + actorUserId: decidedByUserId, + }, + tx, + postCommitActivityPublications, + ); if (effect.comment) await svc.addComment(target.id, interpolate(effect.comment, values), { userId: decidedByUserId }, undefined, 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); + await svc.update( + target.id, + { + blockedByIssueIds: current.map((row) => row.id).filter((id) => !effect.removeBlockedByIssueIds.includes(id)), + actorUserId: decidedByUserId, + }, + tx, + postCommitActivityPublications, + ); result = { removedBlockedByIssueIds: effect.removeBlockedByIssueIds }; } else if (effect.type === "create_issue") { const draft = effect.draft; @@ -418,7 +441,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { 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); + for (const id of cancelled) { + await svc.update( + id, + { status: "cancelled", actorUserId: decidedByUserId }, + tx, + postCommitActivityPublications, + ); + } await svc.addComment(target.id, interpolate(effect.reasonComment, values), { userId: decidedByUserId }, undefined, tx); result = { cancelledIssueIds: cancelled }; } @@ -426,6 +456,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { 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); + 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 }); diff --git a/server/src/services/index.ts b/server/src/services/index.ts index 9b28563655..dd44237841 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -148,7 +148,13 @@ export { executionWorkspaceService } from "./execution-workspaces.js"; export { workspaceOperationService } from "./workspace-operations.js"; export { workspaceFileResourceService } from "./workspace-file-resources.js"; export { workProductService } from "./work-products.js"; -export { logActivity, type LogActivityInput } from "./activity-log.js"; +export { + logActivity, + persistActivity, + publishActivity, + type ActivityPublication, + type LogActivityInput, +} from "./activity-log.js"; export { summarySlotService, SUMMARIZER_BUILT_IN_KEY } from "./summary-slots.js"; export { notifyHireApproved, type NotifyHireApprovedInput } from "./hire-hook.js"; export { publishLiveEvent, subscribeCompanyLiveEvents } from "./live-events.js"; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 83872d6ab6..b42d8382b9 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -115,7 +115,12 @@ import { classifyIssueGraphLiveness, type IssueLivenessFinding } from "./recover import { visibleIssueCondition } from "./issue-visibility.js"; import { finalizeStatusCardsForStalledGeneration } from "./status-card-finalization.js"; import { finalizeSummarySlotsForTerminalIssue } from "./summary-slot-finalization.js"; -import { logActivity } from "./activity-log.js"; +import { + logActivity, + persistActivity, + publishActivity, + type ActivityPublication, +} from "./activity-log.js"; import { buildIssueChanges } from "./issue-change-receipt.js"; const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"]; @@ -1568,6 +1573,11 @@ function inboxVisibleForUserCondition(companyId: string, userId: string) { AND ${activityLog.details}->>'status' IN ('in_review', 'blocked', 'done') AND ${activityLog.details}->'_previous'->>'status' IS DISTINCT FROM ${activityLog.details}->>'status' + AND NOT ( + ${activityLog.details}->>'status' = 'done' + AND ${issues.completedAt} IS NOT NULL + AND ${issueInboxArchives.archivedAt} >= ${issues.completedAt} + ) ) OR EXISTS ( SELECT 1 @@ -4947,6 +4957,45 @@ export function issueService(db: Db) { return { comment, parent }; } + async function archiveInbox( + companyId: string, + issueId: string, + userId: string, + archivedAt: Date = new Date(), + attribution?: { + archivedByActorType: "user" | "agent"; + archivedByAgentId?: string | null; + archivedByRunId?: string | null; + }, + dbOrTx: any = db, + ) { + const now = new Date(); + const [row] = await dbOrTx + .insert(issueInboxArchives) + .values({ + companyId, + issueId, + userId, + archivedByActorType: attribution?.archivedByActorType ?? "user", + archivedByAgentId: attribution?.archivedByAgentId ?? null, + archivedByRunId: attribution?.archivedByRunId ?? null, + archivedAt, + updatedAt: now, + }) + .onConflictDoUpdate({ + target: [issueInboxArchives.companyId, issueInboxArchives.issueId, issueInboxArchives.userId], + set: { + archivedAt, + archivedByActorType: attribution?.archivedByActorType ?? "user", + archivedByAgentId: attribution?.archivedByAgentId ?? null, + archivedByRunId: attribution?.archivedByRunId ?? null, + updatedAt: now, + }, + }) + .returning(); + return row; + } + return { clearExecutionRunIfTerminal, clearCheckoutRunIfTerminal, @@ -5306,43 +5355,7 @@ export function issueService(db: Db) { return deleted.length > 0; }, - archiveInbox: async ( - companyId: string, - issueId: string, - userId: string, - archivedAt: Date = new Date(), - attribution?: { - archivedByActorType: "user" | "agent"; - archivedByAgentId?: string | null; - archivedByRunId?: string | null; - }, - ) => { - const now = new Date(); - const [row] = await db - .insert(issueInboxArchives) - .values({ - companyId, - issueId, - userId, - archivedByActorType: attribution?.archivedByActorType ?? "user", - archivedByAgentId: attribution?.archivedByAgentId ?? null, - archivedByRunId: attribution?.archivedByRunId ?? null, - archivedAt, - updatedAt: now, - }) - .onConflictDoUpdate({ - target: [issueInboxArchives.companyId, issueInboxArchives.issueId, issueInboxArchives.userId], - set: { - archivedAt, - archivedByActorType: attribution?.archivedByActorType ?? "user", - archivedByAgentId: attribution?.archivedByAgentId ?? null, - archivedByRunId: attribution?.archivedByRunId ?? null, - updatedAt: now, - }, - }) - .returning(); - return row; - }, + archiveInbox, /** * Seed inbox archives for a batch of freshly imported issues so a company @@ -7020,7 +7033,10 @@ export function issueService(db: Db) { actorUserId?: string | null; }, dbOrTx: any = db, + postCommitActivityPublications?: ActivityPublication[], ) => { + const ownedActivityPublications: ActivityPublication[] = []; + const activityPublications = postCommitActivityPublications ?? ownedActivityPublications; const existing = await dbOrTx .select() .from(issues) @@ -7357,6 +7373,35 @@ export function issueService(db: Db) { ); } } + if (actorUserId && receiptExisting.status !== "done" && updated.status === "done") { + if (dbOrTx !== db && !postCommitActivityPublications) { + throw new Error("Human completion in an external transaction requires a post-commit activity queue"); + } + const now = new Date(); + const archiveState = await archiveInbox( + updated.companyId, + updated.id, + actorUserId, + now, + undefined, + tx, + ); + const { publication } = await persistActivity(tx as unknown as Db, { + companyId: updated.companyId, + actorType: "user", + actorId: actorUserId, + action: "issue.inbox_archived", + entityType: "issue", + entityId: updated.id, + details: { + userId: actorUserId, + archivedAt: archiveState.archivedAt, + targetResolvedFrom: "responsible_user", + source: "issue_status_done", + }, + }); + activityPublications.push(publication); + } return { ...enriched, ...(nextBlockedByIssueIds !== undefined ? { blockedByIssueIds: nextBlockedByIssueIds } : {}), @@ -7364,7 +7409,11 @@ export function issueService(db: Db) { }; }; - return dbOrTx === db ? db.transaction(runUpdate) : runUpdate(dbOrTx); + const result = await (dbOrTx === db ? db.transaction(runUpdate) : runUpdate(dbOrTx)); + if (dbOrTx === db && !postCommitActivityPublications) { + for (const publication of ownedActivityPublications) publishActivity(publication); + } + return result; }, clearExecutionWorkspaceEnvironmentSelection: async (companyId: string, environmentId: string) => {