feat(storage): add configurable data retention policy

[yasha-dev1]

Summary

• Adds data_retention_days to PyWorkflowConfig (env: PYWORKFLOW_DATA_RETENTION_DAYS, YAML: retention_days); None = keep forever
• New abstract StorageBackend.delete_old_runs(older_than) method implemented across all 7 backends (postgres, sqlite, mysql, memory, file, dynamodb, cassandra); citus inherits from postgres unchanged
• Daily Celery Beat task pyworkflow.run_data_retention — singleton, self-skips when unconfigured, release_lock_on_failure=True to prevent stale locks blocking the next day's run
• 13 unit tests covering deletion semantics, related-data cleanup (events, steps, hooks, cancellation flags), and config loading

Retention scope

Deletes runs in terminal states (completed, failed, cancelled, continued_as_new, interrupted) where updated_at < now - data_retention_days. Active states (running, suspended, pending) are never deleted. Related rows (events, steps, hooks, cancellation flags) are deleted in the same transaction/operation.

Risk tier

Tier 2 — new feature touching storage layer and Celery task registration; no changes to existing execution paths or replay logic.

Test plan

• pytest tests/unit/test_retention.py — 13/13 passing
• pytest tests/unit/ — 769/769 passing (no regressions)
• ruff check — clean
• mypy pyworkflow/ — no new errors introduced

🤖 Generated with Claude Code

[github-actions]

🔄 New commits pushed — requesting re-review.

Commit: 70d2b007f738
sha:70d2b007f7381086ee70125b2d20edc774c31cd2

[github-actions]

🔍 Code Review Agent — Tier 3

Commit: 70d2b007f738
Status: ✅ Complete
Verdict: 🔄 REQUEST_CHANGES — Three blocking bugs found: Cassandra index cleanup missing (data consistency), file backend cancellation flags not deleted (contract violation), and file backend token index not updated (resource leak + invariant violation).

I have now read all the relevant files. Let me compile the review.

---

Summary

This PR adds a configurable data retention policy to PyWorkflow. It introduces delete_old_runs(older_than: datetime) -> int as a new abstract method on StorageBackend, implements it across all seven storage backends (memory, file, SQLite, PostgreSQL, MySQL, Cassandra, DynamoDB), adds a periodic Celery Beat task (pyworkflow.run_data_retention) that fires every 24 hours, and wires the data_retention_days configuration option through env var (PYWORKFLOW_DATA_RETENTION_DAYS) and YAML config.

---

Risk Assessment

Tier 3 — confirmed. The PR touches pyworkflow/storage/base.py (new abstract method, breaking change for any external implementors) and pyworkflow/celery/tasks.py (new periodic task). Both are on the Tier 3 critical-path list. The retention task runs destructive DELETE operations on production data, making correctness of each backend's implementation especially critical.

---

Issues

1. [blocking] pyworkflow/storage/cassandra.py:1712–1776 — runs_by_status index not cleaned up after deletion

The Cassandra delete_old_runs reads from runs_by_status to find candidate runs by status, then deletes from events, steps, hooks, cancellation_flags, and workflow_runs. It never deletes the corresponding row from runs_by_status. Orphaned entries accumulate indefinitely. Any query that reads from runs_by_status (e.g. list_runs with a status filter) will subsequently return run IDs that no longer exist in workflow_runs, producing incorrect results and potentially causing downstream errors whenever callers try to deserialize the missing rows.

2. [blocking] pyworkflow/storage/file.py:963–998 — cancellation flag files (.cancel) are not deleted

delete_old_runs in the file backend unlinks runs/{run_id}.json, the events .jsonl file, step files, and hook files — but it does not unlink runs/{run_id}.cancel. The StorageBackend.delete_old_runs docstring at base.py:708–709 explicitly contracts that "associated events, steps, hooks, and cancellation flags are deleted too." The file backend violates this contract. The InMemoryStorageBackend correctly calls self._cancellation_flags.pop(run_id, None) (line 629), as does the SQL backends (via DELETE FROM cancellation_flags). Orphaned .cancel files pose no data corruption risk by themselves, but they represent a broken invariant and would cause check_cancellation_flag to return True for a run that has already been purged — a logic error for any code that might re-create a run with the same ID.

3. [blocking] pyworkflow/storage/file.py:989–992 — _token_index.json not updated on hook deletion

When hook files are deleted in _sync_delete, the token→(run_id, hook_id) mapping in _token_index.json is left untouched. InMemoryStorageBackend.delete_old_runs correctly removes each token from _token_index (lines 627–629). Over time, stale token entries accumulate in _token_index.json. The file will grow without bound across retention sweeps. While get_hook_by_token degrades gracefully (returning None when the hook file is missing), the ever-growing index is a resource leak and deviates from the documented contract that "hooks are deleted." The index must be updated to remove all tokens belonging to deleted runs.

4. [warning] pyworkflow/config.py:204–210 — no validation that data_retention_days is positive

int(retention_env) is accepted for any integer, including zero and negatives. With data_retention_days=0, the cutoff becomes datetime.now(UTC) and all terminal runs are deleted on every sweep. With a negative value (e.g. -1), cutoff = now + 1 day, which silently deletes runs updated within the last 24 hours. A validation guard (if data_retention_days <= 0: raise ValueError(...)) in _config_from_env_and_yaml and configure_from_yaml would prevent accidental destructive misconfiguration.

---

Architecture

All changes comply with the core ← engine ← celery dependency rule. The new delete_old_runs abstract method in storage/base.py has no upward dependencies. celery/tasks.py imports correctly from pyworkflow.config and pyworkflow.storage. Import order is correct throughout.

---

Test Coverage

Test coverage is inadequate for a Tier 3 PR:

• tests/unit/test_retention.py covers InMemoryStorageBackend.delete_old_runs thoroughly (8 cases including events, steps, hooks, cancellation flags, and mixed run states — well done).
• There are no tests for FileStorageBackend.delete_old_runs, SQLiteStorageBackend.delete_old_runs, PostgresStorageBackend.delete_old_runs, MySQLStorageBackend.delete_old_runs, CassandraStorageBackend.delete_old_runs, or DynamoDBStorageBackend.delete_old_runs.
• There are no tests for the run_data_retention_task Celery task (neither the happy path where data_retention_days is set nor the skip path where it is None).

The two blocking file-backend bugs (issues 2 and 3) would have been caught by a basic integration test against FileStorageBackend. The Cassandra runs_by_status bug (issue 1) would similarly be detected by a test that asserts list_runs returns no results after retention.

---

🤖 Code Review Agent — automated code review.