feat: lock migration, hot reload, cost centers, and refused-call ledger

app/models/usage_record.py

@@ -68,6 +68,8 @@ class UsageRecord(db.Model):
     #: Operation family: chat, analysis, review, tools or project_chat.
     #: Unknown values are preserved so readers can group them safely.
     source = db.Column(db.String(40), nullable=False, default="unknown", index=True)
+    #: Optional cost-center / project tag for chargeback rollups.
+    cost_center = db.Column(db.String(100), nullable=True, index=True)
     provider = db.Column(db.String(50), nullable=True)
     model = db.Column(db.String(100), nullable=True)
     prompt_tokens = db.Column(db.Integer, nullable=False, default=0)
@@ -106,6 +108,7 @@ def to_dict(self) -> dict:
             "project_id": self.project_id,
             "conversation_id": self.conversation_id,
             "source": self.source,
+            "cost_center": self.cost_center,
             "provider": self.provider,
             "model": self.model,
             "prompt_tokens": self.prompt_tokens,

app/services/budgets.py

@@ -396,6 +396,37 @@ def _workspace_id_for_request(kwargs: dict) -> int | None:
     return None
 
 
+def _record_refusal(user, workspace_id: int | None) -> None:
+    """Record why a billed call was refused, without ever failing the request."""
+    from flask import request
+
+    from app.services import usage_ledger
+
+    endpoint = (getattr(request, "endpoint", "") or "").lower()
+    if "review" in endpoint:
+        source = usage_ledger.SOURCE_REVIEW
+    elif "tool" in endpoint:
+        source = usage_ledger.SOURCE_TOOLS
+    elif "chat" in endpoint:
+        source = usage_ledger.SOURCE_CHAT
+    elif "workspace" in endpoint or "project" in endpoint:
+        source = usage_ledger.SOURCE_ANALYSIS
+    else:
+        source = usage_ledger.UNKNOWN_SOURCE
+
+    try:
+        usage_ledger.record_usage(
+            source=source,
+            user=user,
+            workspace_id=workspace_id,
+            status=usage_ledger.STATUS_DENIED,
+            tokens_estimated=True,
+        )
+    except Exception:
+        # Audit/accounting must never turn a refusal into a crash.
+        return
+
+
 def budget_refusal(user, workspace_id: int | None = None):
     """Return a refusal response when a billable call cannot proceed.
 
@@ -408,6 +439,7 @@ def budget_refusal(user, workspace_id: int | None = None):
     try:
         decision = check_budgets(user, workspace_id)
     except BudgetCheckError:
+        _record_refusal(user, workspace_id)
         response = jsonify(
             {
                 "error": "Budget check unavailable; refusing the request.",
@@ -419,6 +451,7 @@ def budget_refusal(user, workspace_id: int | None = None):
 
     g.budget_decision = decision
     if not decision.allowed:
+        _record_refusal(user, workspace_id)
         notify_decision(user, decision, level="blocked")
         return budget_error_response(decision)
     if decision.warned:

app/services/plugin_ops.py

@@ -590,22 +590,61 @@ def rollback_from_catalog(
     )
 
 
+class PluginReloadDisabledError(PluginError):
+    """Plugin hot reload was requested outside development."""
+
+
+def hot_reload_enabled(config=None) -> bool:
+    """Return whether development-only plugin hot reload is enabled."""
+    cfg = config or current_app.config
+    return bool(cfg.get("DEBUG")) or str(cfg.get("APP_ENV", "")).strip().lower() in {
+        "development",
+        "testing",
+    }
+
+
+def reload_plugin(plugin_id: str, *, config=None) -> Plugin:
+    """Reload a plugin's validated manifest from the discovery directory.
+
+    Strictly development-only: production configuration refuses before any
+    filesystem or database access. The manifest is re-read and applied through
+    the normal identity-bound update path, so entry points cannot change and no
+    capability grant is created.
+    """
+    cfg = config or current_app.config
+    if not hot_reload_enabled(cfg):
+        raise PluginReloadDisabledError("Plugin hot reload is disabled outside development.")
+
+    get_plugin(plugin_id)
+    discovery = Path(cfg.get("PLUGIN_DISCOVERY_DIR", "plugins"))
+    manifest_path = discovery / plugin_id / "manifest.json"
+    manifest = PluginManifest.from_file(manifest_path)
+    if manifest.id != plugin_id:
+        raise PluginUpdateError(
+            f"Manifest id {manifest.id!r} does not match reloaded plugin {plugin_id!r}."
+        )
+    return update_plugin(plugin_id, manifest)
+
+
 __all__ = [
     "AtomicUpdateError",
     "ManifestValidationError",
     "PluginDependencyResolutionError",
     "PluginError",
     "PluginManifest",
     "PluginNotFoundError",
+    "PluginReloadDisabledError",
     "PluginUpdateError",
     "RemoteInstallError",
     "atomic_rollback",
     "atomic_update_from_catalog",
     "discover_plugins",
     "get_plugin",
+    "hot_reload_enabled",
     "install_from_catalog",
     "install_from_local_dir",
     "list_plugins",
+    "reload_plugin",
     "resolve_plugin_dependencies",
     "rollback_from_catalog",
     "set_plugin_enabled",

app/services/usage_dashboards.py

@@ -199,6 +199,7 @@ def _payload(records: list[UsageRecord]) -> dict:
         "by_source": [{"source": source, **stats} for source, stats in by_source.items()],
         "by_provider": _group(records, "provider", "provider"),
         "by_model": _group(records, "model", "model"),
+        "by_cost_center": _group(records, "cost_center", "cost_center"),
         "daily": _daily(records),
     }
 

app/services/usage_ledger.py

@@ -87,6 +87,7 @@
     "project_id",
     "conversation_id",
     "source",
+    "cost_center",
     "provider",
     "model",
     "status",
@@ -223,6 +224,7 @@ def estimate_cost(
 def record_usage(
     *,
     source: str,
+    cost_center: str | None = None,
     prompt_tokens=0,
     completion_tokens=0,
     total_tokens=None,
@@ -257,6 +259,7 @@ def record_usage(
     counts = normalize_usage(prompt_tokens, completion_tokens, total_tokens)
     provider = (str(provider).strip()[:50] or None) if provider else None
     model = (str(model).strip()[:100] or None) if model else None
+    cost_center = _optional_text(cost_center, "cost_center", 100)
     request_id = _optional_text(request_id, "request_id", 64)
     latency_ms = _optional_non_negative_int(latency_ms, "latency_ms")
     status = _optional_text(status, "status", 20)
@@ -297,6 +300,7 @@ def record_usage(
         project_id=resolved_project_id,
         conversation_id=resolved_conversation_id,
         source=normalize_source(source) or UNKNOWN_SOURCE,
+        cost_center=cost_center,
         provider=provider,
         model=model,
         prompt_tokens=counts["prompt_tokens"],

migrations/versions/e9d8c7b6a5f4_add_plugin_lock_entries.py

@@ -0,0 +1,62 @@
+"""add plugin lock entries
+
+Revision ID: e9d8c7b6a5f4
+Revises: a1b2c3d4e5f6, aa11bb22cc33
+Create Date: 2026-10-23
+
+Adds the ``plugin_lock_entries`` table from the Phase 9 lockfile model (#394,
+migration tracked by #395). This revision also merges the two existing heads so
+``flask db upgrade`` reaches a single schema.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "e9d8c7b6a5f4"
+down_revision = ("a1b2c3d4e5f6", "aa11bb22cc33")
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "plugin_lock_entries",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("workspace_id", sa.Integer(), nullable=False),
+        sa.Column("owner_id", sa.Integer(), nullable=False),
+        sa.Column("plugin_id", sa.String(length=64), nullable=False),
+        sa.Column("version", sa.String(length=32), nullable=False),
+        sa.Column("source", sa.String(length=512), nullable=False),
+        sa.Column("integrity", sa.String(length=128), nullable=False),
+        sa.Column("dependencies", sa.JSON(), nullable=False),
+        sa.Column("resolved_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["workspace_id"], ["workspaces.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("workspace_id", "plugin_id", name="uq_plugin_lock_entry"),
+    )
+    op.create_index(
+        op.f("ix_plugin_lock_entries_workspace_id"),
+        "plugin_lock_entries",
+        ["workspace_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_plugin_lock_entries_owner_id"),
+        "plugin_lock_entries",
+        ["owner_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_plugin_lock_entries_plugin_id"),
+        "plugin_lock_entries",
+        ["plugin_id"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_index(op.f("ix_plugin_lock_entries_plugin_id"), table_name="plugin_lock_entries")
+    op.drop_index(op.f("ix_plugin_lock_entries_owner_id"), table_name="plugin_lock_entries")
+    op.drop_index(op.f("ix_plugin_lock_entries_workspace_id"), table_name="plugin_lock_entries")
+    op.drop_table("plugin_lock_entries")

migrations/versions/f0e1d2c3b4a5_add_usage_cost_center.py

@@ -0,0 +1,32 @@
+"""add usage cost center tag
+
+Revision ID: f0e1d2c3b4a5
+Revises: e9d8c7b6a5f4, dd44ee55ff66
+Create Date: 2026-10-23
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "f0e1d2c3b4a5"
+down_revision = ("e9d8c7b6a5f4", "dd44ee55ff66")
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "usage_records",
+        sa.Column("cost_center", sa.String(length=100), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_usage_records_cost_center"),
+        "usage_records",
+        ["cost_center"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_index(op.f("ix_usage_records_cost_center"), table_name="usage_records")
+    op.drop_column("usage_records", "cost_center")

tests/test_budget_refusal_ledger.py

@@ -0,0 +1,27 @@
+"""Refused billed calls are recorded with a denied ledger status (#463)."""
+
+from app.models import UsageRecord
+from app.services import budgets
+
+
+def test_refused_call_records_denied_ledger_row(app, db, make_user, monkeypatch):
+    user = make_user(username="budgetowner", email="budgetowner@example.com")
+    decision = budgets.BudgetDecision(
+        allowed=False,
+        warned=False,
+        limits_hit=("tokens",),
+        message="Token limit reached.",
+    )
+    monkeypatch.setattr(budgets, "check_budgets", lambda user, workspace_id=None: decision)
+    monkeypatch.setattr(budgets, "after_usage_recorded", lambda record: None)
+
+    with app.test_request_context("/chat/api/chat", method="POST"):
+        response = budgets.budget_refusal(user)
+
+    assert response.status_code == 429
+    row = UsageRecord.query.order_by(UsageRecord.id.desc()).first()
+    assert row is not None
+    assert row.user_id == user.id
+    assert row.status == "denied"
+    assert row.total_tokens == 0
+    assert row.tokens_estimated is True

tests/test_migrations.py

@@ -16,7 +16,12 @@
 
 REPO_ROOT = Path(__file__).resolve().parent.parent
 
-PLUGIN_TABLES = {"plugins", "plugin_installations", "plugin_capability_grants"}
+PLUGIN_TABLES = {
+    "plugins",
+    "plugin_installations",
+    "plugin_capability_grants",
+    "plugin_lock_entries",
+}
 
 # Column set per plugin table as defined by the models.
 EXPECTED_COLUMNS = {
@@ -57,6 +62,17 @@
         "granted_at",
         "granted_by_id",
     },
+    "plugin_lock_entries": {
+        "id",
+        "workspace_id",
+        "owner_id",
+        "plugin_id",
+        "version",
+        "source",
+        "integrity",
+        "dependencies",
+        "resolved_at",
+    },
 }
 
 

tests/test_plugin_hot_reload.py

@@ -0,0 +1,21 @@
+"""Development-only plugin hot reload (#411)."""
+
+import pytest
+
+from app.services import plugin_ops
+
+
+def test_hot_reload_is_disabled_in_production(app):
+    app.config.update(DEBUG=False, APP_ENV="production")
+
+    assert plugin_ops.hot_reload_enabled(app.config) is False
+    with pytest.raises(plugin_ops.PluginReloadDisabledError):
+        plugin_ops.reload_plugin("example", config=app.config)
+
+
+def test_hot_reload_proceeds_in_development(app):
+    app.config.update(DEBUG=False, APP_ENV="development")
+
+    assert plugin_ops.hot_reload_enabled(app.config) is True
+    with pytest.raises(plugin_ops.PluginNotFoundError):
+        plugin_ops.reload_plugin("example", config=app.config)

tests/test_usage_cost_center.py

@@ -0,0 +1,16 @@
+"""Cost-center tagging and rollups (#484)."""
+
+from app.services import usage_dashboards, usage_ledger
+
+
+def test_usage_rolls_up_by_cost_center(app, db, make_user):
+    user = make_user(username="costowner", email="costowner@example.com")
+
+    usage_ledger.record_usage(source="chat", user=user, cost_center="engineering", total_tokens=10)
+    usage_ledger.record_usage(source="chat", user=user, total_tokens=5)
+
+    payload = usage_dashboards.dashboard(user_id=user.id, days=7)
+    rows = {row["cost_center"]: row for row in payload["by_cost_center"]}
+
+    assert rows["engineering"]["total_tokens"] == 10
+    assert rows["unknown"]["total_tokens"] == 5