fix(worker): bound terminal operation history (#2708)

Add configurable TTL (default 30 days, 0=keep-forever) for terminal async_operations rows. Expired completed/failed/cancelled rows are pruned in bounded batches (1000/cycle) by a background task that never touches pending/processing work. Batch children are protected until their parent is pruned; cancelled-child cleanup atomically cancels a pending parent first. PG uses FOR UPDATE SKIP LOCKED; both PG and Oracle re-check eligibility under the row lock before deleting. Includes indexes, docs, and regenerated SDKs.

184 retention/worker/operation-status tests pass locally. All CI green.

Fixes #2705

Author: Tosko4

.env.example

@@ -105,6 +105,8 @@ HINDSIGHT_API_LOG_LEVEL=info
 # HINDSIGHT_API_MIGRATION_DATABASE_URL=  # Direct PostgreSQL URL for migrations (bypasses PgBouncer). Falls back to DATABASE_URL.
 # HINDSIGHT_API_DATABASE_SCHEMA=public  # PostgreSQL schema name (default: public)
 # HINDSIGHT_API_MIGRATION_CONCURRENCY=1  # Tenant schemas to migrate concurrently (PG only, each in its own process; per-schema work stays sequential). Each worker has ~1-2s startup cost + uses ~3 DB connections, so it only pays off with many schemas (tens+) or slow migrations; keep concurrency*3 <= spare max_connections. Default: 1 (sequential).
+# HINDSIGHT_API_OPERATION_RETENTION_DAYS=30  # Keep terminal operation rows, payloads, and metadata for this many days; 0 disables automatic pruning.
+# HINDSIGHT_API_OPERATION_CLEANUP_BATCH_SIZE=1000  # Maximum expired terminal rows deleted per tenant schema in each cleanup cycle; must be positive.
 
 # Vector Extension (Optional - uses pgvector by default)
 # Options: "pgvector" (default), "vchord", "pgvectorscale" (DiskANN)

hindsight-api-slim/hindsight_api/alembic/versions/a8c1e4f7b0d3_add_operation_retention_indexes.py

@@ -0,0 +1,82 @@
+"""Add indexes for terminal cleanup and newest-first operation listing.
+
+Revision ID: a8c1e4f7b0d3
+Revises: f2a4b6c8d0e2
+Create Date: 2026-07-14
+"""
+
+from collections.abc import Sequence
+
+from alembic import context, op
+
+from hindsight_api.alembic._dialect import run_for_dialect
+
+revision: str = "a8c1e4f7b0d3"
+down_revision: str | Sequence[str] | None = "f2a4b6c8d0e2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _pg_schema_prefix() -> str:
+    """Schema-qualifier for PostgreSQL multi-tenant migration runs."""
+    schema = context.config.get_main_option("target_schema")
+    return f'"{schema}".' if schema else ""
+
+
+def _pg_upgrade() -> None:
+    schema = _pg_schema_prefix()
+    # These can be large tables in long-running installations. Concurrent DDL
+    # keeps operation submission, polling, and status reads available.
+    with op.get_context().autocommit_block():
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_async_operations_terminal_cleanup "
+            f"ON {schema}async_operations (updated_at, operation_id) "
+            "WHERE status IN ('completed', 'failed', 'cancelled')"
+        )
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_async_operations_bank_created_desc "
+            f"ON {schema}async_operations (bank_id, created_at DESC)"
+        )
+
+
+def _pg_downgrade() -> None:
+    schema = _pg_schema_prefix()
+    with op.get_context().autocommit_block():
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_async_operations_bank_created_desc")
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {schema}idx_async_operations_terminal_cleanup")
+
+
+def _oracle_create_index(sql: str) -> None:
+    """Create an index idempotently for rerun-safe Oracle migrations."""
+    block = (
+        "BEGIN "
+        "EXECUTE IMMEDIATE :stmt; "
+        "EXCEPTION WHEN OTHERS THEN "
+        "IF SQLCODE = -955 THEN NULL; ELSE RAISE; END IF; "
+        "END;"
+    )
+    op.get_bind().exec_driver_sql(block, {"stmt": sql})
+
+
+def _oracle_upgrade() -> None:
+    # Oracle migrations run with CURRENT_SCHEMA set to each tenant, so table
+    # and index names intentionally remain unqualified here.
+    _oracle_create_index(
+        "CREATE INDEX idx_async_operations_terminal_cleanup ON async_operations (updated_at, operation_id, status)"
+    )
+    _oracle_create_index(
+        "CREATE INDEX idx_async_operations_bank_created_desc ON async_operations (bank_id, created_at DESC)"
+    )
+
+
+def _oracle_downgrade() -> None:
+    op.execute("DROP INDEX idx_async_operations_bank_created_desc")
+    op.execute("DROP INDEX idx_async_operations_terminal_cleanup")
+
+
+def upgrade() -> None:
+    run_for_dialect(pg=_pg_upgrade, oracle=_oracle_upgrade)
+
+
+def downgrade() -> None:
+    run_for_dialect(pg=_pg_downgrade, oracle=_oracle_downgrade)

