feat(core): ingest docx, pptx, and csv into document sidecar notes

CHANGELOG.md

@@ -4,6 +4,17 @@
 
 ### Features
 
+- **#1512**: Word, PowerPoint, and CSV files get the same sidecar Markdown note a
+  PDF gets. `bm import document <path>` indexes the project, extracts the file,
+  and writes `<file>.<ext>.md` next to it plus a run note under
+  `document-ingestion-runs/`, all through the existing parser-neutral document
+  contract. Office formats run through Microsoft's `markitdown` converters
+  (`basic-memory[documents]`) inside the same killable, byte-capped worker
+  process as pdf-inspector; CSV renders a bounded stdlib preview with strict
+  UTF-8. Re-running is a no-op while the source is unchanged, and the command
+  refuses to overwrite a hand-written or enriched note at the sidecar path.
+  PDF keeps pdf-inspector and works through the same command.
+
 - **#610**: The manual's SYNOPSIS blocks are now generated from the tool registry.
   `just man-regen` renders the MCP call on every section-3 page from the schema
   clients actually receive (required parameters first, then defaults, in schema

pyproject.toml

@@ -75,6 +75,9 @@ redis = [
 pdf = [
     "pdf-inspector>=0.2.6,<2",
 ]
+documents = [
+    "markitdown[docx,pptx]>=0.1.7,<0.2",
+]
 
 [project.urls]
 Homepage = "https://github.com/basicmachines-co/basic-memory"
@@ -149,6 +152,7 @@ dev = [
     "libcst>=1.8.6",
     "pytest-timeout>=2.4.0",
     "pytest-split>=0.11.0",
+    "markitdown[docx,pptx]>=0.1.7,<0.2",
 ]
 
 [tool.hatch.version]

src/basic_memory/cli/commands/import_document.py

@@ -0,0 +1,114 @@
+"""Import command: extract one document file in a project into a sidecar Markdown note."""
+
+# PEP 563 lazy annotations keep the ingestion stack out of module import (#886).
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Annotated, Optional
+
+import typer
+from loguru import logger
+from rich.console import Console
+from rich.panel import Panel
+
+from basic_memory.cli.app import import_app
+from basic_memory.cli.commands.command_utils import run_with_cleanup
+from basic_memory.cli.commands.routing import force_routing
+from basic_memory.config import ConfigManager
+
+if TYPE_CHECKING:
+    from basic_memory.document_ingestion.raw_document import RawDocumentWriteResult
+
+console = Console()
+
+
+async def import_document(path: Path, project: str | None) -> tuple[str, RawDocumentWriteResult]:
+    """Index the source, extract it, and write its sidecar note into the project."""
+    # Deferred: the ingestion stack and API client load only when the command runs (#886).
+    from basic_memory.document_ingestion.local_runtime import (
+        ApiDocumentSourceEntityResolver,
+        LocalDocumentSourceReader,
+        LocalRawDocumentWriter,
+        default_document_extractors,
+    )
+    from basic_memory.document_ingestion.raw_document import RawDocumentRuntime
+    from basic_memory.markdown import EntityParser, MarkdownProcessor
+    from basic_memory.mcp.async_client import get_client
+    from basic_memory.mcp.clients import KnowledgeClient, ProjectClient
+    from basic_memory.mcp.project_context import get_active_project
+    from basic_memory.services.file_service import FileService
+
+    config_manager = ConfigManager()
+    project_name = project or config_manager.default_project
+    async with get_client(project_name=project_name) as client:
+        project_item = await get_active_project(client, project_name, None)
+        project_home = Path(project_item.path).expanduser().resolve()
+        source = path.expanduser().resolve()
+        if not source.is_file():
+            raise typer.BadParameter(f"File not found: {path}")
+        if not source.is_relative_to(project_home):
+            raise typer.BadParameter(
+                f"{path} is not inside project {project_item.name!r} ({project_home})"
+            )
+        relative_path = source.relative_to(project_home).as_posix()
+
+        # The source must be an indexed file entity before it can own a document
+        # note, and a fresh drop may not have been seen by the watcher yet.
+        await ProjectClient(client).index(
+            project_item.external_id, force_full=False, run_in_background=False
+        )
+
+        knowledge = KnowledgeClient(client, project_item.external_id)
+        app_config = config_manager.config
+        markdown_processor = MarkdownProcessor(EntityParser(project_home), app_config=app_config)
+        file_service = FileService(project_home, markdown_processor, app_config=app_config)
+        runtime = RawDocumentRuntime(
+            source_resolver=ApiDocumentSourceEntityResolver(knowledge),
+            source_reader=LocalDocumentSourceReader(project_home),
+            extractors=default_document_extractors(),
+            writer=LocalRawDocumentWriter(file_service, knowledge),
+        )
+        result = await runtime.ingest(file_path=relative_path, observed_etag=None)
+        return project_item.name, result
+
+
+@import_app.command(
+    name="document",
+    help="Extract a PDF, docx, pptx, or csv file in a project into a sidecar Markdown note.",
+)
+def document(
+    path: Annotated[Path, typer.Argument(help="Path to a file inside the project directory")],
+    project: Annotated[
+        Optional[str],
+        typer.Option("--project", "-p", help="Project name (defaults to the default project)"),
+    ] = None,
+) -> None:
+    """Write ``<file>.md`` next to the source and a run note under document-ingestion-runs/."""
+    try:
+        # Trigger: the project may be configured for cloud routing.
+        # Why: this runtime reads the source and writes the sidecar in the local
+        #      project directory, so the API it indexes through must be the local
+        #      one; a cloud client would be asked to index files it cannot see.
+        # Outcome: local ASGI routing for the whole command, whatever the project mode.
+        with force_routing(local=True):
+            project_name, result = run_with_cleanup(import_document(path, project))
+    except typer.BadParameter as error:
+        typer.echo(f"Error: {error}", err=True)
+        raise typer.Exit(1)
+    except Exception as error:
+        logger.error("Document import failed")
+        typer.echo(f"Error during import: {error}", err=True)
+        raise typer.Exit(1)
+
+    document_state = "created" if result.document_created else "already current"
+    run_state = "created" if result.run_created else "already recorded"
+    console.print(
+        Panel(
+            f"[green]Document import complete![/green]\n\n"
+            f"Project: {project_name}\n"
+            f"Document note: {result.document_file_path} ({document_state})\n"
+            f"Run note: {result.run_file_path} ({run_state})\n"
+            f"Run id: {result.run_id}",
+            expand=False,
+        )
+    )

src/basic_memory/cli/main.py

@@ -24,6 +24,7 @@ def _version_only_invocation(argv: list[str]) -> bool:
         import_chatgpt,
         import_claude_conversations,
         import_claude_projects,
+        import_document,
         import_memory_json,
         inspect,
         install,

src/basic_memory/document_ingestion/bounded_process.py

@@ -0,0 +1,166 @@
+"""Run one extractor over untrusted bytes in a killable, byte-capped child process.
+
+Every native or third-party document parser Basic Memory runs is synchronous and
+parses attacker-controlled input. The parent therefore never calls a parser
+in-process: it feeds the bytes to ``python -m <worker>`` on stdin, drains both
+pipes with a hard ceiling, and kills the child at the deadline. The worker
+applies POSIX rlimits to itself before reading any input.
+
+This module owns only the process mechanics. Each extractor owns its argv,
+its output contract, and the error types its callers see.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Sequence
+
+
+class BoundedProcessError(RuntimeError):
+    """Base failure for one bounded child process run."""
+
+
+class BoundedProcessSpawnError(BoundedProcessError):
+    """The running event loop cannot spawn subprocesses."""
+
+
+class BoundedProcessTimeoutError(BoundedProcessError):
+    """The child was killed after exceeding its wall-clock deadline."""
+
+
+class BoundedProcessOutputLimitError(BoundedProcessError):
+    """A child pipe crossed the output byte ceiling while streaming."""
+
+    def __init__(self, stream_name: str) -> None:
+        super().__init__(f"child process exceeded the output byte limit on {stream_name}")
+        self.stream_name = stream_name
+
+
+class BoundedProcessExitError(BoundedProcessError):
+    """The child exited with a non-zero status."""
+
+    def __init__(self, returncode: int, detail: str) -> None:
+        super().__init__(f"child process failed with exit code {returncode}: {detail}")
+        self.returncode = returncode
+        self.detail = detail
+
+
+async def run_bounded_process(
+    argv: Sequence[str],
+    stdin_bytes: bytes,
+    *,
+    timeout_seconds: float,
+    max_output_bytes: int,
+) -> bytes:
+    """Run ``argv`` with ``stdin_bytes`` on stdin and return its capped stdout.
+
+    The caller bounds concurrency; this function bounds one run. It raises a
+    :class:`BoundedProcessError` subclass for every failure mode so the caller
+    can name the failure in its own domain terms.
+    """
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *argv,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    except NotImplementedError as error:
+        # Trigger: the running loop cannot spawn subprocesses. Basic Memory
+        # installs the selector loop on Windows for aiosqlite (see db.py),
+        # and that loop has no subprocess transport.
+        # Outcome: a named error instead of a bare NotImplementedError.
+        raise BoundedProcessSpawnError(
+            "child process needs an event loop with subprocess support; "
+            "Basic Memory runs the selector loop on Windows"
+        ) from error
+    try:
+        async with asyncio.timeout(timeout_seconds):
+            stdout, stderr = await _communicate_capped(process, stdin_bytes, cap=max_output_bytes)
+    except asyncio.CancelledError:
+        await kill_and_wait(process)
+        raise
+    except TimeoutError as error:
+        await kill_and_wait(process)
+        raise BoundedProcessTimeoutError(
+            "child process exceeded the configured deadline"
+        ) from error
+    except BoundedProcessOutputLimitError:
+        # A pipe crossed its ceiling while the child may still be writing.
+        await kill_and_wait(process)
+        raise
+
+    if process.returncode != 0:
+        detail = stderr.decode("utf-8", errors="replace")[-2000:].strip()
+        raise BoundedProcessExitError(
+            process.returncode if process.returncode is not None else -1,
+            detail or "no error detail",
+        )
+    return stdout
+
+
+async def kill_and_wait(process: asyncio.subprocess.Process) -> None:
+    """Terminate and reap one child before releasing worker capacity."""
+    if process.returncode is None:
+        try:
+            process.kill()
+        except ProcessLookupError:
+            # The child can exit between the returncode check and the signal.
+            # Waiting still reaps it and preserves the caller's original outcome.
+            pass
+    await process.wait()
+
+
+# Read pipes in modest chunks so the ceiling applies as bytes arrive rather than
+# after an entire stream has already been buffered.
+_PIPE_CHUNK_BYTES = 64 * 1024
+
+
+async def _communicate_capped(
+    process: asyncio.subprocess.Process,
+    stdin_bytes: bytes,
+    *,
+    cap: int,
+) -> tuple[bytes, bytes]:
+    """Feed stdin and drain both pipes with a hard byte ceiling on each.
+
+    ``Process.communicate()`` buffers stdout and stderr without limit before the
+    caller can look at their size, so a hostile document that makes the worker
+    emit a huge field, or a runaway error stream, would land in parent memory.
+    This variant raises as soon as either pipe crosses ``cap``; the caller kills
+    the child.
+    """
+    stdin, stdout, stderr = process.stdin, process.stdout, process.stderr
+    if stdin is None or stdout is None or stderr is None:  # pragma: no cover - PIPE at spawn
+        raise RuntimeError("child process was started without pipes")
+    try:
+        async with asyncio.TaskGroup() as group:
+            stdout_task = group.create_task(_read_capped(stdout, cap=cap, stream_name="stdout"))
+            stderr_task = group.create_task(_read_capped(stderr, cap=cap, stream_name="stderr"))
+            group.create_task(_feed_stdin(stdin, stdin_bytes))
+    except* BoundedProcessOutputLimitError as overflow:
+        raise overflow.exceptions[0] from None
+    await process.wait()
+    return stdout_task.result(), stderr_task.result()
+
+
+async def _read_capped(stream: asyncio.StreamReader, *, cap: int, stream_name: str) -> bytes:
+    buffer = bytearray()
+    while chunk := await stream.read(_PIPE_CHUNK_BYTES):
+        buffer.extend(chunk)
+        if len(buffer) > cap:
+            raise BoundedProcessOutputLimitError(stream_name)
+    return bytes(buffer)
+
+
+async def _feed_stdin(stdin: asyncio.StreamWriter, stdin_bytes: bytes) -> None:
+    try:
+        stdin.write(stdin_bytes)
+        await stdin.drain()
+    except (BrokenPipeError, ConnectionResetError):
+        # The child exited before consuming its input. Its exit status and
+        # stderr carry the failure; ``Process.communicate()`` ignores the same
+        # pair for the same reason.
+        pass
+    finally:
+        stdin.close()