feat(annotation): tag partial panels via status --tag-partial-panels

src/pragmata/annotation/__init__.py

@@ -21,6 +21,7 @@
     "Locale",
     "SetupResult",
     "StatusReport",
+    "TagResult",
     "Task",
     "UserSpec",
     "compute_iaa",
@@ -39,6 +40,7 @@
     "Locale": ("pragmata.core.schemas.annotation_task", "Locale"),
     "SetupResult": ("pragmata.core.annotation.setup", "SetupResult"),
     "StatusReport": ("pragmata.core.annotation.panel_status", "StatusReport"),
+    "TagResult": ("pragmata.core.annotation.panel_status", "TagResult"),
     "Task": ("pragmata.core.schemas.annotation_task", "Task"),
     "UserSpec": ("pragmata.core.settings.annotation_settings", "UserSpec"),
     "compute_iaa": ("pragmata.api.annotation_iaa", "compute_iaa"),
@@ -75,6 +77,7 @@ def __dir__() -> list[str]:
     from pragmata.core.annotation.export_runner import ExportResult as ExportResult
     from pragmata.core.annotation.panel_status import HeadlineTotals as HeadlineTotals
     from pragmata.core.annotation.panel_status import StatusReport as StatusReport
+    from pragmata.core.annotation.panel_status import TagResult as TagResult
     from pragmata.core.annotation.setup import SetupResult as SetupResult
     from pragmata.core.schemas.annotation_task import Locale as Locale
     from pragmata.core.schemas.annotation_task import Task as Task

src/pragmata/api/annotation_status.py

@@ -1,16 +1,23 @@
-"""Annotation status API - live per-panel completeness report.
+"""Annotation status API - live progress + retrieval panel report + optional tag.
 
-Pure read and config-free: resolves Argilla credentials, then walks the live
-retrieval datasets (optionally narrowed by workspace) with no local topology
-config. The optional ``--tag-partial-panels`` advisory write ships in a
-follow-up PR so the read path carries no Argilla mutation surface.
+Config-free: resolves Argilla credentials, then walks the live datasets (all
+tasks; optionally narrowed by workspace) with no local topology config.
+``tag_partial_panels=True`` additionally stamps the ``needs_completion``
+advisory tag on partial retrieval panels' unresolved chunks - the one live-write
+surface, sharing the same retrieval walk as the panel report.
 """
 
 import logging
 import os
 
 from pragmata.core.annotation.client import resolve_argilla_client
-from pragmata.core.annotation.panel_status import StatusReport, compute_panel_status, compute_task_progress
+from pragmata.core.annotation.panel_status import (
+    StatusReport,
+    _apply_tags,
+    _build_report,
+    _collect_records,
+    compute_task_progress,
+)
 from pragmata.core.settings.settings_base import UNSET, Unset, resolve_api_key
 
 logger = logging.getLogger(__name__)
@@ -21,8 +28,9 @@ def report_status(
     api_url: str | Unset = UNSET,
     api_key: str | Unset = UNSET,
     workspace: str | None = None,
+    tag_partial_panels: bool = False,
 ) -> StatusReport:
-    """Fetch live retrieval panel status from Argilla (config-free).
+    """Fetch live annotation status from Argilla (config-free).
 
     Credential resolution (config-free):
     - ``api_url``: kwarg > ``ARGILLA_API_URL`` env
@@ -32,16 +40,25 @@ def report_status(
         api_url: Argilla server URL.
         api_key: Argilla API key.
         workspace: If set, only datasets in this Argilla workspace.
+        tag_partial_panels: If True, stamp ``needs_completion`` on partial
+            panels' unresolved chunks (and clear stale tags). Opt-in live write.
 
     Returns:
-        ``StatusReport`` with the all-task ``progress`` summary plus the
-        retrieval per-panel facts.
+        ``StatusReport`` with the all-task ``progress`` summary and the
+        retrieval per-panel facts, plus an optional ``tag_result`` populated
+        when ``tag_partial_panels=True``.
     """
     url = api_url if isinstance(api_url, str) else os.environ.get("ARGILLA_API_URL")
     key = api_key if isinstance(api_key, str) else resolve_api_key("argilla")
     client = resolve_argilla_client(url, key)
+
     progress = compute_task_progress(client, workspace=workspace)
