oc_chat
Loading...
Searching...
No Matches
migrate.py File Reference

Go to the source code of this file.

Classes

class  oc_chat.registry.migrate.RoundEvidence
class  oc_chat.registry.migrate.MigrationOptions
class  oc_chat.registry.migrate.ImportResult
class  oc_chat.registry.migrate.VerifyResult
class  oc_chat.registry.migrate.RollbackResult
class  oc_chat.registry.migrate.CutoverResult
class  oc_chat.registry.migrate.StatusSource
class  oc_chat.registry.migrate.PlanHeader
class  oc_chat.registry.migrate.QueueEntry
class  oc_chat.registry.migrate.MigrationProjection

Namespaces

namespace  oc_chat
namespace  oc_chat.registry
namespace  oc_chat.registry.migrate

Functions

MigrationProjection oc_chat.registry.migrate.build_projection (Registry registry, MigrationOptions options)
ImportResult oc_chat.registry.migrate.run_import (Registry registry, MigrationOptions options, *, bool dry_run=False)
VerifyResult oc_chat.registry.migrate.run_verify (Registry registry, MigrationOptions options)
RollbackResult oc_chat.registry.migrate.run_rollback (Registry registry, MigrationOptions options)
CutoverResult oc_chat.registry.migrate.run_cutover_stub ()
dict[str, object] oc_chat.registry.migrate.audit_null_closed_round_backfills (Registry registry)
list[dict[str, object]] oc_chat.registry.migrate._build_agent_rows ()
tuple[dict[str, QueueEntry], dict[str, str]] oc_chat.registry.migrate._read_queue_entries (Path queue_path, Path root)
dict[str, PlanHeader] oc_chat.registry.migrate._read_plan_headers (Path plan_dir, Path root)
dict[str, StatusSource] oc_chat.registry.migrate._read_status_sources (tuple[Path,...]|None explicit_paths, Path root, str observed_at, list[str] warnings)
StatusSource oc_chat.registry.migrate._read_status_source (Path path, Path root, str observed_at)
list[Path] oc_chat.registry.migrate._discover_status_paths (Path root)
dict[str, RoundEvidence] oc_chat.registry.migrate._round_evidence_by_plan (MigrationOptions options, Path root)
list[RoundEvidence] oc_chat.registry.migrate._discover_round_evidence (Path root)
str|None oc_chat.registry.migrate._main_sha (Path root)
str oc_chat.registry.migrate._select_legacy_status (str plan_id, PlanHeader header, StatusSource|None source, Mapping[str, QueueEntry] pending, Mapping[str, str] completed)
set[str] oc_chat.registry.migrate._dedupe_plan_ids_by_number (set[str] plan_ids, Mapping[str, PlanHeader] headers, Mapping[str, QueueEntry] pending, Mapping[str, str] completed, Mapping[str, StatusSource] status_sources, list[str] warnings)
int oc_chat.registry.migrate._plan_number (str plan_id, Mapping[str, PlanHeader] headers)
tuple[int, str] oc_chat.registry.migrate._plan_priority (str plan_id, Mapping[str, QueueEntry] pending, Mapping[str, str] completed, Mapping[str, StatusSource] status_sources)
dict[str, object]|None oc_chat.registry.migrate._map_legacy_status (str plan_id, int plan_number, str legacy_status, StatusSource|None source, RoundEvidence|None evidence, str|None discovered_sha, list[str] errors)
dict[str, object] oc_chat.registry.migrate._plan_payload (str plan_id, PlanHeader header, str legacy_status, Mapping[str, object] mapped, str source_file)
dict[str, object] oc_chat.registry.migrate._round_row (RoundEvidence evidence)
PlanHeader oc_chat.registry.migrate._placeholder_header (str plan_id, Path plan_dir, Path root)
str|None oc_chat.registry.migrate._assigned_slot (QueueEntry|None queue_entry, StatusSource|None source, str|None completed_slot)
tuple[str|None, str|None, str|None] oc_chat.registry.migrate._select_overrides (PlanHeader header, StatusSource|None source, QueueEntry|None queue_entry)
str oc_chat.registry.migrate._parse_title (str text, str plan_id)
str|None oc_chat.registry.migrate._header_value (str text, str name)
tuple[str|None, str|None, str|None] oc_chat.registry.migrate._parse_suggested_model (str text)
str|None oc_chat.registry.migrate._clean_header_value (str|None value)
str|None oc_chat.registry.migrate._normalise_branch (str|None value)
None oc_chat.registry.migrate._clear_duplicate_branches (Sequence[dict[str, object]] plan_rows, list[str] warnings)
str oc_chat.registry.migrate._normalise_legacy_status (object value)
str|None oc_chat.registry.migrate._normalise_header_status (object value)
str oc_chat.registry.migrate._agent_label (str value)
str oc_chat.registry.migrate._agent_label_to_slot (str label)
str|None oc_chat.registry.migrate._first_model (object value)
bool oc_chat.registry.migrate._looks_small_model (str|None model)
str|None oc_chat.registry.migrate._provider_from_model (str|None model)
str oc_chat.registry.migrate._display_path (Path path, Path root)
str|None oc_chat.registry.migrate._string_timestamp (object value)
str|None oc_chat.registry.migrate._optional_text (object value)
str|None oc_chat.registry.migrate._first_text (*object values)
str|None oc_chat.registry.migrate._first_nonempty_line (str text)
tuple[int, str] oc_chat.registry.migrate._plan_sort_key (str plan_id)
Path oc_chat.registry.migrate._backup_dir (Path root)
Path oc_chat.registry.migrate._backup_path (Path root, str now)
SQLiteBackend oc_chat.registry.migrate._sqlite_backend (Registry registry)
list[dict[str, object]] oc_chat.registry.migrate._comparable_agents (Sequence[Mapping[str, object]] rows)
list[dict[str, object]] oc_chat.registry.migrate._comparable_plans (Sequence[Mapping[str, object]] rows)
list[dict[str, object]] oc_chat.registry.migrate._comparable_queue_entries (Sequence[Mapping[str, object]] rows)
list[dict[str, object]] oc_chat.registry.migrate._comparable_rounds (Sequence[Mapping[str, object]] rows)
list[dict[str, object]] oc_chat.registry.migrate._comparable_round_plans (Sequence[Mapping[str, object]] rows)
dict[str, object] oc_chat.registry.migrate._snapshot_diff (Mapping[str, object] expected, Mapping[str, object] actual)
dict[str, object] oc_chat.registry.migrate._result_dict (object value)

Variables

 oc_chat.registry.migrate.PLAN_ID_RE = re.compile(r"^(?P<number>\d{3})-(?P<slug>.+)$")
 oc_chat.registry.migrate.LEGACY_PLAN_ID_RE = re.compile(r"\b\d{3}-[A-Za-z0-9][A-Za-z0-9_.-]*\b")
 oc_chat.registry.migrate.ROUND_RE = re.compile(r"\bround[-\s]*(?P<round>\d+)\b", re.IGNORECASE)
dict oc_chat.registry.migrate.HEADER_PATTERNS
 oc_chat.registry.migrate.H1_RE = re.compile(r"^#\s+(?P<value>.+?)\s*$", re.MULTILINE)
dict oc_chat.registry.migrate.VALID_LEGACY_STATUSES
dict oc_chat.registry.migrate.STATUS_NORMALISATIONS
int oc_chat.registry.migrate.PRE_ROUND_SYSTEM_PLAN_CEILING = 50