Extend playground services¶

def discover_config(start: Path) -> Path | None:
    """Find the nearest playground configuration at or above ``start``."""
    current = start.resolve()
    if current.is_file():
        current = current.parent
    for directory in (current, *current.parents):
        candidate = directory / CONFIG_NAME
        if candidate.is_file():
            return candidate
    return None

def parse_config(path: Path) -> PlaygroundConfig:
    """Parse a configuration while retaining its relative paths."""
    try:
        source = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        raise
    return parse_config_source(source, source_name=str(path))

def parse_config_source(
    source: str,
    *,
    source_name: str = "ormdantic.toml",
) -> PlaygroundConfig:
    """Parse and validate an in-memory playground TOML document."""
    try:
        payload = toml_loads(source)
    except Exception as exc:
        raise ConfigError(f"{source_name}: invalid TOML: {exc}") from exc
    _reject_unknown(payload, {"project", "environments"}, "")
    project_payload = _table(payload.get("project"), "project")
    environments_payload = _table(payload.get("environments"), "environments")
    if not environments_payload:
        raise ConfigError("environments: configure at least one named environment")
    project = _parse_project(project_payload)
    environments = {
        str(name): _parse_environment(
            str(name),
            _table(value, f"environments.{name}"),
        )
        for name, value in environments_payload.items()
    }
    return PlaygroundConfig(project=project, environments=environments)

def load_config(
    path: Path,
    *,
    environment: str | None = None,
    target: str | None = None,
    migrations_dir: Path | None = None,
) -> EffectiveConfig:
    """Load, validate, select, and resolve a playground configuration."""
    config_path = path.resolve()
    config = parse_config(config_path)
    environment_name = environment or "development"
    selected = config.environments.get(environment_name)
    if selected is None:
        raise ConfigError(
            f"environments.{environment_name}: named environment is not configured"
        )

    root = config_path.parent
    project = config.project
    effective_target = _optional_nonempty_string(
        target if target is not None else project.target,
        "project.target",
    )
    if effective_target is None:
        raise ConfigError("project.target: import target is required")
    migration_path = migrations_dir or project.migrations_dir
    resolved_project = replace(
        project,
        target=effective_target,
        migrations_dir=_resolve_path(root, migration_path),
    )
    resolved_environment = replace(
        selected,
        env_file=(
            _resolve_path(root, selected.env_file)
            if selected.env_file is not None
            else None
        ),
        safety="typed" if selected.production else selected.safety,
    )
    return EffectiveConfig(
        path=config_path,
        root=root,
        project=resolved_project,
        environment=resolved_environment,
    )

def write_config(
    path: Path,
    config: PlaygroundConfig,
    *,
    overwrite: bool = False,
) -> None:
    """Atomically write a canonical playground TOML file."""
    destination = path.resolve()
    if destination.exists() and not overwrite:
        raise FileExistsError(destination)
    source = _config_toml(config)
    _atomic_write(destination, source)

def write_config_source(path: Path, source: str) -> None:
    """Validate then atomically replace an existing playground TOML file."""
    destination = path.resolve()
    parse_config_source(source, source_name=str(destination))
    _atomic_write(destination, source)

def resolve_database_url(environment: EnvironmentConfig) -> DatabaseUrlSource:
    """Resolve a URL from the process environment and then its dotenv file."""
    env_value = _clean_value(os.environ.get(environment.url_env))
    if env_value is not None:
        return DatabaseUrlSource(env_value, environment.url_env)
    dotenv_values = _read_env_file(environment.env_file)
    dotenv_value = _clean_value(dotenv_values.get(environment.url_env))
    if dotenv_value is not None:
        return DatabaseUrlSource(
            dotenv_value,
            f"{environment.env_file}:{environment.url_env}",
        )
    raise ConfigError(
        f"environments.{environment.name}.url_env: export {environment.url_env} "
        f"or define it in {environment.env_file or '.env'}"
    )

