vega-core Contract

This page defines the exact boundary between vega-backend and vega-core: how the backend invokes the engine, what it passes in, what it receives back, and how the vega-core Python API is structured internally.

Integration boundary

vega-core is always run as a subprocess — never imported directly by the backend process. The backend adapter (LocalVegaCoreService in adapters/engine/vega_core/) spawns vega-core/tests/manual/e2e_scan_debug.py with CLI arguments, reads a JSON event stream from stdout, and reads artifact JSON files from the artifact root after the process exits.

flowchart LR
    subgraph runner[ECS Runner / Local Process]
        LSvc[LocalVegaCoreService\nadapters/engine/vega_core/]
        Sub[subprocess:\ne2e_scan_debug.py]
        Codex[Codex CLI]
    end

    subgraph vega_core[vega-core internals]
        API[api/entrypoints.py\nplan_repo / run_scan / verify_and_patch]
        Orch[StageOrchestrator]
        Stages[Stages:\nplan, threat_model, audit, verify, patch]
    end

    ArtifactRoot[(artifact root\nlocal filesystem / S3)]

    LSvc -->|spawn subprocess\nCLI args| Sub
    Sub --> API --> Orch --> Stages
    Stages -->|events → stdout| Sub
    Sub -->|JSON lines| LSvc
    Stages --> ArtifactRoot
    Codex --> Stages

How the backend calls vega-core

`VegaCoreEngineAdapter` → `LocalVegaCoreService`

VegaCoreEngineAdapter (the port implementation) delegates to LocalVegaCoreService for all scan phases. LocalVegaCoreService is used in both local development and ECS production — the "Local" in the name refers to running on the same filesystem as the runner task, not that it is a development-only stub.

Each phase maps to a subprocess call with --enabled-stages:

`VegaCoreEngineAdapter` method	subprocess `--enabled-stages`
`run_plan(...)`	`plan`
`run_threat_model(...)`	`threat_model`
`run_audit(...)`	`audit`
`run_verification(...)`	`verify`

Subprocess CLI arguments

python e2e_scan_debug.py \
  --target        <source root directory>    \
  --artifact-root <artifact output dir>      \
  --repo-id       <repository_id>            \
  --scan-id       <scan_id>                  \
  --codex-bin     codex                      \
  --model         <model name>               \
  --max-workers-per-runner   <N>             \
  --max-fast-runners-per-scan <N>            \
  --depth         standard|quick|deep        \
  --enabled-stages plan,threat_model,audit   \
  --stream-json-to-stdout                    \
  [--reuse-plan-artifact <path>]             \
  [--threat-model-strategy codex_direct|llm_bug_class_selection|none]  \
  [--audit-strategy default|variant_analysis]\
  [--bug-class memory-safety ...]            \
  [--include-path src/ ...]

Event stream (stdout)

With --stream-json-to-stdout, the subprocess emits one JSON object per line on stdout as events occur. Each line is a CoreEvent:

{
  "kind": "scan_progress",
  "stage": "audit",
  "payload": { "completed": 3, "total": 10, "component": "src/parser.c" },
  "occurred_at": "2026-01-01T12:00:00Z"
}

LocalVegaCoreService reads these lines and passes each to the EngineEventSink callback, which stores them as DomainEvents.

Artifact output

After the subprocess exits (return code 0), artifacts are written as JSON files under --artifact-root/<repo_id>/<scan_id>/:

Artifact kind	File path	Contents
`plan`	`plan/<hash>.json`	Component manifest (files grouped by component)
`threat_model`	`threat_model/<hash>.json`	Threat model per component
`raw_findings`	`raw_findings/<hash>.json`	Undeduped audit findings
`deduplicated_bugs`	`deduplicated_bugs/<hash>.json`	Deduplicated `BugRecord` list
`audit_component_state`	`audit_component_state/<hash>.json`	Per-component audit progress
`verification_evidence`	`verification_evidence/<hash>.json`	Verification results
`patch`	`patch/<hash>.json`	Patch / remediation results

LocalVegaCoreService._engine_result_from_artifacts() reads these files and wraps them in EngineResult(artifact_refs={...}). The backend then uploads each artifact to S3 and records an ArtifactRecord.

vega-core Python API (internal)

Within the subprocess, e2e_scan_debug.py calls vega-core's public Python API. Backend developers working on the adapter need to understand this API to add or change how arguments are mapped.

Entry points (`api/entrypoints.py`)

from api import configure_core_runtime, plan_repo, run_scan, verify_and_patch

`configure_core_runtime`

