feat(cli): add 'add-all' command and auto_delete_added_files config

openkb/agent/compiler.py

@@ -260,6 +260,71 @@
 # ---------------------------------------------------------------------------
 
 
+def _should_retry_exception(exc: Exception) -> bool:
+    """Determine whether an exception is retryable (transient error).
+
+    Returns True for temporary API/network errors that may succeed on retry:
+    - Timeout (client-side or server-side)
+    - APIError 5xx (server errors)
+    - RateLimitError (429)
+    - ConnectionError / ServiceUnavailableError
+
+    Returns False for permanent errors that won't be fixed by retry:
+    - TruncatedResponseError (model hit max_tokens)
+    - ValueError, TypeError (malformed input/output)
+    - AuthenticationError (credentials issue)
+    - BadRequestError (invalid parameters)
+    - Unknown error types (conservative approach)
+    """
+    exc_type_name = type(exc).__name__
+
+    # ===== RETRYABLE (transient errors) =====
+
+    # Timeout (network/gateway timeout)
+    if "Timeout" in exc_type_name:
+        return True
+
+    # Generic API errors (5xx range, but not 4xx)
+    if "APIError" in exc_type_name:
+        # Don't retry if it's a BadRequest/Invalid error (4xx)
+        if "Invalid" not in exc_type_name and "BadRequest" not in exc_type_name:
+            return True
+
+    # Rate limiting (429)
+    if "RateLimitError" in exc_type_name or "Rate" in exc_type_name:
+        return True
+
+    # Connection errors
+    if "ConnectionError" in exc_type_name:
+        return True
+
+    # Service unavailable
+    if "ServiceUnavailable" in exc_type_name:
+        return True
+
+    # ===== NOT RETRYABLE (permanent errors) =====
+
+    # Model hit max_tokens limit
+    if isinstance(exc, TruncatedResponseError):
+        return False
+
+    # Content validation failures
+    if "ValueError" in exc_type_name or "TypeError" in exc_type_name:
+        return False
+
+    # Authentication failures
+    if "Auth" in exc_type_name or "Permission" in exc_type_name:
+        return False
+
+    # Bad parameters/requests
+    if "BadRequest" in exc_type_name or "Invalid" in exc_type_name:
+        return False
+
+    # ===== UNKNOWN: Conservative approach =====
+    # Don't retry errors we don't recognize
+    return False
+
+
 def _cached_text(text: str) -> list[dict]:
     """Wrap a text payload into a content-block list with an Anthropic
     ephemeral cache_control marker.
@@ -406,7 +471,11 @@ def _llm_call(
     bundle=None,
     **kwargs,
 ) -> str:
-    """Single LLM call with animated progress and debug logging."""
+    """Single LLM call with animated progress, debug logging, and retry support.
+
+    Transient errors (Timeout, 5xx, 429) are automatically retried by LiteLLM.
+    Permanent errors (4xx, truncation, validation) are raised immediately.
+    """
     messages = _prepare_messages(model, messages)
     extra_headers = bundle.extra_headers if bundle is not None else get_extra_headers()
     if extra_headers:
@@ -417,6 +486,10 @@ def _llm_call(
     if bundle is not None:
         kwargs.setdefault("api_key", bundle.api_key)
         kwargs.setdefault("base_url", bundle.base_url)
+
+    # NEW: Retry configuration for transient errors (fixed: 2 retries, base 2)
+    kwargs.setdefault("retries", 2)
+
     logger.debug("LLM request [%s]:\n%s", step_name, _fmt_messages(messages))
     if kwargs:
         logger.debug("LLM kwargs [%s]: %s", step_name, kwargs)
@@ -425,7 +498,27 @@ def _llm_call(
     spinner.start()
     t0 = time.time()
 
-    response = litellm.completion(model=model, messages=messages, **kwargs)
+    try:
+        response = litellm.completion(model=model, messages=messages, **kwargs)
+    except Exception as exc:
+        # NEW: Better error logging with retry context
+        if _should_retry_exception(exc):
+            logger.warning(
+                "LLM [%s] failed with transient error (retries applied by LiteLLM): %s",
+                step_name,
+                exc,
+                exc_info=False,  # Don't spam stack traces for known transient errors
+            )
+        else:
+            logger.warning(
+                "LLM [%s] failed with permanent error (no retry): %s",
+                step_name,
+                exc,
+                exc_info=True,  # Full trace for unexpected errors
+            )
+        spinner.stop("[FAILED]")
+        raise
+
     content = response.choices[0].message.content or ""
     truncated = _warn_if_truncated(response, step_name, kwargs.get("max_tokens"))
 
@@ -449,7 +542,11 @@ async def _llm_call_async(
     bundle=None,
     **kwargs,
 ) -> str:
-    """Async LLM call with timing output and debug logging."""
+    """Async LLM call with timing output, debug logging, and retry support.
+
+    Transient errors (Timeout, 5xx, 429) are automatically retried by LiteLLM.
+    Permanent errors (4xx, truncation, validation) are raised immediately.
+    """
     messages = _prepare_messages(model, messages)
     extra_headers = bundle.extra_headers if bundle is not None else get_extra_headers()
     if extra_headers:
@@ -460,13 +557,36 @@ async def _llm_call_async(
     if bundle is not None:
         kwargs.setdefault("api_key", bundle.api_key)
         kwargs.setdefault("base_url", bundle.base_url)
+
+    # NEW: Retry configuration for transient errors (fixed: 2 retries, base 2)
+    kwargs.setdefault("retries", 2)
+
     logger.debug("LLM request [%s]:\n%s", step_name, _fmt_messages(messages))
     if kwargs:
         logger.debug("LLM kwargs [%s]: %s", step_name, kwargs)
 
     t0 = time.time()
 
-    response = await litellm.acompletion(model=model, messages=messages, **kwargs)
+    try:
+        response = await litellm.acompletion(model=model, messages=messages, **kwargs)
+    except Exception as exc:
+        # NEW: Better error logging with retry context
+        if _should_retry_exception(exc):
+            logger.warning(
+                "LLM [%s] failed with transient error (retries applied by LiteLLM): %s",
+                step_name,
+                exc,
+                exc_info=False,
+            )
+        else:
+            logger.warning(
+                "LLM [%s] failed with permanent error (no retry): %s",
+                step_name,
+                exc,
+                exc_info=True,
+            )
+        raise
+
     content = response.choices[0].message.content or ""
     truncated = _warn_if_truncated(response, step_name, kwargs.get("max_tokens"))
 

openkb/cli.py

@@ -452,6 +452,32 @@ def add_single_file(
         return _add_single_file_locked(file_path, kb_dir, stage=stage, bundle=bundle)
 
 
+def _delete_if_auto_cleanup_enabled(
+    file_path: Path, status: Literal["added", "skipped", "failed"], config: dict
+) -> bool:
+    """Delete file if auto_delete_added_files is enabled and ingestion succeeded/skipped.
+
+    Deletes on both "added" (successful ingestion) and "skipped" (duplicate already
+    in KB) to keep raw/ directory clean. Preserves files on "failed" to allow retries.
+
+    Args:
+        file_path: Path to the file to potentially delete.
+        status: Result status from add_single_file ("added", "skipped", or "failed").
+        config: Configuration dict (typically from resolve_effective_config).
+
+    Returns:
+        True if file was deleted, False otherwise.
+    """
+    if status in ("added", "skipped") and config.get("auto_delete_added_files", False):
+        try:
+            file_path.unlink(missing_ok=True)
+            return True
+        except Exception as exc:
+            logger.warning(f"Failed to delete {file_path.name}: {exc}")
+            return False
+    return False
+
+
 def _add_single_file_locked(
     file_path: Path, kb_dir: Path, *, stage: bool = True, bundle=None
 ) -> Literal["added", "skipped", "failed"]:
@@ -1086,6 +1112,10 @@ def add(ctx, path, from_pageindex_cloud):
     Alternatively, pass --from-pageindex-cloud <DOC_ID> to import a document
     that is already indexed in PageIndex Cloud, with no local file. Requires
     the PAGEINDEX_API_KEY environment variable.
+
+    If ``auto_delete_added_files`` is enabled in config.yaml, files are
+    automatically deleted after ingestion (both on successful addition and
+    on skip/duplicate).
     """
     kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
     if kb_dir is None:
