Phase 9/10: marketplace docs/E2E and tool/usage ledger instrumentation

.env.example

@@ -44,21 +44,30 @@ CURRENCY_CODE=USD
 CURRENCY_SYMBOL=$
 CURRENCY_DECIMAL_PLACES=4
 
-# Plugin manifest trust (optional). Unsigned plugins remain allowed by default.
-# Set PLUGIN_TRUST_POLICY=required to reject unsigned or invalid signatures.
-# JSON object mapping canonical publisher ids to Ed25519 public keys; each key
-# may be base64-encoded raw 32 bytes or a PEM SubjectPublicKeyInfo. Invalid key
-# material aborts startup instead of silently disabling verification.
+# Plugin marketplace, catalog, network, and execution settings (Phase 8-9).
+# Defaults are intentionally fail-closed: remote installation is disabled, no
+# plugin hosts are reachable, and all callbacks are time-bounded.
+
+# Publisher trust. "if-present" accepts unsigned plugins but still refuses a
+# forged or untrusted signature; set "required" to also refuse unsigned ones.
+# PLUGIN_TRUSTED_KEYS is a JSON object mapping canonical publisher ids to
+# Ed25519 public keys (base64 raw 32 bytes or PEM SubjectPublicKeyInfo). Invalid
+# key material aborts startup instead of silently disabling verification.
 PLUGIN_TRUST_POLICY=if-present
 PLUGIN_TRUSTED_KEYS={}
 
 # Operator-controlled catalog for POST .../plugins/install-from-catalog
 # (Phase 9, issue #389). Empty disables remote installation. The file is a JSON
 # catalog of validated entries: id, version, https source, sha256 integrity,
-# optional detached Ed25519 signature, and a full manifest. See
-# docs/plugins.md.
+# optional detached Ed25519 signature, and a full manifest. See docs/plugins.md.
+# The catalog is re-read on each install request; no cache TTL setting is active
+# yet (catalog caching is tracked in #377).
 PLUGIN_CATALOG_FILE=
 
+# Read-only plugin discovery directory used by GET /plugins/discover. It is
+# scanned for manifest.json files only; nothing is installed or executed.
+PLUGIN_DISCOVERY_DIR=plugins
+
 # Plugin execution budget (Phase 9, issue #408): seconds an event handler or
 # lifecycle hook may run before it is abandoned (daemon thread), recorded as a
 # bounded error report, and dispatch continues. 0 disables the bound.

README.md

@@ -56,6 +56,26 @@ Open http://localhost:5000 in your browser.
 - Team collaboration and workspace-level permissions
 - Extensible plugin architecture for future integrations
 