hindsight-api-slim/hindsight_api/api/http.py

@@ -3144,6 +3144,8 @@ async def lifespan(app: FastAPI):
                 max_slots=config.worker_max_slots,
                 slot_reservations=config.worker_slot_reservations,
                 consolidation_bank_priority=config.worker_consolidation_bank_priority or None,
+                operation_retention_days=config.operation_retention_days,
+                operation_cleanup_batch_size=config.operation_cleanup_batch_size,
             )
             poller_task = asyncio.create_task(poller.run())
             logging.info(f"Worker poller started (worker_id={worker_id})")
@@ -5422,8 +5424,9 @@ async def api_list_operations(
         "/v1/default/banks/{bank_id}/operations/{operation_id}",
         response_model=OperationStatusResponse,
         summary="Get operation status",
-        description="Get the status of a specific async operation. Returns 'pending', 'completed', or 'failed'. "
-        "Completed operations are removed from storage, so 'completed' means the operation finished successfully.",
+        description="Get the status of a specific async operation. Returns 'pending', 'processing', 'completed', "
+        "'failed', or 'cancelled'. Completed operations remain queryable with their payload for the configured "
+        "retention window and are pruned afterward.",
         operation_id="get_operation_status",
         tags=["Operations"],
     )

hindsight-api-slim/hindsight_api/config.py

@@ -599,6 +599,8 @@ def _resolve_operation_temperature(operation_env: str, default: float) -> float
 ENV_WORKER_TASK_RETRY_BACKOFF_SECONDS = "HINDSIGHT_API_WORKER_TASK_RETRY_BACKOFF_SECONDS"
 ENV_WORKER_HTTP_PORT = "HINDSIGHT_API_WORKER_HTTP_PORT"
 ENV_WORKER_MAX_SLOTS = "HINDSIGHT_API_WORKER_MAX_SLOTS"
+ENV_OPERATION_RETENTION_DAYS = "HINDSIGHT_API_OPERATION_RETENTION_DAYS"
+ENV_OPERATION_CLEANUP_BATCH_SIZE = "HINDSIGHT_API_OPERATION_CLEANUP_BATCH_SIZE"
 
 # Per-operation-type slot reservations. Each entry maps an operation_type
 # (as stored in async_operations.operation_type) to its env var and default.
@@ -1055,6 +1057,10 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 DEFAULT_WORKER_TASK_RETRY_BACKOFF_SECONDS = 60  # Seconds between retries on transient task failure
 DEFAULT_WORKER_HTTP_PORT = 8889  # HTTP port for worker metrics/health
 DEFAULT_WORKER_MAX_SLOTS = 10  # Total concurrent tasks per worker
+# Terminal rows keep their payload and metadata for one coherent debug/retry TTL.
+# Zero retention days disables automatic pruning entirely.
+DEFAULT_OPERATION_RETENTION_DAYS = 30
+DEFAULT_OPERATION_CLEANUP_BATCH_SIZE = 1000
 DEFAULT_RETAIN_MAX_CONCURRENT = 4  # Max concurrent retain DB phases (HNSW reads + writes). Limits I/O contention.
 
 # Reflect agent settings
@@ -1929,6 +1935,8 @@ class HindsightConfig:
     worker_max_slots: int
     worker_slot_reservations: dict[str, int]
     worker_consolidation_bank_priority: dict[str, int]
+    operation_retention_days: int
+    operation_cleanup_batch_size: int
     retain_max_concurrent: int
 
     # Reflect agent settings
@@ -2293,6 +2301,13 @@ def validate(self) -> None:
                 f"Reduce reservations or increase HINDSIGHT_API_WORKER_MAX_SLOTS."
             )
 
