Domain Types & Contracts¶

Core data structures and exceptions defining the Sovereign Dark Factory domain model.

`dark_factory.domain.types` ¶

Domain data contracts and types for Sovereign Dark Factory.

Classes¶

`RunStatus` ¶

Bases: StrEnum

Lifecycle status of a dark factory run.

Source code in dark_factory/domain/types.py

class RunStatus(StrEnum):
    """Lifecycle status of a dark factory run."""

    PENDING = "PENDING"
    CREATING_SANDBOX = "CREATING_SANDBOX"
    AGENT_RUNNING = "AGENT_RUNNING"
    VERIFYING = "VERIFYING"
    SELF_HEALING = "SELF_HEALING"
    AWAITING_REVIEW = "AWAITING_REVIEW"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"
    FAILED = "FAILED"
    TIMED_OUT = "TIMED_OUT"
    CANCELLED = "CANCELLED"

    @property
    def is_terminal(self) -> bool:
        return self in (
            RunStatus.APPROVED,
            RunStatus.REJECTED,
            RunStatus.FAILED,
            RunStatus.TIMED_OUT,
            RunStatus.CANCELLED,
        )

`TaskSpec` `dataclass` ¶

Specification for an autonomous coding task.

Source code in dark_factory/domain/types.py

@dataclass
class TaskSpec:
    """Specification for an autonomous coding task."""

    repo_path: str
    task_prompt: str
    base_rev: str = "HEAD"
    agent: str = "local-coder"
    model: str = "qwen2.5-coder:14b"
    verification_steps: list[VerificationStep] = field(default_factory=list)
    allow_no_verify: bool = False
    protected_paths: list[str] = field(default_factory=list)
    allow_gate_edits: bool = False
    max_healing_attempts: int = 3
    timeout_minutes: float = 30
    metadata: dict[str, Any] = field(default_factory=dict)
    skip_adversarial: bool = False
    mutate_adversarial: bool = False
    skip_analysis: bool = False
    planner_model: str | None = None
    skip_plan: bool = False
    target_files: list[str] = field(default_factory=list)  # files the agent edits; inlined in full as context
    allow_out_of_scope: bool = False  # bypass the patch-scope gate (placeholder paths, files outside target_files)

`StepExecution` `dataclass` ¶

Outcome of a single verification step or command execution.

Source code in dark_factory/domain/types.py

@dataclass
class StepExecution:
    """Outcome of a single verification step or command execution."""

    step_id: str
    exit_code: int
    stdout: str
    stderr: str
    duration_sec: float
    timed_out: bool = False

    @property
    def passed(self) -> bool:
        return self.exit_code == 0 and not self.timed_out

`VerificationStep` `dataclass` ¶

A single deterministic verification gate.

Source code in dark_factory/domain/types.py

@dataclass(frozen=True)
class VerificationStep:
    """A single deterministic verification gate."""

    id: str
    argv: list[str]
    timeout_sec: int = 300
    mandatory: bool = True
    description: str = ""

`EvidenceManifest` `dataclass` ¶

Immutable audit record and evidence of a completed or paused run.

Source code in dark_factory/domain/types.py

@dataclass
class EvidenceManifest:
    """Immutable audit record and evidence of a completed or paused run."""

    run_id: str
    status: RunStatus
    repo_path: str
    base_rev: str
    created_at: str
    completed_at: str | None = None
    resulting_rev: str | None = None
    patch_path: str | None = None
    patch_size_bytes: int = 0
    patch_sha256: str | None = None
    healing_attempts: int = 0
    verification_results: list[StepExecution] = field(default_factory=list)
    model_telemetry: ModelTelemetry | None = None
    operator_notes: str | None = None
    phase_timings: list[PhaseTiming] = field(default_factory=list)
    repeated_failure_streak: int = 0
    adversarial_report: AdversarialReport | None = None
    execution_plan: ExecutionPlan | None = None
    adversarial_test_code: str | None = None
    analysis_report: AnalysisReport | None = None

    @classmethod
    def create(
        cls,
        run_id: str,
        repo_path: str,
        base_rev: str,
        status: RunStatus = RunStatus.PENDING,
    ) -> EvidenceManifest:
        return cls(
            run_id=run_id,
            status=status,
            repo_path=repo_path,
            base_rev=base_rev,
            created_at=datetime.now(UTC).isoformat(),
        )

`ModelTelemetry` `dataclass` ¶

Telemetry captured from local inference engines (Ollama / Prism).

Source code in dark_factory/domain/types.py

@dataclass
class ModelTelemetry:
    """Telemetry captured from local inference engines (Ollama / Prism)."""

    engine: str
    model_name: str
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    duration_sec: float = 0.0
    tokens_per_sec: float = 0.0
    cost_usd: float = 0.0  # Always 0.0 for local models
    num_ctx: int | None = None  # context window requested from the engine (Ollama only)
    finish_reason: str | None = None  # why the last call stopped: `stop`, `length`, ... (None: engine did not say)
    max_tokens: int | None = None  # completion budget requested from the engine (Prism only)
    reasoning_chars: int = 0  # length of the engine's separate reasoning text; diagnostic only, never executed

`dark_factory.domain.errors` ¶

Domain error definitions for Sovereign Dark Factory.