Skip to main content

Contexta Operations Guide

This guide is for operators and advanced users working with recovery-oriented Contexta workflows.

The current public operations surface is centered on:

  • workspace backup
  • restore planning and verify-only checks
  • outbox replay
  • artifact verification and transfer

At the current prototype stage, the most reliable operator path is the public Python API in contexta.recovery. Complete runnable operational programs are included on this page.

Start Here​

If you are new to the recovery surface, use these in order:

  1. Advanced Usage
  2. Complete Runnable Programs
  3. CLI Reference

That path keeps the operational guidance attached to executable examples and current command behavior.

Operational Flow​

Use recovery actions in this order:

  1. identify the target workspace
  2. create or locate a backup
  3. run a plan or verify-only operation
  4. inspect warnings and loss notes
  5. apply the operation only when the result is expected
  6. validate metadata, records, artifacts, and reports after the operation

Core Principles​

  • Prefer planning before applying.
  • Prefer verify-only restore before destructive restore.
  • Keep recovery work inside one explicit workspace.
  • Use public recovery APIs and examples instead of reaching into internal modules.

Backup​

The current public backup workflow is:

from contexta.recovery import create_workspace_backup, plan_workspace_backup

plan = plan_workspace_backup(config, label="manual")
result = create_workspace_backup(config, plan)

What this gives you:

  • a stable backup reference
  • a backup directory under the configured recovery root
  • a manifest describing included sections

Current operational notes:

  • cache and exports are excluded by default unless explicitly included
  • backup output is workspace-oriented, not a remote snapshot service
  • the backup helper is safe to use as a pre-change checkpoint before more invasive work

For a complete program, see Backup And Verify-Only Restore.

Restore​

The safest current restore posture is verify-only:

from contexta.recovery import plan_restore, restore_workspace

restore_plan = plan_restore(config, backup_ref, verify_only=True)
restore_result = restore_workspace(config, restore_plan)

Use verify-only when you want to confirm:

  • the backup manifest is readable
  • the staged workspace can be materialized
  • metadata, records, and artifacts pass the current verification path

Current operational notes:

  • verify-only does not overwrite the target workspace
  • non-verify restore can replace the target workspace contents
  • if later enabled by config, restore may create a safety backup before applying

Replay​

Replay is for recovery-outbox processing, not for ordinary query workflows.

The public entrypoint is:

from contexta.recovery import replay_outbox

result = replay_outbox(config)

Use replay when you need to:

  • retry failed sink deliveries
  • inspect acknowledged, pending, and dead-lettered counts
  • move failed payloads into a replay target sink

Current operational notes:

  • replay requires config.recovery.outbox_root
  • the default replay sink writes under the workspace exports area
  • replay is a recovery action, so run it intentionally rather than as a hidden side effect

For a complete program, see Outbox Replay.

Artifact Verification And Transfer​

Artifact transfer is currently best handled through the artifact store public surface rather than a top-level recovery facade.

Useful current operations:

  • inspect_store(...)
  • verify_artifact(...)
  • verify_all(...)
  • export_artifact(...)
  • import_export_package(...)

Use these when you need to:

  • verify stored artifact bodies
  • export a self-describing package
  • import that package into another store root

For a complete program, see Artifact Transfer.

Complete Runnable Programs​

These operational examples are complete scripts. Create the named file from the visible code block and run it in a project where contexta is installed.

Backup And Verify-Only Restore Program​

uv add contexta
uv run backup_restore_verify.py
backup_restore_verify.py
"""Backup and verify-only restore example for Contexta."""

from __future__ import annotations

import argparse
import tempfile
from pathlib import Path
from typing import Any

from contexta import Contexta
from contexta.config import RecoveryConfig, UnifiedConfig, WorkspaceConfig
from contexta.contract import MetricPayload, MetricRecord, Project, RecordEnvelope, Run, StageExecution
from contexta.recovery import create_workspace_backup, plan_restore, plan_workspace_backup, restore_workspace


PROJECT_NAME = "recovery-proj"
RUN_NAME = "demo-run"
RUN_REF = f"run:{PROJECT_NAME}.{RUN_NAME}"


def _resolve_root(root: Path | str | None) -> Path:
    if root is None:
        return Path(tempfile.mkdtemp(prefix="contexta-recovery-demo-"))
    return Path(root)


