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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
@@ -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 `<rebaseBaseSha>..HEAD` reports zero own commits, merger must also validate the source `fusion/<id>` 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.
|
||||
|
||||
@@ -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:<taskId>:<deletedAt>", 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.
|
||||
@@ -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 `<rootDir>/.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).
|
||||
|
||||
|
||||
@@ -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<Task> & { id: string }): Task {
|
||||
@@ -80,6 +81,7 @@ function makeDeleteStore(task: Task, children: string[] = []) {
|
||||
backendMode: true,
|
||||
isWatching: true,
|
||||
taskCache: new Map<string, Task>([[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
|
||||
|
||||
@@ -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.");
|
||||
});
|
||||
});
|
||||
@@ -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 };
|
||||
|
||||
@@ -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<TaskStoreEvents> {
|
||||
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<TaskStoreEvents> {
|
||||
public workflowDefinitionsCache: WorkflowDefinition[] | null = null;
|
||||
public _pluginWorkflowStepTemplates: Array<{ pluginId: string; template: WorkflowStepTemplate }> = [];
|
||||
public globalSettingsStore: GlobalSettingsStore;
|
||||
public pollInterval: ReturnType<typeof setInterval> | null = null;
|
||||
public pollingInProgress = false;
|
||||
public lastKnownModified: number = 0;
|
||||
public lastPollTime: string | null = null;
|
||||
public donePauseBackfillDone = false;
|
||||
public startupSlimListMemo = new Map<string, { expiresAt: number; promise: Promise<Task[]> }>();
|
||||
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<TaskStoreEvents> {
|
||||
async watch(): Promise<void> {
|
||||
return watchImpl(this);
|
||||
}
|
||||
public async checkForChanges(): Promise<void> {
|
||||
return checkForChangesImpl(this);
|
||||
}
|
||||
stopWatching(): void {
|
||||
return stopWatchingImpl(this);
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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<void>((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<void>((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<void> {
|
||||
const migrationKey = "agentLogEntriesToFileMigrationVersion";
|
||||
const migrationVersion = "1";
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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<string>(),
|
||||
});
|
||||
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");
|
||||
|
||||
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user