+        if self.operation_retention_days < 0:
+            raise ValueError(f"{ENV_OPERATION_RETENTION_DAYS} must be >= 0, got {self.operation_retention_days}")
+        if self.operation_cleanup_batch_size < 1:
+            raise ValueError(
+                f"{ENV_OPERATION_CLEANUP_BATCH_SIZE} must be >= 1, got {self.operation_cleanup_batch_size}"
+            )
+
     @classmethod
     def from_env(cls) -> "HindsightConfig":
         """Create configuration from environment variables."""
@@ -2958,6 +2973,16 @@ def from_env(cls) -> "HindsightConfig":
             worker_consolidation_bank_priority=_parse_bank_priority(
                 os.getenv(ENV_WORKER_CONSOLIDATION_BANK_PRIORITY, "")
             ),
+            operation_retention_days=_parse_non_negative_int(
+                ENV_OPERATION_RETENTION_DAYS,
+                os.getenv(ENV_OPERATION_RETENTION_DAYS),
+                DEFAULT_OPERATION_RETENTION_DAYS,
+            ),
+            operation_cleanup_batch_size=_parse_positive_int(
+                ENV_OPERATION_CLEANUP_BATCH_SIZE,
+                os.getenv(ENV_OPERATION_CLEANUP_BATCH_SIZE),
+                DEFAULT_OPERATION_CLEANUP_BATCH_SIZE,
+            ),
             retain_max_concurrent=int(os.getenv(ENV_RETAIN_MAX_CONCURRENT, str(DEFAULT_RETAIN_MAX_CONCURRENT))),
             # Reflect agent settings
             reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))),

hindsight-api-slim/hindsight_api/engine/db/ops.py

@@ -18,6 +18,7 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any
 
 from .base import DatabaseConnection