@@ -1106,6 +1136,8 @@ def add(ctx, path, from_pageindex_cloud):
         click.echo("Provide a PATH or use --from-pageindex-cloud <DOC_ID>.")
         return
 
+    config = resolve_effective_config(kb_dir)[0]
+
     # URL ingest: download into raw/ first, then call add_single_file explicitly.
     # Keep staged conversion enabled so converted source artifacts do not touch
     # the live KB before the mutation snapshot exists. The tri-state outcome
@@ -1123,6 +1155,8 @@ def add(ctx, path, from_pageindex_cloud):
         # indexing has already succeeded but compilation didn't.
         if outcome == "skipped":
             fetched.unlink(missing_ok=True)
+        else:
+            _delete_if_auto_cleanup_enabled(fetched, outcome, config)
         return
 
     target = Path(path)
@@ -1143,15 +1177,71 @@ def add(ctx, path, from_pageindex_cloud):
         click.echo(f"Found {total} supported file(s) in {path}.")
         for i, f in enumerate(files, 1):
             click.echo(f"\n[{i}/{total}] ", nl=False)
-            add_single_file(f, kb_dir)
+            outcome = add_single_file(f, kb_dir)
+            _delete_if_auto_cleanup_enabled(f, outcome, config)
     else:
         if target.suffix.lower() not in SUPPORTED_EXTENSIONS:
             click.echo(
                 f"Unsupported file type: {target.suffix}. "
                 f"Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}"
             )
             return
