Give expiring DynamoDB records a TTL

backend/app/config.py

@@ -9,6 +9,16 @@ class Settings(BaseSettings):
     # DynamoDB single table for sessions
     dynamo_table: str = "agent-platform"
 
+    # How long records are kept before DynamoDB TTL may delete them, in days
+    # (0 = keep forever). See app/services/retention.py. Credential rows
+    # (gateway grants, revocations, workspace tokens) follow their own expiry.
+    retention_ledger_days: int = 90  # invocation ledger (PK=INVOCATION)
+    retention_usage_days: int = 90  # daily quota counters (PK=USAGE)
+    retention_service_invocation_days: int = 7  # async service-entry results (PK=SVCINV)
+    retention_terminated_session_days: int = 30  # sessions after they are terminated
+    retention_audit_days: int = 0  # audit trail (PK=AUDIT)
+    retention_pipeline_run_days: int = 0  # pipeline runs (PK=PIPELINERUN)
+
     # AgentCore Runtime ARNs (outputs of the RuntimeStack)
     interactive_runtime_arn: str = ""
     sdk_runtime_arn: str = ""

backend/app/services/audit_service.py

@@ -12,6 +12,7 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -32,17 +33,17 @@ def record(self, user: str, action: str, resource: str = "", detail: str = "") -
         action being audited."""
         try:
             ts = _now()
-            self.table.put_item(
-                Item={
-                    "PK": PK,
-                    "SK": f"{ts}#{uuid.uuid4().hex[:8]}",
-                    "ts": ts,
-                    "user": user,
-                    "action": action,
-                    "resource": resource[:300],
-                    "detail": detail[:500],
-                }
-            )
+            item = {
+                "PK": PK,
+                "SK": f"{ts}#{uuid.uuid4().hex[:8]}",
+                "ts": ts,
+                "user": user,
+                "action": action,
+                "resource": resource[:300],
+                "detail": detail[:500],
+            }
+            retention.with_ttl(item, retention.ttl_after_days(settings.retention_audit_days))
+            self.table.put_item(Item=item)
         except Exception:
             logger.exception("audit record failed: %s %s", action, resource)
 

backend/app/services/governance_service.py

@@ -15,6 +15,7 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -85,11 +86,19 @@ def update_policy(self, patch: dict) -> dict:
     # -------------------------------------------------------------- usage
 
     def _increment(self, sk: str) -> int:
+        ttl = retention.ttl_after_days(settings.retention_usage_days)
+        if ttl is None:
+            update, names, values = "ADD #c :one", {"#c": "count"}, {":one": 1}
+        else:
+            # if_not_exists: the day's first increment fixes the deletion date
+            update = "ADD #c :one SET #t = if_not_exists(#t, :ttl)"
+            names = {"#c": "count", "#t": retention.TTL_ATTR}
+            values = {":one": 1, ":ttl": ttl}
         resp = self.table.update_item(
             Key={"PK": USAGE_PK, "SK": sk},
-            UpdateExpression="ADD #c :one",
-            ExpressionAttributeNames={"#c": "count"},
-            ExpressionAttributeValues={":one": 1},
+            UpdateExpression=update,
+            ExpressionAttributeNames=names,
+            ExpressionAttributeValues=values,
             ReturnValues="UPDATED_NEW",
         )
         return int(resp["Attributes"]["count"])

backend/app/services/llm_credentials_service.py

@@ -38,6 +38,7 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -189,6 +190,7 @@ def mint(
                     # replay the grant.
                     "token_sha256": _sha256(token),
                     "expires_at": expires_at,
+                    retention.TTL_ATTR: retention.ttl_after_expiry(expires_at),
                     "runtime_session_id": runtime_session_id,
                     "user": user,
                     "team": team,
@@ -334,6 +336,7 @@ def mint_agentcore(
                     # No token digest: the credential is an STS session, not a
                     # bearer value this table could be used to replay.
                     "expires_at": expires_at,
+                    retention.TTL_ATTR: retention.ttl_after_expiry(expires_at),
                     "runtime_session_id": runtime_session_id,
                     "user": user,
                     "team": team,
@@ -435,6 +438,8 @@ def revoke_agentcore(self, runtime_session_id: str) -> None:
                     "PK": LLM_REVOKE_PK,
                     "SK": f"RSID#{runtime_session_id}",
                     "expires_at": expires_at,
+                    # A revocation only has to outlive the credential it denies.
+                    retention.TTL_ATTR: retention.ttl_after_expiry(expires_at),
                 }
             )
         except Exception:  # noqa: BLE001
@@ -561,9 +566,15 @@ def rotate(self, runtime_session_id: str) -> dict | None:
         try:
             self.table.update_item(
                 Key={"PK": LLM_TOKEN_PK, "SK": f"RSID#{runtime_session_id}"},
-                UpdateExpression="SET token_sha256 = :h, expires_at = :e",
+                # Rotation extends validity, so it must extend the row's TTL too.
+                UpdateExpression="SET token_sha256 = :h, expires_at = :e, #ttl = :ttl",
                 ConditionExpression="attribute_exists(PK)",
-                ExpressionAttributeValues={":h": _sha256(token), ":e": expires_at},
+                ExpressionAttributeNames={"#ttl": retention.TTL_ATTR},
+                ExpressionAttributeValues={
+                    ":h": _sha256(token),
+                    ":e": expires_at,
+                    ":ttl": retention.ttl_after_expiry(expires_at),
+                },
             )
         except Exception:
             # No grant for this session (never gateway-routed, or revoked).

backend/app/services/observability_service.py

@@ -18,6 +18,7 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -84,6 +85,7 @@ def record(
                 item["num_turns"] = int(num_turns)
             if total_cost_usd is not None:
                 item["total_cost_usd"] = _to_decimal(float(total_cost_usd))
+            retention.with_ttl(item, retention.ttl_after_days(settings.retention_ledger_days))
             self.table.put_item(Item=item)
         except Exception:
             logger.exception("invocation record failed")

backend/app/services/pipeline_service.py

@@ -28,6 +28,7 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -330,6 +331,7 @@ def _create_run(self, pipeline: str, user: str, source: str,
         }
         if parent_run:
             item["parent_run"] = parent_run
+        retention.with_ttl(item, retention.ttl_after_days(settings.retention_pipeline_run_days))
         self.table.put_item(Item=item)
         return sk
 

backend/app/services/retention.py

@@ -0,0 +1,40 @@
+"""When a DynamoDB item may be deleted.
+
+The platform table has DynamoDB TTL enabled on the ``ttl`` attribute (epoch
+seconds). DynamoDB deletes an item some time after that moment, typically
+within a few days, never before it. An item without ``ttl`` is kept forever.
+
+``ttl`` is deliberately separate from ``expires_at``: ``expires_at`` says when
+a credential stops being *valid* (checked in code on every use, refreshed on
+rotation); ``ttl`` only says when the row is no longer needed at all.
+
+Retention per record type is configured in :mod:`app.config`
+(``retention_*_days``; ``0`` keeps the record forever).
+"""
+
+import time
+
+TTL_ATTR = "ttl"
+
+# A credential row is kept a day past its expiry: reads already reject it
+# once expired, and the margin absorbs clock skew and in-flight refreshes.
+CREDENTIAL_GRACE_S = 24 * 3600
+
+
+def ttl_after_days(days: int, *, start: float | None = None) -> int | None:
+    """Epoch seconds ``days`` from ``start`` (default now); None keeps forever."""
+    if not days or days <= 0:
+        return None
+    return int((start if start is not None else time.time()) + days * 86400)
+
+
+def ttl_after_expiry(expires_at: float) -> int:
+    """For a row whose content is useless once ``expires_at`` has passed."""
+    return int(expires_at) + CREDENTIAL_GRACE_S
+
+
+def with_ttl(item: dict, ttl: int | None) -> dict:
+    """Return ``item`` with the TTL attribute set, or unchanged when ``ttl`` is None."""
+    if ttl is not None:
+        item[TTL_ATTR] = ttl
+    return item

backend/app/services/service_invocation_service.py

@@ -27,6 +27,7 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -87,6 +88,7 @@ def submit(
             "created_at": _now(),
             "updated_at": _now(),
         }
+        retention.with_ttl(item, retention.ttl_after_days(settings.retention_service_invocation_days))
         self.table.put_item(Item=item)
         _EXECUTOR.submit(
             self._run, channel_item, channel_id, inv_id, caller, message,

backend/app/services/session_service.py

@@ -16,6 +16,7 @@
 from botocore.awsrequest import AWSRequest
 
 from app.config import runtime_arn, settings
+from app.services import retention
 
 logger = logging.getLogger(__name__)
 
@@ -95,11 +96,22 @@ def set_attachments(
         )
 
     def set_status(self, user: str, session_id: str, status: str) -> None:
+        update = "SET #s = :s, last_activity = :t"
+        names = {"#s": "status"}
+        values = {":s": status, ":t": _now()}
+        ttl = retention.ttl_after_days(settings.retention_terminated_session_days)
+        if status == "terminated" and ttl is not None:
+            # A terminated session is hidden from every listing; keep the row
+            # a while for support, then let TTL drop it. Workspace files in S3
+            # are not touched.
+            update += ", #ttl = :ttl"
+            names["#ttl"] = retention.TTL_ATTR
+            values[":ttl"] = ttl
         self.table.update_item(
             Key={"PK": f"USER#{user}", "SK": f"SESSION#{session_id}"},
-            UpdateExpression="SET #s = :s, last_activity = :t",
-            ExpressionAttributeNames={"#s": "status"},
-            ExpressionAttributeValues={":s": status, ":t": _now()},
+            UpdateExpression=update,
+            ExpressionAttributeNames=names,
+            ExpressionAttributeValues=values,
         )
 
     # ------------------------------------------------- terminal connectivity

backend/app/services/workspace_credentials_service.py

@@ -27,6 +27,9 @@
 import boto3
 
 from app.config import settings
+from app.services import retention
+
+WS_TOKEN_RETENTION_DAYS = 2
 
 logger = logging.getLogger(__name__)
 
@@ -132,6 +135,9 @@ def issue_refresh_token(self, user: str, session_id: str) -> str:
                     "runtime_session_id": runtime_session_id,
                     "user": user,
                     "session_id": session_id,
+                    # AgentCore sessions live at most 8 hours and every connect
+                    # re-issues this token, so the row is dead long before this.
+                    retention.TTL_ATTR: retention.ttl_after_days(WS_TOKEN_RETENTION_DAYS),
                 }
             )
         return token

backend/tests/test_retention.py

@@ -0,0 +1,109 @@
+"""Every record type that should expire carries a DynamoDB ``ttl``; others do not."""
+
+import importlib
+import sys
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import boto3
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+DAY = 86400
+
+
+def load(module: str):
+    """Import a service with boto3 stubbed; the real settings (all defaulted)."""
+    with patch.object(boto3, "resource", return_value=MagicMock()), patch.object(boto3, "client", return_value=MagicMock()):
+        sys.modules.pop(module, None)
+        return importlib.import_module(module)
+
+
+def fresh(module, cls_name: str):
+    service = getattr(module, cls_name).__new__(getattr(module, cls_name))
+    service.table = MagicMock()
+    return service
+
+
+class RetentionHelperTests(unittest.TestCase):
+    def test_helpers(self):
+        from app.services import retention
+
+        self.assertIsNone(retention.ttl_after_days(0))
+        self.assertEqual(retention.ttl_after_days(2, start=1000), 1000 + 2 * DAY)
+        self.assertEqual(retention.ttl_after_expiry(5000), 5000 + DAY)
+        self.assertEqual(retention.with_ttl({}, None), {})
+        self.assertEqual(retention.with_ttl({}, 7), {"ttl": 7})
+
+
+class RecordTtlTests(unittest.TestCase):
+    def assertAbout(self, value, expected):
+        self.assertLess(abs(int(value) - expected), 60)
+
+    def test_ledger_row_expires_after_the_configured_days(self):
+        mod = load("app.services.observability_service")
+        svc = fresh(mod, "ObservabilityService")
+        svc.record(user="u", source="api", target="agent:a", prompt="hi", ok=True)
+        item = svc.table.put_item.call_args.kwargs["Item"]
+        self.assertAbout(item["ttl"], time.time() + 90 * DAY)
+        with patch.object(mod.settings, "retention_ledger_days", 0):
+            svc.record(user="u", source="api", target="agent:a", prompt="hi", ok=True)
+        self.assertNotIn("ttl", svc.table.put_item.call_args.kwargs["Item"])
+
+    def test_quota_counter_sets_ttl_only_on_the_first_increment(self):
+        mod = load("app.services.governance_service")
+        svc = fresh(mod, "GovernanceService")
+        svc.table.update_item.return_value = {"Attributes": {"count": 1}}
+        svc._increment("2026-10-08")
+        kwargs = svc.table.update_item.call_args.kwargs
+        self.assertIn("ADD #c :one", kwargs["UpdateExpression"])
+        self.assertIn("if_not_exists(#t, :ttl)", kwargs["UpdateExpression"])
+        self.assertEqual(kwargs["ExpressionAttributeNames"]["#t"], "ttl")
+        self.assertAbout(kwargs["ExpressionAttributeValues"][":ttl"], time.time() + 90 * DAY)
+
+    def test_only_terminated_sessions_get_a_ttl(self):
+        mod = load("app.services.session_service")
+        svc = fresh(mod, "SessionService")
+        svc.set_status("u", "s1", "active")
+        self.assertNotIn("ttl", svc.table.update_item.call_args.kwargs["UpdateExpression"])
+        svc.set_status("u", "s1", "terminated")
+        kwargs = svc.table.update_item.call_args.kwargs
+        self.assertIn("#ttl = :ttl", kwargs["UpdateExpression"])
+        self.assertAbout(kwargs["ExpressionAttributeValues"][":ttl"], time.time() + 30 * DAY)
+
+    def test_audit_and_pipeline_runs_are_kept_by_default(self):
+        audit = fresh(load("app.services.audit_service"), "AuditService")
+        audit.record("u", "agent.publish", "agent:a")
+        self.assertNotIn("ttl", audit.table.put_item.call_args.kwargs["Item"])
+        pipes = fresh(load("app.services.pipeline_service"), "PipelineService")
+        pipes._create_run("nightly", "u", "api")
+        self.assertNotIn("ttl", pipes.table.put_item.call_args_list[0].kwargs["Item"])
+
+    def test_gateway_grant_expires_a_day_after_the_credential_and_rotation_extends_it(self):
+        mod = load("app.services.llm_credentials_service")
+        svc = fresh(mod, "LlmCredentialsService")
+        spec = {"backend": "litellm", "base_url": "https://gw.example", "secret_name": "s", "model": "m"}
+        with patch.object(mod.settings, "llm_edge_url", "http://edge"):
+            grant = svc.mint("rsid-1", "u", spec)
+            item = svc.table.put_item.call_args.kwargs["Item"]
+            self.assertEqual(item["ttl"], grant["expires_at"] + DAY)
+
+            svc.table.get_item.return_value = {"Item": {"PK": "LLMTOKEN", "SK": "RSID#rsid-1"}}
+            rotated = svc.rotate("rsid-1")
+            kwargs = svc.table.update_item.call_args.kwargs
+            self.assertIn("#ttl = :ttl", kwargs["UpdateExpression"])
+            self.assertEqual(kwargs["ExpressionAttributeValues"][":ttl"], rotated["expires_at"] + DAY)
+
+    def test_workspace_token_lookup_expires(self):
+        mod = load("app.services.workspace_credentials_service")
+        svc = fresh(mod, "WorkspaceCredentialsService")
+        svc.table.update_item.return_value = {"Attributes": {"runtime_session_id": "rsid-1"}}
+        svc.issue_refresh_token("u", "s1")
+        item = svc.table.put_item.call_args.kwargs["Item"]
+        self.assertAbout(item["ttl"], time.time() + 2 * DAY)
+
+
+if __name__ == "__main__":
+    unittest.main()