def configure_core_runtime(
    *,
    config: Any | None = None,        # RuntimeConfig (from env if None)
    event_sink: Any | None = None,    # callable(CoreEvent) -> None
    cancellation_token: Any | None = None,
) -> None: ...

Must be called once per process before any scan function.

`plan_repo`

def plan_repo(request: PlanRequest) -> CoreResult: ...

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

Returns: CoreResult with artifacts["plan"] = ArtifactRef to the plan JSON.

`run_scan`

def run_scan(request: ScanRequest) -> CoreResult: ...

@dataclass(frozen=True)
class ScanRequest:
    repo_id: str
    scan_id: str
    source_snapshot: ArtifactRef
    plan_artifact: ArtifactRef | None = None
    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)

@dataclass(frozen=True)
class ScanIntent:
    include_paths: tuple[str, ...] = ()
    bug_classes: tuple[str, ...] = ()
    depth: Literal["quick", "standard", "deep"] = "standard"
    variant_context: str | None = None

@dataclass(frozen=True)
class ScanExecution:
    threat_model_strategy: str = "codex_direct"
    audit_strategy: str = "default"
    threat_model_artifact: ArtifactRef | None = None
    audit_resume_artifact: ArtifactRef | None = None
    audit_resume_policy: Literal["fresh", "skip_completed"] = "fresh"
    assigned_component_ids: tuple[str, ...] = ()
    audit_shard_id: str | None = None

Returns: CoreResult with: - artifacts["threat_model"], artifacts["raw_findings"], artifacts["deduplicated_bugs"] - data["selected_plugin"], data["plugin_selection"]

`verify_and_patch`

def verify_and_patch(request: VerificationRequest) -> CoreResult: ...

@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)

Returns: CoreResult with: - artifacts["patch"] — remediation results - artifacts["verification_evidence"] — per-finding verification status

`CoreResult`

@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

Always check result.status before accessing result.artifacts.

`ArtifactRef`

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

`RunPolicy`

@dataclass(frozen=True)
class RunPolicy:
    enabled_stages: tuple[StageName, ...] = ()
    agent_provider: str | None = None
    model: str | None = None
    max_workers_per_runner: int = 1
    max_fast_runners_per_scan: int = 1
    max_cost_cents: int | None = None
    stage_timeout_seconds: Mapping[StageName, int] = field(default_factory=dict)
    retry_count: int = 0
    verify_before_patch: bool = True
    metadata: Mapping[str, Any] = field(default_factory=dict)

Legacy fallback

VegaCoreEngineAdapter has a backward-compatibility fallback for older vega-core deployments. If the injected service does not have run_plan() (new API), it falls back to calling the old methods:

New method	Legacy fallback
`run_plan()`	`service.plan_source()`
`run_threat_model()`	`service.threat_model_source()`
`run_audit()`	`service.run_audit()` → `service.scan_source()`
`run_verification()`	`service.verify_findings()`

If no service is injected at all and no legacy_settings are provided, the adapter raises PortDependencyError("vega-core engine service is not configured").

Event contract

CoreEvent values emitted during a scan (see stdout stream above):

Kind	Action backend takes
`scan_started`	Update scan stage status
`scan_progress`	Update scan progress percentage
`scan_log`	Store debug log event
`finding_updated`	Upsert `FindingRecord` via `NormalizeAndUpsertFindingsUseCase`
`finding_verified`	Update finding verification status
`stage_started` / `stage_completed`	Stage-level progress bookmarks
`component_worker_started` / `_completed` / `_failed`	Per-component worker status
`scan_completed`	Engine audit loop finished
`scan_failed`	Engine encountered an error
`scan_cancelled`	Scan was cancelled cooperatively
`vega_core_started` / `vega_core_completed`	Subprocess lifecycle (from `LocalVegaCoreService`)

Unknown kinds are stored as generic domain events. Do not discard unknown events; they may become first-class in future releases.

Contract stability

The subprocess interface (e2e_scan_debug.py CLI args and artifact layout) is the stable backend boundary. The Python API (api/entrypoints.py, api/requests.py, api/results.py) is stable for vega-core internal consumers.

Internal vega-core types (framework/, orchestrator/, stages/) are implementation details and may change without notice.

vega-core Contract

Integration boundary

How the backend calls vega-core

VegaCoreEngineAdapter → LocalVegaCoreService

Subprocess CLI arguments

Event stream (stdout)

Artifact output

vega-core Python API (internal)

Entry points (api/entrypoints.py)

configure_core_runtime

plan_repo

run_scan

verify_and_patch

CoreResult

ArtifactRef

RunPolicy