Deploy Core¶
Deploy Core — mvp.deploy_core
Cluster: Core Infrastructure | Type: component | MCP Tools: None
Overview¶
Shared protocol layer and data types for the entire G6 deployment pipeline. Defines ImageSpec, DeployTarget, DeployResult, SecretRef, and RolloutPolicy frozen dataclasses used by every provider, plus structural Protocol interfaces (ContainerRuntime, Orchestrator, InfraTarget) that enforce a consistent API surface. DeployCoreBlock dispatches stages through a provider registry, persists successful and failed DeployRecord entries, and reports provider capability gaps with completion_state, warning_card, and evidence metadata instead of synthetic success. Also provides port-availability checking and a safe run_command subprocess helper used across all deployment components.
Launch readiness caveat
deploy_core is a provider-facing reliability layer, not a complete self-serve deployment product. It improves operational safety through registry-backed dispatch, structured failures, dry-run command previews, bounded rollout polling, port checks, rollback capability detection, and deployment history, but first-customer readiness still depends on concrete provider adapters, CLI/REST wiring, Dockerfile correctness, environment configuration, and a clean-machine install-to-health-check smoke test.
When to use:
- Building a new deployment provider — implement
ContainerRuntimeorInfraTargetand it slots in automatically - Constructing
ImageSpec/DeployTargetobjects to pass to any provider block - Checking whether a local port is free before starting a service
Example:
from mvp.deploy_core.schema import ImageSpec, DeployTarget, RolloutPolicy
from mvp.deploy_core.ports import find_available_port
port = find_available_port(8000)
image = ImageSpec(name="g6-rest", tag="latest")
target = DeployTarget(
provider="docker",
port=port,
rollout_policy=RolloutPolicy(timeout_sec=120, failure_action="rollback"),
)
Works well with: deploy_docker, deploy_aws, deploy_k8s, cicd
Public API¶
DeployCoreInput(BaseModel)¶
Input for deployment pipeline operations.
| Field | Type | Default |
|---|---|---|
op | str | required |
image_name | str | '' |
image_tag | str | 'latest' |
registry | str | '' |
provider | str | 'docker' |
base_name | str | '' |
stages | list[str] | Field(default_factory=lambda: ['build']) |
dry_run | bool | False |
timeout_sec | int | 300 |
parameters | dict[str, Any] | Field(default_factory=dict) |
DeployCoreOutput(BaseModel)¶
Output from deployment pipeline operations.
| Field | Type | Default |
|---|---|---|
op | str | '' |
ok | bool | True |
message | str | '' |
stage | str | '' |
duration_sec | float | 0.0 |
records | list[dict[str, Any]] | Field(default_factory=list) |
metadata | dict[str, Any] | Field(default_factory=dict) |
DeployCoreBlock(AIBlock)¶
Orchestrates deployment pipeline stages with audit trail.
Constructor:
| Parameter | Type | Default |
|---|---|---|
state_db_path | str | '' |
planner | Any | None |
Methods:
infer(input: DeployCoreInput) -> Result[DeployCoreOutput]¶
PortUnavailableError(Exception)¶
Raised when no port is available in the requested range.
Constructor:
| Parameter | Type | Default |
|---|---|---|
preferred | int | required |
range_size | int | required |
conflicts | list[PortCheck] | required |
ContainerRuntime(Protocol)¶
Build, push, run, and stop container images.
Methods:
build(image: ImageSpec, context_dir: str) -> DeployResult¶
push(image: ImageSpec, registry: str) -> DeployResult¶
run(image: ImageSpec, ports: dict[int, int], env: dict[str, str]) -> DeployResult¶
stop(container_id: str) -> DeployResult¶
Orchestrator(Protocol)¶
Generate and apply orchestration manifests.
Methods:
generate_manifests(image: ImageSpec, target: DeployTarget) -> DeployResult¶
apply(manifest_path: str) -> DeployResult¶
status(namespace: str) -> DeployResult¶
rollback(deployment: str, namespace: str) -> DeployResult¶
InfraTarget(Protocol)¶
Deploy container images to infrastructure targets.
Methods:
push_image(image: ImageSpec) -> DeployResult¶
deploy(image: ImageSpec, target: DeployTarget) -> DeployResult¶
status(target: DeployTarget) -> DeployResult¶
teardown(target: DeployTarget) -> DeployResult¶
wait_healthy(image: ImageSpec, target: DeployTarget) -> DeployResult¶
SecretRef¶
A reference to a secret in an external backend.
| Field | Type | Default |
|---|---|---|
name | str | required |
backend | str | required |
key | str | required |
env_var_name | str | required |
RolloutPolicy¶
Controls how a provider waits for a healthy deployment.
| Field | Type | Default |
|---|---|---|
timeout_sec | int | 300 |
poll_interval_sec | int | 10 |
min_healthy_percent | int | 100 |
failure_action | str | 'rollback' |
ImageSpec¶
Container image specification for a G6 base.
| Field | Type | Default |
|---|---|---|
name | str | required |
tag | str | 'latest' |
registry | str | '' |
dockerfile | str | 'Dockerfile' |
base_name | str | '' |
version | str | '' |
DeployTarget¶
Where to deploy a container image.
| Field | Type | Default |
|---|---|---|
provider | str | required |
region | str | '' |
cluster | str | '' |
namespace | str | 'default' |
host | str | '' |
port | int | 0 |
service_name | str | '' |
secrets | tuple[SecretRef, ...] | () |
rollout_policy | RolloutPolicy | field(default_factory=RolloutPolicy) |
health_endpoint | str | '' |
DeployResult¶
Result of a deployment pipeline stage.
| Field | Type | Default |
|---|---|---|
ok | bool | required |
stage | str | required |
message | str | '' |
artifact | str | '' |
duration_sec | float | 0.0 |
resource_id | str | '' |
failure_category | str | '' |
Methods:
stdout() -> str¶
Compatibility alias for subprocess-style command output.
stderr() -> str¶
Compatibility alias for subprocess-style command errors.
DeployRecord¶
A completed deployment event recorded for audit and agent queries.
| Field | Type | Default |
|---|---|---|
provider | str | required |
image_name | str | required |
image_tag | str | required |
target_id | str | required |
status | str | required |
timestamp | float | field(default_factory=time.time) |
manifest_artifact | str | '' |
rollout_duration_sec | float | 0.0 |
secrets_used | list[str] | field(default_factory=list) |
PortCheck¶
Result of a port availability check.
| Field | Type | Default |
|---|---|---|
port | int | required |
available | bool | required |
pid | int \| None | None |
process_name | str | '' |
reason | str | '' |
DeployStateStore(Protocol)¶
Protocol for recording and querying deployment history.
Methods:
record(deploy_record: DeployRecord) -> None¶
get_latest(provider: str, target_id: str) -> DeployRecord | None¶
list_history(provider: str, target_id: str, limit: int = 10) -> list[DeployRecord]¶
SqliteDeployStateStore¶
SQLite-backed deploy state store. DB created automatically on first use.
Constructor:
| Parameter | Type | Default |
|---|---|---|
db_path | str | '' |
Methods:
record(deploy_record: DeployRecord) -> None¶
get_latest(provider: str, target_id: str) -> DeployRecord | None¶
list_history(provider: str, target_id: str, limit: int = 10) -> list[DeployRecord]¶
Functions¶
check_port(port: int, host: str = '127.0.0.1') -> PortCheck¶
Check if a port is available by attempting a socket bind.
find_available_port(preferred: int, range_size: int = 10, host: str = '127.0.0.1') -> int¶
Return preferred port if free, otherwise scan preferred+1..+range_size.
wait_for_rollout(check_fn: Callable[[], bool], policy: RolloutPolicy, stage: str = 'rollout') -> DeployResult¶
Poll check_fn until it returns True or the timeout is exceeded.
run_command(cmd: list[str], dry_run: bool = False, timeout: int = 300, cwd: str | None = None, stdin_text: str | None = None) -> DeployResult¶
Execute a subprocess command with timeout and structured result.