domain.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/domain.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_domain_queue.py and test_domain_plan.py"
"""
Domain-level registry mutation API.
 
Purpose
-------
Expose typed queue and plan-lifecycle functions that emit canonical
operation-log rows through :class:`oc_chat.registry.Registry`.
 
Responsibilities
----------------
- Build versioned payloads for the queue and plan operation catalogs.
- Enforce domain-level preconditions that belong above the seam.
- Keep callers state-version and operation-uuid unaware.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Every public mutating function calls ``Registry.execute_domain_command``.
- This module performs no SQL, git, subprocess, filesystem, or clock work.
- Payload version is always 1 for the v0.2.3 operation catalog.
 
Dependencies
------------
- typing (literal operation arguments)
- collections.abc (the hub message payload mapping)
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.api (Registry facade)
- oc_chat.registry.backends.base (OperationLogRow return type)
"""
from __future__ import annotations
 
from collections.abc import Mapping
from typing import Literal
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.backends.base import OperationLogRow
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
PlanStatus = Literal[
    "draft",
    "pending",
    "implementing",
    "ready_for_merge",
    "integrating",
    "completed",
    "partial",
    "failed",
    "superseded",
]
ActorType = Literal["human_operator", "orchestrator_agent", "worker_agent"]
QueueEnqueueReason = Literal[
    "initial enqueue",
    "promoted_from_partial",
    "restored_from_aborted_round",
]
QueueDequeueReason = Literal[
    "transition_to_failed",
    "transition_to_integrating",
    "operator_dequeue",
    # Plan 259 P4, closing a plan-258 residual. The set had no member
    # meaning "handed to an instance": ``operator_dequeue`` says WHO acted
    # rather than why, and the other two rewrite the plan's status, which a
    # dispatch must not. Recording a dispatch honestly needs this word.
    "dispatched",
]
PlanStatusSource = Literal["worker_self", "orchestrator", "round_lifecycle"]
 
