Skip to content

Patch Server

Patch Server — mvp.patch_server

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview

Server-side patch management. Receives bug reports from diagnostic_collector via patch_client, generates patches from source analysis, cryptographically signs them with Ed25519 keys, and deploys patches through a CI/CD pipeline (git commit, test, rebuild).

Uses SQLite-backed persistence for bug reports, patches, feedback, and audit rows. Tracks bugs through a full lifecycle: received, analyzing, patch_ready, deployed, closed.

When to use:

  • Operating a self-healing patch pipeline that receives diagnostics and produces signed fixes
  • Deploying patches through an automated test-commit-rebuild cycle
  • Tracking bug-to-fix lifecycle with cryptographic provenance
  • Checking patch-server readiness with the side-effect-free capabilities op before invoking external CLIs
  • Running the operator-controlled flow: receive -> generate -> sign -> approve -> deploy
  • Using validate to produce the canonical reliability label: verified, qualified-draft, or blocked-escalated

Example:

from mvp.patch_server import PatchServerBlock, PatchServerInput

block = PatchServerBlock()
result = block.infer(PatchServerInput(
    op="receive",
    diagnostic_report={"component": "goal_engine", "error": "depth exceeded"},
))
# result.value -> PatchServerOutput with bug_id, status="received"

Caveats and known limitations:

  • Requires G6_SIGNING_KEY_PATH env var pointing to an Ed25519 PEM file; signing fails without it
  • Auto-patch generation delegates to generate_patch(); patch quality depends on source analysis heuristics
  • capabilities reports Claude, Codex, signing key, cryptography, httpx/GitHub, git, pytest, and SQLite availability without invoking patch generation, deployment, network calls, or source mutation
  • Codex review is advisory only; human approval remains the deployment gate
  • approve records reviewer identity and status in the audit trail, but production multi-tenant authentication is not specified here
  • Partial but useful outcomes are surfaced as completion_state="qualified-draft" with warning_card and evidence; refusals and failures must not be treated as verified
  • Deployment strategy is controlled by settings.g6_patch_deploy_strategy; can be set to "disabled"
  • Deploy requires a signed patch and a component test directory; components without tests must be covered before auto-deploy
  • _revert_source_files() uses subprocess git checkout after apply/test/commit failures; may fail on merge commits or submodules
  • There is no MCP tool surface for patch_server; keep operator actions on the block/API surface until a constrained read-only MCP policy is approved

Works well with: patch_client, diagnostic_collector, self_training

Public API

WorktreeCleanupTarget

Field Type Default
path str required
label str required
created_at float required

PatchValidationRequest

Field Type Default
touched_paths tuple[str, ...] required
diff str required
tests_passed bool False
approval bool False
docs_touched bool False
source_vis_touched bool False
phase_id str ''

PatchValidationResult

Field Type Default
label Literal['verified', 'qualified-draft', 'blocked-escalated'] required
diff_summary str required
risk_reasons tuple[str, ...] ()
docs_impact bool False
source_vis_impact bool False
rollback_plan str 'git revert '
block_reasons tuple[str, ...] field(default_factory=tuple)
friendliness_review FriendlinessPatchReview \| None None
required_human_decision str ''
observation_event_id str ''

Methods:

friendliness_badge() -> dict

to_dict() -> dict

PatchRetryState

Field Type Default
run_id str required
parent_run_id str ''
retry_count int 0
worktree_id str ''
label PatchLabel 'qualified-draft'
human_review_required bool False
reason str ''

BugReport(BaseModel)

Serialized bug report received from a client.

Field Type Default
bug_id str required
diagnostic_report dict Field(default_factory=dict)
received_at str ''
status Literal['received', 'analyzing', 'patch_ready', 'deployed', 'closed', 'failed'] 'received'
patch_id str \| None None
notes str ''

PatchGenerationRequest(BaseModel)

Request to generate a patch for a given bug.

Field Type Default
bug_id str required
auto bool True

PatchStatus(BaseModel)

Tracks a generated patch through its lifecycle.

Field Type Default
patch_id str required
bug_id str required
component str required
target_version str required
status Literal['generating', 'ready', 'delivered', 'failed'] 'generating'
created_at str ''
diff str ''
files dict[str, str] Field(default_factory=dict)
sha256 dict[str, str] Field(default_factory=dict)
signature str ''
error str ''
deploy_status Literal['', 'testing', 'committed', 'building', 'deployed', 'deploy_failed'] ''
deploy_branch str ''
deploy_image_tag str ''
review_status Literal['pending', 'approved', 'rejected'] 'pending'
reviewer_id str ''
confidence float 0.0

PatchServerInput(BaseModel)

Input for PatchServerBlock.

Field Type Default
op Literal['receive', 'generate', 'sign', 'list', 'get', 'deploy', 'feedback', 'validate', 'capabilities', 'approve'] required
diagnostic_report dict Field(default_factory=dict)
feedback dict Field(default_factory=dict)
bug_id str ''
auto bool True
patch_id str ''
version str ''
component str ''
touched_paths list[str] Field(default_factory=list)
diff str ''
tests_passed bool False
approval bool False
docs_touched bool False
source_vis_touched bool False
phase_id str ''
reviewer_id str ''
review_status Literal['', 'approved', 'rejected'] ''
confidence float 0.0

PatchServerOutput(BaseModel)

Output from PatchServerBlock.

Field Type Default
message str ''
bug_id str ''
patch_id str ''
bug_report BugReport \| None None
patch_status PatchStatus \| None None
patches list[dict] Field(default_factory=list)
feedback list[dict] Field(default_factory=list)
validation_result dict Field(default_factory=dict)
friendliness_badge dict Field(default_factory=dict)
friendliness_review dict Field(default_factory=dict)
degraded bool False
degradation_reason str \| None None
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict[str, Any] \| None None
evidence dict[str, Any] Field(default_factory=dict)
request_id str ''
task_id str ''
run_id str ''

PatchServerBlock(AIBlock[PatchServerInput, PatchServerOutput, None])

Server-side patch management block.

Field Type Default
name str 'patch_server'
db_path str ''
source_root str ''

Methods:

infer(data: PatchServerInput) -> Result[PatchServerOutput]

SecureConfigError(RuntimeError)

Raised for signing key configuration errors. Masks sensitive paths.

SimilarityPolicy

Field Type Default
block_threshold float 0.8
warn_threshold float 0.6

Functions

check_budget_for_patch(estimated_prompt_chars: int = 8000) -> Result[bool]

Check that sufficient budget remains for a patch generation.

record_patch_usage(prompt_chars: int, response_chars: int, component: str, bug_id: str, session_id: str = '') -> None

Record token usage for a patch generation to the central ledger.

cleanup_worktree(target: WorktreeCleanupTarget, now: float | None = None, ttl_seconds: int = 24 * 60 * 60, force: bool = False) -> WorktreeCleanupResult

sanitize_bug_report(report: dict) -> dict

Sanitize untrusted bug report fields before prompt construction.

validate_patch(request: PatchValidationRequest) -> PatchValidationResult

next_retry_state(parent_run_id: str, previous_retry_count: int, worktree_id: str, new_run_id: str) -> PatchRetryState

normalized_jaccard(left: str, right: str) -> float

similarity_verdict(patch_hunk: str, exemplar: str, copying_allowed: bool, policy: SimilarityPolicy | None = None) -> Literal['allow', 'warn', 'block']