"""Heavy knowledge refresh path run outside request handling."""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import signal
import sys
from contextlib import asynccontextmanager, suppress
from dataclasses import asdict, dataclass, replace
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, TypedDict, cast

from mindroom.config.knowledge import KnowledgeBaseConfig
from mindroom.config.main import Config
from mindroom.constants import RuntimePaths, resolve_runtime_paths, runtime_env_values
from mindroom.knowledge.availability import KnowledgeAvailability
from mindroom.knowledge.index_metadata import state_for_publication
from mindroom.knowledge.manager import KnowledgeManager
from mindroom.knowledge.redaction import redact_credentials_in_text
from mindroom.knowledge.refresh_locks import (
    mark_refresh_active,
    mark_refresh_inactive,
    refresh_source_root_lock,
)
from mindroom.knowledge.registry import (
    PublishedIndexKey,
    PublishedIndexState,
    load_published_index_state,
    mark_knowledge_source_changed_async,
    mark_published_index_refresh_failed_preserving_last_good,
    mark_published_index_refresh_running,
    mark_published_index_refresh_succeeded,
    mark_published_index_stale,
    prune_private_index_bookkeeping,
    publish_knowledge_index_from_state,
    published_index_availability_for_state,
    published_index_collection_exists_for_state,
    published_index_metadata_path,
    published_index_settings_compatible,
    refresh_target_for_published_index_key,
    resolve_published_index_key,
    resolve_refresh_target,
    save_published_index_state,
    source_root_for_published_index_key,
    source_root_for_refresh_target,
)
from mindroom.logging_config import get_logger
from mindroom.runtime_resolution import resolve_knowledge_binding
from mindroom.tool_system.worker_routing import (
    SerializedToolExecutionIdentity,
    ToolExecutionIdentity,
    parse_tool_execution_identity_payload,
    serialize_tool_execution_identity,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator


logger = get_logger(__name__)


@dataclass(frozen=True)
class KnowledgeRefreshResult:
    """Result of one explicit knowledge refresh."""

    key: PublishedIndexKey
    indexed_count: int
    index_published: bool
    availability: KnowledgeAvailability
    last_error: str | None = None


@dataclass(frozen=True)
class _SubprocessRefreshRequest:
    base_id: str
    config_data: dict[str, object]
    config_path: str
    storage_root: str
    runtime_knowledge_base: dict[str, object] | None = None
    execution_identity: SerializedToolExecutionIdentity | None = None
    force_reindex: bool = False


class _SubprocessSessionKwargs(TypedDict, total=False):
    start_new_session: bool


_REFRESH_SUBPROCESS_THREAD_ENV = {
    "OMP_NUM_THREADS": "1",
    "OPENBLAS_NUM_THREADS": "1",
    "MKL_NUM_THREADS": "1",
    "NUMEXPR_NUM_THREADS": "1",
    "VECLIB_MAXIMUM_THREADS": "1",
    "TOKENIZERS_PARALLELISM": "false",
}


async def refresh_knowledge_binding_in_subprocess(
    base_id: str,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None = None,
    force_reindex: bool = False,
) -> None:
    """Run one knowledge refresh in a child interpreter.

    Scheduled refreshes are best-effort maintenance work. Running them in a
    subprocess keeps Chroma, embedding, Git, and reader CPU/I/O away from the
    control-plane event loop and its shared thread pool while preserving the
    last-good published index semantics.
    """
    key = resolve_published_index_key(
        base_id,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        create=True,
    )
    initial_state = await asyncio.to_thread(load_published_index_state, published_index_metadata_path(key))
    request_payload = _serialize_subprocess_refresh_request(
        base_id,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        force_reindex=force_reindex,
    )
    env = dict(runtime_env_values(runtime_paths))
    env.update(_REFRESH_SUBPROCESS_THREAD_ENV)
    env["MINDROOM_KNOWLEDGE_REFRESH_SUBPROCESS"] = "1"
    process = await asyncio.create_subprocess_exec(
        sys.executable,
        "-m",
        "mindroom.knowledge_refresh_runner",
        stdin=asyncio.subprocess.PIPE,
        env=env,
        **_subprocess_session_kwargs(),
    )
    try:
        with suppress(BrokenPipeError, ConnectionResetError):
            await _send_subprocess_refresh_request(process, request_payload)
        return_code = await process.wait()
    except asyncio.CancelledError:
        cleanup_task = asyncio.create_task(
            _cleanup_cancelled_refresh_subprocess(
                process,
                key,
                initial_state=initial_state,
                config=config,
                runtime_paths=runtime_paths,
            ),
        )
        while not cleanup_task.done():
            with suppress(asyncio.CancelledError):
                await asyncio.shield(cleanup_task)
        with suppress(Exception):
            cleanup_task.result()
        raise

    if return_code != 0:
        msg = f"Knowledge refresh subprocess failed for {base_id!r} with exit code {return_code}"
        await _reconcile_failed_refresh_subprocess(key, initial_state=initial_state, error=msg)
        raise RuntimeError(msg)


def _serialize_subprocess_refresh_request(
    base_id: str,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None,
    force_reindex: bool,
) -> bytes:
    runtime_knowledge_base = config.runtime_knowledge_base_overlay(base_id)
    payload = _SubprocessRefreshRequest(
        base_id=base_id,
        config_data=config.authored_model_dump(),
        config_path=str(runtime_paths.config_path),
        storage_root=str(runtime_paths.storage_root),
        runtime_knowledge_base=(
            None
            if runtime_knowledge_base is None
            else cast("dict[str, object]", runtime_knowledge_base.model_dump(mode="json", exclude_unset=True))
        ),
        execution_identity=None
        if execution_identity is None
        else serialize_tool_execution_identity(execution_identity),
        force_reindex=force_reindex,
    )
    return json.dumps(asdict(payload), sort_keys=True).encode()


async def _send_subprocess_refresh_request(
    process: asyncio.subprocess.Process,
    payload: bytes,
) -> None:
    if process.stdin is None:
        msg = "Knowledge refresh subprocess was created without stdin"
        raise RuntimeError(msg)
    process.stdin.write(payload)
    await process.stdin.drain()
    process.stdin.close()
    with suppress(BrokenPipeError, ConnectionResetError):
        await process.stdin.wait_closed()


def _subprocess_session_kwargs() -> _SubprocessSessionKwargs:
    if os.name == "nt":
        return {}
    return {"start_new_session": True}


async def _terminate_refresh_subprocess(process: asyncio.subprocess.Process) -> None:
    if process.returncode is not None:
        return
    if os.name == "nt":
        process.terminate()
    else:
        with suppress(ProcessLookupError):
            os.killpg(process.pid, signal.SIGTERM)
    try:
        await asyncio.wait_for(process.wait(), timeout=10)
    except TimeoutError:
        if os.name == "nt":
            process.kill()
        else:
            with suppress(ProcessLookupError):
                os.killpg(process.pid, signal.SIGKILL)
        await process.wait()


async def _cleanup_cancelled_refresh_subprocess(
    process: asyncio.subprocess.Process,
    key: PublishedIndexKey,
    *,
    initial_state: PublishedIndexState | None,
    config: Config,
    runtime_paths: RuntimePaths,
) -> None:
    try:
        await _terminate_refresh_subprocess(process)
        source_root = source_root_for_published_index_key(key)
        async with refresh_source_root_lock(source_root):
            await _reconcile_cancelled_refresh(
                key,
                initial_state=initial_state,
                config=config,
                runtime_paths=runtime_paths,
            )
    except Exception:
        logger.warning("Failed to reconcile cancelled knowledge refresh subprocess", base_id=key.base_id, exc_info=True)


async def _reconcile_failed_refresh_subprocess(
    key: PublishedIndexKey,
    *,
    initial_state: PublishedIndexState | None,
    error: str,
) -> None:
    try:
        source_root = source_root_for_published_index_key(key)
        async with refresh_source_root_lock(source_root):
            state = await asyncio.to_thread(load_published_index_state, published_index_metadata_path(key))
            if not _failed_subprocess_state_can_be_reconciled(key, state, initial_state):
                return
            await asyncio.to_thread(mark_published_index_refresh_failed_preserving_last_good, key, error=error)
    except Exception:
        logger.warning("Failed to reconcile failed knowledge refresh subprocess", base_id=key.base_id, exc_info=True)


@asynccontextmanager
async def knowledge_binding_mutation_lock(
    base_id: str,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None = None,
    create: bool = False,
) -> AsyncIterator[None]:
    """Serialize source mutations with refresh publishes in this runtime event loop."""
    key = resolve_refresh_target(
        base_id,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        create=create,
    )
    source_root = source_root_for_refresh_target(key)
    async with refresh_source_root_lock(source_root):
        yield


async def refresh_knowledge_binding(
    base_id: str,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None = None,
    force_reindex: bool = False,
) -> KnowledgeRefreshResult:
    """Build and publish one resolved knowledge binding."""
    key = resolve_published_index_key(
        base_id,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        create=True,
    )
    return await _refresh_resolved_knowledge_binding(
        key,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        force_reindex=force_reindex,
    )


async def _refresh_resolved_knowledge_binding(
    key: PublishedIndexKey,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None,
    force_reindex: bool,
) -> KnowledgeRefreshResult:
    refresh_target = refresh_target_for_published_index_key(key)
    source_root = source_root_for_published_index_key(key)
    mark_refresh_active(refresh_target)
    try:
        async with refresh_source_root_lock(source_root):
            initial_state = await asyncio.to_thread(
                load_published_index_state,
                published_index_metadata_path(key),
            )
            try:
                await _save_refreshing_state(key)
                return await _refresh_knowledge_binding_locked(
                    key,
                    config=config,
                    runtime_paths=runtime_paths,
                    execution_identity=execution_identity,
                    force_reindex=force_reindex,
                )
            except asyncio.CancelledError:
                await _reconcile_cancelled_refresh(
                    key,
                    initial_state=initial_state,
                    config=config,
                    runtime_paths=runtime_paths,
                )
                raise
    finally:
        mark_refresh_inactive(refresh_target)
        prune_private_index_bookkeeping()


async def _save_refreshing_state(key: PublishedIndexKey) -> None:
    write_task = asyncio.create_task(asyncio.to_thread(mark_published_index_refresh_running, key))
    try:
        await asyncio.shield(write_task)
    except asyncio.CancelledError:
        write_completed = False
        with suppress(Exception):
            await write_task
            write_completed = True
        if write_completed:
            with suppress(Exception):
                await asyncio.to_thread(mark_published_index_stale, key, reason="refresh_cancelled", refresh_job="idle")
        raise


async def _refresh_knowledge_binding_locked(
    key: PublishedIndexKey,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None = None,
    force_reindex: bool = False,
) -> KnowledgeRefreshResult:
    base_id = key.base_id
    try:
        if config.get_knowledge_base_config(base_id).mode == "files":
            return await _refresh_file_mode_binding_locked(
                key,
                config=config,
                runtime_paths=runtime_paths,
                execution_identity=execution_identity,
            )

        binding = resolve_knowledge_binding(
            base_id,
            config,
            runtime_paths,
            execution_identity=execution_identity,
            start_watchers=False,
            create=True,
        )
        manager = KnowledgeManager(
            base_id=base_id,
            config=config,
            runtime_paths=runtime_paths,
            storage_path=binding.storage_root,
            knowledge_path=binding.knowledge_path,
        )
        if await _should_defer_cold_empty_publication(manager, key):
            await asyncio.to_thread(
                mark_published_index_stale,
                key,
                reason="source_empty",
                refresh_job="idle",
            )
            return KnowledgeRefreshResult(
                key=key,
                indexed_count=0,
                index_published=False,
                availability=KnowledgeAvailability.INITIALIZING,
                last_error=None,
            )
        unchanged_result = await _maybe_publish_unchanged_index(
            manager,
            key,
            execution_identity=execution_identity,
            force_reindex=force_reindex,
        )
        if unchanged_result is not None:
            return unchanged_result
        outcome = await manager.reindex_all(force_reindex=force_reindex)
        if outcome.error is not None:
            await asyncio.to_thread(
                mark_published_index_refresh_failed_preserving_last_good,
                key,
                error=outcome.error,
            )
            return KnowledgeRefreshResult(
                key=key,
                indexed_count=outcome.indexed_count,
                index_published=False,
                availability=KnowledgeAvailability.REFRESH_FAILED,
                last_error=outcome.error,
            )
    except Exception as exc:
        error = redact_credentials_in_text(str(exc))
        await asyncio.to_thread(mark_published_index_refresh_failed_preserving_last_good, key, error=error)
        raise
    return await _refresh_result_from_persisted_state(
        key,
        indexed_count=outcome.indexed_count,
        config=config,
        runtime_paths=runtime_paths,
    )


async def _should_defer_cold_empty_publication(
    manager: KnowledgeManager,
    key: PublishedIndexKey,
) -> bool:
    """Return whether a content-gated cold index has no corpus to publish yet."""
    if not manager.config.get_knowledge_base_config(manager.base_id).require_content_before_publish:
        return False
    files = await asyncio.to_thread(manager.list_files)
    if files:
        return False
    state = await asyncio.to_thread(load_published_index_state, published_index_metadata_path(key))
    return state is None or state.status != "complete" or not state.indexed_count


async def _publish_file_mode_source_metadata(
    key: PublishedIndexKey,
    manager: KnowledgeManager,
    *,
    published_revision: str | None = None,
) -> KnowledgeRefreshResult:
    """Publish current source metadata for a file-only base without building vectors."""
    source_signature = await manager.source_signature()
    await asyncio.to_thread(
        save_published_index_state,
        published_index_metadata_path(key),
        state_for_publication(
            settings=key.indexing_settings,
            collection=None,
            indexed_count=0,
            source_signature=source_signature,
            published_revision=published_revision,
        ),
    )
    return KnowledgeRefreshResult(
        key=key,
        indexed_count=0,
        index_published=True,
        availability=KnowledgeAvailability.READY,
        last_error=None,
    )


async def publish_file_mode_source_metadata_for_base(
    base_id: str,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None = None,
) -> KnowledgeRefreshResult:
    """Resolve and publish current source metadata for a file-only base."""
    key = resolve_published_index_key(
        base_id,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        create=True,
    )
    return await _publish_file_mode_source_metadata_for_resolved(
        key,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
    )


async def _publish_file_mode_source_metadata_for_resolved(
    key: PublishedIndexKey,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None,
) -> KnowledgeRefreshResult:
    base_id = key.base_id
    binding = resolve_knowledge_binding(
        base_id,
        config,
        runtime_paths,
        execution_identity=execution_identity,
        start_watchers=False,
        create=True,
    )
    manager = KnowledgeManager(
        base_id=base_id,
        config=config,
        runtime_paths=runtime_paths,
        storage_path=binding.storage_root,
        knowledge_path=binding.knowledge_path,
    )
    return await _publish_file_mode_source_metadata(key, manager)


async def _refresh_file_mode_binding_locked(
    key: PublishedIndexKey,
    *,
    config: Config,
    runtime_paths: RuntimePaths,
    execution_identity: ToolExecutionIdentity | None,
) -> KnowledgeRefreshResult:
    """Refresh source metadata for a file-only base without building vectors."""
    binding = resolve_knowledge_binding(
        key.base_id,
        config,
        runtime_paths,
        execution_identity=execution_identity,
        start_watchers=False,
        create=True,
    )
    manager = KnowledgeManager(
        base_id=key.base_id,
        config=config,
        runtime_paths=runtime_paths,
        storage_path=binding.storage_root,
        knowledge_path=binding.knowledge_path,
    )
    published_revision: str | None = None
    if manager.git_source.is_configured():
        git_sync_result = await manager.git_source.sync()
        published_revision = git_sync_result.head
        if git_sync_result.updated:
            await mark_knowledge_source_changed_async(
                key.base_id,
                config=manager.config,
                runtime_paths=manager.runtime_paths,
                execution_identity=execution_identity,
                reason="git_source_updated",
            )

    return await _publish_file_mode_source_metadata(key, manager, published_revision=published_revision)


async def _maybe_publish_unchanged_index(
    manager: KnowledgeManager,
    key: PublishedIndexKey,
    *,
    execution_identity: ToolExecutionIdentity | None,
    force_reindex: bool,
) -> KnowledgeRefreshResult | None:
    force_reindex = force_reindex or manager.needs_full_reindex_on_create()
    if manager.git_source.is_configured():
        git_sync_result = await manager.git_source.sync()
        if force_reindex or git_sync_result.updated:
            if git_sync_result.updated:
                await mark_knowledge_source_changed_async(
                    key.base_id,
                    config=manager.config,
                    runtime_paths=manager.runtime_paths,
                    execution_identity=execution_identity,
                    reason="git_source_updated",
                )
            return None
        return await _publish_unchanged_index(
            manager,
            key,
            published_revision=git_sync_result.head,
        )
    if force_reindex:
        await mark_knowledge_source_changed_async(
            key.base_id,
            config=manager.config,
            runtime_paths=manager.runtime_paths,
            execution_identity=execution_identity,
            reason="manual_reindex",
        )
        return None
    return await _publish_unchanged_index(
        manager,
        key,
        mark_stale_on_source_change=True,
        execution_identity=execution_identity,
    )


async def _refresh_result_from_persisted_state(
    key: PublishedIndexKey,
    *,
    indexed_count: int,
    config: Config,
    runtime_paths: RuntimePaths,
) -> KnowledgeRefreshResult:
    state = await asyncio.to_thread(load_published_index_state, published_index_metadata_path(key))
    if state is None:
        error = "Published index metadata was missing after refresh"
        await asyncio.to_thread(mark_published_index_refresh_failed_preserving_last_good, key, error=error)
        return KnowledgeRefreshResult(
            key=key,
            indexed_count=indexed_count,
            index_published=False,
            availability=KnowledgeAvailability.REFRESH_FAILED,
            last_error=error,
        )
    if state.status != "complete":
        error = "Published index metadata was incomplete after refresh"
        await asyncio.to_thread(mark_published_index_refresh_failed_preserving_last_good, key, error=error)
        return KnowledgeRefreshResult(
            key=key,
            indexed_count=indexed_count,
            index_published=False,
            availability=KnowledgeAvailability.REFRESH_FAILED,
            last_error=error,
        )

    availability = published_index_availability_for_state(key=key, state=state)
    if not published_index_settings_compatible(state.settings, key.indexing_settings):
        await asyncio.to_thread(
            mark_published_index_stale,
            key,
            reason="published_index_config_mismatch",
        )
        return KnowledgeRefreshResult(
            key=key,
            indexed_count=indexed_count,
            index_published=False,
            availability=availability,
            last_error=None,
        )
    index = publish_knowledge_index_from_state(
        key,
        state=state,
        config=config,
        runtime_paths=runtime_paths,
        metadata_path=published_index_metadata_path(key),
    )
    if index is None:
        error = "Published index collection was missing after refresh"
        await asyncio.to_thread(mark_published_index_refresh_failed_preserving_last_good, key, error=error)
        return KnowledgeRefreshResult(
            key=key,
            indexed_count=indexed_count,
            index_published=False,
            availability=KnowledgeAvailability.REFRESH_FAILED,
            last_error=error,
        )
    await asyncio.to_thread(mark_published_index_refresh_succeeded, key)
    return KnowledgeRefreshResult(
        key=key,
        indexed_count=indexed_count,
        index_published=True,
        availability=KnowledgeAvailability.READY,
        last_error=None,
    )


def _revision_proves_source_unchanged(state: PublishedIndexState, published_revision: str | None) -> bool:
    """Return whether a matching Git revision already proves the indexed corpus is unchanged.

    MindRoom owns the checkout and realigns it with ``git reset --hard``, so a tracked
    file's content is fully determined by HEAD. Every corpus filter (branch, LFS, hidden
    paths, include/exclude patterns and extensions) lives in ``IndexingSettings``, which
    the caller has already compared. An unmoved revision therefore means byte-identical
    indexed content, and hashing the corpus to learn the same thing costs a full read of
    every file — the dominant cost on a large or network-mounted source.
    """
    return published_revision is not None and state.published_revision == published_revision


async def _publish_unchanged_index(
    manager: KnowledgeManager,
    key: PublishedIndexKey,
    *,
    published_revision: str | None = None,
    mark_stale_on_source_change: bool = False,
    execution_identity: ToolExecutionIdentity | None = None,
) -> KnowledgeRefreshResult | None:
    state = await asyncio.to_thread(load_published_index_state, published_index_metadata_path(key))
    if (
        state is None
        or state.status != "complete"
        or state.source_signature is None
        or state.settings != key.indexing_settings
        or not await asyncio.to_thread(published_index_collection_exists_for_state, key, state)
    ):
        return None

    if _revision_proves_source_unchanged(state, published_revision):
        current_source_signature = state.source_signature
    else:
        current_source_signature = await manager.source_signature()
    if current_source_signature != state.source_signature:
        if mark_stale_on_source_change:
            await mark_knowledge_source_changed_async(
                key.base_id,
                config=manager.config,
                runtime_paths=manager.runtime_paths,
                execution_identity=execution_identity,
                reason="source_changed",
            )
        return None

    # Settings already matched at the top of this function, so only the revision
    # and publish stamp can still move.
    updated_state = state
    if published_revision is not None:
        updated_state = replace(
            updated_state,
            last_published_at=datetime.now(tz=UTC).isoformat(),
            published_revision=published_revision,
        )
    if updated_state != state:
        await asyncio.to_thread(save_published_index_state, published_index_metadata_path(key), updated_state)
    index = publish_knowledge_index_from_state(
        key,
        state=updated_state,
        config=manager.config,
        runtime_paths=manager.runtime_paths,
        metadata_path=published_index_metadata_path(key),
    )
    if index is None:
        error = "Published index collection was missing during unchanged refresh"
        await asyncio.to_thread(mark_published_index_refresh_failed_preserving_last_good, key, error=error)
        return KnowledgeRefreshResult(
            key=key,
            indexed_count=updated_state.indexed_count or 0,
            index_published=False,
            availability=KnowledgeAvailability.REFRESH_FAILED,
            last_error=error,
        )
    await asyncio.to_thread(mark_published_index_refresh_succeeded, key)
    # This path returns before any candidate is opened, so it is the only place
    # that can retire candidate state left by an interrupted forced rebuild.
    try:
        await manager.discard_superseded_candidate(published_collection=updated_state.collection)
    except Exception:
        logger.warning(
            "Failed to retire candidate after unchanged knowledge index publish",
            base_id=manager.base_id,
            collection=updated_state.collection,
            exc_info=True,
        )
    return KnowledgeRefreshResult(
        key=key,
        indexed_count=updated_state.indexed_count or 0,
        index_published=True,
        availability=KnowledgeAvailability.READY,
        last_error=None,
    )


def _published_state_fingerprint(state: PublishedIndexState | None) -> tuple[object, ...] | None:
    if state is None:
        return None
    return (
        state.settings,
        state.status,
        state.collection,
        state.last_published_at,
        state.published_revision,
        state.indexed_count,
        state.source_signature,
        state.refresh_job,
        state.reason,
        state.last_error,
        state.consecutive_refresh_failures,
    )


def _refresh_running_fingerprint(
    key: PublishedIndexKey,
    state: PublishedIndexState | None,
) -> tuple[object, ...] | None:
    if state is None:
        return _published_state_fingerprint(
            PublishedIndexState(
                settings=key.indexing_settings,
                status="indexing",
                refresh_job="running",
                reason="refreshing",
            ),
        )
    return _published_state_fingerprint(
        replace(
            state,
            refresh_job="running",
            reason="refreshing",
            last_error=None,
        ),
    )


def _failed_subprocess_state_can_be_reconciled(
    key: PublishedIndexKey,
    state: PublishedIndexState | None,
    initial_state: PublishedIndexState | None,
) -> bool:
    if _published_state_fingerprint(state) in {
        _published_state_fingerprint(initial_state),
        _refresh_running_fingerprint(key, initial_state),
    }:
        return True
    return state is not None and state.refresh_job == "running" and state.reason == "refreshing"


async def _reconcile_cancelled_refresh(
    key: PublishedIndexKey,
    *,
    initial_state: PublishedIndexState | None,
    config: Config,
    runtime_paths: RuntimePaths,
) -> None:
    state = await asyncio.to_thread(load_published_index_state, published_index_metadata_path(key))
    state_advanced = _published_state_fingerprint(state) != _published_state_fingerprint(initial_state)
    if not state_advanced:
        return
    if (
        state is not None
        and state.status == "complete"
        and published_index_settings_compatible(state.settings, key.indexing_settings)
        and published_index_availability_for_state(key=key, state=state) is KnowledgeAvailability.READY
    ):
        if state.settings.mode == "files":
            await asyncio.to_thread(mark_published_index_refresh_succeeded, key)
            return
        if not await asyncio.to_thread(published_index_collection_exists_for_state, key, state):
            await asyncio.to_thread(mark_published_index_stale, key, reason="refresh_cancelled", refresh_job="idle")
            return
        index = publish_knowledge_index_from_state(
            key,
            state=state,
            config=config,
            runtime_paths=runtime_paths,
            metadata_path=published_index_metadata_path(key),
        )
        if index is not None:
            await asyncio.to_thread(mark_published_index_refresh_succeeded, key)
            return
    await asyncio.to_thread(mark_published_index_stale, key, reason="refresh_cancelled", refresh_job="idle")


def _load_subprocess_refresh_request(payload: bytes) -> _SubprocessRefreshRequest:
    raw_payload = json.loads(payload.decode())
    if not isinstance(raw_payload, dict):
        msg = "Knowledge refresh subprocess request must be a JSON object"
        raise TypeError(msg)
    raw_base_id = raw_payload.get("base_id")
    raw_config_data = raw_payload.get("config_data")
    raw_config_path = raw_payload.get("config_path")
    raw_storage_root = raw_payload.get("storage_root")
    raw_runtime_knowledge_base = raw_payload.get("runtime_knowledge_base")
    raw_execution_identity = raw_payload.get("execution_identity")
    raw_force_reindex = raw_payload.get("force_reindex", False)
    if not isinstance(raw_base_id, str) or not raw_base_id.strip():
        msg = "Knowledge refresh subprocess request is missing base_id"
        raise TypeError(msg)
    if not isinstance(raw_config_data, dict):
        msg = "Knowledge refresh subprocess request is missing config_data"
        raise TypeError(msg)
    if not isinstance(raw_config_path, str) or not raw_config_path.strip():
        msg = "Knowledge refresh subprocess request is missing config_path"
        raise TypeError(msg)
    if not isinstance(raw_storage_root, str) or not raw_storage_root.strip():
        msg = "Knowledge refresh subprocess request is missing storage_root"
        raise TypeError(msg)
    if raw_runtime_knowledge_base is not None and not isinstance(raw_runtime_knowledge_base, dict):
        msg = "Knowledge refresh subprocess request runtime_knowledge_base must be an object when present"
        raise TypeError(msg)
    if raw_execution_identity is not None and not isinstance(raw_execution_identity, dict):
        msg = "Knowledge refresh subprocess request execution_identity must be an object when present"
        raise TypeError(msg)
    return _SubprocessRefreshRequest(
        base_id=raw_base_id,
        config_data=raw_config_data,
        config_path=raw_config_path,
        storage_root=raw_storage_root,
        runtime_knowledge_base=cast("dict[str, object] | None", raw_runtime_knowledge_base),
        execution_identity=raw_execution_identity,
        force_reindex=bool(raw_force_reindex),
    )


async def _run_subprocess_refresh_request(payload: bytes) -> KnowledgeRefreshResult:
    request = _load_subprocess_refresh_request(payload)
    runtime_paths = resolve_runtime_paths(
        config_path=Path(request.config_path),
        storage_path=Path(request.storage_root),
        process_env=dict(os.environ),
    )
    config = Config.validate_with_runtime(request.config_data, runtime_paths, tolerate_plugin_load_errors=True)
    if request.runtime_knowledge_base is not None:
        base_config = KnowledgeBaseConfig.model_validate(request.runtime_knowledge_base)
        config = config.with_runtime_knowledge_base_overlay(request.base_id, base_config)
    execution_identity = (
        None
        if request.execution_identity is None
        else parse_tool_execution_identity_payload(
            request.execution_identity,
            error_prefix="Knowledge refresh execution_identity",
        )
    )
    return await refresh_knowledge_binding(
        request.base_id,
        config=config,
        runtime_paths=runtime_paths,
        execution_identity=execution_identity,
        force_reindex=request.force_reindex,
    )


def _parse_refresh_runner_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run one internal MindRoom knowledge refresh request.")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    """Internal CLI used by scheduled knowledge refresh subprocesses."""
    _parse_refresh_runner_args(argv)
    payload = sys.stdin.buffer.read()
    try:
        result = asyncio.run(_run_subprocess_refresh_request(payload))
    except Exception:
        logger.exception("Knowledge refresh subprocess failed")
        return 1
    logger.info(
        "Knowledge refresh subprocess completed",
        base_id=result.key.base_id,
        indexed_count=result.indexed_count,
        index_published=result.index_published,
        availability=result.availability.value,
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