@@ -484,6 +485,23 @@ async def prune_stale_cooccurrences(
 
     # -- Task claiming operations ------------------------------------------
 
+    @abstractmethod
+    async def prune_terminal_operations(
+        self,
+        conn: DatabaseConnection,
+        table: str,
+        cutoff: datetime,
+        *,
+        batch_size: int,
+    ) -> int:
+        """Delete one deterministic batch of terminal operations older than ``cutoff``.
+
+        Implementations must lock candidates without waiting on rows another
+        worker is pruning, never select pending/processing rows, and return the
+        number deleted. The caller provides a transaction around this method.
+        """
+        ...
+
     @abstractmethod
     async def claim_tasks(
         self,

hindsight-api-slim/hindsight_api/engine/db/ops_oracle.py

@@ -13,6 +13,8 @@
 from .ops import DataAccessOps, TagListingParts
 from .result import DictResultRow as ResultRow
 
+ORACLE_IN_LIST_LIMIT = 1000
+
 
 class OracleOps(DataAccessOps):
     """Oracle-specific data access operations."""
@@ -831,6 +833,157 @@ async def insert_webhook_delivery_task(self, conn, ops_table, operation_id, bank
 
     # -- Task claiming operations ------------------------------------------
 
+    async def prune_terminal_operations(
+        self,
+        conn: DatabaseConnection,
+        table: str,
+        cutoff: datetime,
+        *,
+        batch_size: int,
+    ) -> int:
+        # Oracle rejects a row-limited SELECT ... FOR UPDATE (ORA-02014). Pick
+        # the deterministic bounded IDs first, then lock only that candidate
+        # set and re-check eligibility before deleting in the same transaction.
+        # Clamp to Oracle's 1000-expression IN-list limit because the adapter
+        # expands the candidate UUID list into individual bind variables.
+        # Cancelled children cannot complete parent aggregation, so retain the
+        # parent guard only for completed/failed children. Before removing a
+        # cancelled child, preserve its signal by cancelling a pending parent
+        # in this transaction and refreshing the parent's retention window.
+        # Validate metadata before HEXTORAW: CASE makes malformed UUIDs yield
+        # NULL while keeping the indexed RAW parent.operation_id key unwrapped.
+        effective_batch_size = min(batch_size, ORACLE_IN_LIST_LIMIT)
+        candidates = await conn.fetch(
+            f"""
+            SELECT candidate_operation.operation_id
+            FROM {table} candidate_operation
+            WHERE candidate_operation.status IN ('completed', 'failed', 'cancelled')
+              AND candidate_operation.updated_at < $1
+              AND (
+                  candidate_operation.status = 'cancelled'
+                  OR NOT EXISTS (
+                      SELECT 1
+                      FROM {table} parent
+                      WHERE parent.operation_id = CASE
+                          WHEN REGEXP_LIKE(
+                              JSON_VALUE(
+                                  candidate_operation.result_metadata,
+                                  '$.parent_operation_id' RETURNING VARCHAR2(36) NULL ON ERROR
+                              ),
+                              '^[0-9A-Fa-f]{{8}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{12}}$'
+                          )
+                          THEN HEXTORAW(REPLACE(
+                              JSON_VALUE(
+                                  candidate_operation.result_metadata,
+                                  '$.parent_operation_id' RETURNING VARCHAR2(36) NULL ON ERROR
+                              ),
+                              '-',
+                              ''
+                          ))
+                          ELSE NULL
+                      END
+                        AND parent.bank_id = candidate_operation.bank_id
+                  )
+              )
+            ORDER BY candidate_operation.updated_at, candidate_operation.operation_id
+            LIMIT $2
+            """,
+            cutoff,
+            effective_batch_size,
+        )
+        if not candidates:
+            return 0
+
+        candidate_ids = [row["operation_id"] for row in candidates]
+        locked = await conn.fetch(
+            f"""
+            SELECT candidate_operation.operation_id
+            FROM {table} candidate_operation
+            WHERE candidate_operation.operation_id = ANY($1)
+              AND candidate_operation.status IN ('completed', 'failed', 'cancelled')
+              AND candidate_operation.updated_at < $2
+              AND (
+                  candidate_operation.status = 'cancelled'
+                  OR NOT EXISTS (
+                      SELECT 1
+                      FROM {table} parent
+                      WHERE parent.operation_id = CASE
+                          WHEN REGEXP_LIKE(
+                              JSON_VALUE(
+                                  candidate_operation.result_metadata,
+                                  '$.parent_operation_id' RETURNING VARCHAR2(36) NULL ON ERROR
+                              ),
+                              '^[0-9A-Fa-f]{{8}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{12}}$'
+                          )
+                          THEN HEXTORAW(REPLACE(
+                              JSON_VALUE(
+                                  candidate_operation.result_metadata,
+                                  '$.parent_operation_id' RETURNING VARCHAR2(36) NULL ON ERROR
+                              ),
+                              '-',
+                              ''
+                          ))
+                          ELSE NULL
+                      END
+                        AND parent.bank_id = candidate_operation.bank_id
+                  )
+              )
+            ORDER BY candidate_operation.updated_at, candidate_operation.operation_id
+            FOR UPDATE OF candidate_operation.operation_id SKIP LOCKED
+            """,
+            candidate_ids,
+            cutoff,
+        )
+        if not locked:
+            return 0
+        operation_ids = [row["operation_id"] for row in locked]
+        await conn.execute(
+            f"""
+            UPDATE {table} parent
+            SET status = 'cancelled',
+                updated_at = now(),
+                completed_at = COALESCE(parent.completed_at, now()),
+                error_message = COALESCE(
+                    parent.error_message,
+                    'Cancelled because a child operation was cancelled'
+                )
+            WHERE parent.status = 'pending'
+              AND EXISTS (
+                  SELECT 1
+                  FROM {table} candidate_operation
+                  WHERE candidate_operation.operation_id = ANY($1)
+                    AND candidate_operation.status = 'cancelled'
+                    AND candidate_operation.updated_at < $2
+                    AND candidate_operation.bank_id = parent.bank_id
+                    AND parent.operation_id = CASE
+                        WHEN REGEXP_LIKE(
+                            JSON_VALUE(
+                                candidate_operation.result_metadata,
+                                '$.parent_operation_id' RETURNING VARCHAR2(36) NULL ON ERROR
+                            ),
+                            '^[0-9A-Fa-f]{{8}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{4}}-[0-9A-Fa-f]{{12}}$'
+                        )
+                        THEN HEXTORAW(REPLACE(
+                            JSON_VALUE(
+                                candidate_operation.result_metadata,
+                                '$.parent_operation_id' RETURNING VARCHAR2(36) NULL ON ERROR
+                            ),
+                            '-',
+                            ''
+                        ))
+                        ELSE NULL
+                    END
+              )
+            """,
+            operation_ids,
+            cutoff,
+        )
+        await conn.execute(
+            f"DELETE FROM {table} WHERE operation_id = ANY($1)",
+            operation_ids,
+        )
+        return len(operation_ids)
+
     async def _claim_consolidation_tasks(
         self,
         conn,