Entry Points & API

vega-core exposes three product-facing operations through src/api/. These are the only functions vega-backend should call into this package. All other internal types and classes are implementation details.

Service wiring

Before calling any entrypoint, the caller must wire up the core services once per process. There are two wiring paths:

Option A — environment-driven (production)

from api import configure_core_runtime

configure_core_runtime(
    config=None,            # defaults to RuntimeConfig.from_env()
    event_sink=my_sink,     # callable(CoreEvent) -> None
    cancellation_token=tok, # CancellationToken | None
)

configure_core_runtime builds all services from environment variables (see Configuration) and stores them globally. You can also pass an explicit RuntimeConfig object.

Option B — explicit injection (testing / custom runtimes)

from api import configure_core_services

configure_core_services(
    event_bus=bus,
    artifact_store=store,
    cancellation=cancellation_manager,
    agent_runner=runner,
    plugin_resolver=resolver,
    triage=triage_service,
    policy=policy,          # RunPolicy | None
    stages=None,            # custom stage map | None
)

Call reset_core_services() between test cases to clear the global state.

Entrypoints

All three entrypoints are exported from api/__init__.py:

from api import plan_repo, run_scan, verify_and_patch

---

plan_repo(request: PlanRequest) -> CoreResult

Reads a source snapshot, generates a component plan, and stores a vega.plan artifact.

@dataclass(frozen=True)
class PlanRequest:
    repo_id: str
    source_snapshot: ArtifactRef   # reference to uploaded source archive
    include_paths: tuple[str, ...] = ()
    policy: RunPolicy = field(default_factory=RunPolicy)
    metadata: Mapping[str, Any] = field(default_factory=dict)

The result CoreResult.artifacts["plan"] is the ArtifactRef to store and pass back with future scan requests.

Stage sequence: [PLAN]
Transition: PlanTransition

---

run_scan(request: ScanRequest) -> CoreResult

Selects a plugin, runs threat modeling and auditing, and delivers raw findings to the shared triage service.

@dataclass(frozen=True)
class ScanRequest:
    repo_id: str
    scan_id: str
    source_snapshot: ArtifactRef
    plan_artifact: ArtifactRef
    intent: ScanIntent = field(default_factory=ScanIntent)
    execution: ScanExecution = field(default_factory=ScanExecution)
    policy: RunPolicy = field(default_factory=RunPolicy)
    metadata: Mapping[str, Any] = field(default_factory=dict)

ScanIntent — what to look for

ScanDepth = Literal["quick", "standard", "deep"]

@dataclass(frozen=True)
class ScanIntent:
    include_paths: tuple[str, ...] = ()   # path scope (empty = whole repo)
    bug_classes: tuple[str, ...] = ()     # bug class ids to focus on
    depth: ScanDepth = "standard"
    variant_context: str | None = None    # for variant_analysis audit strategy

ScanExecution — how to run it

AuditResumePolicy = Literal["fresh", "skip_completed"]

@dataclass(frozen=True)
class ScanExecution:
    threat_model_strategy: str = "codex_direct"
    # "codex_direct" | "llm_bug_class_selection" | "default"
    audit_strategy: str = "default"
    # "default" | "variant_analysis"
    threat_model_artifact: ArtifactRef | None = None
    # pass a cached threat model to skip the threat-model stage
    audit_resume_artifact: ArtifactRef | None = None
    # pass a vega.audit_component_state to resume a previous audit
    audit_resume_policy: AuditResumePolicy = "fresh"
    assigned_component_ids: tuple[str, ...] = ()
    # restrict audit to these component ids (sharding)
    audit_shard_id: str | None = None

Threat model caching

When execution.threat_model_artifact is set, the orchestrator validates the cached threat model's fingerprint against the current source, plan, plugin, policy, prompt template, and strategy. A valid cache skips the threat-model stage entirely. A stale cache fails the scan with stale_threat_model_cache.

Audit resume

When execution.audit_resume_artifact is set with audit_resume_policy="skip_completed", the audit stage loads the saved vega.audit_component_state, skips completed components, and preserves their findings.

PR review mode

PR review context is passed as metadata["pr_review"] — a JSON object with PR number, title, URL, base/head refs, changed files, and unified diff patch. AuditStage renders a dedicated review-focus prompt block from this metadata. Domain plugins are selected independently and still contribute their normal sections.

Stage sequence: [PLAN, THREAT_MODEL, AUDIT] when no plan_artifact is provided; [THREAT_MODEL, AUDIT] when plan_artifact is provided; [AUDIT] when a valid cached threat model is supplied.
Transition: ScanTransition
Result: CoreResult.data["deduplicated_bugs"] contains triage output; raw findings are also in CoreResult.data["raw_findings"].

---

verify_and_patch(request: VerificationRequest) -> CoreResult

Verifies deduplicated bugs and produces patch artifacts.

@dataclass(frozen=True)
class VerificationRequest:
    repo_id: str
    verification_id: str
    deduplicated_bugs: tuple[Mapping[str, Any], ...]
    source_snapshot: ArtifactRef
    plan_artifact: ArtifactRef | None = None
    policy: RunPolicy = field(default_factory=RunPolicy)
    metadata: Mapping[str, Any] = field(default_factory=dict)

Pass the deduplicated_bugs from a previous run_scan result (or from the backend's persisted bug records). The selected plugin id/version from the original scan should be carried in metadata["plugin_id"] and metadata["plugin_version"] so the verify/patch stages use consistent plugin behavior without re-resolving.

Stage sequence: [VERIFY, PATCH]
Transition: VerifyPatchTransition
Result: CoreResult.artifacts["patch"] is the vega.remediation_results artifact ref.

---

CoreResult

All three entrypoints return:

@dataclass(frozen=True)
class CoreResult:
    status: StageStatus           # "completed" | "cancelled" | "failed"
    run_id: str
    artifacts: Mapping[str, ArtifactRef] = field(default_factory=dict)
    data: Mapping[str, Any] = field(default_factory=dict)
    error: StageError | None = None

Check result.status first. If status == "failed", inspect result.error for code, message, and retryable.

ArtifactRef

All artifact inputs and outputs use typed references:

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

Refs are opaque to the caller — store them in the database and pass them back to vega-core. Do not construct URIs manually.

Example: full scan flow

from api import configure_core_runtime, plan_repo, run_scan, verify_and_patch
from api.requests import (
    PlanRequest, ScanRequest, ScanIntent, ScanExecution, VerificationRequest
)
from artifacts.refs import ArtifactRef
from framework.policy import RunPolicy

configure_core_runtime(event_sink=my_event_sink)

# 1. Plan the repository (done once per snapshot)
plan_result = plan_repo(PlanRequest(
    repo_id="repo_abc",
    source_snapshot=source_ref,
))
plan_ref = plan_result.artifacts["plan"]

# 2. Scan
scan_result = run_scan(ScanRequest(
    repo_id="repo_abc",
    scan_id="scan_xyz",
    source_snapshot=source_ref,
    plan_artifact=plan_ref,
    intent=ScanIntent(depth="standard"),
    execution=ScanExecution(threat_model_strategy="codex_direct"),
))
bugs = scan_result.data.get("deduplicated_bugs", ())

# 3. Verify and patch
patch_result = verify_and_patch(VerificationRequest(
    repo_id="repo_abc",
    verification_id="verify_xyz",
    deduplicated_bugs=tuple(bugs),
    source_snapshot=source_ref,
    plan_artifact=plan_ref,
    metadata={"plugin_id": scan_result.data.get("selected_plugin_id")},
))
patch_ref = patch_result.artifacts.get("patch")