RBT Document Editor

Instructions

Compare new file with ROOT original file and sync differences to Graphiti.

This tool compares a modified document with its original version in the ROOT
directory, chunks both versions, identifies changes, and syncs the differences
to the Graphiti knowledge graph.

Args:
    new_file_path: Absolute path to the new/modified file
    project_id: Project identifier (e.g., "knowledge-smith")
    feature_id: Feature identifier (required for RBT documents)
    rbt_type: RBT document type ("REQ"/"BP"/"TASK"). Leave as None for general documents.
    file_path:
        - For RBT TASK: task identifier (e.g., "006")
        - For general files: relative path (e.g., "todos/xxx.md" or "docs/todos/xxx.md")
          Note: "docs/" prefix is optional and will be handled automatically.

Returns:
    Sync statistics:
    {
        "status": "success",
        "added": 3,      # Number of chunks added
        "updated": 2,    # Number of chunks updated
        "deleted": 1,    # Number of chunks deleted
        "unchanged": 5,  # Number of chunks unchanged
        "total": 11      # Total chunks
    }

Raises:
    FileNotFoundError: If new file not found
    ValueError: If invalid rbt_type or parameter combination

Examples:
    # RBT TASK document
    add_document(
        new_file_path="/Users/me/workspace/TASK-006-AddDocument.md",
        project_id="knowledge-smith",
        feature_id="graphiti-chunk-mcp",
        rbt_type="TASK",
        file_path="006"
    )

    # RBT BP document
    add_document(
        new_file_path="/Users/me/workspace/BP-graphiti-chunk-mcp.md",
        project_id="knowledge-smith",
        feature_id="graphiti-chunk-mcp",
        rbt_type="BP"
    )

    # General document (both work)
    add_document(
        new_file_path="/Users/me/workspace/TODO-001.md",
        project_id="General",
        file_path="todos/TODO-001.md"  # or "docs/todos/TODO-001.md"
    )

@REQ: REQ-graphiti-chunk-mcp
@BP: BP-graphiti-chunk-mcp
@TASK: TASK-006-AddDocument

Implementation Reference