def _seed_workspace(config: UnifiedConfig) -> None:
    ctx = Contexta(config=config)
    store = ctx.metadata_store
    try:
        store.projects.put_project(
            Project(
                project_ref=f"project:{PROJECT_NAME}",
                name=PROJECT_NAME,
                created_at="2024-06-01T12:00:00Z",
            )
        )
        store.runs.put_run(
            Run(
                run_ref=RUN_REF,
                project_ref=f"project:{PROJECT_NAME}",
                name=RUN_NAME,
                status="completed",
                started_at="2024-06-01T12:00:00Z",
                ended_at="2024-06-01T12:05:00Z",
            )
        )
        store.stages.put_stage_execution(
            StageExecution(
                stage_execution_ref=f"stage:{PROJECT_NAME}.{RUN_NAME}.train",
                run_ref=RUN_REF,
                stage_name="train",
                status="completed",
                started_at="2024-06-01T12:01:00Z",
                ended_at="2024-06-01T12:04:00Z",
                order_index=0,
            )
        )
        ctx.record_store.append(
            MetricRecord(
                envelope=RecordEnvelope(
                    record_ref=f"record:{PROJECT_NAME}.{RUN_NAME}.m0001",
                    record_type="metric",
                    recorded_at="2024-06-01T12:03:00Z",
                    observed_at="2024-06-01T12:03:00Z",
                    producer_ref="contexta.recovery.example",
                    run_ref=RUN_REF,
                ),
                payload=MetricPayload(metric_key="accuracy", value=0.93, value_type="float64"),
            )
        )
    finally:
        store.close()


