operations.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/operations.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_operations.py"
"""
Roadmap registry operation executors.
 
Purpose
-------
Expose typed helpers for the roadmap v0.3 operation catalog.
 
Responsibilities
----------------
- Build versioned payloads for roadmap item, version, phase, section, and
  plan-link operations.
- Require HMAC proof for non-system callers before invoking the registry seam.
- Keep SQL ownership inside the SQLite backend projection layer.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - operation helper lifecycle
    L3 - payload construction details
    L4 - preflight validation details
 
Contracts
---------
- Every mutating helper calls Registry.execute_domain_command.
- The module performs no direct SQL, filesystem, subprocess, or clock work.
 
Dependencies
------------
- oc_chat.registry.api (Registry facade)
- oc_chat.registry.errors (HMAC preflight error)
- oc_chat.registry.backends.base (OperationLogRow type returned by each helper)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from typing import Literal, Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.backends.base import OperationLogRow
from oc_chat.registry.errors import HmacVerifyError
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
ActorType = Literal["human_operator", "orchestrator_agent", "worker_agent"]
RoadmapStatus = Literal["todo", "in_progress", "done", "deferred", "blocked", "superseded"]
RoadmapPriority = Literal["critical", "high", "medium", "lower"]
RoadmapTrackKind = Literal["m_phase", "backlog", "vision", "git_workflow", "pre_m0_era"]
DependencyKind = Literal["blocks", "enables", "gates", "part_of"]
ContributionKind = Literal["full", "partial", "context_only"]
 
 
def roadmap_item_create(
    registry: Registry,
    *,
    item_id: str,
    title: str,
    actor_slot: str,
    actor_type: ActorType,
    track_kind: RoadmapTrackKind = "m_phase",
    phase_code: str | None = None,
    phase_label: str | None = None,
    parent_item_id: str | None = None,
    prose: str | None = None,
    prose_path: str | None = None,
    status: RoadmapStatus = "todo",
    priority: RoadmapPriority | None = None,
    sort_order: int = 0,
    target_version: str | None = None,
    introduced_in_version_id: str | None = None,
    closed_in_version_id: str | None = None,
    closed_round_no: int | None = None,
    superseded_by_item_id: str | None = None,
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Create or upsert a roadmap item through the registry seam."""
 
    _require_auth(auth_proof, system_actor, command or "roadmap_item.create", actor_slot)
    return registry.execute_domain_command(
        "roadmap_item.create",
        actor_slot,
        {
            "version": 1,
            "item_id": item_id,
            "title": title,
            "track_kind": track_kind,
            "phase_code": phase_code,
            "phase_label": phase_label,
            "parent_item_id": parent_item_id,
            "prose": prose,
            "prose_path": prose_path,
            "status": status,
            "priority": priority,
            "sort_order": sort_order,
            "target_version": target_version,
            "introduced_in_version_id": introduced_in_version_id,
            "closed_in_version_id": closed_in_version_id,
            "closed_round_no": closed_round_no,
            "superseded_by_item_id": superseded_by_item_id,
        },
        actor_type=actor_type,
        command=command,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def roadmap_item_set_status(
    registry: Registry,
    *,
    item_id: str,
    from_status: RoadmapStatus | None,
    to_status: RoadmapStatus,
    actor_slot: str,
    actor_type: ActorType,
    in_version_id: str | None = None,
    closed_round_no: int | None = None,
    superseded_by_item_id: str | None = None,
    note: str | None = None,
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Set a roadmap item status and append a roadmap item event."""
 
    _require_auth(auth_proof, system_actor, command or "roadmap_item.set_status", actor_slot)
    return registry.execute_domain_command(
        "roadmap_item.set_status",
        actor_slot,
        {
            "version": 1,
            "item_id": item_id,
            "from_status": from_status,
            "to_status": to_status,
            "in_version_id": in_version_id,
            "closed_round_no": closed_round_no,
            "superseded_by_item_id": superseded_by_item_id,
            "note": note,
        },
        actor_type=actor_type,
        command=command,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def roadmap_item_add_dependency(
    registry: Registry,
    *,
    from_item_id: str,
    to_item_id: str,
    actor_slot: str,
    actor_type: ActorType,
    kind: DependencyKind = "gates",
    in_version_id: str | None = None,
    note: str | None = None,
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Create an item-to-item dependency."""
 
    if from_item_id == to_item_id:
        raise ValueError("roadmap dependency cannot point to itself")
    _require_auth(auth_proof, system_actor, command or "roadmap_item.add_dependency", actor_slot)
    return registry.execute_domain_command(
        "roadmap_item.add_dependency",
        actor_slot,
        {
            "version": 1,
            "from_item_id": from_item_id,
            "to_item_id": to_item_id,
            "kind": kind,
            "in_version_id": in_version_id,
            "note": note,
        },
        actor_type=actor_type,
        command=command,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def roadmap_version_bump(
    registry: Registry,
    *,
    from_version_id: str | None,
    to_version_id: str,
    actor_slot: str,
    actor_type: ActorType,
    bump_kind: Literal["patch", "minor", "major", "out_of_round_revision"] = "patch",
    era_id: str | None = None,
    status: Literal["draft", "current", "superseded"] = "draft",
    released_at: str | None = None,
    last_updated_at: str | None = None,
    source_round_no: int | None = None,
    notes: str | None = None,
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Create a new roadmap version row."""
 
    _require_auth(auth_proof, system_actor, command or "roadmap_version.bump", actor_slot)
    return registry.execute_domain_command(
        "roadmap_version.bump",
        actor_slot,
        {
            "version": 1,
            "from_version_id": from_version_id,
            "to_version_id": to_version_id,
            "era_id": era_id,
            "status": status,
            "released_at": released_at,
            "last_updated_at": last_updated_at,
            "source_round_no": source_round_no,
            "bump_kind": bump_kind,
            "notes": notes,
        },
        actor_type=actor_type,
        command=command,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def roadmap_phase_complete(
    registry: Registry,
    *,
    phase_code: str,
    actor_slot: str,
    actor_type: ActorType,
    in_version_id: str | None = None,
    note: str | None = None,
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Mark a roadmap phase complete."""
 
    _require_auth(auth_proof, system_actor, command or "roadmap_phase.complete", actor_slot)
    return registry.execute_domain_command(
        "roadmap_phase.complete",
        actor_slot,
        {
            "version": 1,
            "phase_code": phase_code,
            "in_version_id": in_version_id,
            "note": note,
        },
        actor_type=actor_type,
        command=command,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def roadmap_section_upsert(
    registry: Registry,
    *,
    file_kind: str,
    section_path: str,
    sort_order: int,
    query_kind: str,
    actor_slot: str,
    actor_type: ActorType,
    query_params_json: str | None = None,
    required_min_version: str | None = None,
    active: bool = True,
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Upsert roadmap render-section configuration."""
 
    _require_auth(auth_proof, system_actor, command or "roadmap_section.upsert", actor_slot)
    return registry.execute_domain_command(
        "roadmap_section.upsert",
        actor_slot,
        {
            "version": 1,
            "file_kind": file_kind,
            "section_path": section_path,
            "sort_order": sort_order,
            "query_kind": query_kind,
            "query_params_json": query_params_json,
            "required_min_version": required_min_version,
            "active": active,
        },
        actor_type=actor_type,
        command=command,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def roadmap_plan_link_create(
    registry: Registry,
    *,
    plan_id: str,
    item_id: str,
    actor_slot: str,
    actor_type: ActorType,
    contribution: ContributionKind = "partial",
    command: str | None = None,
    auth_proof: Mapping[str, object] | None = None,
    system_actor: bool = False,
) -> OperationLogRow:
    """Link a plan to a roadmap item."""
 
    _require_auth(auth_proof, system_actor, command or "roadmap_plan_link.create", actor_slot)
    return registry.execute_domain_command(
        "roadmap_plan_link.create",
        actor_slot,
        {
            "version": 1,
            "plan_id": plan_id,
            "item_id": item_id,
            "contribution": contribution,
        },
        actor_type=actor_type,
        command=command,
        related_plan_id=plan_id,
        auth_proof=auth_proof,
        system_actor=system_actor,
    )
 
 
def _require_auth(
    auth_proof: Mapping[str, object] | None,
    system_actor: bool,
    command: str,
    actor_slot: str,
) -> None:
    if system_actor or auth_proof is not None:
        return
    raise HmacVerifyError(
        attempted_slot=actor_slot,
        hmac_key_id=None,
        command=command,
        failure_reason="missing_auth_proof",
    )