luplo.cli
=========

.. py:module:: luplo.cli

.. autoapi-nested-parse::

   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
----------

.. autoapisummary::

   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


Functions
---------

.. autoapisummary::

   luplo.cli.migrate
   luplo.cli.init
   luplo.cli.items_add
   luplo.cli.items_show
   luplo.cli.items_list
   luplo.cli.items_search
   luplo.cli.work_open
   luplo.cli.work_ls
   luplo.cli.work_resume
   luplo.cli.work_close
   luplo.cli.systems_add
   luplo.cli.systems_list
   luplo.cli.glossary_ls
   luplo.cli.glossary_pending
   luplo.cli.glossary_approve
   luplo.cli.glossary_reject
   luplo.cli.glossary_group_create
   luplo.cli.glossary_add
   luplo.cli.glossary_term_rm
   luplo.cli.brief
   luplo.cli.impact_cmd
   luplo.cli.check_cmd
   luplo.cli.worker_start
   luplo.cli.task_add
   luplo.cli.task_ls
   luplo.cli.task_show
   luplo.cli.task_start
   luplo.cli.task_done
   luplo.cli.task_blocked
   luplo.cli.task_skip
   luplo.cli.task_reorder
   luplo.cli.task_edit
   luplo.cli.task_in_progress
   luplo.cli.qa_add
   luplo.cli.qa_ls
   luplo.cli.qa_show
   luplo.cli.qa_start
   luplo.cli.qa_pass
   luplo.cli.qa_fail
   luplo.cli.qa_block
   luplo.cli.qa_assign
   luplo.cli.capture_add
   luplo.cli.capture_ls
   luplo.cli.capture_find
   luplo.cli.capture_state
   luplo.cli.capture_discard
   luplo.cli.capture_redact
   luplo.cli.capture_annotate
   luplo.cli.capture_promote
   luplo.cli.idea_add
   luplo.cli.idea_ls
   luplo.cli.idea_find
   luplo.cli.idea_redact
   luplo.cli.import_begin
   luplo.cli.import_finalize


Module Contents
---------------

.. py:data:: app

.. py:data:: items_app

.. py:data:: work_app

.. py:data:: systems_app

.. py:data:: glossary_app

.. py:data:: glossary_group_app

.. py:data:: glossary_term_app

.. py:data:: task_app

.. py:data:: qa_app

.. py:data:: idea_app

.. py:data:: capture_app

.. py:data:: import_app

.. py:function:: 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
   ``.luplo`` and does not require a config file. Production deploys
   should invoke this directly (e.g. in ``start.sh``) before launching
   the application.

   
   .. rubric:: Examples

   LUPLO_DB_URL=postgresql://... lp migrate
   lp migrate --db-url postgresql://...


.. py:function:: 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://...


.. py:function:: 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.


.. py:function:: items_show(item_id: str = typer.Argument(..., help='Full UUID or 8-char+ hex prefix. Ambiguous prefixes error out.')) -> None

   Show a single item.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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).


.. py:function:: 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.


.. py:function:: systems_list(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT')) -> None

   List all systems for a project.


.. py:function:: glossary_ls(project: str | None = typer.Option(None, '--project', '-p', envvar='LUPLO_PROJECT'), limit: int = typer.Option(50, '--limit', '-n')) -> None

   List glossary groups.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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`` / ``conflicts``
   edges 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


.. py:function:: 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_rules`` are skipped regardless of
   ``--rule``.

   
   .. rubric:: Examples

   lp check
   lp check --rule missing_rationale --rule dangling_edge
   lp check --severity error
   lp check --list


.. py:function:: worker_start() -> None

   Start the background worker (sync jobs + glossary processing).


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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).


.. py:function:: 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).


.. py:function:: 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'.


.. py:function:: 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).


.. py:function:: 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).


.. py:function:: 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).


.. py:function:: 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.


.. py:function:: task_in_progress(work_unit: str = typer.Option(..., '--wu', '-w')) -> None

   Show the current in_progress task for a work unit, if any.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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).


.. py:function:: qa_start(qa_id: str = typer.Argument(...), actor: str | None = typer.Option(None, '--actor', '-a', envvar='LUPLO_ACTOR_ID')) -> None

.. py:function:: 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

.. py:function:: 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

.. py:function:: 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

.. py:function:: 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

.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.


.. py:function:: 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.

   ``--project`` scopes prefix resolution **and** the SQL predicate so a
   full UUID from another project cannot mutate this row. Aligned with
   the rest of ``lp idea`` — uses ``_cfg_project`` like the others.


.. py:function:: 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 resulting ``ImportManifest`` to stdout. The
   manifest is the contract the calling agent consumes to extract
   decision / knowledge / document items.

   :param from_spec: Optional path to a spec markdown file.
   :param from_plan: Optional path to a plan markdown file. At least one of
                     ``from_spec`` or ``from_plan`` must be provided.
   :param dest_lang: ISO 639-1 target language. Overrides
                     ``[project].language`` from ``.luplo``. ``None`` preserves the
                     source language.
   :param force: When True, archive any prior import work unit covering the
                 same source set and stage a new one in its place.
   :param project: Project ID override (defaults to ``.luplo``/env).
   :param actor: Actor UUID override (defaults to ``.luplo``/env).

   :raises typer.Exit: Code ``1`` when no source files are provided. Code ``2``
       when a duplicate import exists and ``force`` is not set.


.. py:function:: 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 ``ImportResults`` JSON envelope (from a file or stdin),
   validates it, and forwards it to ``finalize_import`` which writes the
   contained items to PostgreSQL under the bundle's work unit.

   :param results: Path to a JSON file produced by the agent, or ``-`` to read
                   the JSON from stdin.
   :param project: Project ID override (defaults to ``.luplo``/env).
   :param actor: Actor UUID override (defaults to ``.luplo``/env).

   :raises typer.Exit: Code ``3`` when the results JSON fails pydantic
       validation.