-        add_single_file(target, kb_dir)
+        outcome = add_single_file(target, kb_dir)
+        _delete_if_auto_cleanup_enabled(target, outcome, config)
+
+
+@cli.command()
+@click.pass_context
+@_with_kb_lock(exclusive=True)
+def add_all(ctx):
+    """Process all files in the ``raw/`` directory and add them to the knowledge base.
+
+    This command walks the ``raw/`` directory recursively for all supported
+    document types and ingests them into the KB. If ``auto_delete_added_files``
+    is enabled in config.yaml, files are automatically deleted after ingestion
+    (both on successful addition and on skip/duplicate).
+
+    Returns a summary of the operation (added, skipped, failed, deleted counts).
+    """
+    kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
+    if kb_dir is None:
+        click.echo("No knowledge base found. Run `openkb init` first.")
+        return
+
+    raw_dir = kb_dir / "raw"
+    if not raw_dir.is_dir():
+        click.echo(f"No raw/ directory found at {raw_dir}")
+        return
+
+    files = [
+        f
+        for f in sorted(raw_dir.rglob("*"))
+        if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS
+    ]
+    if not files:
+        click.echo("No supported files found in raw/ directory.")
+        return
+
+    config = resolve_effective_config(kb_dir)[0]
+    total = len(files)
+    added = skipped = failed = deleted = 0
+
+    click.echo(f"Processing {total} file(s) from raw/ directory...")
+    for i, f in enumerate(files, 1):
+        click.echo(f"\n[{i}/{total}] ", nl=False)
+        outcome = add_single_file(f, kb_dir)
+        if outcome == "added":
+            added += 1
+        elif outcome == "skipped":
+            skipped += 1
+        else:
+            failed += 1
+        if _delete_if_auto_cleanup_enabled(f, outcome, config):
+            deleted += 1
+
+    click.echo(
+        f"\n\nSummary: Added: {added}, Skipped: {skipped}, Failed: {failed}, Deleted: {deleted}"
+    )
 
 
 def _stream_to_tty() -> bool:

openkb/config.py

@@ -36,6 +36,7 @@
     # global/KB list overrides it wholesale; resolve_entity_types cleans the
     # effective value on read.
     "entity_types": list(DEFAULT_ENTITY_TYPES),
+    "auto_delete_added_files": False,
 }
 
 GLOBAL_CONFIG_DIR = Path.home() / ".config" / "openkb"