plan.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/plan.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_domain_plan.py"
"""
Registry-native plan commands.
 
Purpose
-------
Expose plan lifecycle mutation and inspection commands beneath
``python -m oc_chat.registry plan``.
 
Responsibilities
----------------
- Parse plan domain command arguments.
- Resolve the registry and actor slot.
- Dispatch to domain API and reader helpers.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Mutating commands emit one operation-log row through the seam.
- ``show`` is read-only and returns the current projected plan row.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (SQLite backend)
- oc_chat.registry.discovery (registry path discovery)
- oc_chat.registry.domain (plan mutation API)
- oc_chat.registry.readers (plan readers)
- oc_chat.registry (Registry facade for plan mutations and readers)
"""
from __future__ import annotations
 
import argparse
import json
from typing import Callable
 
from oc_diagnostics import _dbg
 
from oc_chat.registry import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.domain import (
    plan_assign,
    plan_create,
    plan_reimport,
    plan_set_status,
    plan_supersede,
)
from oc_chat.registry.readers import show_plan
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_plan_subcommands(plan_parser: argparse.ArgumentParser) -> None:
    """Attach registry-native plan subcommands."""
 
    plan_parser.set_defaults(handler=_cmd_plan_group, plan_parser=plan_parser)
    subcommands = plan_parser.add_subparsers(dest="plan_command")
 
    create = subcommands.add_parser("create", help="Emit plan.create")
    _add_common(create)
    create.add_argument("plan_id")
    create.add_argument("--plan-number", required=True, type=int)
    create.add_argument("--title", required=True)
    create.add_argument("--body-path", required=True)
    create.add_argument("--body-sha256", required=True)
    create.add_argument("--plan-type")
    create.add_argument("--model-override")
    create.add_argument("--thinking-override")
    create.add_argument("--session-override")
    create.set_defaults(handler=_cmd_create)
 
    show = subcommands.add_parser("show", help="Show a projected plan row")
    show.add_argument("plan_id")
    show.add_argument("--registry", help="Registry SQLite path.")
    show.add_argument("--json", action="store_true")
    show.set_defaults(handler=_cmd_show)
 
    set_status = subcommands.add_parser("set-status", help="Emit plan.set_status")
    _add_common(set_status)
    set_status.add_argument("plan_id")
    set_status.add_argument("--from-status", required=True)
    set_status.add_argument("--to-status", required=True)
    set_status.add_argument(
        "--source",
        required=True,
        choices=["worker_self", "orchestrator", "round_lifecycle"],
    )
    set_status.add_argument("--note")
    set_status.set_defaults(handler=_cmd_set_status)
 
    assign = subcommands.add_parser("assign", help="Emit plan.assign")
    _add_common(assign)
    assign.add_argument("plan_id")
    assign.add_argument("--from-slot")
    assign.add_argument("--to-slot", required=True)
    assign.add_argument("--branch", required=True)
    assign.add_argument("--worktree", required=True)
    assign.add_argument("--queue-position", required=True, type=int)
    assign.set_defaults(handler=_cmd_assign)
 
    supersede = subcommands.add_parser("supersede", help="Emit plan.supersede")
    _add_common(supersede)
    supersede.add_argument("plan_id")
    supersede.add_argument("--by", required=True, dest="superseded_by_plan_id")
    supersede.add_argument("--reason", required=True)
    supersede.set_defaults(handler=_cmd_supersede)
 
    reimport = subcommands.add_parser("reimport", help="Emit plan.reimport")
    _add_common(reimport)
    reimport.add_argument("plan_id")
    reimport.add_argument("--previous-body-sha256", required=True)
    reimport.add_argument("--new-body-sha256", required=True)
    reimport.add_argument("--previous-body-observed-at", required=True)
    reimport.add_argument("--new-body-observed-at", required=True)
    reimport.set_defaults(handler=_cmd_reimport)
 
 
def _cmd_plan_group(args: argparse.Namespace) -> int:
    parser = args.plan_parser
    parser.print_help()
    return 1
 
 
def _cmd_create(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: plan_create(
            registry,
            plan_id=args.plan_id,
            plan_number=args.plan_number,
            title=args.title,
            body_path=args.body_path,
            body_sha256=args.body_sha256,
            plan_type=args.plan_type,
            model_override=args.model_override,
            thinking_override=args.thinking_override,
            session_override=args.session_override,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="plan create",
        ),
    )
 
 
def _cmd_show(args: argparse.Namespace) -> int:
    registry, _backend = _registry(args.registry)
    view = show_plan(registry, args.plan_id)
    if view is None:
        print(f"plan not found: {args.plan_id}")
        return 1
    if args.json:
        print(json.dumps(view.__dict__, indent=2, sort_keys=True))
    else:
        print(f"{view.plan_id} | {view.status} | {view.assigned_slot or '-'}")
    return 0
 
 
def _cmd_set_status(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: plan_set_status(
            registry,
            plan_id=args.plan_id,
            from_status=args.from_status,
            to_status=args.to_status,
            source=args.source,
            note=args.note,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="plan set-status",
        ),
    )
 
 
def _cmd_assign(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: plan_assign(
            registry,
            plan_id=args.plan_id,
            from_slot=args.from_slot,
            to_slot=args.to_slot,
            branch=args.branch,
            worktree=args.worktree,
            queue_position=args.queue_position,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="plan assign",
        ),
    )
 
 
def _cmd_supersede(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: plan_supersede(
            registry,
            plan_id=args.plan_id,
            superseded_by_plan_id=args.superseded_by_plan_id,
            reason=args.reason,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="plan supersede",
        ),
    )
 
 
def _cmd_reimport(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: plan_reimport(
            registry,
            plan_id=args.plan_id,
            previous_body_sha256=args.previous_body_sha256,
            new_body_sha256=args.new_body_sha256,
            previous_body_observed_at=args.previous_body_observed_at,
            new_body_observed_at=args.new_body_observed_at,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="plan reimport",
        ),
    )
 
 
def _emit(
    args: argparse.Namespace,
    emit: Callable[[Registry, str], object],
) -> int:
    registry, backend = _registry(args.registry)
    actor_slot = args.actor_slot or backend.active_operator_slot()
    if actor_slot is None:
        print("no active operator slot found")
        return 1
    row = emit(registry, actor_slot)
    print(f"operation {getattr(row, 'id')} state_version={getattr(row, 'state_version')}")
    return 0
 
 
def _registry(registry_path: str | None) -> tuple[Registry, SQLiteBackend]:
    location = discover_registry_location(registry_path=registry_path)
    backend = SQLiteBackend(location.registry_path)
    return Registry(backend), backend
 
 
def _add_common(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--registry", help="Registry SQLite path.")
    parser.add_argument("--actor-slot", help="Actor slot. Defaults to active operator.")