envelope.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/regenerate/envelope.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_envelope.py"
# OCT-LINT: disable=no_hardcoded_secrets reason="SRC_*/REGEN_*/SCHEMA_* code strings are the spec enum"
"""
The durable error-code enum + JSON envelope for ``regenerate`` (§10).
 
Purpose
-------
Define the complete, granular ``SRC_*``/``REGEN_*``/``SCHEMA_*`` code enum
(Registry Source & Storage Spec V0.22 §10 — two conditions never share a code,
AC §8.53), the ``RegenerateError``/``SchemaError`` types, and the single JSON
envelope ``{ok, code, message, records, raw_source_tree_hash, source_hash,
logical_state_hash, hint, scan_error?}`` every non-zero exit emits, plus the
exit-code mapping (§5.3).
 
Responsibilities
----------------
- Re-export the P1.3 ``SRC_*`` codes + add the cross-record / rebuild / schema
  codes; expose ``ALL_CODES`` (the enum-completeness anchor).
- ``build_envelope`` (success or error) + ``exit_code_for``.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - envelope assembly
    L3 - code/exit mapping
    L4 - hash-field nullability
 
Contracts
---------
- ``code`` is the durable machine token, NOT the exception class name.
- The three hash fields follow the §10 stage nullability (via P1.3 stage_hashes).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.source.errors (the P1.3 SourceError hierarchy + SRC_* codes)
"""
from __future__ import annotations
 
