Skip to content

RFC-012: Artifact Lifecycle Protocol (ALP) — 9. Storage

AIGP SpecificationRFC-012: Artifact Lifecycle Protocol (ALP) › 9. Storage

← 8. Compensation · Section index · 10. Integration with AIGP and ACP →

9. Storage

Artifact metadata is stored in DynamoDB. Artifact content is stored in S3. This separation ensures fast metadata queries while supporting arbitrarily large artifact content.

9.1 DynamoDB Single-Table Design

Entity PK SK GSI1PK GSI1SK
Artifact ART#{artifact_id} V#{version} STATE#{state} #{created_at}#{artifact_id}
Transition ART#{artifact_id} TRANS#{timestamp} TRANS#{date} #{timestamp}#{artifact_id}
Approval ART#{artifact_id} APPROVAL#{timestamp} APPROVER#{actor_id} #{timestamp}#{artifact_id}
Execution ART#{artifact_id} EXEC#{run_id} RUN#{date} #{timestamp}#{artifact_id}
Provenance ART#{artifact_id} PROV#{step} PROV#{date} #{timestamp}#{artifact_id}
Compensation ART#{artifact_id} COMP#{comp_artifact_id} COMP_STATE#{state} #{created_at}#{artifact_id}

9.2 S3 Structure

s3://run-prj-artifacts/
artifacts/
{artifact_id}/
v{version}/
{filename} # Primary artifact content
rollback-{filename} # Compensation artifact content
metadata.json # Artifact metadata snapshot
executions/
{run_id}/
output.json # Execution output
stdout.log # Standard output
stderr.log # Standard error
provenance/
{artifact_id}.json # Complete provenance chain
archive/
{year}/{month}/
{artifact_id}/ # Archived artifacts (S3 Glacier)

9.3 S3 Lifecycle Rules

Path Prefix Storage Class Transition
artifacts/ S3 Standard Infrequent Access after 30 days
executions/ S3 Standard Infrequent Access after 14 days
provenance/ S3 Standard No transition (always hot)
archive/ S3 Glacier Permanent

9.4 Content Integrity

Every S3 object MUST have a corresponding content_hash in DynamoDB. On read, the consumer MUST verify:

import hashlib
import boto3
def verify_artifact_content(s3_key: str, expected_hash: str, bucket: str) -> bool:
"""Verify S3 artifact content matches its DynamoDB hash."""
s3 = boto3.client("s3")
obj = s3.get_object(Bucket=bucket, Key=s3_key)
content = obj["Body"].read()
actual_hash = f"sha256:{hashlib.sha256(content).hexdigest()}"
return actual_hash == expected_hash

← 8. Compensation · Section index · 10. Integration with AIGP and ACP →