-    report = compute_panel_status(client, workspace=workspace).with_progress(progress)
+    # One retrieval walk shared between the panel report and the optional tag write.
+    collected = _collect_records(client, workspace=workspace)
+    report = _build_report(collected).with_progress(progress)
+    if tag_partial_panels:
+        report = report.with_tag_result(_apply_tags(collected))
+
     logger.info(
         "Status: %d panels, %d complete (%.0f%%), %d overlap-satisfied, %d integrity warnings",
         report.n_panels,

src/pragmata/cli/commands/annotation.py

@@ -243,19 +243,31 @@ def status_command(
     ),
     by_workspace: bool = typer.Option(False, "--by-workspace", help="Add a per-workspace progress breakdown."),
     by_dataset: bool = typer.Option(False, "--by-dataset", help="Add a per-dataset progress breakdown."),
+    tag_partial_panels: bool = typer.Option(
+        False,
+        "--tag-partial-panels",
+        help=(
+            "Live write: stamp 'needs_completion' on the unresolved chunks of PARTIAL panels "
+            "(some but not all chunks annotated) and clear stale tags, so annotators can filter "
+            "straight to them in the Argilla UI. Off by default (read-only)."
+        ),
+    ),
 ) -> None:
     """Report live annotation progress across all tasks, plus retrieval panel-completeness.
 
     Config-free: walks every Argilla dataset. Record progress (total / completed)
     is shown per task; the retrieval row also carries panel-completeness. Add
-    --by-workspace / --by-dataset for finer breakdowns.
+    --by-workspace / --by-dataset for finer breakdowns. With --tag-partial-panels,
+    also stamps the 'needs_completion' advisory tag on partial panels' unresolved
+    chunks (a live write).
     """
     from pragmata import annotation
 
     report = annotation.report_status(
         api_url=UNSET if api_url is None else api_url,
         api_key=UNSET if api_key is None else api_key,
         workspace=workspace,
+        tag_partial_panels=tag_partial_panels,
     )
 
     def _num(n: int) -> str:
@@ -313,6 +325,11 @@ def _pct(done: int, total: int) -> str:
         )
     if report.n_orphans_skipped:
         typer.echo(f"orphans skipped: {report.n_orphans_skipped} record(s) with empty record_uuid")
+    if report.tag_result is not None:
+        tr = report.tag_result
+        typer.echo(
+            f"tag-partial-panels: tagged={tr.n_tagged} cleared={tr.n_cleared} already_tagged={tr.n_already_tagged}"
+        )
 
 
 @annotation_app.command("iaa")

src/pragmata/core/annotation/metadata_ops.py

@@ -0,0 +1,76 @@
+"""Shared safe metadata operations for live Argilla mutations.
+
+Used by the ``--tag-partial-panels`` write path in ``panel_status``.
+Centralises the two safety invariants that every metadata write must respect
+on Argilla v2.8.0:
+
+1. **Argilla metadata is REPLACE, not merge.** Every ``dataset.records.log``
+   call replaces the record's metadata wholesale. To avoid clobbering
+   existing keys, always fetch the current dict, merge in the update, and
+   send the FULL resulting dict.
+2. **Property declaration is additive and idempotent.** Adding a metadata
+   property to an existing dataset is non-destructive, but the SDK raises
+   if the property already exists (with override warning). Skip the add
+   when the property is already present.
+
+Writes go via ``rg.Record(id=..., metadata={...})`` rather than a raw dict
+payload: the SDK's ``IngestedRecordMapper`` flattens dict keys against the
+dataset schema, so a ``{"id": ..., "metadata": {...}}`` shape would treat
+"metadata" as an unknown top-level attribute and silently send an empty
+metadata dict (wiping the record). Passing an ``rg.Record`` bypasses the
+mapper.
+"""
+
+import logging
+from collections.abc import Iterable, Mapping
+
+import argilla as rg
+
+logger = logging.getLogger(__name__)
+
+
+def ensure_metadata_property(dataset: rg.Dataset, prop: rg.MetadataType) -> bool:
+    """Idempotently declare ``prop`` on ``dataset``.
+
+    Returns True if the property was newly added (and the dataset settings
+    pushed to the server), False if it was already present.
+    """
+    existing = dataset.settings.metadata[prop.name]
+    if existing is not None:
+        return False
+    dataset.settings.add(prop)
+    dataset.settings.update()
+    logger.info("Declared metadata property %r on dataset %s", prop.name, dataset.name)
+    return True
+
+
+def build_metadata_upsert(
+    record: rg.Record,
+    updates: Mapping[str, object],
+    *,
+    remove_keys: Iterable[str] = (),
+) -> rg.Record | None:
+    """Merge ``updates`` into ``record.metadata`` and return an upsert Record.
+
+    Returns ``None`` when the merge produces no change (idempotent no-op).
+    Mutates ``record``'s metadata in place and returns it, so its fields (and
+    suggestions) ride along in the upsert payload: Argilla v2.8.0 rejects a
+    field-less record with 422 "fields cannot be empty" because the required
+    text fields must be present, so ``id`` + metadata alone is not a valid
+    upsert.
+
+    Callers batch the returned Records into a single ``dataset.records.log``
+    call per dataset to amortise the round-trip.
+    """
+    current = dict(record.metadata)
+    merged = dict(current)
+    merged.update(updates)
+    for key in remove_keys:
+        merged.pop(key, None)
+    if merged == current:
+        return None
+    for key, value in updates.items():
+        record.metadata[key] = value
+    for key in remove_keys:
+        record.metadata.pop(key, None)
+    return record