from typing import Any, Optional, Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.source.errors import (
    SRC_BAD_PHASE_PATH,
    SRC_BAD_QUERY_PARAMS,
    SRC_BAD_YAML,
    SRC_BAD_YAML_FEATURE,
    SRC_DUP_OR_STEM_MISMATCH,
    SRC_NOT_CANONICAL,
    SRC_SCAN_FAILED,
    SRC_STRICT_TYPE,
    SRC_UNKNOWN_FILE,
    SourceError,
    SourceScanError,
    SourceValidationError,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# --- Cross-record SRC_* codes added by the P1.4 step-6 validation -----------
SRC_BAD_TRACK_CODES = "SRC_BAD_TRACK_CODES"
SRC_DANGLING_DEPENDS_ON = "SRC_DANGLING_DEPENDS_ON"
SRC_BAD_DEPENDENCY_KIND = "SRC_BAD_DEPENDENCY_KIND"
SRC_DEPENDENCY_CYCLE = "SRC_DEPENDENCY_CYCLE"
SRC_SUCCESSOR_CYCLE = "SRC_SUCCESSOR_CYCLE"
SRC_DONE_PARENT_LIVE_CHILD = "SRC_DONE_PARENT_LIVE_CHILD"
SRC_DONE_MISSING_CLOSED_VERSION = "SRC_DONE_MISSING_CLOSED_VERSION"
SRC_NON_PHASE_HAS_PHASE = "SRC_NON_PHASE_HAS_PHASE"
SRC_DUPLICATE_ORDER = "SRC_DUPLICATE_ORDER"
SRC_DUPLICATE_CODE = "SRC_DUPLICATE_CODE"
SRC_DUPLICATE_RENDER_SECTION = "SRC_DUPLICATE_RENDER_SECTION"
SRC_LIFECYCLE_CROSSFIELD = "SRC_LIFECYCLE_CROSSFIELD"
SRC_EMPTY_PURPOSE = "SRC_EMPTY_PURPOSE"
SRC_VERSION_METADATA_CONFLICT = "SRC_VERSION_METADATA_CONFLICT"
SRC_ITEM_ID_COLLISION = "SRC_ITEM_ID_COLLISION"
# Plan 142 P6: every kind=roadmap changelog record must carry a unique positive-int
# extensions.changelog_seq (fail-closed; prevents the silent-drop render bug).
SRC_ROADMAP_CHANGELOG_SEQ = "SRC_ROADMAP_CHANGELOG_SEQ"
# Plan 142 P2: kind=product changelog seq is ALL-OR-NONE (no seq -> structured
# fallback; any seq -> every selected product record needs a unique positive-int
# seq). Shared validator with the render path; fail-closed (silent-drop protection).
SRC_PRODUCT_CHANGELOG_SEQ = "SRC_PRODUCT_CHANGELOG_SEQ"
# [NEW·V24] Plan 144: the framing-entity anchor validation (§3.3). A dangling
# anchor target (unresolved era/version/position) -> DANGLING; a same-(target,
# placement, order) collision, including both-order-unset -> DUPLICATE.
SRC_FRAMING_DANGLING_ANCHOR = "SRC_FRAMING_DANGLING_ANCHOR"
SRC_FRAMING_DUPLICATE_ANCHOR = "SRC_FRAMING_DUPLICATE_ANCHOR"
# Plan 142j pre-dispatch mint (2026-07-08): the write-mode clean-tree
# precondition. A dirty *resolved source root* (uncommitted tracked or untracked
# files under the dir the write-mode command will rewrite) has no rollback point
# for a destructive rewrite -> fail closed. SHARED: 142j `format` (write-mode) +
# 142m `migrate`; minted here at 142j dispatch (first of the two). Exits 2
# (refusal class). Raise-site is wired by the 142j worker (`cli/source.py`).
SRC_TREE_DIRTY = "SRC_TREE_DIRTY"
# Plan 142m (minted at dispatch prep, 2026-07-09): the ten migrate-specific
# refusal codes -- raise-sites TARGET, wired by the 142m worker; all exit 2 via
# the migrate refusal class (spec SS8). SRC_TREE_DIRTY above is SHARED with
# migrate (142j wired the format raise-site; migrate reuses the code).
SRC_TREE_NOT_VERSIONED = "SRC_TREE_NOT_VERSIONED"
SRC_NO_MIGRATION = "SRC_NO_MIGRATION"
SRC_SOURCE_MUTATED = "SRC_SOURCE_MUTATED"
SRC_UNSAFE_PATH = "SRC_UNSAFE_PATH"
SRC_MIGRATION_INVALID = "SRC_MIGRATION_INVALID"
SRC_MIGRATION_OUTPUT_INVALID = "SRC_MIGRATION_OUTPUT_INVALID"
SRC_JOURNAL_UNRESOLVED = "SRC_JOURNAL_UNRESOLVED"
SRC_JOURNAL_WRITE_FAILED = "SRC_JOURNAL_WRITE_FAILED"
SRC_MIGRATION_WRITE_FAILED = "SRC_MIGRATION_WRITE_FAILED"
SRC_MIGRATION_TARGET_INVALID = "SRC_MIGRATION_TARGET_INVALID"
# Reserved 142/P2 cutover-ledger codes (dormant in v1).
SRC_LEDGER_ENTRY_INVALID = "SRC_LEDGER_ENTRY_INVALID"
SRC_LEDGER_BASELINE_DRIFT = "SRC_LEDGER_BASELINE_DRIFT"
SRC_ACTIVE_EXEMPTION_NOT_TERMINAL = "SRC_ACTIVE_EXEMPTION_NOT_TERMINAL"
SRC_LEDGER_HISTORY_UNAVAILABLE = "SRC_LEDGER_HISTORY_UNAVAILABLE"
SRC_LEDGER_REOPEN_DETECTED_NULL_CLOSE = "SRC_LEDGER_REOPEN_DETECTED_NULL_CLOSE"
SRC_LEDGER_LIFECYCLE_LOG_DIVERGENCE = "SRC_LEDGER_LIFECYCLE_LOG_DIVERGENCE"
 
# --- Rebuild / schema codes -------------------------------------------------
REGEN_BUILD_FAILED = "REGEN_BUILD_FAILED"
REGEN_NOT_BYTE_STABLE = "REGEN_NOT_BYTE_STABLE"
REGEN_SCOPE_CONFLICT = "REGEN_SCOPE_CONFLICT"
REGEN_LOCK_HELD = "REGEN_LOCK_HELD"
# Plan 142i R5: the §5.3 pre-swap TOCTOU abort — source/ mutated between the
# step-1 scan and the pre-swap re-check (distinct from REGEN_NOT_BYTE_STABLE, an
# external retryable race, not an internal determinism bug). The one code newly
# added to ALL_CODES by this plan.
REGEN_SOURCE_MUTATED = "REGEN_SOURCE_MUTATED"
REGEN_UNVERIFIED_PROJECTION = "REGEN_UNVERIFIED_PROJECTION"
REGEN_UNSUPPORTED_BACKEND = "REGEN_UNSUPPORTED_BACKEND"
SCHEMA_TOO_OLD = "SCHEMA_TOO_OLD"
SCHEMA_TOO_NEW = "SCHEMA_TOO_NEW"
 
#: The durable enum (§10 / AC §8.53) — every emitted code is a member; distinct.
ALL_CODES: frozenset[str] = frozenset(
    {
        SRC_SCAN_FAILED, SRC_UNKNOWN_FILE, SRC_BAD_PHASE_PATH, SRC_BAD_YAML,
        SRC_BAD_YAML_FEATURE, SRC_STRICT_TYPE, SRC_NOT_CANONICAL,
        SRC_DUP_OR_STEM_MISMATCH, SRC_BAD_TRACK_CODES, SRC_DANGLING_DEPENDS_ON,
        SRC_BAD_DEPENDENCY_KIND, SRC_DEPENDENCY_CYCLE, SRC_SUCCESSOR_CYCLE,
        SRC_DONE_PARENT_LIVE_CHILD, SRC_DONE_MISSING_CLOSED_VERSION,
        SRC_LEDGER_ENTRY_INVALID, SRC_LEDGER_BASELINE_DRIFT,
        SRC_ACTIVE_EXEMPTION_NOT_TERMINAL, SRC_LEDGER_HISTORY_UNAVAILABLE,
        SRC_LEDGER_REOPEN_DETECTED_NULL_CLOSE, SRC_LEDGER_LIFECYCLE_LOG_DIVERGENCE,
        SRC_NON_PHASE_HAS_PHASE, SRC_DUPLICATE_ORDER, SRC_DUPLICATE_CODE,
        SRC_DUPLICATE_RENDER_SECTION, SRC_LIFECYCLE_CROSSFIELD, SRC_EMPTY_PURPOSE,
        SRC_VERSION_METADATA_CONFLICT, SRC_BAD_QUERY_PARAMS, SRC_ITEM_ID_COLLISION,
        SRC_ROADMAP_CHANGELOG_SEQ, SRC_PRODUCT_CHANGELOG_SEQ,
        SRC_FRAMING_DANGLING_ANCHOR, SRC_FRAMING_DUPLICATE_ANCHOR, SRC_TREE_DIRTY,
        SRC_TREE_NOT_VERSIONED, SRC_NO_MIGRATION, SRC_SOURCE_MUTATED,
        SRC_UNSAFE_PATH, SRC_MIGRATION_INVALID, SRC_MIGRATION_OUTPUT_INVALID,
        SRC_JOURNAL_UNRESOLVED, SRC_JOURNAL_WRITE_FAILED,
        SRC_MIGRATION_WRITE_FAILED, SRC_MIGRATION_TARGET_INVALID,
        REGEN_BUILD_FAILED, REGEN_NOT_BYTE_STABLE, REGEN_SCOPE_CONFLICT,
        REGEN_LOCK_HELD, REGEN_SOURCE_MUTATED, REGEN_UNVERIFIED_PROJECTION,
        REGEN_UNSUPPORTED_BACKEND, SCHEMA_TOO_OLD, SCHEMA_TOO_NEW,
    }
)
 
 
class RegenerateError(Exception):
    """A rebuild-stage (``REGEN_*``) failure carrying the §10 envelope fields."""
 
    def __init__(
        self,
        message: str,
        *,
        code: str,
        records: Optional[Sequence[str]] = None,
        hint: Optional[str] = None,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.records: list[str] = list(records) if records else []
        self.hint = hint
 
 
class SchemaError(Exception):
    """A ``schema_version`` (``SCHEMA_TOO_OLD``/``SCHEMA_TOO_NEW``) failure."""
 
    def __init__(
        self,
        message: str,
        *,
        code: str,
        records: Optional[Sequence[str]] = None,
        hint: Optional[str] = None,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.records: list[str] = list(records) if records else []
        self.hint = hint
 
 
# Exit codes (§5.3): 1 source-validation · 2 build/unsupported-backend/
# scope-refusal · 3 round-trip not byte-stable · 4 reserved (vacated 2026-07-06,
# Plan 142i R4 — was "authored/runtime conflict", whose only code REGEN_SCOPE_CONFLICT
# now exits 2 per the §5.3 fail-loud contract) · 5 stale schema.
_EXIT_BY_CODE: dict[str, int] = {
    REGEN_NOT_BYTE_STABLE: 3,
    REGEN_SCOPE_CONFLICT: 2,
    SRC_TREE_DIRTY: 2,  # Plan 142j: write-mode clean-tree refusal (refusal class)
    # Plan 142m (dispatch-prep mint): the migrate refusal class -- all exit 2.
    SRC_TREE_NOT_VERSIONED: 2,
    SRC_NO_MIGRATION: 2,
    SRC_SOURCE_MUTATED: 2,
    SRC_UNSAFE_PATH: 2,
    SRC_MIGRATION_INVALID: 2,
    SRC_MIGRATION_OUTPUT_INVALID: 2,
    SRC_JOURNAL_UNRESOLVED: 2,
    SRC_JOURNAL_WRITE_FAILED: 2,
    SRC_MIGRATION_WRITE_FAILED: 2,
    SRC_MIGRATION_TARGET_INVALID: 2,
    SCHEMA_TOO_OLD: 5,
    SCHEMA_TOO_NEW: 5,
}
 
 
def exit_code_for(error: BaseException) -> int:
    """Map an exception to the §5.3 exit code."""
    if isinstance(error, (SourceScanError, SourceValidationError)):
        return 1
    code = getattr(error, "code", None)
    if code in _EXIT_BY_CODE:
        return _EXIT_BY_CODE[code]
    if isinstance(error, (RegenerateError, SchemaError)):
        return 5 if isinstance(error, SchemaError) else 2
    return 2
 
 
def build_envelope(
    error: Optional[BaseException],
    *,
    raw_source_tree_hash: Optional[str] = None,
    source_hash: Optional[str] = None,
    logical_state_hash: Optional[str] = None,
    message: Optional[str] = None,
    rollback: Optional[str] = None,
    journal_path: Optional[str] = None,
    remaining_paths: Optional[Sequence[str]] = None,
    journal_unresolved: Optional[bool] = None,
) -> dict[str, Any]:
    """Assemble the §10 JSON envelope for a success (``error=None``) or failure.
 
    The migrate (Plan 142m) recovery fields ``rollback`` / ``journal_path`` /
    ``remaining_paths`` / ``journal_unresolved`` are **additive keyword
    arguments** (Sonnet5-R02-P1 / DeepSeek-R03-NP1-3 Option B): each is emitted
    into the envelope **only when non-``None``** (``journal_unresolved`` only when
    ``True``), so a non-migrate caller — which never passes them — sees a
    byte-identical envelope to before. They describe the recovery *attempt* and
    are passed explicitly by the migrate runner (NOT ``getattr``-read off the
    exception, which carries only the error condition).
    """
    if error is None:
        return {
            "ok": True,
            "code": None,
            "message": message or "regenerate ok",
            "records": [],
            "raw_source_tree_hash": raw_source_tree_hash,
            "source_hash": source_hash,
            "logical_state_hash": logical_state_hash,
            "hint": None,
        }
    envelope: dict[str, Any] = {
        "ok": False,
        "code": getattr(error, "code", None),
        "message": str(error),
        "records": list(getattr(error, "records", []) or []),
        "raw_source_tree_hash": raw_source_tree_hash,
        "source_hash": source_hash,
        "logical_state_hash": logical_state_hash,
        "hint": getattr(error, "hint", None),
    }
    scan_error = getattr(error, "scan_error", None)
    if scan_error is not None:
        envelope["scan_error"] = scan_error
    # Migrate recovery fields — emit-only-when-non-None (the byte-identical
    # invariant for every non-migrate caller).
    if rollback is not None:
        envelope["rollback"] = rollback
    if journal_path is not None:
        envelope["journal_path"] = journal_path
    if remaining_paths is not None:
        envelope["remaining_paths"] = list(remaining_paths)
    if journal_unresolved:
        envelope["journal_unresolved"] = True
    return envelope
 
 
# Re-export the P1.3 hierarchy so callers import errors from one place.
__all__ = [
    "SourceError",
    "SourceScanError",
    "SourceValidationError",
    "RegenerateError",
    "SchemaError",
    "ALL_CODES",
    "build_envelope",
    "exit_code_for",
]