def accept_refresh(
    state: PlaygroundState,
    result: RefreshResult,
) -> PlaygroundState:
    """Publish a refresh unless a newer generation already won the race."""
    if result.generation < state.generation:
        return state
    changes: dict[str, object] = {
        "generation": result.generation,
        "status": result.status,
        "schema": result.schema,
        "connection_label": result.connection_label,
        "dialect": result.dialect,
        "diagnostics": result.diagnostics,
    }
    if result.migrations is not None:
        changes["migrations"] = result.migrations
    return replace(state, **changes)

def redact_text(value: str) -> str:
    """Redact credentials embedded in URLs or secret query parameters."""
    value = _CREDENTIAL_URL.sub(
        lambda match: f"{match.group('scheme')}{match.group('user')}:<redacted>@",
        value,
    )
    return _SECRET_PARAMETER.sub(
        lambda match: f"{match.group('prefix')}<redacted>",
        value,
    )

def redact_value(value: Any, *, key: str | None = None) -> Any:
    """Recursively redact values carried by diagnostics and logs."""
    if key is not None and key.casefold() in _SECRET_KEYS:
        return "<redacted>"
    if isinstance(value, str):
        return redact_text(value)
    if isinstance(value, Mapping):
        return {
            str(item_key): redact_value(item, key=str(item_key))
            for item_key, item in value.items()
        }
    if isinstance(value, Sequence) and not isinstance(value, bytes | bytearray | str):
        return tuple(redact_value(item) for item in value)
    return value

async def inspect_models(
    target: str,
    *,
    cwd: Path,
    timeout: float = 15.0,
    process_factory: ProcessFactory | None = None,
) -> InspectionResult:
    """Inspect models in a subprocess without polluting the TUI process."""
    spawn = process_factory or cast(ProcessFactory, asyncio.create_subprocess_exec)
    process = await spawn(
        sys.executable,
        "-m",
        "ormdantic.playground.inspect_worker",
        target,
        cwd=str(cwd),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout)
    except asyncio.CancelledError:
        await asyncio.shield(_terminate_process(process))
        raise
    except asyncio.TimeoutError:
        await _terminate_process(process)
        message = f"Model inspection exceeded {timeout:g} seconds"
        return InspectionResult(
            snapshot=None,
            error=InspectionError("TimeoutError", message),
            diagnostics=(
                Diagnostic.create(
                    Severity.ERROR,
                    "model.timeout",
                    message,
                    hint="Fix slow import-time work or increase the inspection timeout.",
                ),
            ),
        )

    stdout_text = stdout.decode("utf-8", errors="replace").strip()
    stderr_text = stderr.decode("utf-8", errors="replace").strip()
    if process.returncode not in {0, None}:
        message = stderr_text or f"Inspection worker exited with {process.returncode}"
        return InspectionResult(
            snapshot=None,
            error=InspectionError("WorkerError", message),
            diagnostics=(
                Diagnostic.create(
                    Severity.ERROR,
                    "model.worker_failed",
                    message,
                    details={"returncode": process.returncode},
                ),
            ),
        )
    try:
        payload = json.loads(stdout_text)
        if not isinstance(payload, Mapping):
            raise TypeError("worker payload must be a JSON object")
        if payload.get("protocol") != PROTOCOL_VERSION:
            raise ValueError(f"unsupported worker protocol {payload.get('protocol')!r}")
        diagnostics = _diagnostics(payload.get("diagnostics", []))
        if payload.get("ok") is True:
            snapshot_payload = payload.get("snapshot")
            if not isinstance(snapshot_payload, Mapping):
                raise TypeError("successful worker payload has no snapshot")
            return InspectionResult(
                snapshot=SchemaSnapshot.from_dict(snapshot_payload),
                diagnostics=diagnostics,
            )
        error = _inspection_error(payload.get("error"))
        return InspectionResult(
            snapshot=None,
            diagnostics=(
                Diagnostic.create(
                    Severity.ERROR,
                    "model.import_failed",
                    f"{error.type}: {error.message}",
                    hint="Fix the target or its import error, then refresh.",
                ),
                *diagnostics,
            ),
            error=error,
        )
    except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
        message = f"Invalid model-inspection response: {exc}"
        if stderr_text:
            message = f"{message}. Worker stderr: {stderr_text}"
        return InspectionResult(
            snapshot=None,
            error=InspectionError(type(exc).__name__, str(exc)),
            diagnostics=(
                Diagnostic.create(
                    Severity.ERROR,
                    "model.protocol_error",
                    message,
                    hint="Reinstall matching Ormdantic package versions.",
                ),
            ),
        )