def run_example(root: Path | str | None = None) -> dict[str, Any]:
    base_root = _resolve_root(root)
    workspace = base_root / ".contexta"
    backup_root = base_root / "backups"
    restore_staging_root = base_root / "restore-staging"

    config = UnifiedConfig(
        project_name=PROJECT_NAME,
        workspace=WorkspaceConfig(root_path=workspace),
        recovery=RecoveryConfig(
            backup_root=backup_root,
            restore_staging_root=restore_staging_root,
            create_backup_before_restore=False,
        ),
    )

    _seed_workspace(config)
    backup_plan = plan_workspace_backup(config, label="ops-demo")
    backup_result = create_workspace_backup(config, backup_plan)
    restore_plan = plan_restore(config, backup_result.backup_ref, verify_only=True)
    restore_result = restore_workspace(config, restore_plan)

    return {
        "workspace": str(workspace),
        "backup_ref": backup_result.backup_ref,
        "backup_location": str(backup_result.location),
        "included_sections": list(backup_result.included_sections),
        "restore_status": restore_result.status,
        "restore_applied": restore_result.applied,
        "verification_notes": list(restore_result.verification_notes),
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the Contexta backup/restore recovery example.")
    parser.add_argument("--root", type=Path, default=None, help="Optional demo root directory.")
    args = parser.parse_args()

    result = run_example(args.root)
    print(f"Workspace: {result['workspace']}")
    print(f"Backup ref: {result['backup_ref']}")
    print(f"Backup location: {result['backup_location']}")
    print(f"Included sections: {', '.join(result['included_sections'])}")
    print(f"Restore status: {result['restore_status']}")
    print(f"Restore applied: {result['restore_applied']}")


if __name__ == "__main__":
    main()

Outbox Replay Program​

uv add contexta
uv run replay_outbox_demo.py
replay_outbox_demo.py
"""Outbox replay example for Contexta."""

from __future__ import annotations

import argparse
import json
import tempfile
from pathlib import Path
from typing import Any

from contexta.config import RecoveryConfig, UnifiedConfig, WorkspaceConfig
from contexta.recovery import replay_outbox


def _resolve_root(root: Path | str | None) -> Path:
    if root is None:
        return Path(tempfile.mkdtemp(prefix="contexta-replay-demo-"))
    return Path(root)


def _write_failed_delivery(outbox_root: Path) -> Path:
    outbox_root.mkdir(parents=True, exist_ok=True)
    path = outbox_root / "failed_deliveries.jsonl"
    entry = {
        "replay_ref": "replay:record.demo.0001",
        "family": "RECORD",
        "sink_name": "local-jsonl-replay",
        "payload": {"run_id": "recovery-proj.demo-run", "metric_key": "loss", "value": 0.12},
        "attempts": 0,
    }
    path.write_text(json.dumps(entry) + "\n", encoding="utf-8")
    return path


def run_example(root: Path | str | None = None) -> dict[str, Any]:
    base_root = _resolve_root(root)
    workspace = base_root / ".contexta"
    outbox_root = base_root / "outbox"

    config = UnifiedConfig(
        project_name="recovery-proj",
        workspace=WorkspaceConfig(root_path=workspace),
        recovery=RecoveryConfig(outbox_root=outbox_root),
    )

    failed_deliveries_path = _write_failed_delivery(outbox_root)
    result = replay_outbox(config)
    replayed_path = config.workspace.exports_path / "replayed" / "record.jsonl"

    return {
        "outbox_path": str(failed_deliveries_path),
        "status": result.status,
        "acknowledged_count": result.acknowledged_count,
        "pending_count": result.pending_count,
        "dead_lettered_count": result.dead_lettered_count,
        "replayed_path": str(replayed_path),
        "replayed_exists": replayed_path.exists(),
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the Contexta outbox replay example.")
    parser.add_argument("--root", type=Path, default=None, help="Optional demo root directory.")
    args = parser.parse_args()

    result = run_example(args.root)
    print(f"Outbox path: {result['outbox_path']}")
    print(f"Replay status: {result['status']}")
    print(f"Acknowledged: {result['acknowledged_count']}")
    print(f"Pending: {result['pending_count']}")
    print(f"Dead-lettered: {result['dead_lettered_count']}")
    print(f"Replayed output: {result['replayed_path']}")


if __name__ == "__main__":
    main()

Artifact Transfer Program​

uv add contexta
uv run artifact_transfer_demo.py
artifact_transfer_demo.py
"""Artifact export/import recovery example for Contexta."""

from __future__ import annotations

import argparse
import tempfile
from pathlib import Path
from typing import Any

from contexta.contract import ArtifactManifest
from contexta.store.artifacts import ArtifactStore, VaultConfig, export_artifact, import_export_package, inspect_store


def _resolve_root(root: Path | str | None) -> Path:
    if root is None:
        return Path(tempfile.mkdtemp(prefix="contexta-artifact-transfer-demo-"))
    return Path(root)


def run_example(root: Path | str | None = None) -> dict[str, Any]:
    base_root = _resolve_root(root)
    base_root.mkdir(parents=True, exist_ok=True)
    source_root = base_root / "source-artifacts"
    target_root = base_root / "target-artifacts"
    export_root = base_root / "exports"
    model_path = base_root / "model.bin"
    model_path.write_bytes(b"artifact content for recovery example")

    source_store = ArtifactStore(VaultConfig(root_path=source_root))
    target_store = ArtifactStore(VaultConfig(root_path=target_root))

    manifest = ArtifactManifest(
        artifact_ref="artifact:my-proj.run-01.model",
        artifact_kind="checkpoint",
        created_at="2024-01-01T00:00:00Z",
        producer_ref="contexta.recovery.example",
        run_ref="run:my-proj.run-01",
        location_ref="vault://my-proj/run-01/model.bin",
    )

    put_receipt = source_store.put_artifact(manifest, model_path)
    export_receipt = export_artifact(source_store, "artifact:my-proj.run-01.model", export_root=export_root)
    import_receipt = import_export_package(target_store, export_receipt.export_directory)
    target_summary = inspect_store(target_store)

    return {
        "source_binding": put_receipt.binding.artifact_ref,
        "export_directory": str(export_receipt.export_directory),
        "import_outcome": import_receipt.outcome.value,
        "target_artifact_count": target_summary.artifact_count,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the Contexta artifact transfer example.")
    parser.add_argument("--root", type=Path, default=None, help="Optional demo root directory.")
    args = parser.parse_args()

    result = run_example(args.root)
    print(f"Source artifact: {result['source_binding']}")
    print(f"Export directory: {result['export_directory']}")
    print(f"Import outcome: {result['import_outcome']}")
    print(f"Target artifact count: {result['target_artifact_count']}")


if __name__ == "__main__":
    main()

Safety Checklist​

Before a risky recovery action:

  • confirm the target workspace path
  • create a fresh backup if data matters
  • prefer verify-only restore first
  • inspect warnings, loss notes, and verification notes instead of ignoring them

Command-Line Notes​

The embedded CLI already exposes a small maintenance surface:

  • contexta backup create
  • contexta restore apply
  • contexta recover replay

At the current prototype stage:

  • public docs use canonical contexta naming
  • the Python APIs and executable examples remain the clearest current operator contract

See:

Validation​

If you change operational docs or examples, rerun:

uv run pytest tests/e2e/test_recovery_examples.py -q

If your change also touches replay behavior or restore logic, expand validation with the nearest recovery suites from the Testing Guide.