• rbt_mcp_server/server.py:229-330 (handler)

  MCP tool handler for 'add_document'. Validates inputs, creates Graphiti client, calls core implementation, formats and returns sync results.

  @mcp.tool()
  async def add_document(
      new_file_path: str,
      project_id: str,
      feature_id: Optional[str] = None,
      rbt_type: Optional[str] = None,
      file_path: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Compare new file with ROOT original file and sync differences to Graphiti.
  
      This tool compares a modified document with its original version in the ROOT
      directory, chunks both versions, identifies changes, and syncs the differences
      to the Graphiti knowledge graph.
  
      Args:
          new_file_path: Absolute path to the new/modified file
          project_id: Project identifier (e.g., "knowledge-smith")
          feature_id: Feature identifier (required for RBT documents)
          rbt_type: RBT document type ("REQ"/"BP"/"TASK"). Leave as None for general documents.
          file_path:
              - For RBT TASK: task identifier (e.g., "006")
              - For general files: relative path (e.g., "todos/xxx.md" or "docs/todos/xxx.md")
                Note: "docs/" prefix is optional and will be handled automatically.
  
      Returns:
          Sync statistics:
          {
              "status": "success",
              "added": 3,      # Number of chunks added
              "updated": 2,    # Number of chunks updated
              "deleted": 1,    # Number of chunks deleted
              "unchanged": 5,  # Number of chunks unchanged
              "total": 11      # Total chunks
          }
  
      Raises:
          FileNotFoundError: If new file not found
          ValueError: If invalid rbt_type or parameter combination
  
      Examples:
          # RBT TASK document
          add_document(
              new_file_path="/Users/me/workspace/TASK-006-AddDocument.md",
              project_id="knowledge-smith",
              feature_id="graphiti-chunk-mcp",
              rbt_type="TASK",
              file_path="006"
          )
  
          # RBT BP document
          add_document(
              new_file_path="/Users/me/workspace/BP-graphiti-chunk-mcp.md",
              project_id="knowledge-smith",
              feature_id="graphiti-chunk-mcp",
              rbt_type="BP"
          )
  
          # General document (both work)
          add_document(
              new_file_path="/Users/me/workspace/TODO-001.md",
              project_id="General",
              file_path="todos/TODO-001.md"  # or "docs/todos/TODO-001.md"
          )
  
      @REQ: REQ-graphiti-chunk-mcp
      @BP: BP-graphiti-chunk-mcp
      @TASK: TASK-006-AddDocument
      """
      # Read environment variable
      root_dir = os.environ.get("RBT_ROOT_DIR")
      if not root_dir:
          raise ValueError("Environment variable RBT_ROOT_DIR is not set")
  
      # Validate new file exists
      if not os.path.exists(new_file_path):
          raise FileNotFoundError(f"New file does not exist: {new_file_path}")
  
      # Create GraphitiClient
      client = graphiti_tools.get_graphiti_client()
  
      async with client:
          result = await add_document_impl(
              new_file_path=new_file_path,
              project_id=project_id,
              feature_id=feature_id,
              doc_type=rbt_type,  # Internal implementation still uses doc_type
              file_path=file_path,
              root_dir=root_dir,
              graphiti_client=client
          )
  
      # Return simplified statistics
      return {
          "status": "success",
          "added": len(result.added),
          "updated": len(result.updated),
          "deleted": len(result.deleted),
          "unchanged": result.unchanged,
          "total": result.total_chunks
      }

• rbt_mcp_server/chunking/add_document.py:89-296 (handler)

  Core implementation of document addition: reads new/ROOT files, chunks them, compares changes, updates ROOT file, syncs chunks to Graphiti knowledge graph.

  async def add_document(
      new_file_path: str,
      project_id: str,
      feature_id: Optional[str],
      doc_type: Optional[str],
      file_path: Optional[str],
      root_dir: str,
      graphiti_client: GraphitiClient,
      sync_mode: bool = False
  ) -> SyncResult:
      """
      Compare new file with ROOT original file and sync differences to Graphiti.
  
      @REQ: REQ-graphiti-chunk-mcp
      @BP: BP-graphiti-chunk-mcp
      @TASK: TASK-006-AddDocument
  
      This is the main entry point for document comparison and syncing pipeline.
      It performs the following steps:
      1. Reads new file content
      2. Uses PathResolver to locate ROOT original file
      3. Reads ROOT original file content
      4. Chunks both new and old files
      5. Compares chunks to identify changes
      6. Synchronizes changes to Graphiti (add/update/delete episodes)
      7. Returns sync results
  
      Args:
          new_file_path: Absolute path to the new/modified file
          project_id: Project identifier (e.g., 'knowledge-smith')
          feature_id: Feature identifier (e.g., 'graphiti-chunk-mcp'), optional for general docs
          doc_type: Document type (e.g., 'REQ', 'BP', 'TASK', or None for general docs)
          file_path: TASK identifier or relative path for general documents
          root_dir: Root directory for document resolution
          graphiti_client: GraphitiClient instance for Graphiti operations
          sync_mode: If True, wait for Graphiti sync to complete (for testing).
                     If False (default), schedule Graphiti sync in background and return immediately.
  
      Returns:
          SyncResult containing statistics about added, updated, deleted, and unchanged chunks.
          Note: In background mode (sync_mode=False), the sync is scheduled but may not be
          complete when this function returns.
  
      Raises:
          FileNotFoundError: If new file or ROOT original file cannot be found
          ValueError: If invalid parameters provided
          RuntimeError: If Graphiti operations fail
  
      Examples:
          >>> # Sync TASK document
          >>> async with GraphitiClient(...) as client:
          ...     result = await add_document(
          ...         new_file_path="/path/to/TASK-006-AddDocument.md",
          ...         project_id="knowledge-smith",
          ...         feature_id="graphiti-chunk-mcp",
          ...         doc_type="TASK",
          ...         file_path="006",
          ...         root_dir="/path/to/root",
          ...         graphiti_client=client
          ...     )
          ...     print(f"Added: {len(result.added)}, Updated: {len(result.updated)}")
      """
      logger.info(f"Starting add_document: new_file={new_file_path}, "
                  f"project_id={project_id}, feature_id={feature_id}, "
                  f"doc_type={doc_type}, file_path={file_path}")
  
      try:
          # Step 0: Validate parameters
          logger.debug("Step 0: Validating parameters")
  
          # Validate doc_type
          if doc_type is not None and doc_type not in VALID_RBT_TYPES:
              raise ValueError(
                  f"Invalid doc_type: '{doc_type}'. "
                  f"Must be one of {sorted(VALID_RBT_TYPES)} for RBT documents, or None for general documents.\n"
                  f"For general documents (e.g., todos, guides), set doc_type=None and use file_path "
                  f"(e.g., file_path='todos/TODO-001.md')."
              )
  
          # Step 1: Validate and read new file
          logger.debug("Step 1: Reading new file")
          if not os.path.exists(new_file_path):
              raise FileNotFoundError(f"New file does not exist: {new_file_path}")
  
          with open(new_file_path, 'r', encoding='utf-8') as f:
              new_content = f.read()
          logger.info(f"Read {len(new_content)} characters from new file")
  
          # Step 1.5: For TASK documents, extract full task identifier from frontmatter if needed
          if doc_type == "TASK" and file_path:
              # Check if file_path is a simple number (e.g., "001")
              # If so, try to extract full task ID from document frontmatter
              if file_path.replace("TASK-", "").replace("-", "").isdigit():
                  logger.debug("Step 1.5: Extracting full TASK identifier from frontmatter")
                  doc_id = _extract_document_id_from_frontmatter(new_content)
                  if doc_id and doc_id.startswith("TASK-"):
                      # Extract the full task identifier (e.g., "TASK-001-AddDocument" → "001-AddDocument")
                      task_full_id = doc_id.replace("TASK-", "", 1)
                      logger.info(f"Extracted full TASK identifier from frontmatter: {task_full_id}")
                      file_path = task_full_id
                  else:
                      logger.warning(f"Could not extract TASK ID from frontmatter, using original file_path: {file_path}")
  
          # Step 2: Use PathResolver to locate ROOT original file
          logger.debug("Step 2: Resolving ROOT original file")
          resolver = PathResolver(root_dir=root_dir)
  
          try:
              old_path_info = resolver.resolve(
                  project_id=project_id,
                  feature_id=feature_id,
                  doc_type=doc_type,
                  file_path=file_path
              )
          except FileNotFoundError as e:
              # Enhance error message
              raise FileNotFoundError(
                  f"Could not find ROOT original file\n"
                  f"Please check parameters:\n"
                  f"  project_id: {project_id}\n"
                  f"  feature_id: {feature_id}\n"
                  f"  doc_type: {doc_type}\n"
                  f"  file_path: {file_path}\n"
                  f"Original error: {e}"
              ) from e
  
          logger.info(f"Resolved ROOT file path: {old_path_info.file_path} (exists={old_path_info.file_exists})")
  
          # Step 3: Read ROOT original file content (if exists)
          if not old_path_info.file_exists:
              # ROOT file does not exist - treat as new document
              logger.info(f"ROOT file does not exist, treating as new document: {old_path_info.file_path}")
              old_chunks = []
          else:
              # ROOT file exists - read and chunk it
              logger.debug("Step 3: Reading ROOT original file")
              with open(old_path_info.file_path, 'r', encoding='utf-8') as f:
                  old_content = f.read()
              logger.info(f"Read {len(old_content)} characters from ROOT file")
  
              # Chunk ROOT file
              logger.debug("Step 3b: Chunking ROOT file")
              old_chunks = _chunk_document(
                  document_content=old_content,
                  project_id=project_id,
                  feature_id=feature_id or old_path_info.feature_id,
                  doc_type=doc_type or old_path_info.doc_type
              )
              logger.info(f"Created {len(old_chunks)} chunks from ROOT file")
  
          # Step 4: Chunk new file
          logger.debug("Step 4: Chunking new file")
          new_chunks = _chunk_document(
              document_content=new_content,
              project_id=project_id,
              feature_id=feature_id or old_path_info.feature_id,
              doc_type=doc_type or old_path_info.doc_type
          )
          logger.info(f"Created {len(new_chunks)} chunks from new file")
  
          # Step 5: Compare chunks
          logger.debug("Step 5: Comparing chunks")
          comparator = ChunkComparator()
          sync_result = comparator.compare(old_chunks=old_chunks, new_chunks=new_chunks)
          logger.info(f"Comparison result: {len(sync_result.added)} added, "
                     f"{len(sync_result.updated)} updated, {len(sync_result.deleted)} deleted, "
                     f"{sync_result.unchanged} unchanged")
  
          # Step 6: Update ROOT file immediately after comparison
          logger.debug("Step 6: Updating ROOT file")
          _update_root_file(new_file_path, old_path_info.file_path)
          logger.info(f"Updated ROOT file: {old_path_info.file_path}")
  
          # Step 7: Synchronize changes to Graphiti
          if sync_mode:
              # Synchronous mode: Wait for Graphiti sync to complete
              logger.debug("Step 7: Synchronizing changes to Graphiti (sync mode)")
              await _sync_chunks_to_graphiti(
                  graphiti_client=graphiti_client,
                  new_chunks=new_chunks,
                  sync_result=sync_result,
                  project_id=project_id
              )
              logger.info("Successfully synchronized all changes to Graphiti (sync mode)")
          else:
              # Asynchronous mode: Schedule background synchronization
              logger.debug("Step 7: Scheduling background synchronization to Graphiti")
              asyncio.create_task(
                  _sync_chunks_to_graphiti_background(
                      new_chunks=new_chunks,
                      sync_result=sync_result,
                      project_id=project_id
                  )
              )
              logger.info("Background synchronization task scheduled")
  
          # Step 8: Return sync results
          logger.info(f"add_document completed successfully. Total chunks: {sync_result.total_chunks}")
          return sync_result
  
      except FileNotFoundError:
          raise
      except ValueError:
          raise
      except Exception as e:
          logger.error(f"Unexpected error in add_document: {e}", exc_info=True)
          raise RuntimeError(f"Failed to add document: {e}") from e

• rbt_mcp_server/server.py:16-16 (registration)

  Import of the add_document implementation aliased as add_document_impl for use in the MCP tool handler.

  from .chunking import add_document as add_document_impl

• rbt_mcp_server/chunking/models.py:72-119 (schema)

  Pydantic model for SyncResult used internally to represent chunk sync statistics, which are returned by the tool.

  class SyncResult(BaseModel):
      """
      Result of comparing old and new chunks to identify changes.
  
      @REQ: REQ-graphiti-chunk-mcp
      @BP: BP-graphiti-chunk-mcp
      @TASK: TASK-004-ChunkComparator
  
      Attributes:
          added: List of chunk_ids that are new in new_chunks
          updated: List of chunk_ids that exist in both but have different content
          deleted: List of chunk_ids that exist in old_chunks but not in new_chunks
          unchanged: Count of chunks that are identical
          total_chunks: Total number of chunks in new_chunks
      """
  
      added: List[str] = Field(
          default_factory=list,
          description="List of chunk_ids that are new"
      )
      updated: List[str] = Field(
          default_factory=list,
          description="List of chunk_ids that have been updated"
      )
      deleted: List[str] = Field(
          default_factory=list,
          description="List of chunk_ids that have been deleted"
      )
      unchanged: int = Field(
          0,
          description="Count of chunks that are identical"
      )
      total_chunks: int = Field(
          0,
          description="Total number of chunks in new_chunks"
      )
  
      model_config = ConfigDict(
          json_schema_extra={
              "example": {
                  "added": ["chunk-id-1", "chunk-id-2"],
                  "updated": ["chunk-id-3"],
                  "deleted": ["chunk-id-4"],
                  "unchanged": 5,
                  "total_chunks": 8
              }
          }
      )