def reflect_database(
    url: str,
    *,
    database_factory: Callable[[str], Any] = Ormdantic,
    history_exists: Callable[[SchemaSnapshot], bool] | None = None,
) -> DatabaseRefresh:
    """Reflect a live schema and read existing migration history in a worker."""
    database = database_factory(url)
    live_snapshot = database.migrations.live_snapshot()
    if history_exists is not None:
        has_history = history_exists(live_snapshot)
    else:
        has_history = bool(asyncio.run(database.migrations.history_table_exists()))
    if not has_history:
        return DatabaseRefresh(live_snapshot=live_snapshot)
    try:
        history = tuple(asyncio.run(database.migrations.history()))
    except Exception as exc:
        return DatabaseRefresh(
            live_snapshot=live_snapshot,
            diagnostics=(
                Diagnostic.create(
                    Severity.ERROR,
                    "database.history_failed",
                    str(exc),
                    hint="The schema is available; inspect migration history permissions.",
                ),
            ),
        )
    current = next(
        (
            entry.revision
            for entry in reversed(history)
            if entry.status == "applied" and not entry.dirty
        ),
        None,
    )
    return DatabaseRefresh(
        live_snapshot=live_snapshot,
        history=history,
        current_revision=current,
        dirty=any(entry.dirty for entry in history),
    )

def plan_migration(
    url: str,
    before: SchemaSnapshot,
    after: SchemaSnapshot,
) -> MigrationPlan:
    """Generate drift SQL using the existing migration manager."""
    return Ormdantic(url).migrations.generate_plan(before, after, dialect=url)

def load_workspace(directory: Path) -> MigrationWorkspace:
    """Load every top-level TOML and JSON migration without hiding errors."""
    root = directory.resolve()
    if not root.is_dir():
        return MigrationWorkspace()
    paths = sorted(
        (
            path
            for path in root.iterdir()
            if path.is_file() and path.suffix.casefold() in {".toml", ".json"}
        ),
        key=lambda path: path.name,
    )
    documents = [_load_document(path) for path in paths]
    documents = _annotate_revision_graph(documents)
    return MigrationWorkspace(documents=tuple(documents))

def select_document(
    workspace: MigrationWorkspace,
    path: Path,
) -> MigrationWorkspace:
    """Select a known document without mutating the workspace."""
    resolved = path.resolve()
    if not any(document.path == resolved for document in workspace.documents):
        raise ValueError(f"migration document is not in the workspace: {path}")
    return replace(workspace, selected_path=resolved)

def update_source(document: ArtifactDocument, source: str) -> ArtifactDocument:
    """Parse an edited TOML source while retaining the last valid artifact."""
    if document.format == "json":
        raise ValueError("Convert to TOML before editing a JSON migration artifact")
    last_valid = document.artifact or document.last_valid_artifact
    try:
        artifact = _parse_edited_source(source, document.format)
    except Exception as exc:
        return replace(
            document,
            source=source,
            artifact=None,
            last_valid_artifact=last_valid,
            diagnostics=(
                Diagnostic.create(
                    Severity.ERROR,
                    "artifact.edit_invalid",
                    str(exc),
                    source=str(document.path),
                    hint="Fix the TOML error before saving or running this migration.",
                ),
            ),
            dirty=True,
        )
    return replace(
        document,
        source=source,
        artifact=artifact,
        last_valid_artifact=artifact,
        diagnostics=(),
        dirty=True,
    )