Artifact Store

Stages exchange ArtifactRef values — opaque typed references to stored artifacts. The backing store may be local disk or S3. Stage code is identical for both.

Location

src/artifacts/

ArtifactRef

@dataclass(frozen=True)
class ArtifactRef:
    uri: str             # "file:///abs/path" or "s3://bucket/key"
    kind: str            # artifact kind string (e.g. "vega.plan")
    content_type: str    # MIME type (e.g. "application/json")
    metadata: Mapping[str, Any] = field(default_factory=dict)

    @property
    def scheme(self) -> str: ...          # "file" or "s3"

    @property
    def is_s3(self) -> bool: ...

    @property
    def is_file(self) -> bool: ...

    def with_metadata(self, metadata: Mapping) -> "ArtifactRef": ...
    def require_s3(self) -> "S3ArtifactLocation": ...

@dataclass(frozen=True)
class S3ArtifactLocation:
    bucket: str
    key: str

ArtifactRef values are serializable. Store them in the backend database as JSON blobs and pass them back to vega-core in subsequent requests.

ArtifactStore protocol

All stores implement:

class ArtifactStore(Protocol):
    def read(self, ref: ArtifactRef) -> bytes: ...
    def read_json(self, ref: ArtifactRef) -> Mapping[str, Any]: ...
    def write(
        self,
        *,
        context: RunContext,
        kind: str,
        content: bytes,
        content_type: str,
        metadata: Mapping[str, Any] | None = None,
    ) -> ArtifactRef: ...
    def write_json(
        self,
        *,
        context: RunContext,
        kind: str,
        value: Mapping[str, Any],
        metadata: Mapping[str, Any] | None = None,
    ) -> ArtifactRef: ...
    def exists(self, ref: ArtifactRef) -> bool: ...

Stages access the store through BaseStage.read_json_artifact(context, input, key) and BaseStage.write_json_artifact(context, kind=..., value=...).

LocalArtifactStore

Stores artifacts as JSON files under a local directory. Used in development and single-machine deployments.

Key layout:

<artifact_root>/
  <repo_id>/
    <run_id>/
      plan.json
      threat_model.json
      raw_findings.json
      audit_component_state.json
      deduplicated_bugs.json
      verification_results.json
      remediation_results.json
      codex_run_<hash>.log

URIs use file:// scheme:

file:///artifacts/repo_abc/scan_xyz/plan.json

Configure with VEGA_ARTIFACT_STORE=local and VEGA_ARTIFACT_ROOT=./artifacts.

S3ArtifactStore

Stores artifacts in S3. Used in production.

s3://<bucket>/<prefix>/<repo_id>/<run_id>/plan.json

For scan-scoped isolation, S3ArtifactStore can assume an IAM role via STS before writing artifacts for a scan. This uses the credentials variables:

VEGA_S3_SCAN_ROLE_ARN=arn:aws:iam::123456789:role/vega-scan
VEGA_S3_SCAN_POLICY_JSON='{"Version":"2012-10-17","Statement":[...]}'
VEGA_S3_SCAN_SESSION_DURATION_SECONDS=3600

URIs use s3:// scheme:

s3://vega-artifacts-prod/scans/repo_abc/scan_xyz/plan.json

Artifact kinds

Kind	Artifact key	content_type
vega.plan	plan	application/json
vega.threat_model	threat_model	application/json
vega.raw_findings	raw_findings	application/json
vega.audit_component_state	audit_component_state	application/json
vega.deduplicated_bugs	deduplicated_bugs	application/json
vega.verification_results	verification_evidence	application/json
vega.remediation_results	patch	application/json
vega.codex_run_log	codex_run_log	text/plain

URI helpers

from artifacts.refs import (
    parse_s3_uri,           # "s3://bucket/key" → S3ArtifactLocation
    build_s3_uri,           # bucket, key → "s3://bucket/key"
    join_s3_uri,            # base_uri, *parts → extended URI
    normalize_artifact_key, # sanitize key segment
    normalize_artifact_prefix,
    safe_artifact_segment,  # assert no path traversal
)

Do not construct artifact URIs manually in backend code. Backend should treat ArtifactRef.uri as opaque and pass it back verbatim.

How BaseStage accesses artifacts

class BaseStage:
    def get_artifact_ref(self, input: StageInput, key: str) -> ArtifactRef:
        # looks up key in input.artifacts or input.source_snapshot/plan_artifact
        ...

    def read_json_artifact(
        self, context: RunContext, input: StageInput, key: str
    ) -> Mapping[str, Any]:
        ref = self.get_artifact_ref(input, key)
        return context.artifacts.read_json(ref)

    def write_json_artifact(
        self, context: RunContext, *, kind: str, value: Mapping, metadata=None
    ) -> ArtifactRef:
        return context.artifacts.write_json(
            context=context, kind=kind, value=value, metadata=metadata
        )

Stages should use these helpers rather than accessing context.artifacts directly.

Artifact validation

BaseStage validates:

• Required artifacts — declared in required_artifacts must be present in input.artifacts before execute(...) is called. Missing required artifacts raise a non-retryable StageArtifactError.
• Produced artifacts — declared in produced_artifacts must be present in the returned StageOutput.artifacts. Missing produced artifacts raise a StageArtifactError after execute(...) returns.

Path safety

safe_artifact_segment(value) rejects values containing .., /, or other path-traversal characters. All artifact key segments are sanitized before use in URIs. Stages should not construct paths from user-controlled strings.

Switching between local and S3

# Local (dev)
VEGA_ARTIFACT_STORE=local
VEGA_ARTIFACT_ROOT=./data/artifacts

# S3 (prod)
VEGA_ARTIFACT_STORE=s3
VEGA_ARTIFACT_S3_BUCKET=vega-artifacts-prod
VEGA_ARTIFACT_S3_PREFIX=scans/

No code change required — the store is constructed by build_artifact_store(config).