luplo.cli¶
luplo CLI — lp command-line interface.
All commands delegate to LocalBackend. Async core functions are
bridged via asyncio.run(). Configuration is loaded from .luplo
file → env vars → CLI flags (highest priority wins).
Attributes¶
Functions¶
|
Run alembic migrations against |
|
Initialise luplo in the current directory. |
|
Add a new item. |
|
Show a single item. |
|
List items for a project. |
|
Search items using glossary-expanded tsquery. |
|
Open a new work unit. |
|
List work units for a project, ordered by created_at DESC. |
|
Find in-progress work units by title. |
|
Close a work unit. Refuses if an in_progress task remains (use --force). |
|
Add a new system. |
|
List all systems for a project. |
|
List glossary groups. |
|
Show terms awaiting curation. |
|
Approve a pending term into a group. |
|
Reject a term permanently. |
|
Create a glossary group plus its canonical surface term. |
|
Add a new alias (or canonical) term to an existing group. |
|
Permanently remove a glossary term. |
|
Get a project brief — active work + recent decisions. |
|
Traverse typed edges to find an item's blast radius. |
|
Run the deterministic rule pack and report findings. |
|
Start the background worker (sync jobs + glossary processing). |
|
Add a new task in 'proposed' status. |
|
List tasks (chain heads) for a work unit, ordered by sort_order. |
|
Show a single task (resolved to chain head). |
|
Transition task to 'in_progress' (enforces 1 in_progress per WU). |
|
Transition task to 'done'. |
|
Transition task to 'blocked' (auto-creates a decision item). |
|
Transition task to 'skipped' (terminal). |
|
Reorder tasks (in-place sort_order update — P10). |
|
Edit a task's title / body / sort_order via a supersede row. |
|
Show the current in_progress task for a work unit, if any. |
|
Add a new qa_check in 'pending' status. |
|
List qa_checks. With --task / --item shows pending qa for that target. |
|
Show a single qa_check (chain head). |
|
|
|
|
|
|
|
|
|
|
|
Save raw text into the capture backlog. |
|
List recent captures, newest first. |
|
Full-text search over captures, newest first within rank. |
|
Move a capture to another review state. |
|
Discard a capture so default list/search hides it. |
|
Redact capture content from normal storage. |
|
Store caller-supplied BYOLLM annotation hints on a capture. |
|
Explicitly promote a capture into a curated item. |
|
Append an ideation note to a work unit (append-only, redact-only). |
|
List ideas for a work unit, newest first. |
|
Full-text search over ideas in a project. All filters are optional. |
|
Mark an idea redacted (idempotent). |
|
Stage a new import bundle and emit its manifest as JSON. |
|
Apply agent-produced results to a staged import bundle. |
Module Contents¶
- luplo.cli.app¶
- luplo.cli.items_app¶
- luplo.cli.work_app¶
- luplo.cli.systems_app¶
- luplo.cli.glossary_app¶
- luplo.cli.glossary_group_app¶
- luplo.cli.glossary_term_app¶
- luplo.cli.task_app¶
- luplo.cli.qa_app¶
- luplo.cli.idea_app¶
- luplo.cli.capture_app¶
- luplo.cli.import_app¶
- luplo.cli.migrate(db_url: str = typer.Option('', '--db-url', envvar='LUPLO_DB_URL', help='PostgreSQL connection URL. Defaults to $LUPLO_DB_URL.')) None¶
Run alembic migrations against
LUPLO_DB_URL.Idempotent — safe to call from container boot scripts. Does not read
.luploand does not require a config file. Production deploys should invoke this directly (e.g. instart.sh) before launching the application... rubric:: Examples
LUPLO_DB_URL=postgresql://… lp migrate lp migrate –db-url postgresql://…
- luplo.cli.init(project: str = typer.Option(..., '--project', '-p', help="Project ID (e.g. 'hearthward'). Stored in .luplo and created in DB."), email: str = typer.Option(..., '--email', '-e', help='Your email (required — primary identifier after v0.5.1).'), project_name: str | None = typer.Option(None, '--project-name', help='Human-readable project name. Defaults to project ID.'), actor_name: str | None = typer.Option(None, '--name', help='Your display name. Defaults to the local-part of email.'), actor_id: str | None = typer.Option(None, '--actor-id', help='Explicit actor UUID. Auto-generated (uuid4) if omitted.'), db_url: str = typer.Option('postgresql://localhost/luplo', '--db-url', help='PostgreSQL connection string.', envvar='LUPLO_DB_URL'), server_url: str = typer.Option('', '--server-url', help='Optional remote server URL (for `lp login`).', envvar='LUPLO_SERVER_URL')) None¶
Initialise luplo in the current directory.
Creates a .luplo config file, runs database migrations, and seeds the project and actor. After init, all other commands read from .luplo automatically.
.. rubric:: Examples
lp init -p hearthward -e me@example.com lp init -p hearthward -e me@example.com –name “Ryan” lp init -p myapp -e me@example.com –db-url postgresql://…
- luplo.cli.items_add(title: str = typer.Argument(..., help='Item title.'), item_type: str = typer.Option('decision', '--type', '-t', help='Item type.'), body: str | None = typer.Option(None, '--body', '-b', help='Item body.'), rationale: str | None = typer.Option(None, '--rationale', '-r'), system: list[str] | None = typer.Option(None, '--system', '-s'), work_unit: str | None = typer.Option(None, '--wu', '-w', help='Attach this item to a work unit (full UUID or 8+ char hex prefix).'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Add a new item.
- luplo.cli.items_show(item_id: str = typer.Argument(..., help='Full UUID or 8-char+ hex prefix. Ambiguous prefixes error out.')) None¶
Show a single item.
- luplo.cli.items_list(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), item_type: str | None = typer.Option(None, '--type', '-t'), system: str | None = typer.Option(None, '--system', '-s'), work_unit: str | None = typer.Option(None, '--wu', '-w', help='Filter to items attached to this work unit (full UUID or 8+ char hex prefix).'), limit: int = typer.Option(20, '--limit', '-n')) None¶
List items for a project.
- luplo.cli.items_search(query: str = typer.Argument(..., help='Search query.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), limit: int = typer.Option(10, '--limit', '-n')) None¶
Search items using glossary-expanded tsquery.
- luplo.cli.work_open(title: str = typer.Argument(..., help='Work unit title.'), description: str | None = typer.Option(None, '--desc', '-d'), system: list[str] | None = typer.Option(None, '--system', '-s'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Open a new work unit.
- luplo.cli.work_ls(status: str | None = typer.Option(None, '--status', '-s', help='Filter by status (in_progress | done | abandoned). Default: all.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
List work units for a project, ordered by created_at DESC.
- luplo.cli.work_resume(query: str = typer.Argument(..., help='Title keyword to search.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
Find in-progress work units by title.
- luplo.cli.work_close(work_id: str = typer.Argument(..., help='Work unit ID.'), status: str = typer.Option('done', '--status', help='done or abandoned.'), force: bool = typer.Option(False, '--force', '-f', help='Close even if an in_progress task remains.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Close a work unit. Refuses if an in_progress task remains (use –force).
- luplo.cli.systems_add(name: str = typer.Argument(..., help='System name.'), description: str | None = typer.Option(None, '--desc', '-d'), depends: list[str] | None = typer.Option(None, '--depends'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
Add a new system.
- luplo.cli.systems_list(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
List all systems for a project.
- luplo.cli.glossary_ls(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), limit: int = typer.Option(50, '--limit', '-n')) None¶
List glossary groups.
- luplo.cli.glossary_pending(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), limit: int = typer.Option(20, '--limit', '-n')) None¶
Show terms awaiting curation.
- luplo.cli.glossary_approve(term_id: str = typer.Argument(..., help='Term ID to approve.'), group_id: str = typer.Option(..., '--group', '-g', help='Target group ID.'), canonical: bool = typer.Option(False, '--canonical', '-c', help='Set as canonical.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Approve a pending term into a group.
- luplo.cli.glossary_reject(term_id: str = typer.Argument(..., help='Term ID to reject.'), reason: str | None = typer.Option(None, '--reason', '-r'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Reject a term permanently.
- luplo.cli.glossary_group_create(canonical: str = typer.Argument(..., help='Canonical surface form for the new group.'), definition: str | None = typer.Option(None, '--def', '-d', help='One-line definition stored on the group.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Create a glossary group plus its canonical surface term.
- luplo.cli.glossary_add(surface: str = typer.Argument(..., help='New surface form to add to a group.'), group: str = typer.Option(..., '--group', '-g', help='Target group ID (full UUID or ≥8-char hex prefix).'), canonical: bool = typer.Option(False, '--canonical', '-c', help='Promote this term to canonical, demoting any existing canonical to alias.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Add a new alias (or canonical) term to an existing group.
- luplo.cli.glossary_term_rm(term_id: str = typer.Argument(..., help='Term ID (full UUID or ≥8-char hex prefix).'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Permanently remove a glossary term.
Removing the last canonical/alias term in a group cascades — the group (plus its rejection records) is dropped as well. Removing the canonical while aliases remain is refused; promote one alias first.
- luplo.cli.brief(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), system: str | None = typer.Option(None, '--system', '-s')) None¶
Get a project brief — active work + recent decisions.
- luplo.cli.impact_cmd(item_id: str = typer.Argument(..., help='Root item (full UUID or 8+ char hex prefix).'), depth: int = typer.Option(5, '--depth', '-d', min=1, max=5, help='Maximum traversal depth (1-5, capped at 5 server-side).'), output_format: str = typer.Option('tree', '--format', '-f', help='Output format: tree | flat | json.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
Traverse typed edges to find an item’s blast radius.
Walks outgoing
depends/blocks/supersedes/conflictsedges up to the given depth and prints every item reachable from the root. Cycles are broken automatically; each item appears once, at its shortest-path depth... rubric:: Examples
lp impact 5d4b04c4 lp impact 5d4b04c4 –depth 2 lp impact 5d4b04c4 –format json
- luplo.cli.check_cmd(rule: list[str] = typer.Option([], '--rule', '-r', help='Run only these rules (by name). Repeat for multiple. Default: all enabled.'), severity: str = typer.Option('warn', '--severity', '-s', help='Show findings at or above this severity. One of: error, warn, info.'), list_rules: bool = typer.Option(False, '--list', help='Print every registered rule and its default severity, then exit.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
Run the deterministic rule pack and report findings.
Exits non-zero if any finding has severity=error. Rules disabled in
.luplo [checks] disabled_rulesare skipped regardless of--rule... rubric:: Examples
lp check lp check –rule missing_rationale –rule dangling_edge lp check –severity error lp check –list
- luplo.cli.task_add(title: str = typer.Argument(..., help='Task title.'), work_unit: str = typer.Option(..., '--wu', '-w', help='Work unit full UUID or 8-char+ hex prefix.'), body: str | None = typer.Option(None, '--body', '-b'), system: list[str] | None = typer.Option(None, '--system', '-s'), sort_order: int | None = typer.Option(None, '--sort'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Add a new task in ‘proposed’ status.
- luplo.cli.task_ls(work_unit: str = typer.Option(..., '--wu', '-w'), status: str | None = typer.Option(None, '--status', '-s')) None¶
List tasks (chain heads) for a work unit, ordered by sort_order.
- luplo.cli.task_show(task_id: str = typer.Argument(..., help='Full UUID or 8-char+ hex prefix. Ambiguous prefixes error out.')) None¶
Show a single task (resolved to chain head).
- luplo.cli.task_start(task_id: str = typer.Argument(...), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Transition task to ‘in_progress’ (enforces 1 in_progress per WU).
- luplo.cli.task_done(task_id: str = typer.Argument(...), summary: str | None = typer.Option(None, '--summary'), propose_decision: bool = typer.Option(False, '--propose-decision', help='After completion, print a draft decision item derived from this task. The draft is NOT inserted — copy-paste the lp command shown to save it.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Transition task to ‘done’.
- luplo.cli.task_blocked(task_id: str = typer.Argument(...), reason: str = typer.Option(..., '--reason', '-r'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Transition task to ‘blocked’ (auto-creates a decision item).
- luplo.cli.task_skip(task_id: str = typer.Argument(...), reason: str | None = typer.Option(None, '--reason', '-r'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Transition task to ‘skipped’ (terminal).
- luplo.cli.task_reorder(work_unit: str = typer.Argument(..., help='Work unit ID.'), task_ids: list[str] = typer.Argument(..., help='Task IDs in desired order.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Reorder tasks (in-place sort_order update — P10).
- luplo.cli.task_edit(task_id: str = typer.Argument(...), title: str | None = typer.Option(None, '--title', '-t', help='New title.'), body: str | None = typer.Option(None, '--body', '-b', help='New body.'), sort_order: int | None = typer.Option(None, '--sort', '-s', help='New sort_order.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Edit a task’s title / body / sort_order via a supersede row.
Status is preserved — use lp task start / done / blocked / skip to change status. Passing no flags is a no-op that just returns the current head.
- luplo.cli.task_in_progress(work_unit: str = typer.Option(..., '--wu', '-w')) None¶
Show the current in_progress task for a work unit, if any.
- luplo.cli.qa_add(title: str = typer.Argument(...), coverage: str = typer.Option(..., '--coverage', '-c', help='auto_partial | human_only'), area: list[str] | None = typer.Option(None, '--area', help='vfx, sfx, ux, edge_case, perf, a11y, sec'), tasks_target: list[str] | None = typer.Option(None, '--task', '-t', help='Target task IDs.'), items_target: list[str] | None = typer.Option(None, '--item', '-i', help='Target item IDs.'), work_unit: str | None = typer.Option(None, '--wu', '-w'), body: str | None = typer.Option(None, '--body'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Add a new qa_check in ‘pending’ status.
- luplo.cli.qa_ls(status: str | None = typer.Option(None, '--status', '-s'), work_unit: str | None = typer.Option(None, '--wu', '-w'), task: str | None = typer.Option(None, '--task', '-t', help='Filter to qa_checks targeting this task.'), item_id_filter: str | None = typer.Option(None, '--item', '-i'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
List qa_checks. With –task / –item shows pending qa for that target.
- luplo.cli.qa_show(qa_id: str = typer.Argument(..., help='Full UUID or 8-char+ hex prefix. Ambiguous prefixes error out.')) None¶
Show a single qa_check (chain head).
- luplo.cli.qa_start(qa_id: str = typer.Argument(...), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
- luplo.cli.qa_pass(qa_id: str = typer.Argument(...), evidence: str | None = typer.Option(None, '--evidence', '-e'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
- luplo.cli.qa_fail(qa_id: str = typer.Argument(...), reason: str = typer.Option(..., '--reason', '-r'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
- luplo.cli.qa_block(qa_id: str = typer.Argument(...), reason: str = typer.Option(..., '--reason', '-r'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
- luplo.cli.qa_assign(qa_id: str = typer.Argument(...), assignee: str = typer.Option(..., '--to', help='Assignee actor UUID.'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
- luplo.cli.capture_add(text: list[str] = typer.Argument(..., help='Raw capture text (joined with spaces).'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Save raw text into the capture backlog.
- luplo.cli.capture_ls(state: str | None = typer.Option(None, '--state', help='Filter by capture state.'), limit: int = typer.Option(100, '--limit'), include_discarded: bool = typer.Option(False, '--include-discarded'), include_redacted: bool = typer.Option(False, '--include-redacted')) None¶
List recent captures, newest first.
- luplo.cli.capture_find(query: list[str] | None = typer.Argument(None, help='Search query (joined with spaces). Omit for filter-only search.'), state: str | None = typer.Option(None, '--state', help='Filter by capture state.'), limit: int = typer.Option(50, '--limit'), include_discarded: bool = typer.Option(False, '--include-discarded'), include_redacted: bool = typer.Option(False, '--include-redacted')) None¶
Full-text search over captures, newest first within rank.
- luplo.cli.capture_state(capture_id: str = typer.Argument(...), state: str = typer.Argument(...), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Move a capture to another review state.
- luplo.cli.capture_discard(capture_id: str = typer.Argument(...), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Discard a capture so default list/search hides it.
- luplo.cli.capture_redact(capture_id: str = typer.Argument(...), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Redact capture content from normal storage.
- luplo.cli.capture_annotate(capture_id: str = typer.Argument(...), summary: str | None = typer.Option(None, '--summary'), sensitivity_hint: str | None = typer.Option(None, '--sensitivity-hint'), signals: str | None = typer.Option(None, '--signals', help='JSON object of caller-supplied annotation signals.')) None¶
Store caller-supplied BYOLLM annotation hints on a capture.
- luplo.cli.capture_promote(capture_id: str = typer.Argument(...), item_type: str = typer.Option('knowledge', '--type', '-t'), title: str = typer.Option(..., '--title'), body: str | None = typer.Option(None, '--body'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Explicitly promote a capture into a curated item.
- luplo.cli.idea_add(text: list[str] = typer.Argument(..., help='Idea text (joined with spaces).'), work_unit: str | None = typer.Option(None, '--wu', '-w', help='Work unit (UUID or 8+ hex prefix). Defaults to active WU.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Append an ideation note to a work unit (append-only, redact-only).
Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.
- luplo.cli.idea_ls(work_unit: str | None = typer.Option(None, '--wu', '-w', help='Work unit (UUID or 8+ hex prefix). Defaults to active WU.'), limit: int = typer.Option(100, '--limit'), include_redacted: bool = typer.Option(False, '--include-redacted', help='Include redacted ideas in the listing.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
List ideas for a work unit, newest first.
Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.
- luplo.cli.idea_find(query: list[str] | None = typer.Argument(None, help='Search query (joined with spaces). Omit for filter-only search.'), work_unit: str | None = typer.Option(None, '--wu', '-w', help='Narrow to one work unit.'), author: str | None = typer.Option(None, '--author', help='Filter by actor id.'), since: str | None = typer.Option(None, '--since', help='ISO datetime, Nd/Nw, or this_week/this_month/this_quarter.'), until: str | None = typer.Option(None, '--until', help='ISO datetime or Nd/Nw. Anchors (this_week / this_month / this_quarter) are since-only and rejected here.'), include_redacted: bool = typer.Option(False, '--include-redacted', help='Include redacted rows in the listing (filter-only — cannot be combined with a text query, which would leak via match/no-match).'), limit: int = typer.Option(50, '--limit'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
Full-text search over ideas in a project. All filters are optional.
Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.
- luplo.cli.idea_redact(idea_id: str = typer.Argument(..., help='Idea id (full UUID or 8+ hex prefix).'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) None¶
Mark an idea redacted (idempotent).
Deprecated for raw intake. Use capture for unstructured backlog entries. Ideas remain for compatibility with work-unit-scoped ideation notes.
--projectscopes prefix resolution and the SQL predicate so a full UUID from another project cannot mutate this row. Aligned with the rest oflp idea— uses_cfg_projectlike the others.
- luplo.cli.import_begin(from_spec: pathlib.Path | None = typer.Option(None, '--from-spec', help='Path to a spec markdown file.'), from_plan: pathlib.Path | None = typer.Option(None, '--from-plan', help='Path to a plan markdown file.'), dest_lang: str | None = typer.Option(None, '--dest-lang', help='Target language (ISO 639-1). Overrides .luplo [project].language.'), force: bool = typer.Option(False, '--force', help='Replace any prior import of the same source set.'), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Stage a new import bundle and emit its manifest as JSON.
Opens a fresh
import-typed work unit, reads the spec/plan source files, and prints the resultingImportManifestto stdout. The manifest is the contract the calling agent consumes to extract decision / knowledge / document items.- Parameters:
from_spec – Optional path to a spec markdown file.
from_plan – Optional path to a plan markdown file. At least one of
from_specorfrom_planmust be provided.dest_lang – ISO 639-1 target language. Overrides
[project].languagefrom.luplo.Nonepreserves the source language.force – When True, archive any prior import work unit covering the same source set and stage a new one in its place.
project – Project ID override (defaults to
.luplo/env).actor – Actor UUID override (defaults to
.luplo/env).
- Raises:
typer.Exit – Code
1when no source files are provided. Code2when a duplicate import exists andforceis not set.
- luplo.cli.import_finalize(results: pathlib.Path = typer.Option(..., '--results', help="Path to results JSON file (or '-' for stdin)."), project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) None¶
Apply agent-produced results to a staged import bundle.
Reads the
ImportResultsJSON envelope (from a file or stdin), validates it, and forwards it tofinalize_importwhich writes the contained items to PostgreSQL under the bundle’s work unit.- Parameters:
results – Path to a JSON file produced by the agent, or
-to read the JSON from stdin.project – Project ID override (defaults to
.luplo/env).actor – Actor UUID override (defaults to
.luplo/env).
- Raises:
typer.Exit – Code
3when the results JSON fails pydantic validation.