본문으로 건너뛰기

Contexta 운영 가이드

이 가이드는 복구 중심의 Contexta 워크플로를 다루는 운영자 및 고급 사용자를 위한 것입니다.

현재 공용 운영 페이스(surface)는 다음 사항들을 중심으로 구성되어 있습니다:

  • 워크스페이스 백업
  • 복구 계획 및 검증 전용(verify-only) 체크
  • Outbox 메시지 재처리(Replay)
  • 아티팩트 검증 및 전송

현재의 프로토타입 단계에서 가장 신뢰할 수 있는 운영자 경로는 contexta.recovery의 공용 Python API입니다. 완결 실행 가능한 운영 프로그램은 이 페이지 안에서 제공합니다.

여기서 시작하세요​

복구 인터페이스가 처음이라면 다음 순서대로 확인하세요:

  1. 고급 사용법
  2. 완결 실행 프로그램
  3. CLI 레퍼런스

이 경로는 운영 가이드를 실행 가능한 예제 및 현재 명령 동작과 연결하여 제공합니다.

운영 흐름​

복구 작업은 다음 순서로 진행하세요:

  1. 대상 workspace 확인
  2. backup 생성 또는 기존 backup 선택
  3. plan 또는 verify-only 작업 실행
  4. warning과 loss note 확인
  5. 결과가 예상과 맞을 때만 apply
  6. 작업 후 metadata, records, artifacts, reports 검증

핵심 원칙​

  • 적용하기 전에 먼저 계획을 세우는 것을 권장합니다.
  • 파괴적인 복구를 수행하기 전에 검증 전용 복구(verify-only restore)를 먼저 수행하는 것을 권장합니다.
  • 복구 작업은 하나의 명시적인 워크스페이스 내부에서 유지하세요.
  • 내부 모듈을 직접 사용하는 대신 공용 복구 API와 예제를 사용하세요.

백업 (Backup)​

현재 공식 백업 워크플로는 다음과 같습니다:

from contexta.recovery import create_workspace_backup, plan_workspace_backup

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

이 작업을 통해 얻을 수 있는 것:

  • 안정적인 백업 참조(reference)
  • 설정된 복구 루트 아래의 백업 디렉터리
  • 포함된 섹션들을 설명하는 매니페스트(manifest)

현재 운영 참고 사항:

  • 캐시 및 내보내기는 명시적으로 포함하지 않는 한 기본적으로 제외됩니다.
  • 백업 출력물은 워크스페이스 지향적이며, 원격 스냅샷 서비스가 아닙니다.
  • 백업 헬퍼는 더 침습적인 작업을 수행하기 전의 사전 변경 체크포인트로 사용하기에 안전합니다.

백업 및 검증 전용 복구 프로그램에서 전체 코드를 확인할 수 있습니다.

복구 (Restore)​

현재 가장 안전한 복구 방식은 검증 전용(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)

다음 사항을 확인하고 싶을 때 검증 전용 방식을 사용하세요:

  • 백업 매니페스트를 읽을 수 있는지 여부
  • 준비된 워크스페이스가 구체화될 수 있는지 여부
  • 메타데이터, 레코드 및 아티팩트가 현재의 검증 경로를 통과하는지 여부

현재 운영 참고 사항:

  • 검증 전용 방식은 대상 워크스페이스를 덮어쓰지 않습니다.
  • 검증 전용이 아닌 복구는 대상 워크스페이스의 내용을 대체할 수 있습니다.
  • 설정에 의해 활성화된 경우, 복구 작업은 적용 전에 안전 백업을 생성할 수 있습니다.

재생 (Replay)​

재생(Replay)은 일반적인 쿼리 워크플로가 아닌 Outbox 복구(recovery-outbox) 처리를 위한 것입니다.

공식 진입점은 다음과 같습니다:

from contexta.recovery import replay_outbox

result = replay_outbox(config)

다음의 경우에 재생 기능을 사용하세요:

  • 실패한 싱크(sink) 전달을 재시도할 때
  • 승인됨, 대기 중, 데드 레터(dead-lettered) 카운트를 조사할 때
  • 실패한 페이로드를 재생 대상 싱크로 이동할 때

현재 운영 참고 사항:

  • 재생 기능을 사용하려면 config.recovery.outbox_root 설정이 필요합니다.
  • 기본 재생 싱크는 워크스페이스의 내보내기(exports) 영역 아래에 기록됩니다.
  • 재생은 복구 동작이므로 숨겨진 부작용으로 실행하기보다는 의도적으로 실행해야 합니다.

Outbox 메시지 재처리 프로그램에서 전체 코드를 확인할 수 있습니다.

아티팩트 검증 및 전송 (Artifact Verification And Transfer)​

아티팩트 전송은 현재 최상위 복구 퍼사드(facade)보다는 아티팩트 스토어의 공용 인터페이스를 통해 처리하는 것이 가장 좋습니다.

유용한 현재 작업들:

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

다음의 경우에 이 기능들을 사용하세요:

  • 저장된 아티팩트 본문을 검증할 때
  • 자기 설명적 패키지(self-describing package)를 내보낼 때
  • 해당 패키지를 다른 스토어 루트로 가져올 때

아티팩트 전송 프로그램에서 전체 코드를 확인할 수 있습니다.

완결 실행 프로그램​

아래 운영 예제는 완결된 스크립트입니다. 보이는 코드 블록으로 지정된 파일을 만든 뒤 contexta가 설치된 프로젝트에서 실행하세요.

백업 및 검증 전용 복구 프로그램​

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 메시지 재처리 프로그램​

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()

아티팩트 전송 프로그램​

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()

안전 체크리스트​

위험한 복구 작업을 시작하기 전:

  • 대상 워크스페이스 경로를 확인하세요.
  • 데이터가 중요하다면 새로운 백업을 생성하세요.
  • 검증 전용 복구를 먼저 수행하는 것을 권장합니다.
  • 경고, 손실 정보 및 검증 노트를 무시하지 말고 면밀히 조사하세요.

명령줄 참고 사항 (Command-Line Notes)​

내장된 CLI는 이미 작은 유지보수 기능을 제공합니다:

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

현재 프로토타입 단계에서:

  • 공용 문서는 표준 contexta 명칭을 사용합니다.
  • Python API와 실행 가능한 예제는 여전히 가장 명확한 운영자 계약을 유지합니다.

참고:

유효성 검사​

운영 문서나 예제를 변경한 경우 다음 명령을 다시 실행하세요:

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

변경 사항이 재생 동작이나 복구 로직에도 영향을 미치는 경우, 테스트 가이드에서 가장 가까운 복구 스위트로 유효성 검사를 확장하세요.