diff --git a/backend/app/config.py b/backend/app/config.py index 26cf219..f41bd3f 100644 --- a/backend/app/config.py +++ b/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 = "" diff --git a/backend/app/services/audit_service.py b/backend/app/services/audit_service.py index f38ea6f..934e587 100644 --- a/backend/app/services/audit_service.py +++ b/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) diff --git a/backend/app/services/governance_service.py b/backend/app/services/governance_service.py index 371fbd4..eb610fc 100644 --- a/backend/app/services/governance_service.py +++ b/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"]) diff --git a/backend/app/services/llm_credentials_service.py b/backend/app/services/llm_credentials_service.py index 938856c..889e89f 100644 --- a/backend/app/services/llm_credentials_service.py +++ b/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). diff --git a/backend/app/services/observability_service.py b/backend/app/services/observability_service.py index 3855052..d459205 100644 --- a/backend/app/services/observability_service.py +++ b/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") diff --git a/backend/app/services/pipeline_service.py b/backend/app/services/pipeline_service.py index df21d28..e1d4cce 100644 --- a/backend/app/services/pipeline_service.py +++ b/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 diff --git a/backend/app/services/retention.py b/backend/app/services/retention.py new file mode 100644 index 0000000..f88c951 --- /dev/null +++ b/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 diff --git a/backend/app/services/service_invocation_service.py b/backend/app/services/service_invocation_service.py index e0b7a3d..66943a5 100644 --- a/backend/app/services/service_invocation_service.py +++ b/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, diff --git a/backend/app/services/session_service.py b/backend/app/services/session_service.py index 0301338..8b0ddbb 100644 --- a/backend/app/services/session_service.py +++ b/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 diff --git a/backend/app/services/workspace_credentials_service.py b/backend/app/services/workspace_credentials_service.py index f748e9d..77284e9 100644 --- a/backend/app/services/workspace_credentials_service.py +++ b/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 diff --git a/backend/tests/test_retention.py b/backend/tests/test_retention.py new file mode 100644 index 0000000..466d18e --- /dev/null +++ b/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() diff --git a/deploy-cli/scripts/20-platform.sh b/deploy-cli/scripts/20-platform.sh index c0979cf..c9b8e9b 100755 --- a/deploy-cli/scripts/20-platform.sh +++ b/deploy-cli/scripts/20-platform.sh @@ -106,8 +106,16 @@ fi aws dynamodb update-continuous-backups --table-name "$TABLE" \ --point-in-time-recovery-specification PointInTimeRecoveryEnabled=true >/dev/null 2>&1 \ || log "PITR already enabled" +# TTL on the `ttl` attribute (see backend/app/services/retention.py). Like PITR, +# enabling it again errors, so only enable when it is not already on. +if [ "$(aws dynamodb describe-time-to-live --table-name "$TABLE" \ + --query 'TimeToLiveDescription.TimeToLiveStatus' --output text 2>/dev/null)" != "ENABLED" ]; then + aws dynamodb update-time-to-live --table-name "$TABLE" \ + --time-to-live-specification Enabled=true,AttributeName=ttl >/dev/null 2>&1 \ + || log "TTL already enabled or being enabled" +fi save TABLE "$TABLE" -log "table ready (PITR + SSE)" +log "table ready (PITR + SSE + TTL)" # ------------------------------------------------------------------ ecr for r in "${KERNEL_REPOS[@]}" "${SERVICE_REPOS[@]}"; do diff --git a/docs/deployment.md b/docs/deployment.md index 3d1465c..dc87382 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -667,6 +667,33 @@ Kernel images can be pinned individually (`claude_code_image_tag`, forwarding needs a headless-kernel build that supports per-attachment MCP headers. +## Data retention + +The platform table has DynamoDB TTL enabled on the `ttl` attribute. DynamoDB +deletes an item some time after its `ttl`, usually within a few days and +never before it, at no cost. Items without `ttl` are kept. The backend sets +`ttl` per record type: + +| Record | Kept for | Setting (backend env) | +|---|---|---| +| Invocation ledger | 90 days | `RETENTION_LEDGER_DAYS` | +| Daily quota counters | 90 days after the day | `RETENTION_USAGE_DAYS` | +| Async service-entry results | 7 days | `RETENTION_SERVICE_INVOCATION_DAYS` | +| Terminated sessions | 30 days after termination; workspace files in S3 are not touched | `RETENTION_TERMINATED_SESSION_DAYS` | +| Model gateway grants and revocations | 1 day past the credential's expiry | fixed | +| Workspace refresh-token lookups | 2 days; AgentCore sessions last at most 8 hours | fixed | +| Audit trail | forever | `RETENTION_AUDIT_DAYS` | +| Pipeline runs | forever | `RETENTION_PIPELINE_RUN_DAYS` | + +`0` means keep forever. Agents, the registry, schedules, channels, +pipelines, evaluation datasets and runs, governance policy and model +configuration are never given a `ttl`. + +Retention applies to items written after the change. Older items have no +`ttl` and stay until you delete them. The model gateway grants matter most +here: a headless invocation routed through the gateway mints one per call, +and before TTL nothing removed them. + ## Troubleshooting | Symptom | Likely cause | diff --git a/infrastructure/stacks/platform_stack.py b/infrastructure/stacks/platform_stack.py index 41f6fcd..f1651a5 100644 --- a/infrastructure/stacks/platform_stack.py +++ b/infrastructure/stacks/platform_stack.py @@ -31,6 +31,8 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None: partition_key=dynamodb.Attribute(name="PK", type=dynamodb.AttributeType.STRING), sort_key=dynamodb.Attribute(name="SK", type=dynamodb.AttributeType.STRING), billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST, + # see backend/app/services/retention.py + time_to_live_attribute="ttl", removal_policy=RemovalPolicy.DESTROY, ) diff --git a/terraform/modules/platform/main.tf b/terraform/modules/platform/main.tf index 4818f57..f93d428 100644 --- a/terraform/modules/platform/main.tf +++ b/terraform/modules/platform/main.tf @@ -100,6 +100,16 @@ resource "aws_dynamodb_table" "platform" { point_in_time_recovery { enabled = true } + + # Items that carry a `ttl` (epoch seconds) are deleted by DynamoDB some time + # after it, at no cost: the ledger, quota counters, credential rows, + # service-entry results and terminated sessions. Items without it are kept. + # Per-type retention is backend configuration (RETENTION_*_DAYS); see + # backend/app/services/retention.py. + ttl { + attribute_name = "ttl" + enabled = true + } } # ------------------------------ ECR repos ----------------------------------