From 95d33e1788361b59fe08491b6a81e1890f7f5a0d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:43:36 -0500 Subject: [PATCH] fix(inbox): archive tasks completed by human users (#10668) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The Mine inbox gives each human user a personal work queue. > - A human user can complete a task from the board. > - The status update did not update the user's inbox archive state. > - The completed task therefore stayed in the user's Mine inbox. > - This pull request archives the task for the human user who completes it. > - Agent completion does not change another user's inbox archive state. > - The benefit is that manual completion removes the task from Mine without an extra action or comment. ## Linked Issues or Issue Description **What happened?** When a human user changed a task status to `done`, the task stayed in that user's Mine inbox. **Expected behavior** The status change should archive the task from that user's Mine inbox. The operation should not add a task comment. **Steps to reproduce** 1. Open a task that appears in Mine. 2. Change the task status to `done`. 3. Return to Mine. 4. Observe that the completed task is still present. **Paperclip version or commit** `master` before this change. **Deployment mode** All deployment modes with a board user and the Mine inbox. **Access context** Board (human operator). ## What Changed - Archive the completed task for the board user who changes its status to `done`. - Persist the status update, inbox archive, and archive audit atomically. - Publish live and plugin activity only after the owning transaction commits, including recovery, decision, and approval completion paths. - Keep agent-driven completion from changing a human user's inbox archive state. - Add database-backed regression tests for the archive, audit, Mine filter, rollback, event publication, recovery, and agent paths. ## Verification - `pnpm exec vitest run server/src/__tests__/inbox-archive-routes.test.ts` passed all 7 tests. - `pnpm exec vitest run server/src/__tests__/issue-recovery-actions.test.ts` passed all 44 tests. - Five directly affected server test files passed all 74 tests; decision and comment-route suites passed all 105 tests. - `pnpm --filter @paperclipai/server typecheck` passed after the final fix. - `pnpm -r typecheck` and `pnpm build` passed. - The initial repo-wide `pnpm test:run` passed 3,223 tests; one unrelated plugin orchestration wake-reason test failed and reproduced in isolation. - All latest-head GitHub Actions gates are green, including all server and E2E shards. - Greptile reviewed the latest commit at 5/5 with no unresolved review threads. - Public GitHub search found no duplicate issue or pull request. ## Risks - Low risk. The behavior only runs on a board user's transition into `done`. - Reopening and completing the task again updates the existing per-user archive row. - Agent status updates do not archive a human user's inbox. - Transactional callers must supply a post-commit activity queue; covered completion paths do so and tests exercise rollback and publication order. - This bug fix does not duplicate planned core work in `ROADMAP.md`. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI Codex from the GPT-5 model family assisted with this change. The runtime does not expose the exact serving model ID or context window. Reasoning, repository tools, code execution, GitHub CLI, and Paperclip API access were enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [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: Paperclip --- .../__tests__/inbox-archive-routes.test.ts | 109 ++++++++++++++- .../issue-comment-reopen-routes.test.ts | 1 + .../__tests__/issue-recovery-actions.test.ts | 8 ++ .../low-trust-red-team-routes.test.ts | 2 + server/src/routes/issues.ts | 52 ++++--- server/src/services/activity-log.ts | 94 ++++++++----- server/src/services/decisions.ts | 46 ++++++- server/src/services/index.ts | 8 +- server/src/services/issues.ts | 127 ++++++++++++------ 9 files changed, 343 insertions(+), 104 deletions(-) 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) => {