core.migration

Migration modules: legacy JSON-to-SQLite migration planning, execution, and rollback.

Migration Manifest

core.migration.migration_manifest

Migration state tracking and manifests.

Defines the data structures for migration planning: - MigrationKind: enum of migratable artifact types - MigrationStep: a single atomic migration operation - MigrationManifest: the full set of migration steps for a kind - MigrationPlan: user-facing plan (dry_run, backup, rollback) - MigrationStatus: current state of a migration

BackgroundTaskResult dataclass

Result of a background task.

BackgroundTaskStatus

Bases: StrEnum

Background task status.

BackgroundTaskType

Bases: StrEnum

Background task type.

MigrationKind

Bases: StrEnum

Types of artifacts that can be migrated.

MigrationManifest dataclass

The full set of migration steps for a given kind.

MigrationPlan dataclass

A migration plan for execution.

MigrationReport dataclass

Report after migration execution.

MigrationSource dataclass

Describes a source artifact to migrate.

from_path(path: pathlib.Path, description: str) -> MigrationSource classmethod

Create from an existing path.

MigrationStatus

Bases: StrEnum

Migration status.

MigrationStep dataclass

A single atomic migration operation.

Migration Manager

core.migration.migration_manager

Migration manager for Vector Companion runtime data.

Provides migration/import routines for existing config, logs, memory data, agent settings, and runtime artifacts. Supports: - Dry run (preview changes without applying) - Backup (create backup before migration) - Rollback (restore from backup if migration fails) - Dual-write (write to both old and new formats during transition)

Migration kinds: - CONFIG: config.json schema evolution - CONVERSATION_HISTORY: conversation_history.json format migration - CHAT_LOG: chat.log text to structured transcript - CHROMA_DB: ChromaDB collection schema migration - TOOL_CALLS: tool_calls_active.json migration - BACKGROUND_TASKS: background task result migration - RUN_LOGS: unstructured run logs to structured logs - TRAINING_DATA: training data format migration

MigrationError

Bases: Exception

Migration execution error.

MigrationManager

Orchestrates migrations for all runtime data artifacts.

Usage

manager = MigrationManager( project_root=pathlib.Path("."), structured_logger=structured_logger, ) plan = manager.build_plan( kinds=[MigrationKind.CONFIG, MigrationKind.CONVERSATION_HISTORY], dry_run=True, ) report = manager.execute(plan)

build_plan(*, kinds: list[MigrationKind] | None = None, dry_run: bool = False, skip_if_already_migrated: bool = True) -> MigrationPlan

Build a migration plan for the specified kinds.

Parameters:

Name	Type	Description	Default
kinds	list[MigrationKind] | None	Migration kinds to include. If None, scan all.	None
dry_run	bool	Whether to run in dry-run mode.	False
skip_if_already_migrated	bool	Skip kinds that have already been migrated.	True

Returns:

Type	Description
MigrationPlan	MigrationPlan with all steps populated.

enable_dual_write(kinds: list[MigrationKind]) -> list[BackgroundTaskResult]

Enable dual-write mode for the specified migration kinds.

In dual-write mode, writes go to both old and new formats. This allows for a smooth transition period.

Parameters:

Name	Type	Description	Default
kinds	list[MigrationKind]	Migration kinds to enable dual-write for.	required

Returns:

Type	Description
list[BackgroundTaskResult]	List of background task results.

execute(plan: MigrationPlan) -> MigrationReport

Execute a migration plan.

Parameters:

Name	Type	Description	Default
plan	MigrationPlan	The migration plan to execute.	required

Returns:

Type	Description
MigrationReport	MigrationReport with results.

Raises:

Type	Description
MigrationError	If a required step fails.

execute_async(plan: MigrationPlan) -> MigrationReport async

Execute a migration plan in a thread pool (non-blocking).

rollback(backup_path: pathlib.Path) -> MigrationReport

Restore from a backup.

Parameters:

Name	Type	Description	Default
backup_path	Path	Path to the backup zip file.	required

Returns:

Type	Description
MigrationReport	MigrationReport for the rollback operation.

scan_sources() -> dict[str, MigrationSource]

Scan all runtime data sources and report their status.