# TASK 5: Add Import Orchestration, Durable Recovery, Diagnostics, And Cleanup

## Summary

Compose reservation, immutable snapshot, Redshift replacement, and Athena publication into one resumable import workflow driven by durable batch details.

## Dependencies

- [TASK 1](./TASK1.md)
- [TASK 2](./TASK2.md)
- [TASK 3](./TASK3.md)
- [TASK 4](./TASK4.md)

## Files To Create

Recommended locations:

- `backend/Kds.Ai.Integrations/FleetS3/IFleetS3ImportOrchestrator.cs`
- `backend/Kds.Ai.Integrations/FleetS3/FleetS3ImportOrchestrator.cs`
- `backend/Kds.Ai.Integrations/FleetS3/IFleetS3WorkspaceCleanupService.cs`
- `backend/Kds.Ai.Integrations/FleetS3/FleetS3WorkspaceCleanupService.cs`
- bounded failure/diagnostic helpers under the Fleet namespace.

## Execution State Machine

The orchestrator receives a previously reserved `batchId`. It never creates an unscoped replacement batch inside the worker.

Normal flow:

```text
reserved
  -> snapshotting
  -> snapshot_ready
  -> loading_redshift
  -> validating
  -> replacing_redshift
  -> replacement_intent_recorded
  -> redshift_committed
  -> publishing_athena
  -> completed / Succeeded
```

Persist phase details at every durable boundary, especially:

- after immutable manifest index creation;
- immediately before Redshift work;
- after old/new/affected dates and expected counts are durably recorded but before Redshift permanent deletes;
- immediately after Redshift commit/reconciliation;
- after each Athena activation journal update;
- at final completion/failure.

Use generic repository status/details operations, not the business-date stage lifecycle service.

## Resume Matrix

| Durable state | Resume behavior |
|---|---|
| `reserved` or `snapshotting` | rebuild/validate the same batch workspace, then continue |
| `snapshot_ready` | reuse validated manifest index; do not relist customer source |
| `loading_redshift`, `validating`, `replacing_redshift`, `replacement_intent_recorded` | acquire the worker lease, run Task 3 reconciliation, and continue or safely restart pre-commit work |
| `redshift_committed` or `failed_publication` | verify current Redshift slice batch lineage, then publish only |
| `publishing_athena` | resume incomplete publication journal entries |
| `completed` | return idempotent already-completed result |
| `superseded` | never publish or mutate; return non-retryable result |
| `publication_expired` | artifacts are no longer retryable; require a new full override import |

Never relist the mutable customer folder after `snapshot_ready` for the same batch.

## Failure Classification

Classify and persist bounded errors:

- request/identity configuration;
- source inventory/snapshot mutation;
- source contract validation;
- Redshift load/validation/replacement;
- Redshift commit reconciliation;
- Athena staging/activation;
- superseded by newer import;
- cleanup.

Rules:

- pre-commit failure: mark operational batch `Failed`, retain old Redshift/Athena data;
- post-commit publication failure: mark `Failed`, retain snapshot/staging, set retryable publication state;
- cleanup failure after successful publication: keep business batch success and record/schedule cleanup retry separately; do not republish data;
- cancellation follows the same phase-aware rules and must not erase diagnostic state;
- never store exception stack traces or payloads unbounded in PostgreSQL.
- a handled `failed_pre_commit` batch is not resumed; the operator starts a new import batch;
- duplicate or interrupted delivery for a still-`Running` batch must acquire/renew Task 1's worker lease before continuing;
- automatic delivery for a terminal batch performs no data work.

## Publication Resume Safety

Before a failed publication batch is resumed:

1. reacquire Task 1's tenant/source reservation;
2. resolve the same active tenant warehouse;
3. verify the complete current Fleet slice is stamped with the failed batch ID;
4. verify snapshot/index and publication journal still exist;
5. continue Task 4 without invoking Task 3 replacement.

If the current Fleet slice belongs to a newer batch, persist `superseded`, release the reservation, and do not touch Athena.

## Diagnostics

Final details must report:

- source objects/bytes and skipped non-order object count;
- manifest index and snapshot workspace URIs;
- min/max/populated business dates;
- raw/order/item candidate counts;
- bounded validation issue counts/samples;
- old/new/affected date counts;
- Redshift deleted/inserted counts and commit/reconciliation state;
- publication expected/exported/activated/removed counts by entity;
- publication retry count and outstanding journal count;
- phase timings and final status.

Structured logs use batch/customer/tenant/source folder/counts only. Do not log customer names, notes, modifiers, raw JSON, credentials, auth clauses, or full SQL.

## Cleanup

- retain immutable snapshots and staged publication artifacts until the configured diagnostic/retry retention expires;
- cleanup operates only on the exact batch workspace/staging prefixes returned by storage services;
- never derive a delete prefix from unvalidated caller input;
- enqueue or schedule a TickerQ cleanup function rather than blocking import completion until retention elapses;
- a failed publication workspace remains retryable only until `publicationRetryExpiresAt`;
- after expiry, atomically mark the batch `publication_expired`, alert operators that Athena may be incomplete, and clean the workspace after retention;
- recovery after publication expiry is a new full override import, whose old/new affected-date intent repairs/removes the current source publication;
- cleanup is idempotent and records completion.

## Testing

Add orchestration tests for every state transition and restart point:

- success from reserved state;
- snapshot retry without duplicate permanent effects;
- validation failure before mutation;
- crash immediately before/after Redshift commit;
- publication failure and publication-only retry;
- partial activation journal resume;
- newer import superseding an old failed publisher;
- duplicate TickerQ delivery of the same batch;
- concurrent delivery lease exclusion and expired-lease takeover;
- cancellation in every major phase;
- cleanup scheduling, retention, and exact-prefix deletion;
- publication retry expiry followed by a repairing full import;
- bounded details/logs with no payload data.

## Acceptance Criteria

- the workflow is deterministic from durable batch state;
- post-commit retry never repeats the Redshift replacement;
- duplicate worker delivery cannot create duplicate business data or active Parquet generations;
- all failures preserve the strongest valid prior state and expose an actionable bounded category;
- cleanup cannot delete customer source or another batch/source publication.