QUEUE_ENQUEUE_REASONS = {
    "initial enqueue",
    "promoted_from_partial",
    "restored_from_aborted_round",
}
QUEUE_DEQUEUE_REASONS = {
    "transition_to_failed",
    "transition_to_integrating",
    "operator_dequeue",
    "dispatched",
}
PLAN_STATUS_SOURCES = {"worker_self", "orchestrator", "round_lifecycle"}
PLAN_STATUSES = {
    "draft",
    "pending",
    "implementing",
    "ready_for_merge",
    "integrating",
    "completed",
    "partial",
    "failed",
    "superseded",
}
 
 
def queue_enqueue(
    registry: Registry,
    *,
    plan_id: str,
    slot: str,
    position: int,
    reason: QueueEnqueueReason,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``queue.enqueue`` operation."""
 
    if reason not in QUEUE_ENQUEUE_REASONS:
        raise ValueError(f"invalid queue enqueue reason: {reason}")
    _require_positive_position(position)
    return registry.execute_domain_command(
        "queue.enqueue",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "slot": slot,
            "position": position,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def queue_move(
    registry: Registry,
    *,
    plan_id: str,
    from_slot: str,
    to_slot: str,
    from_position: int,
    to_position: int,
    reason: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``queue.move`` operation."""
 
    _require_positive_position(from_position)
    _require_positive_position(to_position)
    if not reason.strip():
        raise ValueError("queue move reason is required")
    return registry.execute_domain_command(
        "queue.move",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "from_slot": from_slot,
            "to_slot": to_slot,
            "from_position": from_position,
            "to_position": to_position,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def queue_dequeue(
    registry: Registry,
    *,
    plan_id: str,
    slot: str,
    position: int,
    reason: QueueDequeueReason,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``queue.dequeue`` operation."""
 
    if reason not in QUEUE_DEQUEUE_REASONS:
        raise ValueError(f"invalid queue dequeue reason: {reason}")
    _require_positive_position(position)
    return registry.execute_domain_command(
        "queue.dequeue",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "slot": slot,
            "position": position,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def queue_complete(
    registry: Registry,
    *,
    plan_id: str,
    round_no: int,
    merge_commit_sha: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``queue.complete`` operation."""
 
    if round_no <= 0:
        raise ValueError("round_no must be positive")
    return registry.execute_domain_command(
        "queue.complete",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "round_no": round_no,
            "merge_commit_sha": merge_commit_sha,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
        related_round=round_no,
    )
 
 
def plan_create(
    registry: Registry,
    *,
    plan_id: str,
    plan_number: int,
    title: str,
    body_path: str,
    body_sha256: str,
    body_observed_at: str | None = None,
    plan_type: str | None = None,
    model_override: str | None = None,
    thinking_override: str | None = None,
    session_override: str | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``plan.create`` operation."""
 
    if plan_number <= 0:
        raise ValueError("plan_number must be positive")
    return registry.execute_domain_command(
        "plan.create",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "plan_number": plan_number,
            "title": title,
            "plan_type": plan_type,
            "body_path": body_path,
            "body_sha256": body_sha256,
            "body_observed_at": body_observed_at,
            "model_override": model_override,
            "thinking_override": thinking_override,
            "session_override": session_override,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def plan_set_status(
    registry: Registry,
    *,
    plan_id: str,
    from_status: PlanStatus,
    to_status: PlanStatus,
    source: PlanStatusSource,
    note: str | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``plan.set_status`` operation."""
 
    _require_status(from_status, "from_status")
    _require_status(to_status, "to_status")
    if source not in PLAN_STATUS_SOURCES:
        raise ValueError(f"invalid plan status source: {source}")
    if to_status in {"partial", "failed"} and not (note and note.strip()):
        raise ValueError("note is required for partial and failed status transitions")
    return registry.execute_domain_command(
        "plan.set_status",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "from_status": from_status,
            "to_status": to_status,
            "source": source,
            "note": note,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def plan_assign(
    registry: Registry,
    *,
    plan_id: str,
    from_slot: str | None,
    to_slot: str,
    branch: str,
    worktree: str,
    queue_position: int,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``plan.assign`` operation."""
 
    _require_positive_position(queue_position)
    return registry.execute_domain_command(
        "plan.assign",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "from_slot": from_slot,
            "to_slot": to_slot,
            "branch": branch,
            "worktree": worktree,
            "queue_position": queue_position,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def plan_supersede(
    registry: Registry,
    *,
    plan_id: str,
    superseded_by_plan_id: str,
    reason: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``plan.supersede`` operation."""
 
    if not reason.strip():
        raise ValueError("supersede reason is required")
    return registry.execute_domain_command(
        "plan.supersede",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "superseded_by_plan_id": superseded_by_plan_id,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def plan_reimport(
    registry: Registry,
    *,
    plan_id: str,
    previous_body_sha256: str,
    new_body_sha256: str,
    previous_body_observed_at: str,
    new_body_observed_at: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``plan.reimport`` operation."""
 
    return registry.execute_domain_command(
        "plan.reimport",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "previous_body_sha256": previous_body_sha256,
            "new_body_sha256": new_body_sha256,
            "previous_body_observed_at": previous_body_observed_at,
            "new_body_observed_at": new_body_observed_at,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
    )
 
 
def auth_register_key(
    registry: Registry,
    *,
    slot: str,
    hmac_key_id: str,
    secret_fingerprint: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit an ``auth.register_key`` audit operation."""
 
    return registry.execute_domain_command(
        "auth.register_key",
        actor_slot,
        {
            "version": 1,
            "slot": slot,
            "hmac_key_id": hmac_key_id,
            "secret_fingerprint": secret_fingerprint,
        },
        actor_type=actor_type,
        command=command,
    )
 
 
def auth_reset(
    registry: Registry,
    *,
    slot: str,
    previous_key_ids_revoked: list[str],
    new_key_id: str,
    reason: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit an ``auth.reset`` audit operation."""
 
    return registry.execute_domain_command(
        "auth.reset",
        actor_slot,
        {
            "version": 1,
            "slot": slot,
            "previous_key_ids_revoked": previous_key_ids_revoked,
            "new_key_id": new_key_id,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command,
    )
 
 
def handover_initiate(
    registry: Registry,
    *,
    from_slot: str,
    to_slot: str,
    reason: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a handover-start operation."""
 
    return registry.execute_domain_command(
        "role.handover.start",
        actor_slot,
        {
            "version": 1,
            "from_slot": from_slot,
            "to_slot": to_slot,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command or "handover.initiate",
    )
 
 
def handover_accept(
    registry: Registry,
    *,
    from_slot: str,
    to_slot: str,
    started_at: str,
    completed_at: str,
    parent_operation_id: int,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a handover-complete operation."""
 
    return registry.execute_domain_command(
        "role.handover.complete",
        actor_slot,
        {
            "version": 1,
            "from_slot": from_slot,
            "to_slot": to_slot,
            "started_at": started_at,
            "completed_at": completed_at,
            "parent_operation_id": parent_operation_id,
        },
        actor_type=actor_type,
        command=command or "handover.accept",
        parent_operation_id=parent_operation_id,
    )
 
 
RoundCleanupOutcome = Literal["clean", "failed"]
 
 
def round_open(
    registry: Registry,
    *,
    round_no: int,
    orchestrator_slot: str,
    plan_id: str | None = None,
    merge_order: int | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.open`` operation (idempotent-additive per SF-001)."""
 
    if round_no <= 0:
        raise ValueError("round_no must be positive")
    payload: dict[str, object] = {
        "version": 1,
        "round_no": round_no,
        "orchestrator_slot": orchestrator_slot,
    }
    if plan_id is not None:
        payload["plan_id"] = plan_id
        if merge_order is not None:
            payload["merge_order"] = merge_order
    return registry.execute_domain_command(
        "round.open",
        actor_slot,
        payload,
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
        related_round=round_no,
    )
 
 
def round_merge(
    registry: Registry,
    *,
    round_no: int,
    plan_id: str,
    merge_order: int,
    merge_commit_sha: str,
    base_sha: str | None = None,
    staging_branch: str | None = None,
    merge_base_sha: str | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.merge`` operation populating ``merge_commit_sha``."""
 
    if round_no <= 0:
        raise ValueError("round_no must be positive")
    if merge_order <= 0:
        raise ValueError("merge_order must be positive")
    return registry.execute_domain_command(
        "round.merge",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "plan_id": plan_id,
            "merge_order": merge_order,
            "base_sha": base_sha,
            "merge_commit_sha": merge_commit_sha,
            "staging_branch": staging_branch,
            "merge_base_sha": merge_base_sha,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
        related_round=round_no,
    )
 
 
def round_test(
    registry: Registry,
    *,
    round_no: int,
    test_count_before: int | None,
    test_count_after: int | None,
    input_hash: str,
    result: Literal["pass", "fail"],
    duration_seconds: float | None = None,
    plan_results: list[dict[str, object]] | None = None,
    artifact_paths: list[str] | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.test`` operation recording the full-suite result."""
 
    return registry.execute_domain_command(
        "round.test",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "test_count_before": test_count_before,
            "test_count_after": test_count_after,
            "input_hash": input_hash,
            "result": result,
            "duration_seconds": duration_seconds,
            "plan_results": plan_results or [],
            "artifact_paths": artifact_paths or [],
        },
        actor_type=actor_type,
        command=command,
        related_round=round_no,
    )
 
 
def round_lint(
    registry: Registry,
    *,
    round_no: int,
    staging_sha: str,
    errors: int,
    warnings: int,
    input_hash: str,
    result: Literal["pass", "fail"],
    artifact_paths: list[str] | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.lint`` operation recording the strict-profile result."""
 
    return registry.execute_domain_command(
        "round.lint",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "staging_sha": staging_sha,
            "errors": errors,
            "warnings": warnings,
            "input_hash": input_hash,
            "result": result,
            "artifact_paths": artifact_paths or [],
        },
        actor_type=actor_type,
        command=command,
        related_round=round_no,
    )
 
 
def round_consolidate(
    registry: Registry,
    *,
    round_no: int,
    final_commit_sha: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.consolidate`` operation; flips status to ``committed``."""
 
    return registry.execute_domain_command(
        "round.consolidate",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "final_commit_sha": final_commit_sha,
        },
        actor_type=actor_type,
        command=command,
        related_round=round_no,
    )
 
 
def round_ff_main(
    registry: Registry,
    *,
    round_no: int,
    from_sha: str,
    to_sha: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.ff_main`` operation recording the fast-forward."""
 
    return registry.execute_domain_command(
        "round.ff_main",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "from_sha": from_sha,
            "to_sha": to_sha,
        },
        actor_type=actor_type,
        command=command,
        related_round=round_no,
    )
 
 
def round_cleanup(
    registry: Registry,
    *,
    round_no: int,
    outcome: RoundCleanupOutcome,
    branches_deleted: list[str] | None = None,
    worktrees_removed: list[str] | None = None,
    error_summary: str | None = None,
    roadmap_version: str | None = None,
    app_version: str | None = None,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.cleanup`` operation; closes the round when ``clean``."""
 
    if outcome not in {"clean", "failed"}:
        raise ValueError(f"invalid cleanup outcome: {outcome}")
    return registry.execute_domain_command(
        "round.cleanup",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "outcome": outcome,
            "branches_deleted": branches_deleted or [],
            "worktrees_removed": worktrees_removed or [],
            "error_summary": error_summary,
            "roadmap_version": roadmap_version,
            "app_version": app_version,
        },
        actor_type=actor_type,
        command=command,
        related_round=round_no,
    )
 
 
def round_abort(
    registry: Registry,
    *,
    round_no: int,
    reason: str,
    actor_slot: str,
    actor_type: ActorType,
    command: str | None = None,
) -> OperationLogRow:
    """Emit a ``round.abort`` operation; returns plans to ``ready_for_merge``."""
 
    if not reason.strip():
        raise ValueError("abort reason is required")
    return registry.execute_domain_command(
        "round.abort",
        actor_slot,
        {
            "version": 1,
            "round_no": round_no,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command,
        related_round=round_no,
    )
 
 
def hub_message(
    registry: Registry,
    *,
    message: Mapping[str, object],
    kind: str,
    actor_slot: str,
    actor_type: ActorType,
    related_plan_id: str | None = None,
) -> OperationLogRow:
    """Emit a ``hub.message`` operation carrying one addressed hub message.
 
    Plan 259 P4. The hub's registry backend records a message as ONE
    operation-log row and nothing else -- there is no projection, because
    a message changes no projected state; it IS the state.
 
    ``command`` is the hub kind, so the log is filterable by kind without
    decoding a payload. ``related_plan_id`` is the message's subject when
    that subject is a plan id, and ``None`` otherwise: the column is
    plan-shaped, which the design note §5 names as the seam a second work
    item kind will find, and smuggling a ticket id into it would make the
    column lie before that seam is even reached.
 
    **The ROW's ``actor_type`` and ``hmac_key_id`` are the signature; the
    message's own ``provenance`` is the CLAIM.** Those are different
    things and the hub keeps them apart on purpose: an agent can write
    any provenance it likes into a payload, but the row's actor type is
    derived from the slot by the storage layer and cannot be chosen.
    """
    if not kind.strip():
        raise ValueError("a hub message must name its kind")
    return registry.execute_domain_command(
        "hub.message",
        actor_slot,
        {
            "version": 1,
            "message": dict(message),
        },
        actor_type=actor_type,
        command=kind,
        related_plan_id=related_plan_id,
    )
 
 
def _require_positive_position(position: int) -> None:
    if position <= 0:
        raise ValueError("queue position must be positive")
 
 
def _require_status(value: str, field: str) -> None:
    if value not in PLAN_STATUSES:
        raise ValueError(f"invalid {field}: {value}")