From 4009eb34cb818ebf6d2b9478dfb4e71dd20dcbb2 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 1 Aug 2026 03:24:36 -0700 Subject: [PATCH] FN-8683: remove unreachable SQLite task polling replica Document PostgreSQL task-deletion observation and remove the obsolete SQLite polling path. - Remove polling state, replica emissions, and activity-log suppression from TaskStore. - Retain backend-aware cache warming while documenting the transactional-outbox follow-up. - Add tombstone and soft-delete abort coverage across core and engine lanes. Files changed: docs/architecture.md | 3 +- ...tgres-cross-process-task-deleted-observation.md | 128 ++++++++++++++++ docs/storage.md | 3 +- .../task-delete-nonblocking-cleanup.test.ts | 54 +++++++ .../task-deleted-polling-replica-tombstone.test.ts | 57 +++++++ .../task-updated-lanes-emit-surfaces.test.ts | 26 ---- packages/core/src/store.ts | 13 +- packages/core/src/task-store/lifecycle-ops.ts | 168 ++------------------- packages/core/src/task-store/task-artifacts-ops.ts | 4 - .../__tests__/executor-soft-delete-abort.test.ts | 15 ++ .../src/__tests__/triage-soft-delete-abort.test.ts | 14 ++ 11 files changed, 284 insertions(+), 201 deletions(-) Fusion-Task-Id: FN-8683 Fusion-Task-Lineage: a052db0c-b6fc-4b05-b6b2-f8217b56ded0 Co-authored-by: Fusion (runfusion.ai) --- docs/architecture.md | 3 +- ...-cross-process-task-deleted-observation.md | 128 +++++++++++++ docs/storage.md | 3 +- .../task-delete-nonblocking-cleanup.test.ts | 54 ++++++ ...-deleted-polling-replica-tombstone.test.ts | 57 ++++++ .../task-updated-lanes-emit-surfaces.test.ts | 26 --- packages/core/src/store.ts | 13 +- packages/core/src/task-store/lifecycle-ops.ts | 168 ++---------------- .../core/src/task-store/task-artifacts-ops.ts | 4 - .../executor-soft-delete-abort.test.ts | 15 ++ .../triage-soft-delete-abort.test.ts | 14 ++ 11 files changed, 284 insertions(+), 201 deletions(-) create mode 100644 docs/solutions/architecture/postgres-cross-process-task-deleted-observation.md create mode 100644 packages/core/src/__tests__/task-deleted-polling-replica-tombstone.test.ts diff --git a/docs/architecture.md b/docs/architecture.md index f399923f02..fb674d26d2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1398,7 +1398,7 @@ Limits are controlled by project settings (`maxSpawnedAgentsPerParent`, `maxSpaw - FN-7939: the FN-7645/FN-7718 repair audit is itself self-supervised. `HeartbeatTriggerScheduler` records audit-loop liveness and runs an independent watchdog that re-arms the 60s audit interval plus immediately executes one audit when the driver is stale for a bounded multiple of the audit cadence, so a dropped auditor cannot strand active agents for hours while timer entries remain present. Repeated `zombie-timer-rearmed` repairs that do not advance `lastHeartbeatAt` are also bounded: after the non-advancing count crosses the scheduler threshold, the repair metadata records the count/escalation and logs the greppable `reason=heartbeat-rearm-nonadvancing-escalated` signal instead of silently churning forever; intentional `globalPause`/`enginePaused` suppression and healthy active runs remain non-escalating. - FN-7718: CLI-driven `fn agent stop`/`start` mutate the agent row from a SEPARATE process, so the in-process `agent:updated` listener never fires for those transitions — the 60s audit is the ONLY cross-process reconciliation path. The audit now invalidates a stopped/non-eligible agent's lingering timer entry (state made non-tickable, `runtimeConfig.enabled === false`, or ephemeral/`!isHeartbeatManaged`) instead of bare-`continue`ing past it, so the entry never survives to become an orphaned/"zombie" registration. `syncTimerForAgent` mirrors this for the in-process start seam: an eligible agent whose present timer entry is already stale beyond the same repair threshold is force-cleared and re-armed rather than left in place by the "already ticking" no-op. Net effect: a `stop`/`start` cycle durably clears the zombie-timer condition in one audit cycle instead of deferring repair to the FN-7645 stale-repair path minutes later. - FN-7723 (follow-up from FN-7718): `AgentStore` (`packages/core/src/agent-store.ts`) now supports an opt-in cross-process change-detection fast-path over the FN-7645/FN-7718 audit backstop — `startWatching()`/`stopWatching()`/`checkForChanges()`, modeled directly on `TaskStore`'s `fs.watch`+poll pattern (`packages/core/src/store.ts`): an `fs.watch` on the project's `.fusion` dir as a fail-soft fast-path nudge, plus an always-on poll fallback (default 2s) gated by `db.getLastModified()` so an unchanged DB costs one cheap `__meta` read. On a detected change it diffs current agent rows against a last-seen per-instance snapshot (comparing `state` explicitly, not just `updatedAt`, since two rapid writes can land in the same ISO-millisecond and mask a genuine transition) and re-emits the EXISTING `agent:updated`/`agent:stateChanged` events — no new event names, so `HeartbeatTriggerScheduler.watchAgentLifecycle`'s current listener reacts unchanged, funneling through the same `syncTimerForAgent` seam (including FN-7718's stale-present-entry force-re-arm). Only the long-lived engine `AgentStore` instance opts in (started/stopped alongside `HeartbeatTriggerScheduler` in `packages/engine/src/runtimes/in-process-runtime.ts`); the CLI's short-lived `AgentStore` (`packages/cli/src/commands/agent.ts`) and per-request dashboard stores never call `startWatching()`. The 60s `auditTimerRegistrations` sweep is UNCHANGED and remains the durable backstop — this is a purely additive latency improvement, not a replacement: a `fn agent stop`/`start` is now typically observed within one poll interval (~2s) instead of up to 60s. -- FN-7726 (follow-up from FN-7723): the mechanical fs.watch+poll lifecycle `TaskStore.watch()` and `AgentStore.startWatching()` each hand-rolled (fail-soft `fs.watch` setup with its two canonical warn strings, the `setInterval` poll fallback, and idempotent teardown) is now a single shared `FsWatchPollController` (`packages/core/src/fs-watch-poll-controller.ts`). The controller owns ONLY that mechanism — `start({dir, recursive?, pollIntervalMs, onPoll, log, errorContext?})`/`stop()`/`isWatching()`/`watcher` (a live-handle getter kept for test seams). It does NOT own diff/emit logic or gating: `getLastModified()`-vs-`lastKnownModified` gating and the `pollingInProgress` re-entrancy guard remain private fields on each store, evaluated inside each store's own `checkForChanges()` (the function passed to the controller as `onPoll`) exactly as before extraction — a deliberate scoping decision (see FN-7726's task `plan` document) to avoid coupling the shared controller to each store's very different diff bodies (TaskStore's delete/archive/artifact-cursor diff vs. AgentStore's snapshot state compare). `TaskStore` and `AgentStore` each hold a private `watchPoll: FsWatchPollController` instance and pass their own logger (`storeLog`/`agentStoreLog`) so the `[task-store]`-prefixed and agent-prefixed fail-soft warn strings are unchanged. Public method names/signatures (`watch()`/`stopWatching()` on `TaskStore`; `startWatching()`/`stopWatching()`/`isWatching()`/`checkForChanges()` on `AgentStore`) and all existing events/latencies/gating are unchanged — this is a behavior-preserving internal refactor, not a new mechanism. +- FN-7726's `FsWatchPollController` remains the shared fail-soft watcher/poll primitive for the live `AgentStore` cross-process fast path. FN-8683 removed the unreachable SQLite `TaskStore` polling replica: supported PostgreSQL TaskStores warm only their local cache in `watch()` and do not claim cross-process task lifecycle observation. See [PostgreSQL cross-process `task:deleted` observation](./solutions/architecture/postgres-cross-process-task-deleted-observation.md) for the transactional-outbox decision and its planned durable cursor/replay contract. ### Custom instructions `packages/engine/src/agent-instructions.ts` resolves per-agent instruction text/path with path-traversal and extension validation. @@ -2282,6 +2282,7 @@ This section preserves the detailed lifecycle/self-healing contracts that were f - **Landed-files attribution (FN-5103)**: Rebase-strategy `mergeDetails.landedFiles` / `filesChanged` / `insertions` / `deletions` are captured from task-attributable commits only via `filterFilesToOwnTaskCommits` (subject-prefix + trailer + bracket-prefix evidence), tagged `landedFilesAttributionRestricted: true`. Zero own commits → `landedFiles: []` and `noOpVerifiedShortCircuit: true`. FN-5304 guard: when `..HEAD` reports zero own commits, merger must also validate the source `fusion/` tip; if that source tip still has attributable own commits relative to `rebaseBaseSha`, throw `SilentNoOpAttributionMismatchError`, refuse writing `mergeConfirmed: true`, park the task in `in-review` with `status: "failed"`, and emit `merge:no-op-attribution-mismatch`. If source ref is unavailable, skip with diagnostic + `merge:no-op-attribution-mismatch-skipped` (`reason: "source-ref-unavailable"`). Attribution-helper failures fall back to the unrestricted `rebaseBaseSha..sha` walk and set `landedFilesCaptureFallback: 'attribution-failed'`. Self-healing `recoverDoneTaskMergeMetadata` skips reconcile when `landedFilesAttributionRestricted` or `noOpVerifiedShortCircuit` is set so the narrower set is not overwritten with the full range. Squash-strategy capture is unchanged. - **Soft-delete scheduler invalidation (FN-5137)**: `task:deleted` events must invalidate `AutoClaimSnapshotManager` and clear scheduler bookkeeping (`pausedTaskIds`, `failedTaskIds`, `wasNodeDispatchValidationBlocked`, `wasNodeBlocked`); `executor.execute()` / `resumeOrphaned()` / `resumeTaskForAgent()` refuse any task with `deletedAt` set. - **Soft-delete in-flight abort (FN-5142)**: `task:deleted` must immediately abort/dispose active executor work (`activeSessions`, `activeStepExecutors`, `activeWorkflowStepSessions`, reviewer subagents), interrupt active merge state (`mergeAbortController`, `activeMergeSession`, `activeMergeTaskId`, `mergeActive`, `mergeQueue`, `pausedReviewTaskIds`), and abort triage specify/subagent sessions for that id. Handlers are per-task and idempotent. +- **Cross-process soft-delete boundary (FN-8683)**: `TaskStore` lifecycle events are currently process-local. The former SQLite polling replica was unreachable in PostgreSQL `AsyncDataLayer` mode and was removed rather than implying remote `task:deleted` delivery exists. [The transactional-outbox decision](./solutions/architecture/postgres-cross-process-task-deleted-observation.md) defines the future durable cursor, replay, poison, and retention contract; until it ships, cross-process consumers must reconcile explicitly and must not repeat writer-owned delete audit or mailbox side effects. - **Archive releases active-session locks (FN-7717)**: `task:moved` with `to === "archived"` (from ANY column, including in-progress via a direct single-hop `fn_task_archive`, and triage/planning where Plan Review and other workflow-step sessions run) now disposes in-flight session surfaces via `awaitAbortInFlightTaskWork` and sweeps any remaining `activeSessionRegistry` paths for the task, so a leaked lock can never survive archive and block a successor task's `registerPath` on the same session path. The `to === "archived"` check is ordered BEFORE the `from === "in-progress"` branch so a direct in-progress→archived move gets the same full cleanup instead of falling into the narrower in-progress-only branch. `to === "done"`/`"in-review"` are deliberately excluded — those columns legitimately hold `ai-merge`/`workspace-repo-land` merge leases. - **Soft-delete audit + column reconcile (FN-5175)**: `TaskStore.deleteTask` records a `runAuditEvents` row (`mutationType: "task:deleted"`, `domain: "database"`) inside the same transaction that sets `deletedAt`, and sets `"column" = 'archived'` on the row. Callers without a heartbeat run context (`fn task delete`, pi extension, dashboard delete route) pass an `auditContext` with `agentId: "system"` and a synthetic `runId`. The watcher cross-instance emit path does NOT re-record the audit event. The row stays in `tasks` (not `archivedTasks`); `archiveTask` is unchanged. - **Soft-delete fast path cleanup (FN-7968)**: user-visible `TaskStore.deleteTask` completion is bounded by the soft-delete transaction, audit row, cache/event emission, dependency/lineage gates, and near-duplicate cleanup. Potentially slow branch cleanup (`cleanupBranchForTask` git subprocesses) is scheduled after the row is already soft-deleted; it must still clear stale execution-start branch references and persist the cleaned-branch log entry on the deleted row. Dashboard `DELETE /tasks/:id` likewise responds after `deleteTask` and schedules execution-agent binding release off the HTTP critical path; the release remains observable through warning logs on failure. diff --git a/docs/solutions/architecture/postgres-cross-process-task-deleted-observation.md b/docs/solutions/architecture/postgres-cross-process-task-deleted-observation.md new file mode 100644 index 0000000000..2b29007776 --- /dev/null +++ b/docs/solutions/architecture/postgres-cross-process-task-deleted-observation.md @@ -0,0 +1,128 @@ +--- +category: architecture +module: PostgreSQL lifecycle observation +tags: + - postgresql + - task-lifecycle + - cross-process + - outbox +problem_type: architecture-decision +applies_when: A Fusion deployment has more than one TaskStore process sharing a PostgreSQL project and a consumer needs to observe task deletion. +--- + +# PostgreSQL cross-process `task:deleted` observation + +## Decision + +Use a **transactional PostgreSQL outbox with per-consumer durable cursors** for cross-process lifecycle observation. The initial scope is `task:deleted`; its event contract must be deliberately extended before other lifecycle events join the stream. + +This is a design decision, not an implementation. The unreachable polling replica is removed by FN-8683. Implementation is tracked by the linked follow-up tasks recorded in the FN-8683 task documents. + +## Problem and current boundary + +`TaskStore` lifecycle events are process-local `EventEmitter` events. The former SQLite polling replica in `packages/core/src/task-store/lifecycle-ops.ts` attempted to compare a cache with task rows and re-emit deletions. It cannot operate in the supported runtime: `dbImpl` in `packages/core/src/task-store/task-id-integrity.ts` always throws for `store.db`, with no backend-mode branch, and `watchImpl` returns after asynchronously warming `taskCache` for an injected `AsyncDataLayer`. + +Consequently, a delete in one PostgreSQL process is observed only by that writer's store. A dashboard, CLI, child process, or other node holding another `TaskStore` does not receive `task:deleted`, and must not be described as having cross-process lifecycle observation until the outbox exists. + +The mechanism must ultimately feed these existing bridge and listener surfaces: + +- `packages/engine/src/project-runtime.ts` event contract; +- `packages/engine/src/runtimes/in-process-runtime.ts` TaskStore-to-runtime forwarding; +- `packages/engine/src/runtimes/child-process-runtime.ts` IPC receipt and `child-process-worker.ts` IPC send; +- `packages/engine/src/runtimes/remote-node-runtime.ts` remote-node transport; +- `packages/engine/src/project-manager.ts` project-runtime forwarding; +- live consumers including the scheduler, executor, triage, project engine, and dashboard SSE endpoint. + +## Existing delete ownership + +The delete writer owns the transactional soft-delete (`deletedAt`) and its one `task:deleted` run-audit row. After commit it also owns the best-effort operator mailbox side effect in `task-delete-notice.ts`. An observing process must **not** invoke either writer-owned operation again. Observers only update their local cache and deliver an explicitly observed lifecycle notification to safe consumers. + +`deletedAt` remains the deletion identity: a repeated delete of an already soft-deleted task creates no new lifecycle event. An archived task is a separate archive flow; its delete event retains the pre-delete task snapshot and does not turn an archive move into a second delete. + +## Options considered + +### PostgreSQL LISTEN/NOTIFY + +- **Delivery:** at-most-once. Notifications are delivered only to connected listeners after transaction commit; payloads are not a durable queue. +- **Restart/drop behavior:** a disconnected or restarting consumer misses notifications. It can run a full reconciliation against `tasks`/`deletedAt`, but cannot recover an ordered, bounded event history from NOTIFY itself. +- **Ordering:** commit-order notification is separate from the synchronous in-process emitter. The writer can emit locally after commit, while remote delivery has no exactly-once or global listener order. +- **Soft deletes and side effects:** payloads can name a task and `deletedAt`, but subscribers must query to distinguish stale/duplicate data. A subscriber must not write mailbox notices or delete run-audit rows, or it double-fires writer-owned side effects. +- **Migration:** no table migration is required, but a dedicated connection lifecycle, channel authorization, payload-size handling, and reconciliation path are required. +- **Delete-write cost:** one `pg_notify` call in the delete transaction; small, but it provides no durable retry contract. + +Rejected: it improves promptness but cannot meet durable delivery across an engine restart or dropped connection without separately building a durable reconciliation mechanism. Adding that mechanism makes a durable outbox the clearer primary record. + +### Transactional outbox (recommended) + +- **Delivery:** at-least-once to each registered durable consumer; consumers make handling idempotent. The delete and outbox insert commit atomically, so a committed delete cannot be silently absent from the event stream. +- **Restart/drop behavior:** consumers resume from their stored cursors and scan missed rows. A lost live wake-up only delays the scan; it does not lose the event. +- **Ordering:** outbox sequence order is the authoritative cross-process order per project. It does not redefine synchronous in-process `EventEmitter` ordering; local writer listeners remain immediate, while remote consumers process committed sequence order. +- **Soft deletes and side effects:** one `task:deleted` event is inserted only for the first `deletedAt` transition. The observed-event dispatcher must carry `originProcessId`/writer identity and `observed: true`, and must forbid mailbox notice, delete run-audit, or delete mutation calls. +- **Migration:** yes. An outbox table, consumer-cursor table, and dead-letter/attempt state require a numbered migration explicitly registered in `packages/core/src/postgres/schema-applier.ts` (version constant plus bookkeeping check). +- **Delete-write cost:** one additional indexed insert in the existing delete transaction. This is bounded and fails atomically with the delete instead of creating a hidden loss window. + +Selected because its durable record supplies replay, ordered catch-up, operational visibility, and a bounded recovery path without relying on a continuously connected process. + +### Engine → dashboard bridge + +- **Delivery:** typically at-most-once unless the bridge develops persistence, acknowledgement, replay, and retry. IPC/SSE alone loses messages on engine/dashboard restart or reconnect. +- **Restart/drop behavior:** the dashboard can refetch board data, but an engine restart or independent CLI/remote-node consumer has no shared durable history. The bridge also cannot reach a dashboard process that is not connected to that engine. +- **Ordering:** can preserve a single engine's send order, but not globally across multiple engine nodes or direct API/CLI writers. +- **Soft deletes and side effects:** the bridge can forward the writer event, but receivers must still avoid repeating writer-owned notice/audit work. +- **Migration:** no migration for a simple bridge; a durable bridge would need storage equivalent to an outbox. +- **Delete-write cost:** low for send-only IPC, but it moves reliability cost into process supervision and reconnect handling. + +Rejected: it is useful as a transport adapter after durable observation exists, but is not the system-of-record for a multi-node PostgreSQL project and excludes non-engine writers. + +## Recommended outbox contract + +### Event shape and identity + +Within the delete transaction, insert a project-scoped outbox row such as: + +```text +sequence (monotonic per project), project_id, event_type = "task:deleted", +event_id = "task-deleted::", task_id, deleted_at, +writer_process_id, occurred_at, payload_version +``` + +The unique `(project_id, event_id)` key prevents concurrent delete attempts from producing two events for one `deletedAt` transition. The payload contains only the minimum post-commit observer data: task ID, `deletedAt`, prior column/status if consumers need them, event identity, writer identity, and schema version. It never contains mailbox prose, credentials, or free-text run-audit metadata. + +### Cursor, acknowledgement, and idempotent handling + +Create a project-scoped `lifecycle_consumer_cursors` row keyed by `(project_id, consumer_id)`, holding `last_acked_sequence`, lease/heartbeat metadata, and current failure state. Every independently durable receiver gets a stable consumer ID (for example dashboard instance role, engine runtime role, or remote-node bridge role); ephemeral SSE clients are fed from their local durable receiver and do not own cursors. + +A consumer reads rows with `sequence > last_acked_sequence` in ascending order. It updates local cache and invokes the **observed-event-safe** listener path, then advances the cursor in the same consumer transaction where possible. Handler idempotency is mandatory because a process can crash after side effects and before cursor acknowledgement: deduplicate on `(consumer_id, event_id)` or make every permitted observed handler a no-op when the task is already absent at the recorded `deletedAt`. + +This prevents duplicate delivery from becoming duplicate local work while preserving an at-least-once retry record. + +### Reconnect catch-up + +On startup, connection recovery, or a missed wake-up, a consumer scans from `last_acked_sequence + 1`. `LISTEN/NOTIFY` may later wake consumers for low latency, but it is only a hint; cursor scanning is authoritative. Retain normal outbox rows for **30 days** after all active durable consumers acknowledge them. A consumer whose cursor is older than the retained floor, missing, or invalid must run full reconciliation: refresh its live task cache from `tasks WHERE deleted_at IS NULL`, remove locally cached IDs absent from that result, and record its cursor at the current committed sequence before resuming. + +This prevents a dropped connection or restart from creating an unrecoverable observation gap; the full reconciliation is the bounded fallback once retained history is no longer available. + +### Retry and poison handling + +For a failed event, do not advance its cursor. Retry with exponential backoff (1s, 5s, 30s, 5m, then 15m) for at most **10 attempts**. Persist attempt count, next-attempt time, and last failure class (bounded code only) on a consumer-delivery row or dead-letter table. After attempt 10, atomically park the delivery as dead-letter, emit `lifecycle-observation:consumer-poisoned` run-audit metadata containing only project/consumer/event IDs and counts, and advance the main cursor only after recording that park. + +The consumer continues with later events; a poison event cannot wedge the stream or disappear silently. A repair tool replays a dead-letter event explicitly after the underlying defect is fixed and emits `lifecycle-observation:consumer-replayed`. + +### Retention and pruning ownership + +`SelfHealingManager` owns pruning during its scheduled/store-open lifecycle-maintenance sweep. It prunes an acknowledged normal outbox row only when it is older than 30 days **and** every non-retired durable consumer cursor is at or beyond its sequence. It prunes resolved dead letters after 30 days; unresolved dead letters are retained until an operator resolves or retires their consumer. Consumer retirement is an explicit audited operation, never inferred from a transient missed heartbeat. + +This prevents pruning events that an active consumer has not acknowledged and leaves poison evidence available for repair. + +## Rollout sketch + +1. Add schema migration and schema-applier registration for outbox, cursor/delivery, and dead-letter state; include fresh/upgrade/RLS tests. +2. Add a transactional delete-outbox writer beside the existing PostgreSQL soft-delete/audit transaction. Guard the unique event identity and prove idempotent re-delete writes nothing. +3. Implement consumer polling/cursor replay and an observed-event dispatcher. Keep it separate from the writer emitter so it cannot call `task-delete-notice.ts`, write the delete audit row, or mutate deletion state. +4. Adapt in-process, child-process, remote-node, project-manager, and dashboard bridges to consume observed events from the durable dispatcher. Preserve current payload compatibility for live listeners. +5. Add a NOTIFY wake-up only as an optimization after cursor catch-up is proven; it is never the delivery authority. +6. Enable per deployment, observe cursor lag/dead letters, and retain the old full-cache reconciliation fallback during rollout. The removed SQLite polling replica is not a rollback path because it cannot run in PostgreSQL mode. + +## Consequences + +Until the rollout ships, cross-process `task:deleted` observation is a known PostgreSQL gap. In-process deletes and their existing listeners remain supported and unchanged. Any feature that needs cross-process lifecycle correctness must depend on the outbox implementation rather than reintroducing `store.db` polling. diff --git a/docs/storage.md b/docs/storage.md index 47e951bffd..3edb6be649 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -44,6 +44,7 @@ See the [2026-07-14 PostgreSQL runtime cutover review](./postgres-migration-revi - Active task readers (`getTask`, `listTasks`, search, dependency scans, scheduler/watcher reads, mission task aggregations) must filter with `deletedAt IS NULL`. - Archived-task flows (`archiveTask`, archived cleanup/migration) hard-delete from the active `tasks` table after copying to PostgreSQL cold storage. Legacy `archive.db` files are import-only. - ID reservation is unchanged: soft-deleted IDs remain reserved. `distributed-task-id` and `task-id-integrity` intentionally scan all task rows (including soft-deleted rows), and must not filter on `deletedAt`. +- The legacy SQLite polling replica for cross-process task lifecycle observation no longer exists. In PostgreSQL mode `TaskStore.watch()` only warms its local cache; `task:deleted` is process-local until the transactional outbox defined in [PostgreSQL cross-process `task:deleted` observation](./solutions/architecture/postgres-cross-process-task-deleted-observation.md) is implemented. Observing processes must not recreate writer-owned delete run-audit or mailbox effects. ### Orphaned task-dir reconciliation (FN-6783) @@ -137,7 +138,7 @@ The PostgreSQL writer locks the active `(project_id, task_id)` parent row before - Task-linked artifact registration requires an active, non-archived task. Archived tasks are read-only for artifact writes; soft-deleted or missing tasks are rejected. - Retention follows the existing task lifecycle rather than a separate artifact policy: soft-deleted parent tasks keep artifact rows/files for forensics but normal live-reader APIs hide them; hard deletion from the active `tasks` table cascades artifact metadata through the `taskId` foreign key, and archive cleanup removes the task directory that contains task-scoped artifact binaries. Task-less artifacts live under `/.fusion/artifacts/` and are not tied to task archival cleanup. - Worktree DB hydration copies task-scoped artifact metadata for the current task/dependency graph alongside task rows and `task_documents`. It intentionally does not copy binary payload files, and it intentionally excludes task-less registry artifacts because dependency hydration is scoped to the active task graph. -- **Cross-instance live refresh (FN-7544).** A project can have more than one `TaskStore` instance open against the same DB at once (e.g. the dashboard's cached `getOrCreateProjectStore` instance vs. the engine's own internally-constructed store, or two dashboard processes). `registerArtifact()` calls `bumpLastModified()` and `checkForChanges()`'s 1s polling loop diffs the `artifacts` table by a strictly-increasing `rowid` cursor (not a timestamp, to avoid millisecond ties), re-emitting `artifact:registered` on any instance that did not perform the write itself. Without this, an already-open Documents/task Artifacts gallery served by a different store instance than the one an agent wrote through would never receive the live event and would show a stale list until a full reload re-ran the initial fetch. +- **Cross-instance live refresh boundary (FN-8683).** A project can have more than one `TaskStore` instance open against the same PostgreSQL DB, but TaskStore no longer has a SQLite polling replica to re-emit artifact or task lifecycle events in another process. A receiver refreshes through its normal API reads today; the planned durable outbox in [PostgreSQL cross-process `task:deleted` observation](./solutions/architecture/postgres-cross-process-task-deleted-observation.md) is the only approved path for future cross-process lifecycle delivery, not a revived `checkForChanges()` loop. Agent-facing registration tools are documented in [Artifact registry tools](./agents.md#artifact-registry-tools), and the dashboard browsing surface is documented in [Artifacts View](./dashboard-guide.md#artifacts-view). diff --git a/packages/core/src/__tests__/task-delete-nonblocking-cleanup.test.ts b/packages/core/src/__tests__/task-delete-nonblocking-cleanup.test.ts index 2687c4431c..34c095a10c 100644 --- a/packages/core/src/__tests__/task-delete-nonblocking-cleanup.test.ts +++ b/packages/core/src/__tests__/task-delete-nonblocking-cleanup.test.ts @@ -45,6 +45,7 @@ vi.mock("../async-mission-store-queries.js", () => ({ })); import { deleteTaskImpl } from "../task-store/archive-lifecycle.js"; +import { setupActivityLogListenersImpl } from "../task-store/lifecycle-ops.js"; import { deleteTaskBackendImpl } from "../task-store/archive-lifecycle-2.js"; function createTask(overrides: Partial & { id: string }): Task { @@ -80,6 +81,7 @@ function makeDeleteStore(task: Task, children: string[] = []) { backendMode: true, isWatching: true, taskCache: new Map([[task.id, task]]), + laneCache: { invalidate: vi.fn() }, asyncLayer: { db: {}, projectId: "project-1", @@ -109,6 +111,58 @@ function makeDeleteStore(task: Task, children: string[] = []) { } describe("deleteTask gates that survived the PostgreSQL cutover", () => { + it("keeps in-process task:deleted emissions and activity rows exact for each delete data state", async () => { + const activityRows: Array<{ type: string; taskId?: string }> = []; + const deletedEvents: string[] = []; + const liveStore = makeDeleteStore(createTask({ id: "FN-LIVE" })); + liveStore.activityListenersWired = false; + liveStore.recordActivityFromListener = vi.fn((entry: { type: string; taskId?: string }) => activityRows.push(entry)); + setupActivityLogListenersImpl(liveStore as never); + liveStore.on("task:deleted", (task: Task) => deletedEvents.push(task.id)); + + await deleteTaskImpl(liveStore as never, "FN-LIVE"); + expect(deletedEvents).toEqual(["FN-LIVE"]); + expect(activityRows.filter((entry) => entry.type === "task:deleted")).toEqual([{ type: "task:deleted", taskId: "FN-LIVE", taskTitle: "FN-LIVE", details: "Task FN-LIVE deleted: FN-LIVE" }]); + + const archivedRows: Array<{ type: string; taskId?: string }> = []; + const archivedEvents: string[] = []; + const archivedStore = makeDeleteStore(createTask({ id: "FN-ARCHIVED", column: "archived" })); + archivedStore.activityListenersWired = false; + archivedStore.recordActivityFromListener = vi.fn((entry: { type: string; taskId?: string }) => archivedRows.push(entry)); + setupActivityLogListenersImpl(archivedStore as never); + archivedStore.on("task:deleted", (task: Task) => archivedEvents.push(task.id)); + + await deleteTaskImpl(archivedStore as never, "FN-ARCHIVED"); + expect(archivedEvents).toEqual(["FN-ARCHIVED"]); + expect(archivedRows.filter((entry) => entry.type === "task:deleted")).toHaveLength(1); + + const alreadyDeletedRows: Array<{ type: string }> = []; + const alreadyDeletedEvents: string[] = []; + const alreadyDeletedStore = makeDeleteStore(createTask({ id: "FN-ALREADY-DELETED", deletedAt: "2026-07-15T09:01:00.000Z", column: "archived" })); + alreadyDeletedStore.activityListenersWired = false; + alreadyDeletedStore.recordActivityFromListener = vi.fn((entry: { type: string }) => alreadyDeletedRows.push(entry)); + setupActivityLogListenersImpl(alreadyDeletedStore as never); + alreadyDeletedStore.on("task:deleted", (task: Task) => alreadyDeletedEvents.push(task.id)); + + await deleteTaskImpl(alreadyDeletedStore as never, "FN-ALREADY-DELETED"); + expect(alreadyDeletedEvents).toEqual([]); + expect(alreadyDeletedRows.filter((entry) => entry.type === "task:deleted")).toEqual([]); + + const unknownRows: Array<{ type: string }> = []; + const unknownEvents: string[] = []; + const unknownStore = makeDeleteStore(createTask({ id: "FN-UNKNOWN-SEED" })); + pgRow = null; + unknownStore.activityListenersWired = false; + unknownStore.recordActivityFromListener = vi.fn((entry: { type: string }) => unknownRows.push(entry)); + setupActivityLogListenersImpl(unknownStore as never); + unknownStore.on("task:deleted", (task: Task) => unknownEvents.push(task.id)); + + await expect(deleteTaskImpl(unknownStore as never, "FN-UNKNOWN")).rejects.toMatchObject({ name: "TaskNotFoundError", taskId: "FN-UNKNOWN" }); + expect(unknownEvents).toEqual([]); + expect(unknownRows.filter((entry) => entry.type === "task:deleted")).toEqual([]); + }); + + /* FNXC:TaskDeletion 2026-07-30-20:15 (PR #2697 review — greptile): The module mock is shared across this file and the config clears nothing, so a call-count diff --git a/packages/core/src/__tests__/task-deleted-polling-replica-tombstone.test.ts b/packages/core/src/__tests__/task-deleted-polling-replica-tombstone.test.ts new file mode 100644 index 0000000000..ec2ac9c18a --- /dev/null +++ b/packages/core/src/__tests__/task-deleted-polling-replica-tombstone.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; +import { readFileSync } from "node:fs"; +import { join, resolve } from "node:path"; + +const REPO_ROOT = resolve(import.meta.dirname, "../../../.."); + +const SOURCES = { + lifecycle: "packages/core/src/task-store/lifecycle-ops.ts", + store: "packages/core/src/store.ts", +} as const; + +/** The old replica is documented in comments; executable source must not restore it. */ +function stripComments(source: string): string { + return source + .replace(/\/\*[\s\S]*?\*\//g, " ") + .replace(/(^|[^:])\/\/[^\n]*/g, "$1 "); +} + +function source(key: keyof typeof SOURCES): string { + return stripComments(readFileSync(join(REPO_ROOT, SOURCES[key]), "utf8")); +} + +/* +FNXC:TaskDeletedObservation 2026-08-01-09:59: +The SQLite polling replica could never run for the supported PostgreSQL AsyncDataLayer: its first +`store.db` read always threw. FN-8683 deliberately removes every replica-only TaskStore symbol, +rather than leaving a fake cross-process lifecycle capability. The documented transactional-outbox +replacement is owned by FN-8684/FN-8685; comments may name this history, so assertions strip them. +*/ +const REMOVED_SYMBOLS: ReadonlyArray<{ symbol: string; source: keyof typeof SOURCES; why: string }> = [ + { symbol: "checkForChangesImpl", source: "lifecycle", why: "the SQLite polling implementation cannot access PostgreSQL through store.db" }, + { symbol: "checkForChanges", source: "store", why: "the public façade exposed only the unreachable TaskStore polling path" }, + { symbol: "pollInterval", source: "store", why: "TaskStore no longer schedules a polling replica" }, + { symbol: "pollingInProgress", source: "store", why: "the re-entrancy guard belonged only to the removed replica" }, + { symbol: "lastKnownModified", source: "store", why: "the SQLite modified-stamp cursor belonged only to the removed replica" }, + { symbol: "lastPollTime", source: "store", why: "the SQLite changed-row cursor belonged only to the removed replica" }, + { symbol: "suppressActivityLogForPollingEmit", source: "store", why: "only replica re-emits needed to suppress duplicate activity rows" }, +]; + +describe("task:deleted SQLite polling replica tombstone", () => { + it("keeps every classified replica-only symbol out of executable source", () => { + const violations = REMOVED_SYMBOLS + .filter(({ symbol, source: sourceKey }) => new RegExp(`\\b${symbol}\\b`).test(source(sourceKey))) + .map(({ symbol, why }) => `${symbol} was restored — ${why}`); + + expect(violations).toEqual([]); + }); + + it("does not reintroduce synchronous SQLite access into the watch path", () => { + const lifecycle = source("lifecycle"); + const start = lifecycle.indexOf("export async function watchImpl("); + const watchBody = lifecycle.slice(start, lifecycle.indexOf("export async function migrateAgentLogEntriesImpl(", start)); + + expect(start).toBeGreaterThan(-1); + expect(watchBody).not.toContain("store.db."); + }); +}); diff --git a/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts b/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts index 138f34ce42..0865e66c93 100644 --- a/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts +++ b/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts @@ -22,7 +22,6 @@ const PRODUCERS = { "packages/core/src/task-store/audit-ops.ts": ["emit", "safe"], "packages/core/src/task-store/branch-group-ops.ts": ["emit"], "packages/core/src/task-store/comments-ops.ts": ["emit"], - "packages/core/src/task-store/lifecycle-ops.ts": ["emit"], "packages/core/src/task-store/merge-queue-ops.ts": ["emit"], "packages/core/src/task-store/moves.ts": ["emit"], "packages/core/src/task-store/project-store-ops.ts": ["emit"], @@ -198,31 +197,6 @@ pgDescribe("task:updated producer integration", () => { } }); - it("executes lifecycle polling's real update producer", async () => { - const store = new TaskStore(process.cwd()); - const polledTask = { ...task }; - const received: Array<{ lanes?: { wip?: string } } | undefined> = []; - store.on("task:updated", (_task, meta) => received.push(meta)); - Object.defineProperty(store, "db", { - value: { - getLastModified: vi.fn().mockReturnValueOnce(1).mockReturnValueOnce(2), - prepare: vi.fn((query: string) => query.includes("SELECT id FROM tasks") - ? { all: () => [{ id: task.id }] } - : { all: () => [{}] }), - }, - }); - vi.spyOn(store, "rowToTask").mockReturnValue(polledTask); - store.taskCache.set(task.id, { ...task }); - - store.laneCache.set(task.id, { wip: "building" }); - await store.checkForChanges(); - store.laneCache.invalidate(task.id); - store.taskCache.set(task.id, { ...task }); - await store.checkForChanges(); - - expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]); - }); - it("executes workflow-integrity's real safe update producer", async () => { const store = new TaskStore(process.cwd()); const stampableTask = { ...task, column: "in-review", autoMerge: true }; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 2f139a4070..a68ac701b6 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -118,7 +118,7 @@ import { clearWorkflowRunBranchesImpl, projectMergeRequestToWorkflowWorkItemImpl import { flushAgentLogBufferImpl, appendAgentLogBatchImpl } from "./task-store/agent-logs.js"; import { refineTaskImpl, updateTaskDependenciesImpl } from "./task-store/update-task-deps.js"; import { createWorkflowStepImpl, updateWorkflowStepImpl, updateWorkflowDefinitionImpl, deleteWorkflowDefinitionImpl, setDefaultWorkflowIdImpl, selectTaskWorkflowImpl } from "./task-store/workflow-ops.js"; -import { initImpl, setupActivityLogListenersImpl, reconcileOrphanedTaskDirsImpl, watchImpl, checkForChangesImpl, migrateAgentLogEntriesImpl, migrateMovedSettingsImpl, recoverStaleTransitionPendingImpl, migrateLegacyWorkflowStepsImpl, emitTaskLifecycleEventSafelyImpl } from "./task-store/lifecycle-ops.js"; +import { initImpl, setupActivityLogListenersImpl, reconcileOrphanedTaskDirsImpl, watchImpl, migrateAgentLogEntriesImpl, migrateMovedSettingsImpl, recoverStaleTransitionPendingImpl, migrateLegacyWorkflowStepsImpl, emitTaskLifecycleEventSafelyImpl } from "./task-store/lifecycle-ops.js"; import { updateStepImpl, startStepImpl, acquireMergeQueueLeaseImpl, mergeTaskImpl } from "./task-store/merge-queue-ops.js"; import { addCommentImpl, publishArchivedTaskDocumentAdditionImpl, upsertTaskDocumentImpl } from "./task-store/comments-ops.js"; import { deleteTaskImpl, archiveTaskImpl, type DeleteTaskIfResult } from "./task-store/archive-lifecycle.js"; @@ -383,8 +383,6 @@ export class TaskStore extends EventEmitter { public configPath: string; public _db: Database | null = null; public activityListenersWired = false; - /** When true, activity-log listeners skip recording (set by checkForChanges polling so re-emitted events don't double-log). In-process emit path remains sole source of truth. */ - public suppressActivityLogForPollingEmit = false; public _archiveDb: ArchiveDatabase | null = null; /** @@ -458,16 +456,12 @@ export class TaskStore extends EventEmitter { public workflowDefinitionsCache: WorkflowDefinition[] | null = null; public _pluginWorkflowStepTemplates: Array<{ pluginId: string; template: WorkflowStepTemplate }> = []; public globalSettingsStore: GlobalSettingsStore; - public pollInterval: ReturnType | null = null; - public pollingInProgress = false; - public lastKnownModified: number = 0; - public lastPollTime: string | null = null; public donePauseBackfillDone = false; public startupSlimListMemo = new Map }>(); public static readonly STARTUP_SLIM_LIST_MEMO_TTL_MS = 2_500; public get isWatching(): boolean { - return this.watcher !== null || this.pollInterval !== null; + return this.watcher !== null; } public missionStore: MissionStore | AsyncMissionStore | null = null; public ideationStore: AsyncIdeationStore | null = null; @@ -2306,9 +2300,6 @@ export class TaskStore extends EventEmitter { async watch(): Promise { return watchImpl(this); } - public async checkForChanges(): Promise { - return checkForChangesImpl(this); - } stopWatching(): void { return stopWatchingImpl(this); } diff --git a/packages/core/src/task-store/lifecycle-ops.ts b/packages/core/src/task-store/lifecycle-ops.ts index 2311a8931a..ac265b279b 100644 --- a/packages/core/src/task-store/lifecycle-ops.ts +++ b/packages/core/src/task-store/lifecycle-ops.ts @@ -17,7 +17,7 @@ import { import {mkdir, readdir, readFile, stat} from "node:fs/promises"; import {join} from "node:path"; import {existsSync, type Dirent} from "node:fs"; -import type {Task, AgentLogEntry, Column, GlobalSettings} from "../types.js"; +import type {Task, AgentLogEntry, GlobalSettings} from "../types.js"; import {MOVED_SETTINGS_KEYS, SETTINGS_MIGRATION_VERSION, SETTINGS_MIGRATION_MARKER_KEY} from "../moved-settings.js"; import {stepsToWorkflowIr, stepToFragmentIr, layoutForIr} from "../workflow-steps-to-ir.js"; import {getTraitRegistry} from "../trait-registry.js"; @@ -29,7 +29,6 @@ import {validateSettingValuePatch} from "../workflow-settings.js"; import "../builtin-traits.js"; import {appendAgentLogEntriesSync} from "../agent-log-file-store.js"; import {getErrorMessage} from "../error-message.js"; -import {type TaskRow} from "../task-store/persistence.js"; import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; import {reconcileTaskIdStateAsync} from "../task-store/async-allocator.js"; import {ACTIVE_TASK_FILTER, insertTaskRowInTransaction, isTaskIdConflictError as isPgTaskIdConflictError, readTaskRow} from "./async-persistence.js"; @@ -277,7 +276,6 @@ export function setupActivityLogListenersImpl(store: TaskStore): void { // Task created store.on("task:created", (task) => { - if (store.suppressActivityLogForPollingEmit) return; store.recordActivityFromListener( { type: "task:created", @@ -291,7 +289,6 @@ export function setupActivityLogListenersImpl(store: TaskStore): void { // Task moved store.on("task:moved", (data) => { - if (store.suppressActivityLogForPollingEmit) return; if (data.from === data.to) return; store.recordActivityFromListener( { @@ -322,7 +319,6 @@ export function setupActivityLogListenersImpl(store: TaskStore): void { // Task updated (check for failures) store.on("task:updated", (task) => { - if (store.suppressActivityLogForPollingEmit) return; if (task.status === "failed") { store.recordActivityFromListener( { @@ -367,7 +363,6 @@ export function setupActivityLogListenersImpl(store: TaskStore): void { // Task deleted store.on("task:deleted", (task) => { - if (store.suppressActivityLogForPollingEmit) return; store.recordActivityFromListener( { type: "task:deleted", @@ -592,168 +587,25 @@ export async function reconcileOrphanedTaskDirsImpl(store: TaskStore, opts: { ig } export async function watchImpl(store: TaskStore): Promise { - if (store.watcher || store.pollInterval) return; // already watching + if (store.watcher) return; // already watching store.clearStartupSlimListMemo(); /* - * FNXC:BackendFlip 2026-06-26-16:00: - * In backend mode (PostgreSQL), the entire watch() body below is - * SQLite-specific: it reads store.db.getLastModified(), sets up an fs.watch - * sentinel + a 1s polling interval whose checkForChanges() cycle queries - * store.db.prepare('SELECT ... FROM tasks'), and runs SQLite-only stamp - * markers. All of those throw "SQLite Database is not available in backend - * mode" because store.db is not constructed when an AsyncDataLayer is - * injected. - * - * The async backend does not rely on this SQLite polling loop for change - * detection — runtime mutations go through the async layer and emit their - * own events. Populate the in-memory task cache (so the HTTP layer has a - * snapshot) via the backend-aware listTasks(), then return without - * installing the SQLite watcher/poller. This keeps `fn serve` / boot smoke - * booting against embedded PG. + * FNXC:TaskDeletedObservation 2026-08-01-09:59: + * PostgreSQL has no cross-process `task:deleted` observer today. FN-8683 removed the + * unreachable SQLite poller because `store.db` always throws for AsyncDataLayer stores; + * cache warming below is the only supported watch behavior. The transactional-outbox design is + * documented in docs/solutions/architecture/postgres-cross-process-task-deleted-observation.md. + * FN-8684 owns the transactional writer and FN-8685 owns durable consumer delivery; do not + * reintroduce a local polling replica as a substitute for their cursor/replay contract. */ - const tasks = await store.listTasks({ slim: true, startupMemo: false }); + const tasks = await store.listTasks({ slim: true, startupMemo: false }); store.taskCache.clear(); for (const task of tasks) { store.taskCache.set(task.id, { ...task }); } - return; } -export async function checkForChangesImpl(store: TaskStore): Promise { - const startTime = Date.now(); - - // Guard against overlapping poll cycles - if (store.pollingInProgress) return; - store.pollingInProgress = true; - - try { - const currentModified = store.db.getLastModified(); - if (currentModified <= store.lastKnownModified) return; - store.lastKnownModified = currentModified; - - // Detect deletions cheaply: compare ID sets without loading full rows. - // A row missing from `tasks` can mean two things: the task was actually - // deleted, OR it was archived (archiveTask removes it from `tasks` after - // copying into `archived_tasks`). Other TaskStore instances polling the - // same DB can't tell the difference from this view alone — without the - // archive check below they emit spurious task:deleted events for every - // archived task, which the activity log records as a deletion. - // FN-5105: intentionally include soft-deleted rows here so a deletedAt - // transition can be observed and emit task:deleted exactly once. - const idRows = store.db.prepare('SELECT id FROM tasks').all() as Array<{ id: string }>; - const currentIds = new Set(idRows.map((r) => r.id)); - const missingIds: string[] = []; - for (const id of store.taskCache.keys()) { - if (!currentIds.has(id)) missingIds.push(id); - } - if (missingIds.length > 0) { - const archivedSet = store.archiveDb.filterArchived(missingIds); - for (const id of missingIds) { - const cached = store.taskCache.get(id); - if (!cached) continue; - store.taskCache.delete(id); - store.suppressActivityLogForPollingEmit = true; - try { - if (archivedSet.has(id)) { - // Task moved to archive — emit task:moved (matching what - // archiveTask emits in-process) so other subscribers can react. - // Skip already-archived cache entries to avoid no-op emits. - // Activity-log listeners skip polling emits; the originating - // TaskStore instance wrote the row in-process. - /* - FNXC:WorkflowResolvedColumns 2026-07-31-20:15 (audited — DEAD SYNC PATH, do not convert): - Both the guard and the `to: "archived"` it emits are literals, so on a renamed board a - polling replica would emit a move to a column the board does not declare. It cannot: - `checkForChangesImpl` opens with `store.db.getLastModified()` and `store.db.prepare`, - which throw in PostgreSQL backend mode, so this whole polling replica path is legacy - SQLite only. - - DELIBERATE-LITERAL — recorded rather than converted, for the same reason as `mission-store.ts` and - `project-store-ops.ts`: an unconverted literal in dead code is not debt a fleet pass - should spend a signature change on, but it must not read as missed either. - */ - if (cached.column !== "archived") { - store.emit("task:moved", { task: cached, from: cached.column, to: "archived" as Column, source: "engine" }); - } - } else { - // Polling replicas only mirror the originating delete signal. - // Do not record run-audit here; the writer already owns that row. - store.emit("task:deleted", cached); - } - } finally { - store.suppressActivityLogForPollingEmit = false; - } - } - } - - // Yield to event loop before the expensive SELECT query - await new Promise((resolve) => setImmediate(resolve)); - - // Only load tasks modified since our last known timestamp. - // Use lastKnownPollTime (ISO string) to filter — much cheaper than full scan. - const selectClause = store.getTaskSelectClause(true); - const changedRows = store.lastPollTime - ? store.db.prepare(`SELECT ${selectClause} FROM tasks WHERE updatedAt > ? OR columnMovedAt > ?`).all(store.lastPollTime, store.lastPollTime) as unknown as TaskRow[] - : store.db.prepare(`SELECT ${selectClause} FROM tasks`).all() as unknown as TaskRow[]; - store.lastPollTime = new Date().toISOString(); - - for (let i = 0; i < changedRows.length; i++) { - const row = changedRows[i]; - const task = store.rowToTask(row); - const cached = store.taskCache.get(task.id); - - store.suppressActivityLogForPollingEmit = true; - try { - if (task.deletedAt) { - if (cached) { - store.taskCache.delete(task.id); - // Polling replicas only re-emit task:deleted for subscribers. - // They must not insert duplicate run-audit rows cross-instance. - store.emit("task:deleted", cached); - } - continue; - } - - if (!cached) { - store.taskCache.set(task.id, { ...task }); - store.emit("task:created", task); - } else if (cached.column !== task.column) { - const from = cached.column; - store.taskCache.set(task.id, { ...task }); - store.emit("task:moved", { task, from, to: task.column, source: "engine" }); - } else { - store.taskCache.set(task.id, { ...task }); - store.emit("task:updated", task); - } - } finally { - store.suppressActivityLogForPollingEmit = false; - } - - // Yield every ~50 rows to prevent blocking the event loop during large updates - if (i > 0 && i % 50 === 0) { - await new Promise((resolve) => setImmediate(resolve)); - } - } - - const elapsed = Date.now() - startTime; - if (elapsed > 750) { - storeLog.warn("checkForChanges took longer than expected", { - elapsedMs: elapsed, - thresholdMs: 750, - }); - } - } catch (err) { - storeLog.warn("checkForChanges poll cycle failed", { - lastKnownModified: store.lastKnownModified, - lastPollTime: store.lastPollTime, - error: err instanceof Error ? err.message : String(err), - }); - } finally { - store.pollingInProgress = false; - } - } - export async function migrateAgentLogEntriesImpl(store: TaskStore): Promise { const migrationKey = "agentLogEntriesToFileMigrationVersion"; const migrationVersion = "1"; diff --git a/packages/core/src/task-store/task-artifacts-ops.ts b/packages/core/src/task-store/task-artifacts-ops.ts index e3caf50f45..bfb7cdb706 100644 --- a/packages/core/src/task-store/task-artifacts-ops.ts +++ b/packages/core/src/task-store/task-artifacts-ops.ts @@ -612,10 +612,6 @@ export function stopWatchingImpl(store: TaskStore): void { store.watcher.close(); store.watcher = null; } - if (store.pollInterval) { - clearInterval(store.pollInterval); - store.pollInterval = null; - } for (const timer of store.debounceTimers.values()) { clearTimeout(timer); } diff --git a/packages/engine/src/__tests__/executor-soft-delete-abort.test.ts b/packages/engine/src/__tests__/executor-soft-delete-abort.test.ts index ad334c3cdf..d2497fe8c4 100644 --- a/packages/engine/src/__tests__/executor-soft-delete-abort.test.ts +++ b/packages/engine/src/__tests__/executor-soft-delete-abort.test.ts @@ -124,6 +124,21 @@ describe("TaskExecutor soft-delete aborts", () => { expect((executor as any).activeSubagentSessions.has("FN-TEST-4")).toBe(false); }); + it("receives the unchanged TaskStore delete payload when metadata is present", async () => { + const { store, emit } = createEventedStore(); + const executor = new TaskExecutor(store, "/tmp/test"); + const abort = vi.fn().mockResolvedValue(undefined); + + (executor as any).activeSessions.set("FN-TEST-META", { + session: { abort, dispose: vi.fn() }, + seenSteeringIds: new Set(), + }); + emit("task:deleted", { ...makeTask("FN-TEST-META"), deletedAt: "2026-08-01T09:59:00.000Z" }, { githubIssueAction: "auto" }); + await (executor as any).pendingTaskDisposals.get("FN-TEST-META"); + + expect(abort).toHaveBeenCalledTimes(1); + }); + it("is a silent no-op when the deleted task has no active surfaces", () => { const { store, emit } = createEventedStore(); const executor = new TaskExecutor(store, "/tmp/test"); diff --git a/packages/engine/src/__tests__/triage-soft-delete-abort.test.ts b/packages/engine/src/__tests__/triage-soft-delete-abort.test.ts index b7f54dccb7..b5ef4521e2 100644 --- a/packages/engine/src/__tests__/triage-soft-delete-abort.test.ts +++ b/packages/engine/src/__tests__/triage-soft-delete-abort.test.ts @@ -73,6 +73,20 @@ describe("TriageProcessor soft-delete aborts", () => { processor.stop(); }); + it("receives the unchanged TaskStore delete payload when metadata is present", async () => { + const { store, emit } = createEventedStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + const abort = vi.fn().mockResolvedValue(undefined); + + processor.start(); + (processor as any).activeSessions.set("FN-TEST-META", { abort, dispose: vi.fn() }); + emit("task:deleted", { id: "FN-TEST-META", deletedAt: "2026-08-01T09:59:00.000Z" }, { githubIssueAction: "auto" }); + await Promise.resolve(); + + expect(abort).toHaveBeenCalledTimes(1); + processor.stop(); + }); + it("is a no-op for unknown soft-deleted ids", () => { const { store, emit } = createEventedStore(); const processor = new TriageProcessor(store, "/tmp/root");