+## Marketplace and plugin configuration
+
+Plugin installation and outbound requests are opt-in. The defaults are
+fail-closed, so a fresh install cannot reach a marketplace or the network:
+
+| Setting | Default | Why it is safe |
+| ------- | ------- | -------------- |
+| `PLUGIN_CATALOG_FILE` | empty | Remote/catalog installation is disabled until an operator points at a vetted catalog. |
+| `PLUGIN_TRUST_POLICY` | `if-present` | Accepts unsigned local plugins, but always refuses forged or untrusted signatures. Use `required` in production. |
+| `PLUGIN_TRUSTED_KEYS` | `{}` | No publisher is trusted by default; keys are validated at startup. |
+| `PLUGIN_NETWORK_ALLOWLIST` | empty | Every plugin outbound request is denied. Add only the hosts a plugin needs. |
+| `PLUGIN_NETWORK_HTTPS_ONLY` | `1` | Plugin traffic must use HTTPS. |
+| `PLUGIN_NETWORK_ALLOW_PRIVATE` | `0` | Blocks private, loopback, and link-local targets. |
+| `PLUGIN_EXECUTION_TIMEOUT` | `5` | Abandons a plugin callback after five seconds so it cannot stall a request. |
+
+The catalog is read from disk for each install request; there is no active
+cache TTL setting yet (caching is tracked in #377). See
+[the plugin guide](docs/plugins.md#configuration-reference) for the full
+reference, including signature and catalog formats.
+
 ## Documentation
 
 - [Chat feature guide](CONTRIBUTING.md#chat-feature-developer-guide)

app/__init__.py

@@ -133,4 +133,10 @@ def init_db_command():
 
     register_plugins_cli(app)
 
+    # Usage-ledger maintenance (historical chat backfill and future retention
+    # commands). The backfill is bounded and idempotent.
+    from app.services.usage_cli import register_usage_cli
+
+    register_usage_cli(app)
+
     return app

app/services/usage_cli.py

@@ -0,0 +1,93 @@
+"""Usage-ledger maintenance commands.
+
+Usage::
+
+    flask usage backfill [--batch-size N] [--limit N] [--dry-run] [--json]
+
+The backfill copies historical ``Message`` token columns into the append-only
+usage ledger. It never updates message rows, is safe to re-run, and processes
+messages in bounded keyset pages.
+"""
+
+from __future__ import annotations
+
+import json
+
+import click
+from flask import Flask
+
+from app.services.usage_ledger import BACKFILL_DEFAULT_BATCH_SIZE, BACKFILL_MAX_BATCH_SIZE
+
+EXIT_OK = 0
+EXIT_ERROR = 2
+
+
+def _write_json(data) -> None:
+    click.echo(json.dumps(data, indent=2, sort_keys=True))
+
+
+def _fail(message: str, as_json: bool) -> None:
+    if as_json:
+        _write_json({"error": message})
+    else:
+        click.echo(message, err=True)
+    raise click.exceptions.Exit(EXIT_ERROR)
+
+
+def register_usage_cli(app: Flask) -> None:
+    """Register the ``usage`` CLI command group on ``app``."""
+
+    @app.cli.group("usage")
+    def usage_group():
+        """Usage-ledger maintenance commands."""
+
+    @usage_group.command("backfill")
+    @click.option(
+        "--batch-size",
+        type=click.IntRange(1, BACKFILL_MAX_BATCH_SIZE),
+        default=BACKFILL_DEFAULT_BATCH_SIZE,
+        show_default=True,
+        help="Messages processed per database batch.",
+    )
+    @click.option(
+        "--limit",
+        type=click.IntRange(min=0),
+        default=None,
+        help="Maximum messages to scan in this run (default: all).",
+    )
+    @click.option(
+        "--dry-run",
+        is_flag=True,
+        help="Report what would be written without committing ledger rows.",
+    )
+    @click.option("--json", "as_json", is_flag=True, help="Emit machine-readable JSON.")
+    def backfill(batch_size: int, limit: int | None, dry_run: bool, as_json: bool) -> int:
+        """Backfill ledger rows from historical chat-message token counts."""
+        from app.services.usage_ledger import backfill_from_messages
+
+        try:
+            result = backfill_from_messages(
+                batch_size=batch_size,
+                limit=limit,
+                dry_run=dry_run,
+            )
+        except ValueError as exc:
+            _fail(str(exc), as_json=as_json)
+
+        if as_json:
+            _write_json(result)
+            return EXIT_OK
+
+        action = "Dry run" if dry_run else "Backfill"
+        click.echo(
+            f"{action} complete: scanned={result['scanned']} "
+            f"created={result['created']} "
+            f"skipped_existing={result['skipped_existing']} "
+            f"skipped_incomplete={result['skipped_incomplete']} "
+            f"skipped_orphan={result['skipped_orphan']} "
+            f"batches={result['batches']}"
+        )
+        return EXIT_OK
+
+
+__all__ = ["EXIT_ERROR", "EXIT_OK", "register_usage_cli"]

app/services/usage_ledger.py

@@ -12,8 +12,10 @@
 
 from typing import Any
 
+from sqlalchemy import or_
+
 from app.extensions import db
-from app.models import UsageRecord
+from app.models import Message, UsageRecord
 from app.services import pricing
 
 #: Operation families understood by the ledger. Unknown values are grouped
@@ -32,6 +34,12 @@
 )
 UNKNOWN_SOURCE = "unknown"
 
+#: Default/upper bounds for the historical chat-message backfill. The work is
+#: paginated by message id so a large conversation history is never loaded at
+#: once.
+BACKFILL_DEFAULT_BATCH_SIZE = 500
+BACKFILL_MAX_BATCH_SIZE = 5000
+
 #: Operation outcomes understood by the ledger (see docs/observability.md).
 STATUS_SUCCESS = "success"
 STATUS_ERROR = "error"
@@ -321,7 +329,166 @@ def breakdown_payload(
     }
 
 
+def _message_usage(message: Message) -> dict[str, int] | None:
+    """Return validated token counts for a message, or ``None`` when absent."""
+    values = (
+        getattr(message, "prompt_tokens", None),
+        getattr(message, "completion_tokens", None),
+        getattr(message, "total_tokens", None),
+    )
+    if all(value is None for value in values):
+        return None
+    try:
+        return normalize_usage(*values)
+    except ValueError:
+        return None
+
+
+def _backfill_record_exists(
+    message: Message,
+    conversation_id: int,
+    user_id: int | None,
+    counts: dict[str, int],
+) -> bool:
+    """Return ``True`` when this message was already copied into the ledger.
+
+    The ledger predates a durable ``message_id`` column, so a backfilled row is
+    identified by its source, conversation, token counts, and the original
+    message timestamp. That tuple is stable across runs, which makes the
+    command idempotent without changing or annotating historical message rows.
+    """
+    if message.created_at is None:
+        return False
+    return (
+        UsageRecord.query.filter_by(
+            source=SOURCE_CHAT,
+            conversation_id=conversation_id,
+            user_id=user_id,
+            prompt_tokens=counts["prompt_tokens"],
+            completion_tokens=counts["completion_tokens"],
+            total_tokens=counts["total_tokens"],
+        )
+        .filter(UsageRecord.created_at == message.created_at)
+        .first()
+        is not None
+    )
+
+
+def backfill_from_messages(
+    *,
+    batch_size: int = BACKFILL_DEFAULT_BATCH_SIZE,
+    limit: int | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Copy historical chat-message token counts into the usage ledger.
+
+    The operation is keyset-paginated and idempotent: existing message rows
+    are never modified, and re-running the command only skips rows that were
+    already backfilled. ``limit`` bounds the number of messages considered in a
+    single invocation, which is useful for a staged rollout. ``dry_run`` reports
+    what would be written without committing anything.
+
+    Raises:
+        ValueError: If ``batch_size``/``limit`` is outside the supported range.
+    """
+    if not isinstance(batch_size, int) or not 1 <= batch_size <= BACKFILL_MAX_BATCH_SIZE:
+        raise ValueError(f"batch_size must be an integer between 1 and {BACKFILL_MAX_BATCH_SIZE}.")
+    if limit is not None and (not isinstance(limit, int) or limit < 0):
+        raise ValueError("limit must be a non-negative integer.")
+
+    stats = {
+        "source": SOURCE_CHAT,
+        "scanned": 0,
+        "created": 0,
+        "skipped_existing": 0,
+        "skipped_incomplete": 0,
+        "skipped_orphan": 0,
+        "batches": 0,
+        "batch_size": batch_size,
+        "limit": limit,
+        "dry_run": bool(dry_run),
+    }
+    last_id = 0
+    processed = 0
+
+    try:
+        while limit is None or processed < limit:
+            size = batch_size if limit is None else min(batch_size, limit - processed)
+            if size <= 0:
+                break
+            messages = (
+                Message.query.filter(Message.id > last_id)
+                .filter(
+                    or_(
+                        Message.prompt_tokens.isnot(None),
+                        Message.completion_tokens.isnot(None),
+                        Message.total_tokens.isnot(None),
+                    )
+                )
+                .order_by(Message.id)
+                .limit(size)
+                .all()
+            )
+            if not messages:
+                break
+
+            stats["batches"] += 1
+            for message in messages:
+                if limit is not None and processed >= limit:
+                    break
+                last_id = message.id
+                processed += 1
+                stats["scanned"] += 1
+
+                counts = _message_usage(message)
+                if counts is None:
+                    stats["skipped_incomplete"] += 1
+                    continue
+                conversation = message.conversation
+                if conversation is None:
+                    stats["skipped_orphan"] += 1
+                    continue
+                if _backfill_record_exists(
+                    message,
+                    message.conversation_id,
+                    conversation.user_id,
+                    counts,
+                ):
+                    stats["skipped_existing"] += 1
+                    continue
+                if dry_run:
+                    stats["created"] += 1
+                    continue
+
+                record = record_usage(
+                    source=SOURCE_CHAT,
+                    prompt_tokens=counts["prompt_tokens"],
+                    completion_tokens=counts["completion_tokens"],
+                    total_tokens=counts["total_tokens"],
+                    # Historical columns do not record whether the provider
+                    # reported the counts or they were estimated.
+                    tokens_estimated=True,
+                    user_id=conversation.user_id,
+                    conversation_id=message.conversation_id,
+                    commit=False,
+                )
+                record.created_at = message.created_at
+                stats["created"] += 1
+
+            if not dry_run:
+                db.session.commit()
+            if len(messages) < size:
+                break
+    except Exception:
+        db.session.rollback()
+        raise
+
+    return stats
+
+
 __all__ = [
+    "BACKFILL_DEFAULT_BATCH_SIZE",
+    "BACKFILL_MAX_BATCH_SIZE",
     "SOURCES",
     "SOURCE_ANALYSIS",
     "SOURCE_CHAT",
@@ -332,6 +499,7 @@ def breakdown_payload(
     "STATUS_ERROR",
     "STATUS_SUCCESS",
     "UNKNOWN_SOURCE",
+    "backfill_from_messages",
     "breakdown_by_source",
     "breakdown_payload",
     "current_provider_metadata",