jarvis deep-research-setup — auto-detect local sources, ingest, and chat.
Walks the user through connecting local data sources (Apple Notes, iMessage,
Obsidian), ingesting them into a shared KnowledgeStore, and launching an
interactive Deep Research chat session with Qwen3.5 via Ollama.
Classes
Functions
detect_local_sources(*, notes_db_path: Optional[Path] = None, imessage_db_path: Optional[Path] = None, obsidian_vault_path: Optional[Path] = None) -> List[Dict[str, Any]]
Return a list of available local sources with their config.
Each entry is a dict with keys: connector_id, display_name,
config (kwargs for the connector constructor).
Source code in src/openjarvis/cli/deep_research_setup_cmd.py
| def detect_local_sources(
*,
notes_db_path: Optional[Path] = None,
imessage_db_path: Optional[Path] = None,
obsidian_vault_path: Optional[Path] = None,
) -> List[Dict[str, Any]]:
"""Return a list of available local sources with their config.
Each entry is a dict with keys: ``connector_id``, ``display_name``,
``config`` (kwargs for the connector constructor).
"""
sources: List[Dict[str, Any]] = []
notes_path = notes_db_path or _DEFAULT_NOTES_DB
if notes_path.exists():
sources.append(
{
"connector_id": "apple_notes",
"display_name": "Apple Notes",
"config": {"db_path": str(notes_path)},
}
)
imessage_path = imessage_db_path or _DEFAULT_IMESSAGE_DB
if imessage_path.exists():
sources.append(
{
"connector_id": "imessage",
"display_name": "iMessage",
"config": {"db_path": str(imessage_path)},
}
)
if obsidian_vault_path and obsidian_vault_path.is_dir():
sources.append(
{
"connector_id": "obsidian",
"display_name": "Obsidian / Markdown",
"config": {"vault_path": str(obsidian_vault_path)},
}
)
return sources
|
detect_token_sources(*, connectors_dir: Optional[Path] = None) -> List[Dict[str, Any]]
Return token-based sources that already have valid credentials.
Source code in src/openjarvis/cli/deep_research_setup_cmd.py
| def detect_token_sources(
*,
connectors_dir: Optional[Path] = None,
) -> List[Dict[str, Any]]:
"""Return token-based sources that already have valid credentials."""
cdir = connectors_dir or (DEFAULT_CONFIG_DIR / "connectors")
sources: List[Dict[str, Any]] = []
for ts in _TOKEN_SOURCES:
creds_file = cdir / ts["creds_file"]
if not creds_file.exists():
continue
try:
data = json.loads(creds_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
continue
if not data or not any(v for v in data.values() if v):
continue
sources.append(
{
"connector_id": ts["connector_id"],
"display_name": ts["display_name"],
"config": {},
}
)
return sources
|
ingest_sources(sources: List[Dict[str, Any]], store: KnowledgeStore, *, state_db: str = '') -> int
Connect and ingest all sources into the KnowledgeStore.
| PARAMETER |
DESCRIPTION |
state_db
|
Path for the SyncEngine checkpoint database. Defaults to
~/.openjarvis/sync_state.db when empty.
TYPE:
str
DEFAULT:
''
|
Returns
|
|
Source code in src/openjarvis/cli/deep_research_setup_cmd.py
| def ingest_sources(
sources: List[Dict[str, Any]],
store: KnowledgeStore,
*,
state_db: str = "",
) -> int:
"""Connect and ingest all sources into the KnowledgeStore.
Parameters
----------
state_db:
Path for the SyncEngine checkpoint database. Defaults to
``~/.openjarvis/sync_state.db`` when empty.
Returns total chunks indexed across all sources.
"""
pipeline = IngestionPipeline(store)
engine = SyncEngine(pipeline, state_db=state_db)
total = 0
for src in sources:
connector = _instantiate_connector(src["connector_id"], src["config"])
chunks = engine.sync(connector)
total += chunks
return total
|
deep_research_setup(obsidian_vault: Optional[str], skip_chat: bool) -> None
Auto-detect local data sources, ingest, and launch Deep Research chat.
Source code in src/openjarvis/cli/deep_research_setup_cmd.py
| @click.command("deep-research-setup")
@click.option(
"--obsidian-vault",
type=click.Path(exists=True, file_okay=False),
default=None,
help="Path to an Obsidian vault directory.",
)
@click.option("--skip-chat", is_flag=True, help="Ingest only, don't launch chat.")
def deep_research_setup(obsidian_vault: Optional[str], skip_chat: bool) -> None:
"""Auto-detect local data sources, ingest, and launch Deep Research chat."""
console = Console()
console.print("\n[bold]Deep Research Setup[/bold]\n")
# 1. Detect local sources
vault_path = Path(obsidian_vault) if obsidian_vault else None
local_sources = detect_local_sources(obsidian_vault_path=vault_path)
# 2. Detect already-connected token sources
token_sources = detect_token_sources()
all_sources = local_sources + token_sources
# 3. Show what we found
if all_sources:
table = Table(title="Detected Sources")
table.add_column("Source", style="bold")
table.add_column("Type", style="dim")
table.add_column("Status", style="green")
for src in local_sources:
table.add_row(src["display_name"], "local", "ready")
for src in token_sources:
table.add_row(src["display_name"], "token", "connected")
console.print(table)
console.print()
# 4. Offer to connect new token sources
newly_connected = _prompt_connect_sources(console)
all_sources.extend(newly_connected)
if not all_sources:
console.print(
"[yellow]No data sources detected or connected.[/yellow]\n"
"On macOS, ensure Full Disk Access is granted in "
"System Settings > Privacy & Security."
)
sys.exit(1)
# 5. Confirm and ingest
if not click.confirm("Ingest these sources?", default=True):
sys.exit(0)
db_path = DEFAULT_CONFIG_DIR / "knowledge.db"
db_path.parent.mkdir(parents=True, exist_ok=True)
store = KnowledgeStore(str(db_path))
console.print("\n[bold]Ingesting...[/bold]")
for src in all_sources:
try:
connector = _instantiate_connector(
src["connector_id"],
src["config"],
)
pipeline = IngestionPipeline(store)
engine = SyncEngine(pipeline)
chunks = engine.sync(connector)
console.print(f" {src['display_name']}: [green]{chunks} chunks[/green]")
except Exception as exc: # noqa: BLE001
console.print(f" {src['display_name']}: [red]error: {exc}[/red]")
total = store.count()
console.print(
f"\n[bold green]Done![/bold green] {total} total chunks in {db_path}\n"
)
# 6. Chat
if skip_chat:
return
_launch_chat(store, console)
|