feat: add backup CLI commands, AI correction/modifier services, and timezone-aware date display

- Add `devplace backups` CLI subcommands (list, run, prune, clear) with job enqueueing and orphan cleanup
- Introduce `BACKUPS_DIR` and `BACKUP_STAGING_DIR` config paths for backup storage
- Implement `schedule_correction` and `schedule_modification` calls in content creation, comment creation, and comment editing flows
- Add `DEFAULT_CORRECTION_PROMPT` and `DEFAULT_MODIFIER_PROMPT` config constants for AI content processing
- Document timezone-aware date display using `local_dt`/`dt_ago` Jinja globals with client-side `Intl` localization
- Update README with AI content correction/modifier support in direct messages via `@ai` inline instructions
- Add `track_action(user["uid"], "vote")` call on upvote in `apply_vote`

devplacepy/services/ai_context.py

@@ -0,0 +1,186 @@
+# retoor <retoor@molodetz.nl>
+
+import logging
+from datetime import datetime, timezone
+
+from devplacepy.database import (
+    db,
+    get_follow_counts,
+    get_table,
+    get_user_post_count,
+    get_user_rank,
+)
+
+logger = logging.getLogger(__name__)
+
+MAX_BIO = 280
+MAX_EXCERPT = 400
+MAX_SOURCE = 1500
+MAX_THREAD_MESSAGES = 6
+MAX_MESSAGE = 240
+
+TARGET_TABLES = {
+    "post": "posts",
+    "project": "projects",
+    "gist": "gists",
+    "news": "news",
+}
+
+
+def _truncate(text: str, limit: int) -> str:
+    text = " ".join((text or "").split())
+    if len(text) <= limit:
+        return text
+    return text[:limit].rstrip() + "..."
+
+
+def _today() -> str:
+    return datetime.now(timezone.utc).strftime("%d/%m/%Y")
+
+
+def _author_block(user_uid: str) -> str:
+    user = get_table("users").find_one(uid=user_uid)
+    if not user:
+        return ""
+    stats = [
+        f"role {user.get('role') or 'Member'}",
+        f"level {user.get('level', 1)}",
+        f"{user.get('stars', 0)} stars",
+    ]
+    try:
+        stats.append(f"{get_user_post_count(user_uid)} posts")
+    except Exception:
+        pass
+    try:
+        rank = get_user_rank(user_uid)
+        if rank:
+            stats.append(f"rank #{rank}")
+    except Exception:
+        pass
+    try:
+        follows = get_follow_counts(user_uid)
+        stats.append(f"{follows.get('followers', 0)} followers")
+    except Exception:
+        pass
+    joined = (user.get("created_at") or "")[:10]
+    if joined:
+        stats.append(f"member since {joined}")
+    line = f"Author: {user.get('username', '')} ({', '.join(stats)})."
+    bio = _truncate(user.get("bio") or "", MAX_BIO)
+    if bio:
+        line += f' Bio: "{bio}".'
+    return line
+
+
+def _target_excerpt(target_type: str, target_uid: str) -> str:
+    table = TARGET_TABLES.get(target_type)
+    if not table or table not in db.tables or not target_uid:
+        return ""
+    row = get_table(table).find_one(uid=target_uid)
+    if not row:
+        return ""
+    title = (row.get("title") or "").strip()
+    body = _truncate(row.get("content") or row.get("description") or "", MAX_EXCERPT)
+    label = f'the {target_type} "{title}"' if title else f"a {target_type}"
+    text = f"This is a comment on {label}."
+    if body:
+        text += f' Excerpt: "{body}".'
+    return text
+
+
+def _comment_block(row: dict) -> str:
+    target_type = row.get("target_type") or "post"
+    target_uid = row.get("target_uid") or row.get("post_uid") or ""
+    text = _target_excerpt(target_type, target_uid)
+    parent_uid = row.get("parent_uid")
+    if parent_uid and "comments" in db.tables:
+        parent = get_table("comments").find_one(uid=parent_uid)
+        if parent:
+            text += f' In reply to: "{_truncate(parent.get("content") or "", MAX_EXCERPT)}".'
+    return text
+
+
+def _post_block(row: dict) -> str:
+    topic = row.get("topic") or "general"
+    text = f"This is a post (topic: {topic})."
+    project_uid = row.get("project_uid")
+    if project_uid and "projects" in db.tables:
+        project = get_table("projects").find_one(uid=project_uid)
+        if project and project.get("title"):
+            text += f' Attached to the project "{project["title"]}".'
+    return text
+
+
+def _item_block(table: str, row: dict) -> str:
+    kind = "project" if table == "projects" else "gist"
+    title = (row.get("title") or "").strip()
+    text = f'This is a {kind}' + (f' titled "{title}".' if title else ".")
+    description = _truncate(row.get("description") or "", MAX_EXCERPT)
+    if description:
+        text += f' Description: "{description}".'
+    if table == "gists":
+        language = row.get("language") or "plaintext"
+        source = _truncate(row.get("source_code") or "", MAX_SOURCE)
+        text += f" Language: {language}."
+        if source:
+            text += f" Source code (reference only, do not include unless asked): {source}"
+    return text
+
+
+def _message_block(row: dict) -> str:
+    sender_uid = row.get("sender_uid")
+    receiver_uid = row.get("receiver_uid")
+    receiver = get_table("users").find_one(uid=receiver_uid) if receiver_uid else None
+    receiver_name = (receiver or {}).get("username") or "the recipient"
+    text = f"This is a direct message to {receiver_name}."
+    if "messages" not in db.tables or not sender_uid or not receiver_uid:
+        return text
+    rows = list(
+        db.query(
+            "SELECT sender_uid, content FROM messages "
+            "WHERE ((sender_uid = :a AND receiver_uid = :b) "
+            "OR (sender_uid = :b AND receiver_uid = :a)) "
+            "AND uid != :current AND deleted_at IS NULL "
+            "ORDER BY id DESC LIMIT :lim",
+            a=sender_uid,
+            b=receiver_uid,
+            current=row.get("uid") or "",
+            lim=MAX_THREAD_MESSAGES,
+        )
+    )
+    if not rows:
+        return text
+    lines = []
+    for entry in reversed(rows):
+        who = "you" if entry["sender_uid"] == sender_uid else receiver_name
+        lines.append(f"{who}: {_truncate(entry['content'] or '', MAX_MESSAGE)}")
+    return text + " Recent conversation:\n" + "\n".join(lines)
+
+
+def _location_block(table: str, row: dict) -> str:
+    if table == "comments":
+        return _comment_block(row)
+    if table == "posts":
+        return _post_block(row)
+    if table in ("projects", "gists"):
+        return _item_block(table, row)
+    if table == "messages":
+        return _message_block(row)
+    return ""
+
+
+def build_context(table: str, uid: str, row: dict, user_uid: str) -> str:
+    parts = [f"Today is {_today()} on the DevPlace developer network."]
+    try:
+        author = _author_block(user_uid)
+        if author:
+            parts.append(author)
+    except Exception as exc:
+        logger.warning("ai context author block failed: %s", exc)
+    try:
+        location = _location_block(table, row)
+        if location:
+            parts.append(location)
+    except Exception as exc:
+        logger.warning("ai context location block failed: %s", exc)
+    return "\n".join(parts)

devplacepy/services/ai_modifier.py

@@ -0,0 +1,132 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import logging
+import re
+
+from devplacepy.config import DEFAULT_MODIFIER_PROMPT
+from devplacepy.database import add_modifier_usage, get_table
+from devplacepy.services.ai_context import build_context
+from devplacepy.services.background import background
+from devplacepy.services.correction import (
+    CORRECTABLE_FIELDS,
+    PENDING_SCOPE_KEY,
+    gateway_complete,
+)
+
+logger = logging.getLogger(__name__)
+
+MODIFIER_TIMEOUT_SECONDS = 60.0
+AI_DIRECTIVE = re.compile(r"@ai\s+\S", re.IGNORECASE)
+
+
+def has_ai_directive(text: str | None) -> bool:
+    return bool(text) and AI_DIRECTIVE.search(text) is not None
+
+
+def modify_text(
+    api_key: str, prompt: str, text: str, context: str = ""
+) -> tuple[str, dict | None]:
+    system = (
+        "The user's message contains an inline instruction marked with @ai. "
+        + (prompt or DEFAULT_MODIFIER_PROMPT).strip()
+        + " Return ONLY the resulting text, with no preamble, no explanation, no "
+        "quotes and no code fences."
+    )
+    if context:
+        system += (
+            "\n\n# Context (use it to inform the result; never echo this block)\n"
+            + context
+        )
+    return gateway_complete(api_key, system, text, MODIFIER_TIMEOUT_SECONDS, None)
+
+
+def schedule_modification(
+    user: dict | None, table: str, uid: str, request: object = None
+) -> None:
+    if not user or table not in CORRECTABLE_FIELDS or not uid:
+        return
+    if not user.get("ai_modifier_enabled"):
+        return
+    api_key = (user.get("api_key") or "").strip()
+    if not api_key:
+        return
+    row = get_table(table).find_one(uid=uid)
+    if not row:
+        return
+    if not any(
+        has_ai_directive(row.get(field) or "") for field in CORRECTABLE_FIELDS[table]
+    ):
+        return
+    prompt = (user.get("ai_modifier_prompt") or DEFAULT_MODIFIER_PROMPT).strip()
+    user_uid = user.get("uid") or ""
+    if user.get("ai_modifier_sync") and _run_inline_awaited(
+        api_key, prompt, table, uid, user_uid, request
+    ):
+        return
+    background.submit(_run_modification, api_key, prompt, table, uid, user_uid)
+
+
+def _run_inline_awaited(
+    api_key: str,
+    prompt: str,
+    table: str,
+    uid: str,
+    user_uid: str,
+    request: object,
+) -> bool:
+    scope = getattr(request, "scope", None)
+    if scope is None:
+        return False
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return False
+    future = loop.run_in_executor(
+        None, _run_modification, api_key, prompt, table, uid, user_uid
+    )
+    scope.setdefault(PENDING_SCOPE_KEY, []).append(future)
+    return True
+
+
+def _run_modification(
+    api_key: str, prompt: str, table: str, uid: str, user_uid: str
+) -> None:
+    fields = CORRECTABLE_FIELDS.get(table)
+    if not fields:
+        return
+    row = get_table(table).find_one(uid=uid)
+    if not row:
+        return
+    updates: dict = {}
+    totals = {
+        "calls": 0,
+        "prompt_tokens": 0,
+        "completion_tokens": 0,
+        "total_tokens": 0,
+        "cost_usd": 0.0,
+        "upstream_latency_ms": 0.0,
+        "total_latency_ms": 0.0,
+    }
+    context = None
+    for field in fields:
+        original = row.get(field) or ""
+        if not has_ai_directive(original):
+            continue
+        if context is None:
+            context = build_context(table, uid, row, user_uid)
+        modified, usage = modify_text(api_key, prompt, original, context)
+        if usage:
+            for key in totals:
+                totals[key] += usage[key]
+        if modified and modified != original:
+            updates[field] = modified
+    if updates:
+        updates["uid"] = uid
+        get_table(table).update(updates, ["uid"])
+        if table == "users" and user_uid:
+            from devplacepy.utils import clear_user_cache
+
+            clear_user_cache(user_uid)
+    if totals["calls"] and user_uid:
+        add_modifier_usage(user_uid, totals)

devplacepy/services/audit/categories.py

@@ -20,6 +20,8 @@ CATEGORY_BY_PREFIX: dict[str, str] = {
     "dir": "project_files",
     "files": "project_files",
     "job.zip": "project_files",
+    "job.backup": "backup",
+    "backup": "backup",
     "attachment": "attachment",
     "news": "news",
     "admin": "admin",

devplacepy/services/backup/__init__.py

@@ -0,0 +1,5 @@
+# retoor <retoor@molodetz.nl>
+
+from devplacepy.services.backup.service import BackupService
+
+__all__ = ["BackupService"]

devplacepy/services/backup/service.py

@@ -0,0 +1,238 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import json
+import logging
+import shutil
+import sqlite3
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+from devplacepy import config
+from devplacepy.attachments import _directory_for
+from devplacepy.services.backup import store
+from devplacepy.services.devii.tasks.schedule import next_run as schedule_next_run
+from devplacepy.services.devii.tasks.schedule import now_utc, to_iso
+from devplacepy.services.jobs import queue
+from devplacepy.services.jobs.base import JobService, _human_bytes
+from devplacepy.utils import generate_uid
+
+logger = logging.getLogger(__name__)
+
+WORKER_MODULE = "devplacepy.services.jobs.backup_worker"
+
+
+class BackupService(JobService):
+    kind = "backup"
+    title = "Backup"
+    description = (
+        "Builds compressed tar.gz backups of selected data targets in a subprocess off the "
+        "request path, snapshots the database consistently, records full statistics, fires "
+        "scheduled backups, and rotates them by retention."
+    )
+
+    def __init__(self):
+        super().__init__(name="backup", interval_seconds=15)
+
+    async def run_once(self) -> None:
+        await super().run_once()
+        try:
+            self._fire_due_schedules()
+        except Exception as exc:
+            self.log(f"Schedule pass failed: {exc}")
+
+    async def process(self, job: dict) -> dict:
+        from devplacepy.services.audit import record as audit
+
+        payload = job["payload"]
+        target = payload.get("target", "")
+        schedule_uid = payload.get("schedule_uid", "")
+        keep_last = int(payload.get("keep_last") or 0)
+        job_uid = job["uid"]
+        record = store.get_backup_by_job(job_uid)
+        if record is None:
+            backup_uid = store.create_backup(
+                target=target,
+                created_by=job.get("owner_id", ""),
+                job_uid=job_uid,
+                schedule_uid=schedule_uid,
+            )
+            record = store.get_backup(backup_uid)
+        backup_uid = record["uid"]
+
+        if not store.is_valid_target(target):
+            store.fail_backup(backup_uid, f"unknown backup target: {target}")
+            raise ValueError(f"unknown backup target: {target}")
+
+        store.mark_running(backup_uid)
+        staging = config.BACKUP_STAGING_DIR / job_uid
+        try:
+            sources = await asyncio.to_thread(self._materialize, target, staging)
+            spec_path = staging / "spec.json"
+            await asyncio.to_thread(
+                spec_path.write_text, json.dumps({"sources": sources})
+            )
+            final_dir = config.BACKUPS_DIR / _directory_for(backup_uid)
+            await asyncio.to_thread(final_dir.mkdir, parents=True, exist_ok=True)
+            filename = self._archive_name(target, backup_uid)
+            tmp_path = final_dir / f"{generate_uid()}.partial"
+            stats = await self._run_worker(spec_path, tmp_path)
+            final_path = final_dir / filename
+            await asyncio.to_thread(tmp_path.replace, final_path)
+        except Exception as exc:
+            store.fail_backup(backup_uid, str(exc) or exc.__class__.__name__)
+            audit.record_system(
+                "job.backup.failed",
+                actor_kind="system",
+                origin="scheduler" if schedule_uid else "web",
+                result="failure",
+                target_type="backup",
+                target_uid=backup_uid,
+                metadata={"target": target, "schedule_uid": schedule_uid},
+                summary=f"backup {target} failed",
+                links=[audit.job(job_uid)],
+            )
+            raise
+        finally:
+            await asyncio.to_thread(shutil.rmtree, staging, ignore_errors=True)
+
+        store.finalize_backup(
+            backup_uid,
+            filename=filename,
+            local_path=str(final_path),
+            stats=stats,
+        )
+        removed = store.rotate_schedule(schedule_uid, keep_last)
+        audit.record_system(
+            "job.backup.complete",
+            actor_kind="system",
+            origin="scheduler" if schedule_uid else "web",
+            target_type="backup",
+            target_uid=backup_uid,
+            metadata={
+                "target": target,
+                "schedule_uid": schedule_uid,
+                "bytes_out": stats["bytes_out"],
+                "sha256": stats["sha256"],
+                "file_count": stats["file_count"],
+                "rotated": removed,
+            },
+            summary=f"backup {target} completed ({_human_bytes(stats['bytes_out'])})",
+            links=[audit.job(job_uid)],
+        )
+        return {
+            "backup_uid": backup_uid,
+            "target": target,
+            "filename": filename,
+            "local_path": str(final_path),
+            "sha256": stats["sha256"],
+            "file_count": stats["file_count"],
+            "dir_count": stats["dir_count"],
+            "bytes_in": stats["bytes_in"],
+            "bytes_out": stats["bytes_out"],
+            "item_count": stats["file_count"],
+        }
+
+    def cleanup(self, job: dict) -> None:
+        shutil.rmtree(config.BACKUP_STAGING_DIR / job["uid"], ignore_errors=True)
+
+    def _materialize(self, target: str, staging: Path) -> list[dict]:
+        staging.mkdir(parents=True, exist_ok=True)
+        sources: list[dict] = []
+        if target in ("database", "full"):
+            snapshot = staging / "db_snapshot"
+            self._snapshot_databases(snapshot)
+            sources.append({"root": "database", "path": str(snapshot)})
+        if target in ("uploads", "full"):
+            sources.append({"root": "uploads", "path": str(config.UPLOADS_DIR)})
+        if target in ("keys", "full"):
+            sources.append({"root": "keys", "path": str(config.KEYS_DIR)})
+        return sources
+
+    def _snapshot_databases(self, dest: Path) -> None:
+        dest.mkdir(parents=True, exist_ok=True)
+        database_file = Path(str(config.DATABASE_URL).replace("sqlite:///", ""))
+        for source in (database_file, config.DEVII_TASKS_DB, config.DEVII_LESSONS_DB):
+            if Path(source).exists():
+                self._sqlite_backup(Path(source), dest / Path(source).name)
+
+    def _sqlite_backup(self, source: Path, destination: Path) -> None:
+        origin = sqlite3.connect(f"file:{source}?mode=ro", uri=True)
+        try:
+            target = sqlite3.connect(str(destination))
+            try:
+                origin.backup(target)
+            finally:
+                target.close()
+        finally:
+            origin.close()
+
+    async def _run_worker(self, spec_path: Path, output_path: Path) -> dict:
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable,
+            "-m",
+            WORKER_MODULE,
+            str(spec_path),
+            str(output_path),
+            cwd=str(config.BASE_DIR),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        out, err = await proc.communicate()
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"backup worker exited {proc.returncode}: "
+                f"{err.decode('utf-8', 'replace')[:500]}"
+            )
+        return json.loads(out.decode("utf-8"))
+
+    def _archive_name(self, target: str, backup_uid: str) -> str:
+        stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
+        tail = backup_uid.replace("-", "")[-8:]
+        return f"{target}-{stamp}-{tail}.tar.gz"
+
+    def _fire_due_schedules(self) -> None:
+        reference = to_iso(now_utc())
+        for schedule in store.list_due_schedules(reference):
+            self._enqueue_scheduled(schedule, reference)
+
+    def _enqueue_scheduled(self, schedule: dict, reference: str) -> None:
+        target = schedule["target"]
+        if not store.is_valid_target(target):
+            self.log(f"Schedule {schedule['uid']} skipped: unknown target {target}")
+            return
+        keep_last = int(schedule.get("keep_last") or 0)
+        created_by = schedule.get("created_by", "")
+        job_uid = queue.enqueue(
+            "backup",
+            {
+                "target": target,
+                "schedule_uid": schedule["uid"],
+                "created_by": created_by,
+                "keep_last": keep_last,
+            },
+            owner_kind="system",
+            owner_id=created_by,
+            preferred_name=f"{schedule.get('name', target)} ({target})",
+        )
+        store.create_backup(
+            target=target,
+            created_by=created_by,
+            job_uid=job_uid,
+            schedule_uid=schedule["uid"],
+        )
+        moment = schedule_next_run(
+            schedule["kind"],
+            int(schedule.get("every_seconds") or 0),
+            schedule.get("cron") or None,
+            now_utc(),
+        )
+        store.set_schedule_runtime(
+            schedule["uid"],
+            next_run_at=to_iso(moment) if moment else "",
+            last_run_at=reference,
+            last_job_uid=job_uid,
+            run_count=int(schedule.get("run_count") or 0) + 1,
+        )
+        self.log(f"Scheduled backup '{schedule.get('name')}' enqueued ({job_uid})")

devplacepy/services/backup/store.py

@@ -0,0 +1,437 @@
+# retoor <retoor@molodetz.nl>
+
+import shutil
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from devplacepy import config
+from devplacepy.database import _index, db, get_table
+from devplacepy.utils import generate_uid
+
+BACKUP_TARGETS: dict[str, dict] = {
+    "database": {
+        "label": "Database",
+        "description": "Consistent snapshot of the SQLite database and the Devii task and lesson databases.",
+    },
+    "uploads": {
+        "label": "Uploads",
+        "description": "All attachments and project files under the uploads directory.",
+    },
+    "keys": {
+        "label": "Keys and config",
+        "description": "VAPID notification keys and other small config artifacts.",
+    },
+    "full": {
+        "label": "Full data directory",
+        "description": "Database, uploads, and keys in one archive, excluding regenerable staging, locks, and caches.",
+    },
+}
+
+STATUS_PENDING = "pending"
+STATUS_RUNNING = "running"
+STATUS_DONE = "done"
+STATUS_FAILED = "failed"
+
+STORAGE_CACHE_TTL_SECONDS = 30
+
+_storage_cache: dict = {"at": 0.0, "data": None}
+
+
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def is_valid_target(key: str) -> bool:
+    return key in BACKUP_TARGETS
+
+
+def target_label(key: str) -> str:
+    return BACKUP_TARGETS.get(key, {}).get("label", key)
+
+
+def human_bytes(size: int) -> str:
+    value = float(max(0, int(size or 0)))
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if value < 1024 or unit == "TB":
+            return f"{value:.0f} {unit}" if unit == "B" else f"{value:.1f} {unit}"
+        value /= 1024
+    return f"{value:.1f} TB"
+
+
+def ensure_tables() -> None:
+    backups = get_table("backups")
+    for column, example in (
+        ("uid", ""),
+        ("job_uid", ""),
+        ("target", ""),
+        ("label", ""),
+        ("status", STATUS_PENDING),
+        ("filename", ""),
+        ("local_path", ""),
+        ("size_bytes", 0),
+        ("bytes_in", 0),
+        ("file_count", 0),
+        ("dir_count", 0),
+        ("sha256", ""),
+        ("schedule_uid", ""),
+        ("created_by", ""),
+        ("created_at", ""),
+        ("completed_at", ""),
+        ("error", ""),
+    ):
+        if not backups.has_column(column):
+            backups.create_column_by_example(column, example)
+
+    schedules = get_table("backup_schedules")
+    for column, example in (
+        ("uid", ""),
+        ("name", ""),
+        ("target", ""),
+        ("kind", "interval"),
+        ("every_seconds", 0),
+        ("cron", ""),
+        ("enabled", 1),
+        ("keep_last", 0),
+        ("next_run_at", ""),
+        ("last_run_at", ""),
+        ("last_job_uid", ""),
+        ("run_count", 0),
+        ("created_by", ""),
+        ("created_at", ""),
+        ("updated_at", ""),
+        ("deleted_at", None),
+        ("deleted_by", None),
+    ):
+        if not schedules.has_column(column):
+            schedules.create_column_by_example(column, example)
+
+    _index(db, "backups", "idx_backups_status", ["status"])
+    _index(db, "backups", "idx_backups_job", ["job_uid"])
+    _index(db, "backups", "idx_backups_schedule", ["schedule_uid", "created_at"])
+    _index(db, "backup_schedules", "idx_backup_schedules_enabled", ["enabled"])
+
+
+def create_backup(
+    *, target: str, created_by: str, job_uid: str, schedule_uid: str = ""
+) -> str:
+    uid = generate_uid()
+    get_table("backups").insert(
+        {
+            "uid": uid,
+            "job_uid": job_uid,
+            "target": target,
+            "label": target_label(target),
+            "status": STATUS_PENDING,
+            "filename": "",
+            "local_path": "",
+            "size_bytes": 0,
+            "bytes_in": 0,
+            "file_count": 0,
+            "dir_count": 0,
+            "sha256": "",
+            "schedule_uid": schedule_uid,
+            "created_by": created_by,
+            "created_at": now_iso(),
+            "completed_at": "",
+            "error": "",
+        }
+    )
+    return uid
+
+
+def get_backup(uid: str) -> dict | None:
+    if "backups" not in db.tables:
+        return None
+    return get_table("backups").find_one(uid=uid)
+
+
+def get_backup_by_job(job_uid: str) -> dict | None:
+    if "backups" not in db.tables:
+        return None
+    return get_table("backups").find_one(job_uid=job_uid)
+
+
+def list_backups(limit: int = 100) -> list[dict]:
+    if "backups" not in db.tables:
+        return []
+    return list(get_table("backups").find(order_by=["-created_at"], _limit=limit))
+
+
+def mark_running(uid: str) -> None:
+    get_table("backups").update(
+        {"uid": uid, "status": STATUS_RUNNING}, ["uid"]
+    )
+
+
+def finalize_backup(uid: str, *, filename: str, local_path: str, stats: dict) -> None:
+    get_table("backups").update(
+        {
+            "uid": uid,
+            "status": STATUS_DONE,
+            "filename": filename,
+            "local_path": local_path,
+            "size_bytes": int(stats.get("bytes_out", 0)),
+            "bytes_in": int(stats.get("bytes_in", 0)),
+            "file_count": int(stats.get("file_count", 0)),
+            "dir_count": int(stats.get("dir_count", 0)),
+            "sha256": stats.get("sha256", ""),
+            "completed_at": now_iso(),
+            "error": "",
+        },
+        ["uid"],
+    )
+
+
+def fail_backup(uid: str, error: str) -> None:
+    get_table("backups").update(
+        {
+            "uid": uid,
+            "status": STATUS_FAILED,
+            "completed_at": now_iso(),
+            "error": error[:2000],
+        },
+        ["uid"],
+    )
+
+
+def delete_backup(uid: str) -> dict | None:
+    row = get_backup(uid)
+    if not row:
+        return None
+    _unlink_archive(row)
+    get_table("backups").delete(uid=uid)
+    return row
+
+
+def rotate_schedule(schedule_uid: str, keep_last: int) -> int:
+    if keep_last <= 0 or not schedule_uid:
+        return 0
+    rows = list(
+        get_table("backups").find(
+            schedule_uid=schedule_uid,
+            status=STATUS_DONE,
+            order_by=["-created_at"],
+        )
+    )
+    removed = 0
+    for row in rows[keep_last:]:
+        delete_backup(row["uid"])
+        removed += 1
+    return removed
+
+
+def prune_orphans() -> int:
+    removed = 0
+    for row in list_backups(limit=100000):
+        if row.get("status") != STATUS_DONE:
+            continue
+        local_path = row.get("local_path") or ""
+        if not local_path or not Path(local_path).is_file():
+            get_table("backups").delete(uid=row["uid"])
+            removed += 1
+    return removed
+
+
+def clear_all() -> int:
+    rows = list_backups(limit=100000)
+    for row in rows:
+        _unlink_archive(row)
+    get_table("backups").delete()
+    return len(rows)
+
+
+def _unlink_archive(row: dict) -> None:
+    local_path = row.get("local_path") or ""
+    if local_path:
+        Path(local_path).unlink(missing_ok=True)
+
+
+def create_schedule(
+    *,
+    name: str,
+    target: str,
+    kind: str,
+    every_seconds: int,
+    cron: str,
+    keep_last: int,
+    created_by: str,
+    next_run_at: str,
+) -> str:
+    uid = generate_uid()
+    get_table("backup_schedules").insert(
+        {
+            "uid": uid,
+            "name": name,
+            "target": target,
+            "kind": kind,
+            "every_seconds": every_seconds,
+            "cron": cron,
+            "enabled": 1,
+            "keep_last": keep_last,
+            "next_run_at": next_run_at,
+            "last_run_at": "",
+            "last_job_uid": "",
+            "run_count": 0,
+            "created_by": created_by,
+            "created_at": now_iso(),
+            "updated_at": now_iso(),
+            "deleted_at": None,
+            "deleted_by": None,
+        }
+    )
+    return uid
+
+
+def get_schedule(uid: str) -> dict | None:
+    if "backup_schedules" not in db.tables:
+        return None
+    return get_table("backup_schedules").find_one(uid=uid, deleted_at=None)
+
+
+def list_schedules() -> list[dict]:
+    if "backup_schedules" not in db.tables:
+        return []
+    return list(
+        get_table("backup_schedules").find(
+            deleted_at=None, order_by=["-created_at"]
+        )
+    )
+
+
+def list_due_schedules(reference: str) -> list[dict]:
+    if "backup_schedules" not in db.tables:
+        return []
+    rows = get_table("backup_schedules").find(deleted_at=None, enabled=1)
+    return [
+        row
+        for row in rows
+        if (row.get("next_run_at") or "") and row["next_run_at"] <= reference
+    ]
+
+
+def update_schedule(uid: str, changes: dict) -> None:
+    changes = {**changes, "uid": uid, "updated_at": now_iso()}
+    get_table("backup_schedules").update(changes, ["uid"])
+
+
+def set_schedule_runtime(
+    uid: str, *, next_run_at: str, last_run_at: str, last_job_uid: str, run_count: int
+) -> None:
+    get_table("backup_schedules").update(
+        {
+            "uid": uid,
+            "next_run_at": next_run_at,
+            "last_run_at": last_run_at,
+            "last_job_uid": last_job_uid,
+            "run_count": run_count,
+            "updated_at": now_iso(),
+        },
+        ["uid"],
+    )
+
+
+def delete_schedule(uid: str, deleted_by: str) -> bool:
+    row = get_schedule(uid)
+    if not row:
+        return False
+    get_table("backup_schedules").update(
+        {"uid": uid, "deleted_at": now_iso(), "deleted_by": deleted_by}, ["uid"]
+    )
+    return True
+
+
+def _path_size(path: Path) -> tuple[int, int]:
+    if not path.exists():
+        return 0, 0
+    if path.is_file():
+        return path.stat().st_size, 1
+    total = 0
+    files = 0
+    for entry in path.rglob("*"):
+        try:
+            if entry.is_file() and not entry.is_symlink():
+                total += entry.stat().st_size
+                files += 1
+        except OSError:
+            continue
+    return total, files
+
+
+def _storage_paths() -> list[tuple[str, str, Path]]:
+    database_file = Path(str(config.DATABASE_URL).replace("sqlite:///", ""))
+    return [
+        ("database", "Database file", database_file),
+        ("devii_tasks", "Devii tasks DB", config.DEVII_TASKS_DB),
+        ("devii_lessons", "Devii lessons DB", config.DEVII_LESSONS_DB),
+        ("uploads", "Uploads", config.UPLOADS_DIR),
+        ("attachments", "Attachments", config.ATTACHMENTS_DIR),
+        ("project_files", "Project files", config.PROJECT_FILES_DIR),
+        ("keys", "Keys", config.KEYS_DIR),
+        ("zips", "Zip archives", config.ZIPS_DIR),
+        ("deepsearch", "DeepSearch", config.DEEPSEARCH_DIR),
+        ("container_workspaces", "Container workspaces", config.CONTAINER_WORKSPACES_DIR),
+        ("backups", "Backups", config.BACKUPS_DIR),
+    ]
+
+
+def compute_storage_stats() -> dict:
+    now = time.monotonic()
+    if (
+        _storage_cache["data"] is not None
+        and (now - _storage_cache["at"]) < STORAGE_CACHE_TTL_SECONDS
+    ):
+        return _storage_cache["data"]
+
+    paths = []
+    for key, label, path in _storage_paths():
+        size, files = _path_size(path)
+        paths.append(
+            {
+                "key": key,
+                "label": label,
+                "path": str(path),
+                "size_bytes": size,
+                "size_human": human_bytes(size),
+                "file_count": files,
+                "exists": path.exists(),
+            }
+        )
+
+    data_size, data_files = _path_size(config.DATA_DIR)
+    backups_size, backups_files = _path_size(config.BACKUPS_DIR)
+    backup_count = len(
+        [b for b in list_backups(limit=100000) if b.get("status") == STATUS_DONE]
+    )
+    usage = shutil.disk_usage(str(config.DATA_DIR))
+
+    data = {
+        "paths": paths,
+        "data_dir": {
+            "path": str(config.DATA_DIR),
+            "size_bytes": data_size,
+            "size_human": human_bytes(data_size),
+            "file_count": data_files,
+        },
+        "backups_total": {
+            "count": backup_count,
+            "size_bytes": backups_size,
+            "size_human": human_bytes(backups_size),
+            "file_count": backups_files,
+        },
+        "disk": {
+            "total_bytes": usage.total,
+            "used_bytes": usage.used,
+            "free_bytes": usage.free,
+            "total_human": human_bytes(usage.total),
+            "used_human": human_bytes(usage.used),
+            "free_human": human_bytes(usage.free),
+            "used_percent": round(usage.used / usage.total * 100, 1)
+            if usage.total
+            else 0.0,
+        },
+        "generated_at": now_iso(),
+    }
+    _storage_cache["data"] = data
+    _storage_cache["at"] = now
+    return data

devplacepy/services/containers/api.py

@@ -269,6 +269,10 @@ async def create_instance(
     store.record_event(
         instance, "created", actor[0], actor[1], {"image": config.CONTAINER_IMAGE}
     )
+    if actor and actor[0] == "user":
+        from devplacepy.utils import track_action
+
+        track_action(actor[1], "container")
     return instance
 
 

devplacepy/services/correction.py

@@ -0,0 +1,192 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import logging
+
+import httpx
+
+from devplacepy import stealth
+from devplacepy.config import (
+    DEFAULT_CORRECTION_PROMPT,
+    INTERNAL_GATEWAY_URL,
+    INTERNAL_MODEL,
+)
+from devplacepy.database import add_correction_usage, get_table
+from devplacepy.services.background import background
+
+logger = logging.getLogger(__name__)
+
+CORRECTABLE_FIELDS: dict[str, tuple[str, ...]] = {
+    "posts": ("title", "content"),
+    "projects": ("title", "description"),
+    "gists": ("title", "description"),
+    "comments": ("content",),
+    "messages": ("content",),
+    "users": ("bio",),
+}
+
+CORRECTION_TIMEOUT_SECONDS = 20.0
+MAX_GROWTH_FACTOR = 3
+PENDING_SCOPE_KEY = "devplace_pending_corrections"
+
+
+def _usage_from_headers(response_headers) -> dict:
+    def _int(name: str) -> int:
+        try:
+            return int(response_headers.get(name) or 0)
+        except (TypeError, ValueError):
+            return 0
+
+    def _float(name: str) -> float:
+        try:
+            return float(response_headers.get(name) or 0.0)
+        except (TypeError, ValueError):
+            return 0.0
+
+    return {
+        "calls": 1,
+        "prompt_tokens": _int("X-Gateway-Prompt-Tokens"),
+        "completion_tokens": _int("X-Gateway-Completion-Tokens"),
+        "total_tokens": _int("X-Gateway-Total-Tokens"),
+        "cost_usd": _float("X-Gateway-Cost-USD"),
+        "upstream_latency_ms": _float("X-Gateway-Upstream-Latency-Ms"),
+        "total_latency_ms": _float("X-Gateway-Total-Latency-Ms"),
+    }
+
+
+def gateway_complete(
+    api_key: str,
+    system: str,
+    text: str,
+    timeout: float,
+    max_growth_factor: int | None = None,
+) -> tuple[str, dict | None]:
+    text = text or ""
+    if not text.strip():
+        return text, None
+    payload = {
+        "model": INTERNAL_MODEL,
+        "messages": [
+            {"role": "system", "content": system},
+            {"role": "user", "content": text},
+        ],
+        "temperature": 0.1,
+    }
+    headers = {"Content-Type": "application/json"}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    try:
+        with stealth.stealth_sync_client(timeout=timeout) as client:
+            response = client.post(INTERNAL_GATEWAY_URL, json=payload, headers=headers)
+            response.raise_for_status()
+            usage = _usage_from_headers(response.headers)
+            content = (
+                response.json()
+                .get("choices", [{}])[0]
+                .get("message", {})
+                .get("content", "")
+                .strip()
+            )
+    except (httpx.HTTPError, ValueError, KeyError, IndexError) as exc:
+        logger.warning("AI gateway completion failed, keeping original: %s", exc)
+        return text, None
+    if not content:
+        return text, usage
+    if max_growth_factor and len(content) > len(text) * max_growth_factor + 200:
+        logger.warning("AI gateway output too large, keeping original")
+        return text, usage
+    return content, usage
+
+
+def correct_text(api_key: str, prompt: str, text: str) -> tuple[str, dict | None]:
+    system = (
+        "You are a text correction engine. Apply the correction instruction below to "
+        "the user's message and return ONLY the resulting text, with no preamble, no "
+        "explanation, no quotes and no code fences. Preserve the original meaning, "
+        "language, line breaks and markdown. Correction instruction: "
+        + (prompt or DEFAULT_CORRECTION_PROMPT).strip()
+    )
+    return gateway_complete(
+        api_key, system, text, CORRECTION_TIMEOUT_SECONDS, MAX_GROWTH_FACTOR
+    )
+
+
+def schedule_correction(
+    user: dict | None, table: str, uid: str, request: object = None
+) -> None:
+    if not user or table not in CORRECTABLE_FIELDS or not uid:
+        return
+    if not user.get("ai_correction_enabled"):
+        return
+    api_key = (user.get("api_key") or "").strip()
+    if not api_key:
+        return
+    prompt = (user.get("ai_correction_prompt") or DEFAULT_CORRECTION_PROMPT).strip()
+    user_uid = user.get("uid") or ""
+    if user.get("ai_correction_sync") and _run_inline_awaited(
+        api_key, prompt, table, uid, user_uid, request
+    ):
+        return
+    background.submit(_run_correction, api_key, prompt, table, uid, user_uid)
+
+
+def _run_inline_awaited(
+    api_key: str,
+    prompt: str,
+    table: str,
+    uid: str,
+    user_uid: str,
+    request: object,
+) -> bool:
+    scope = getattr(request, "scope", None)
+    if scope is None:
+        return False
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return False
+    future = loop.run_in_executor(
+        None, _run_correction, api_key, prompt, table, uid, user_uid
+    )
+    scope.setdefault(PENDING_SCOPE_KEY, []).append(future)
+    return True
+
+
+def _run_correction(
+    api_key: str, prompt: str, table: str, uid: str, user_uid: str
+) -> None:
+    fields = CORRECTABLE_FIELDS.get(table)
+    if not fields:
+        return
+    row = get_table(table).find_one(uid=uid)
+    if not row:
+        return
+    updates: dict = {}
+    totals = {
+        "calls": 0,
+        "prompt_tokens": 0,
+        "completion_tokens": 0,
+        "total_tokens": 0,
+        "cost_usd": 0.0,
+        "upstream_latency_ms": 0.0,
+        "total_latency_ms": 0.0,
+    }
+    for field in fields:
+        original = row.get(field) or ""
+        if not original.strip():
+            continue
+        corrected, usage = correct_text(api_key, prompt, original)
+        if usage:
+            for key in totals:
+                totals[key] += usage[key]
+        if corrected and corrected != original:
+            updates[field] = corrected
+    if updates:
+        updates["uid"] = uid
+        get_table(table).update(updates, ["uid"])
+        if table == "users" and user_uid:
+            from devplacepy.utils import clear_user_cache
+
+            clear_user_cache(user_uid)
+    if totals["calls"] and user_uid:
+        add_correction_usage(user_uid, totals)

devplacepy/services/devii/actions/ai_correction_actions.py

@@ -0,0 +1,68 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+from .spec import Action, Param
+
+
+def arg(
+    name: str, description: str, required: bool = False, kind: str = "string"
+) -> Param:
+    return Param(
+        name=name,
+        location="body",
+        description=description,
+        required=required,
+        type=kind,
+    )
+
+
+AI_CORRECTION_ACTIONS: tuple[Action, ...] = (
+    Action(
+        name="ai_correction_get",
+        method="LOCAL",
+        path="",
+        summary="Show the user's AI content correction setting and prompt",
+        description=(
+            "Returns whether automatic AI content correction is enabled for the user and the "
+            "correction instruction in use. Use this before changing the setting."
+        ),
+        handler="ai_correction",
+        requires_auth=True,
+        read_only=True,
+    ),
+    Action(
+        name="ai_correction_set",
+        method="LOCAL",
+        path="",
+        summary="Enable or disable AI content correction and set its prompt",
+        description=(
+            "Turns automatic AI content correction on or off for the user. When enabled, prose the "
+            "user authors (posts, comments, projects, gists, direct messages, profile bio) is "
+            "rewritten using the supplied instruction and the user's own API key. By default the "
+            "correction runs in the background (applied just after saving); pass sync=true to apply it "
+            "synchronously so the save waits for the correction. "
+            "Pass 'prompt' to change the correction instruction; omit it to keep the current one."
+        ),
+        handler="ai_correction",
+        requires_auth=True,
+        params=(
+            arg(
+                "enabled",
+                "true to enable automatic correction, false to disable it.",
+                required=True,
+                kind="boolean",
+            ),
+            arg(
+                "sync",
+                "true to apply correction synchronously (the save waits), false for background. "
+                "Omit to keep the current mode.",
+                kind="boolean",
+            ),
+            arg(
+                "prompt",
+                "The correction instruction (max 2000 chars). Omit to keep the current one.",
+            ),
+        ),
+    ),
+)

devplacepy/services/devii/actions/ai_modifier_actions.py

@@ -0,0 +1,71 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+from .spec import Action, Param
+
+
+def arg(
+    name: str, description: str, required: bool = False, kind: str = "string"
+) -> Param:
+    return Param(
+        name=name,
+        location="body",
+        description=description,
+        required=required,
+        type=kind,
+    )
+
+
+AI_MODIFIER_ACTIONS: tuple[Action, ...] = (
+    Action(
+        name="ai_modifier_get",
+        method="LOCAL",
+        path="",
+        summary="Show the user's AI modifier setting and prompt",
+        description=(
+            "Returns whether the AI modifier is enabled for the user and the instruction in use. "
+            "The modifier runs only when authored text contains an inline '@ai <instruction>' "
+            "directive: it executes that instruction and replaces the marked part. Use this before "
+            "changing the setting."
+        ),
+        handler="ai_modifier",
+        requires_auth=True,
+        read_only=True,
+    ),
+    Action(
+        name="ai_modifier_set",
+        method="LOCAL",
+        path="",
+        summary="Enable or disable the AI modifier and set its prompt",
+        description=(
+            "Turns the AI modifier on or off for the user. When enabled, prose the user authors "
+            "(posts, comments, projects, gists, direct messages, profile bio) is rewritten using the "
+            "supplied instruction and the user's own API key, but ONLY where the text contains an "
+            "inline '@ai <instruction>' directive: that part is executed and the '@ai' marker removed. "
+            "By default the modification runs synchronously (the save waits); pass sync=false to apply "
+            "it in the background just after saving. Pass 'prompt' to change the instruction; omit it to "
+            "keep the current one."
+        ),
+        handler="ai_modifier",
+        requires_auth=True,
+        params=(
+            arg(
+                "enabled",
+                "true to enable the AI modifier, false to disable it.",
+                required=True,
+                kind="boolean",
+            ),
+            arg(
+                "sync",
+                "true to apply the modification synchronously (the save waits), false for background. "
+                "Omit to keep the current mode.",
+                kind="boolean",
+            ),
+            arg(
+                "prompt",
+                "The modifier instruction (max 2000 chars). Omit to keep the current one.",
+            ),
+        ),
+    ),
+)

devplacepy/services/devii/actions/catalog.py

@@ -1415,6 +1415,96 @@ ACTIONS: tuple[Action, ...] = (
         params=(confirm(),),
         requires_admin=True,
     ),
+    Action(
+        name="backups_overview",
+        method="GET",
+        path="/admin/backups/data",
+        summary="Backup dashboard: storage usage, backups, and schedules (admin only)",
+        description=(
+            "Returns JSON: per-path storage usage (database, uploads, attachments, project files, "
+            "keys, zips, deepsearch, container workspaces, backups), total data-directory size, total "
+            "stored-backup size and count, disk usage (total/used/free/percent), every backup archive "
+            "(target, status, size, file count, sha256 checksum, created/completed time, download_url), "
+            "and every configured backup schedule. Use this for any question about backups, data "
+            "footprint, disk space, or what is scheduled."
+        ),
+        requires_admin=True,
+    ),
+    Action(
+        name="backup_run",
+        method="POST",
+        path="/admin/backups/run",
+        summary="Start a backup of a data target (admin only)",
+        description=(
+            "Enqueues an async backup job and returns its job uid and status_url. Target is one of: "
+            "database (consistent SQLite snapshot plus Devii databases), uploads (attachments and "
+            "project files), keys (VAPID keys and config), or full (database, uploads, and keys in one "
+            "archive). Poll backup_status with the returned uid until status is done, then read the "
+            "download_url. The job runs off the request path and never blocks the server."
+        ),
+        params=(
+            body("target", "One of database, uploads, keys, full.", required=True),
+        ),
+        requires_admin=True,
+    ),
+    Action(
+        name="backup_status",
+        method="GET",
+        path="/admin/backups/jobs/{uid}",
+        summary="Check the status of a backup job (admin only)",
+        description=(
+            "Returns JSON for one backup job: status (pending, running, done, failed), target, "
+            "backup_uid, download_url (when done), sha256, archive size, file count, and timestamps. "
+            "Poll this after backup_run."
+        ),
+        params=(path("uid", "Backup job uid returned by backup_run."),),
+        requires_admin=True,
+    ),
+    Action(
+        name="backup_delete",
+        method="POST",
+        path="/admin/backups/{uid}/delete",
+        summary="Permanently delete a backup archive (admin only)",
+        description=(
+            "Removes a backup archive file and its record. This is irreversible and reclaims disk "
+            "space. The uid is a backup uid (from backups_overview), not a job uid."
+        ),
+        params=(path("uid", "Backup uid to delete."), confirm()),
+        requires_admin=True,
+    ),
+    Action(
+        name="backup_schedule_create",
+        method="POST",
+        path="/admin/backups/schedules/create",
+        summary="Create a recurring backup schedule (admin only)",
+        description=(
+            "Creates a schedule that fires backups automatically. kind is 'interval' (use every_seconds, "
+            "minimum 60) or 'cron' (use a 5-field cron expression in 'minute hour dom month dow'). "
+            "keep_last rotates older backups of this schedule, keeping only the newest N (0 keeps all). "
+            "target is one of database, uploads, keys, full."
+        ),
+        params=(
+            body("name", "Human-readable schedule name.", required=True),
+            body("target", "One of database, uploads, keys, full.", required=True),
+            body("kind", "interval or cron.", required=True),
+            body("every_seconds", "Seconds between runs when kind=interval (min 60)."),
+            body("cron", "Cron expression when kind=cron, e.g. '0 3 * * *'."),
+            body("keep_last", "Keep only the newest N backups of this schedule (0 = all)."),
+        ),
+        requires_admin=True,
+    ),
+    Action(
+        name="backup_schedule_delete",
+        method="POST",
+        path="/admin/backups/schedules/{uid}/delete",
+        summary="Delete a backup schedule (admin only)",
+        description=(
+            "Removes a backup schedule so it stops firing. Existing backup archives are kept; only the "
+            "schedule is deleted."
+        ),
+        params=(path("uid", "Backup schedule uid."), confirm()),
+        requires_admin=True,
+    ),
     Action(
         name="restore_media",
         method="POST",

devplacepy/services/devii/actions/dispatcher.py

@@ -50,6 +50,8 @@ CONFIRM_REQUIRED = {
     "admin_reset_all_ai_quota",
     "admin_reset_guest_ai_quota",
     "admin_reset_user_ai_quota",
+    "backup_delete",
+    "backup_schedule_delete",
     "notification_reset",
     "db_insert_row",
     "db_update_row",
@@ -285,6 +287,12 @@ class Dispatcher:
         from ..notification import NotificationController
 
         self._notification = NotificationController(owner_kind, owner_id)
+        from ..ai_correction import AiCorrectionController
+
+        self._ai_correction = AiCorrectionController(owner_kind, owner_id)
+        from ..ai_modifier import AiModifierController
+
+        self._ai_modifier = AiModifierController(owner_kind, owner_id)
         self._virtual_tools = virtual_tools
         self._behavior = behavior
         self._read_files: set[tuple[str, str]] = set()
@@ -434,6 +442,12 @@ class Dispatcher:
         if action.handler == "notification":
             return await self._notification.dispatch(action.name, arguments)
 
+        if action.handler == "ai_correction":
+            return await self._ai_correction.dispatch(action.name, arguments)
+
+        if action.handler == "ai_modifier":
+            return await self._ai_modifier.dispatch(action.name, arguments)
+
         if action.handler == "behavior":
             if self._behavior is None:
                 return error_result(

devplacepy/services/devii/ai_correction/__init__.py

@@ -0,0 +1,5 @@
+# retoor <retoor@molodetz.nl>
+
+from .controller import AiCorrectionController
+
+__all__ = ["AiCorrectionController"]

devplacepy/services/devii/ai_correction/controller.py

@@ -0,0 +1,94 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from devplacepy.config import DEFAULT_CORRECTION_PROMPT
+from devplacepy.database import get_table
+from devplacepy.services.devii.errors import ToolInputError
+
+logger = logging.getLogger("devii.ai_correction")
+
+
+class AiCorrectionController:
+    def __init__(self, owner_kind: str, owner_id: str) -> None:
+        self._owner_kind = owner_kind
+        self._owner_id = owner_id
+
+    async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
+        if name == "ai_correction_get":
+            return self._get()
+        if name == "ai_correction_set":
+            return self._set(arguments)
+        raise ToolInputError(f"Unknown AI correction tool: {name}")
+
+    def _require_user(self) -> None:
+        if self._owner_kind != "user":
+            raise ToolInputError(
+                "AI content correction is only available for signed-in users."
+            )
+
+    def _coerce_bool(self, raw: Any) -> bool:
+        if isinstance(raw, bool):
+            return raw
+        return str(raw).strip().lower() in ("true", "1", "yes", "on")
+
+    def _value(self, arguments: dict[str, Any]) -> bool:
+        return self._coerce_bool(arguments.get("enabled"))
+
+    def _user(self) -> dict[str, Any]:
+        user = get_table("users").find_one(uid=self._owner_id)
+        if not user:
+            raise ToolInputError("User not found.")
+        return user
+
+    def _get(self) -> str:
+        self._require_user()
+        user = self._user()
+        return json.dumps(
+            {
+                "status": "success",
+                "enabled": bool(user.get("ai_correction_enabled")),
+                "sync": bool(user.get("ai_correction_sync")),
+                "prompt": user.get("ai_correction_prompt") or DEFAULT_CORRECTION_PROMPT,
+            },
+            ensure_ascii=False,
+        )
+
+    def _set(self, arguments: dict[str, Any]) -> str:
+        self._require_user()
+        user = self._user()
+        enabled = 1 if self._value(arguments) else 0
+        if "sync" in arguments and arguments.get("sync") is not None:
+            sync = 1 if self._coerce_bool(arguments.get("sync")) else 0
+        else:
+            sync = 1 if user.get("ai_correction_sync") else 0
+        prompt_raw = arguments.get("prompt")
+        if prompt_raw is None:
+            prompt = user.get("ai_correction_prompt") or DEFAULT_CORRECTION_PROMPT
+        else:
+            prompt = str(prompt_raw).strip()[:2000] or DEFAULT_CORRECTION_PROMPT
+        get_table("users").update(
+            {
+                "uid": self._owner_id,
+                "ai_correction_enabled": enabled,
+                "ai_correction_sync": sync,
+                "ai_correction_prompt": prompt,
+            },
+            ["uid"],
+        )
+        from devplacepy.utils import clear_user_cache
+
+        clear_user_cache(self._owner_id)
+        return json.dumps(
+            {
+                "status": "success",
+                "enabled": bool(enabled),
+                "sync": bool(sync),
+                "prompt": prompt,
+            },
+            ensure_ascii=False,
+        )

devplacepy/services/devii/ai_modifier/__init__.py

@@ -0,0 +1,5 @@
+# retoor <retoor@molodetz.nl>
+
+from .controller import AiModifierController
+
+__all__ = ["AiModifierController"]

devplacepy/services/devii/ai_modifier/controller.py

@@ -0,0 +1,94 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from devplacepy.config import DEFAULT_MODIFIER_PROMPT
+from devplacepy.database import get_table
+from devplacepy.services.devii.errors import ToolInputError
+
+logger = logging.getLogger("devii.ai_modifier")
+
+
+class AiModifierController:
+    def __init__(self, owner_kind: str, owner_id: str) -> None:
+        self._owner_kind = owner_kind
+        self._owner_id = owner_id
+
+    async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
+        if name == "ai_modifier_get":
+            return self._get()
+        if name == "ai_modifier_set":
+            return self._set(arguments)
+        raise ToolInputError(f"Unknown AI modifier tool: {name}")
+
+    def _require_user(self) -> None:
+        if self._owner_kind != "user":
+            raise ToolInputError(
+                "AI modifier is only available for signed-in users."
+            )
+
+    def _coerce_bool(self, raw: Any) -> bool:
+        if isinstance(raw, bool):
+            return raw
+        return str(raw).strip().lower() in ("true", "1", "yes", "on")
+
+    def _value(self, arguments: dict[str, Any]) -> bool:
+        return self._coerce_bool(arguments.get("enabled"))
+
+    def _user(self) -> dict[str, Any]:
+        user = get_table("users").find_one(uid=self._owner_id)
+        if not user:
+            raise ToolInputError("User not found.")
+        return user
+
+    def _get(self) -> str:
+        self._require_user()
+        user = self._user()
+        return json.dumps(
+            {
+                "status": "success",
+                "enabled": bool(user.get("ai_modifier_enabled")),
+                "sync": bool(user.get("ai_modifier_sync")),
+                "prompt": user.get("ai_modifier_prompt") or DEFAULT_MODIFIER_PROMPT,
+            },
+            ensure_ascii=False,
+        )
+
+    def _set(self, arguments: dict[str, Any]) -> str:
+        self._require_user()
+        user = self._user()
+        enabled = 1 if self._value(arguments) else 0
+        if "sync" in arguments and arguments.get("sync") is not None:
+            sync = 1 if self._coerce_bool(arguments.get("sync")) else 0
+        else:
+            sync = 1 if user.get("ai_modifier_sync") else 0
+        prompt_raw = arguments.get("prompt")
+        if prompt_raw is None:
+            prompt = user.get("ai_modifier_prompt") or DEFAULT_MODIFIER_PROMPT
+        else:
+            prompt = str(prompt_raw).strip()[:2000] or DEFAULT_MODIFIER_PROMPT
+        get_table("users").update(
+            {
+                "uid": self._owner_id,
+                "ai_modifier_enabled": enabled,
+                "ai_modifier_sync": sync,
+                "ai_modifier_prompt": prompt,
+            },
+            ["uid"],
+        )
+        from devplacepy.utils import clear_user_cache
+
+        clear_user_cache(self._owner_id)
+        return json.dumps(
+            {
+                "status": "success",
+                "enabled": bool(enabled),
+                "sync": bool(sync),
+                "prompt": prompt,
+            },
+            ensure_ascii=False,
+        )

devplacepy/services/devii/registry.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from .actions.ai_correction_actions import AI_CORRECTION_ACTIONS
+from .actions.ai_modifier_actions import AI_MODIFIER_ACTIONS
 from .actions.avatar_actions import AVATAR_ACTIONS
 from .actions.behavior_actions import BEHAVIOR_ACTIONS
 from .actions.catalog import ACTIONS
@@ -34,6 +36,8 @@ CATALOG = Catalog(
     + CUSTOMIZATION_ACTIONS
     + BEHAVIOR_ACTIONS
     + NOTIFICATION_ACTIONS
+    + AI_CORRECTION_ACTIONS
+    + AI_MODIFIER_ACTIONS
     + VIRTUAL_TOOL_ACTIONS
 )
 

devplacepy/services/devii/session.py

@@ -401,6 +401,10 @@ class DeviiSession:
         finally:
             if not cancelled and epoch == self._turn_epoch:
                 self._record_turn(turn_id, started_at, text, reply, error, before)
+                if self.owner_kind == "user" and self.channel != "docs" and not error:
+                    from devplacepy.utils import track_action
+
+                    track_action(self.owner_id, "devii")
 
     def _builtin_tools(self) -> list[dict[str, Any]]:
         schemas = CATALOG.tool_schemas_for(self.client.authenticated, self.is_admin)

devplacepy/services/jobs/backup_worker.py

@@ -0,0 +1,83 @@
+# retoor <retoor@molodetz.nl>
+
+import hashlib
+import json
+import sys
+import tarfile
+from pathlib import Path
+
+CHUNK_SIZE = 1024 * 1024
+
+
+def _reset(info: tarfile.TarInfo) -> tarfile.TarInfo:
+    info.uid = 0
+    info.gid = 0
+    info.uname = ""
+    info.gname = ""
+    return info
+
+
+def _add_file(archive: tarfile.TarFile, full: Path, arcname: str) -> int:
+    size = full.stat().st_size
+    info = _reset(archive.gettarinfo(str(full), arcname=arcname))
+    with full.open("rb") as handle:
+        archive.addfile(info, handle)
+    return size
+
+
+def _build(spec: dict, output_path: str) -> dict:
+    bytes_in = 0
+    file_count = 0
+    dir_count = 0
+    with tarfile.open(output_path, "w:gz") as archive:
+        for source in spec.get("sources", []):
+            root = source["root"]
+            base = Path(source["path"])
+            if not base.exists():
+                continue
+            if base.is_file():
+                bytes_in += _add_file(archive, base, f"{root}/{base.name}")
+                file_count += 1
+                continue
+            for entry in sorted(base.rglob("*"), key=lambda p: str(p)):
+                arcname = f"{root}/{entry.relative_to(base)}"
+                if entry.is_symlink():
+                    continue
+                if entry.is_dir():
+                    info = _reset(archive.gettarinfo(str(entry), arcname=arcname))
+                    archive.addfile(info)
+                    dir_count += 1
+                elif entry.is_file():
+                    try:
+                        bytes_in += _add_file(archive, entry, arcname)
+                        file_count += 1
+                    except OSError:
+                        continue
+
+    digest = hashlib.sha256()
+    with open(output_path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(CHUNK_SIZE), b""):
+            digest.update(chunk)
+
+    return {
+        "bytes_in": bytes_in,
+        "bytes_out": Path(output_path).stat().st_size,
+        "file_count": file_count,
+        "dir_count": dir_count,
+        "sha256": digest.hexdigest(),
+    }
+
+
+def main(argv: list) -> int:
+    if len(argv) != 3:
+        sys.stderr.write("usage: backup_worker <spec_json> <output_tar_gz>\n")
+        return 2
+    spec_path, output_path = argv[1], argv[2]
+    spec = json.loads(Path(spec_path).read_text())
+    stats = _build(spec, output_path)
+    sys.stdout.write(json.dumps(stats))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))

devplacepy/services/live_view_relay.py

@@ -0,0 +1,164 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+import logging
+import re
+import time
+from typing import Optional
+
+from devplacepy.services.base import BaseService
+from devplacepy.services.manager import service_manager
+from devplacepy.services.pubsub import publish as pubsub_publish
+from devplacepy.services.pubsub.hub import pubsub
+
+logger = logging.getLogger(__name__)
+
+_SEGMENT = r"[A-Za-z0-9_-]+"
+LOG_TAIL = 400
+
+
+async def _container_list(_match: re.Match) -> dict:
+    from devplacepy.routers.admin.containers import _decorate
+    from devplacepy.services.containers import store
+
+    return {"instances": _decorate(store.all_instances())}
+
+
+async def _project_containers(match: re.Match) -> Optional[dict]:
+    from devplacepy.database import get_table, resolve_by_slug
+    from devplacepy.services.containers import store
+
+    project = resolve_by_slug(get_table("projects"), match.group("slug"))
+    if not project:
+        return None
+    return {"instances": store.list_instances(project["uid"])}
+
+
+async def _container_detail(match: re.Match) -> Optional[dict]:
+    from devplacepy.services.containers import api, store
+
+    uid = match.group("uid")
+    inst = store.get_instance(uid)
+    if not inst:
+        return None
+    return {
+        "instance": inst,
+        "events": store.list_events(uid),
+        "schedules": store.list_schedules(uid),
+        "stats": api.instance_stats(uid),
+        "runtime": api.instance_runtime(inst),
+    }
+
+
+async def _container_logs(match: re.Match) -> Optional[dict]:
+    from devplacepy.services.containers import store
+    from devplacepy.services.containers.runtime import get_backend
+
+    uid = match.group("uid")
+    inst = store.get_instance(uid)
+    if not inst:
+        return None
+    if not inst.get("container_id"):
+        return {"logs": ""}
+    lines: list = []
+
+    async def collect(line: str) -> None:
+        lines.append(line)
+
+    await get_backend().logs(
+        inst["container_id"], follow=False, tail=LOG_TAIL, on_log=collect
+    )
+    return {"logs": "\n".join(lines)}
+
+
+async def _bots(_match: re.Match) -> dict:
+    from devplacepy.routers.admin.bots import _frames_payload
+
+    return _frames_payload()
+
+
+async def _services(_match: re.Match) -> dict:
+    return {"services": service_manager.describe_all()}
+
+
+async def _service_detail(match: re.Match) -> Optional[dict]:
+    svc = service_manager.get_service(match.group("name"))
+    if svc is None:
+        return None
+    return {"service": svc.describe()}
+
+
+async def _ai_usage(match: re.Match) -> dict:
+    from devplacepy.services.openai_gateway.analytics import build_analytics
+    from devplacepy.services.openai_gateway.usage import pricing_from_cfg
+
+    hours = int(match.group("hours"))
+    svc = service_manager.get_service("openai")
+    pricing = pricing_from_cfg(svc.get_config()) if svc is not None else None
+    return build_analytics(hours, top_n=10, pricing=pricing)
+
+
+VIEWS = [
+    (re.compile(r"^container\.list$"), _container_list, 4.0),
+    (re.compile(rf"^project\.(?P<slug>{_SEGMENT})\.containers$"), _project_containers, 3.0),
+    (re.compile(rf"^container\.(?P<uid>{_SEGMENT})\.detail$"), _container_detail, 4.0),
+    (re.compile(rf"^container\.(?P<uid>{_SEGMENT})\.logs$"), _container_logs, 3.0),
+    (re.compile(r"^fleet\.bots$"), _bots, 2.0),
+    (re.compile(r"^admin\.services$"), _services, 5.0),
+    (re.compile(rf"^admin\.services\.(?P<name>{_SEGMENT})$"), _service_detail, 5.0),
+    (re.compile(r"^admin\.ai-usage\.(?P<hours>\d+)$"), _ai_usage, 15.0),
+]
+
+
+class LiveViewRelayService(BaseService):
+    title = "Live view relay"
+    description = (
+        "Pushes admin live-view snapshots (container list and instances, bot fleet, "
+        "background services, AI usage) onto the pub/sub bus, computed only for topics "
+        "that currently have subscribers. Runs on the service lock owner where pub/sub "
+        "subscribers converge, replacing per-client HTTP polling with server push."
+    )
+    default_enabled = True
+
+    def __init__(self):
+        super().__init__(name="live_view_relay", interval_seconds=1)
+        self._last: dict[str, float] = {}
+
+    def _handler_for(self, topic: str):
+        for pattern, compute, interval in VIEWS:
+            match = pattern.match(topic)
+            if match is not None:
+                return compute, interval, match
+        return None
+
+    async def run_once(self) -> None:
+        now = time.monotonic()
+        active: set[str] = set()
+        published = 0
+        for entry in pubsub.topics():
+            topic = entry["topic"]
+            if not entry["subscribers"] or "*" in topic:
+                continue
+            handler = self._handler_for(topic)
+            if handler is None:
+                continue
+            active.add(topic)
+            compute, interval, match = handler
+            if now - self._last.get(topic, 0.0) < interval:
+                continue
+            self._last[topic] = now
+            try:
+                payload = await compute(match)
+            except Exception:
+                logger.exception("live view relay failed for %s", topic)
+                continue
+            if payload is None:
+                continue
+            published += await pubsub_publish(topic, payload)
+        self._last = {topic: ts for topic, ts in self._last.items() if topic in active}
+        if published:
+            self.log(f"pushed {published} live-view frame(s)")
+
+    def collect_metrics(self) -> dict:
+        return {"tracked_topics": len(self._last)}

devplacepy/services/messaging/persist.py

@@ -12,8 +12,11 @@ from devplacepy.utils import (
     create_notification,
     generate_uid,
     time_ago,
+    track_action,
 )
 from devplacepy.services.audit import record as audit
+from devplacepy.services.correction import schedule_correction
+from devplacepy.services.ai_modifier import schedule_modification
 
 logger = logging.getLogger("messaging.persist")
 
@@ -86,6 +89,8 @@ def persist_message(
     )
 
     link_attachments(attachment_uids, "message", msg_uid)
+    schedule_correction(sender, "messages", msg_uid, request)
+    schedule_modification(sender, "messages", msg_uid, request)
 
     if sender_uid != receiver_uid:
         create_notification(
@@ -100,6 +105,7 @@ def persist_message(
     create_mention_notifications(
         content, sender_uid, f"/messages?with_uid={receiver_uid}"
     )
+    track_action(sender_uid, "message")
 
     logger.info(
         "Message %s sent from %s to %s via %s",

devplacepy/services/notification_relay.py

@@ -0,0 +1,92 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+from devplacepy.database import db
+from devplacepy.services.base import BaseService
+from devplacepy.services.pubsub import publish as pubsub_publish
+
+BATCH_LIMIT = 500
+
+
+class NotificationRelayService(BaseService):
+    title = "Notification relay"
+    description = (
+        "Bridges newly persisted in-app notifications onto the pub/sub bus so the "
+        "recipient's browser raises a live toast. Polls the notifications table on the "
+        "service lock owner, where every pub/sub subscriber converges. A toast fires "
+        "exactly when the in-app channel is enabled, since the row only exists then."
+    )
+    default_enabled = True
+
+    def __init__(self):
+        super().__init__(name="notification_relay", interval_seconds=1)
+        self._watermark = 0
+        self._primed = False
+
+    def _max_id(self) -> int:
+        if "notifications" not in db.tables:
+            return 0
+        rows = list(db.query("SELECT MAX(id) AS max_id FROM notifications"))
+        value = rows[0]["max_id"] if rows else None
+        return int(value or 0)
+
+    def _counts(self, user_uid: str) -> dict:
+        notifications = 0
+        messages = 0
+        if "notifications" in db.tables:
+            rows = list(
+                db.query(
+                    "SELECT COUNT(*) AS c FROM notifications "
+                    "WHERE user_uid = :u AND read = 0",
+                    u=user_uid,
+                )
+            )
+            notifications = int(rows[0]["c"]) if rows else 0
+        if "messages" in db.tables:
+            rows = list(
+                db.query(
+                    "SELECT COUNT(*) AS c FROM messages "
+                    "WHERE receiver_uid = :u AND read = 0",
+                    u=user_uid,
+                )
+            )
+            messages = int(rows[0]["c"]) if rows else 0
+        return {"notifications": notifications, "messages": messages}
+
+    async def run_once(self) -> None:
+        if "notifications" not in db.tables:
+            return
+        if not self._primed:
+            self._watermark = self._max_id()
+            self._primed = True
+            return
+        rows = list(
+            db.query(
+                "SELECT id, uid, user_uid, type, message, target_url "
+                "FROM notifications WHERE id > :wm ORDER BY id ASC LIMIT :lim",
+                wm=self._watermark,
+                lim=BATCH_LIMIT,
+            )
+        )
+        if not rows:
+            return
+        delivered = 0
+        for row in rows:
+            delivered += await pubsub_publish(
+                f"user.{row['user_uid']}.notifications",
+                {
+                    "uid": row["uid"],
+                    "type": row["type"],
+                    "message": row["message"],
+                    "target_url": row["target_url"],
+                },
+            )
+        for user_uid in {row["user_uid"] for row in rows}:
+            await pubsub_publish(f"user.{user_uid}.counts", self._counts(user_uid))
+        self._watermark = max(row["id"] for row in rows)
+        if delivered:
+            self.log(f"relayed {len(rows)} notification(s), {delivered} live toast(s)")
+
+    def collect_metrics(self) -> dict:
+        return {"watermark": self._watermark, "primed": self._primed}

devplacepy/services/openai_gateway/gateway.py

@@ -20,6 +20,7 @@ from devplacepy.services.openai_gateway.usage import (
     extract_params,
     parse_context_map,
     pricing_from_cfg,
+    usage_response_headers,
 )
 from devplacepy.services.openai_gateway.vision import VisionAugmenter, VisionCache
 
@@ -294,10 +295,12 @@ class GatewayRuntime:
             base["success"] = success
             base["error_category"] = category
             base["usage"] = usage
-            self._ledger.record(base, pricing, context_map)
+            return usage_response_headers(
+                self._ledger.record(base, pricing, context_map)
+            )
 
         if timing["circuit_open"]:
-            finalize(503, False, "circuit_open")
+            resp_headers = finalize(503, False, "circuit_open")
             return JSONResponse(
                 status_code=503,
                 content={
@@ -306,9 +309,10 @@ class GatewayRuntime:
                         "type": "circuit_open",
                     }
                 },
+                headers=resp_headers,
             )
         if exc is not None:
-            finalize(502, False, classify_error(0, exc))
+            resp_headers = finalize(502, False, classify_error(0, exc))
             return JSONResponse(
                 status_code=502,
                 content={
@@ -317,9 +321,10 @@ class GatewayRuntime:
                         "type": "upstream_error",
                     }
                 },
+                headers=resp_headers,
             )
         if resp.status_code != 200:
-            finalize(
+            resp_headers = finalize(
                 resp.status_code,
                 False,
                 classify_error(resp.status_code, None, resp.text),
@@ -328,12 +333,13 @@ class GatewayRuntime:
             return JSONResponse(
                 status_code=resp.status_code,
                 content={"error": {"message": resp.text, "type": "upstream_error"}},
+                headers=resp_headers,
             )
         try:
             data = resp.json()
         except ValueError:
             self.errors += 1
-            finalize(502, False, "gateway")
+            resp_headers = finalize(502, False, "gateway")
             log("chat upstream returned 200 but body was not valid JSON")
             return JSONResponse(
                 status_code=502,
@@ -343,14 +349,17 @@ class GatewayRuntime:
                         "type": "upstream_error",
                     }
                 },
+                headers=resp_headers,
             )
-        finalize(200, True, None, data.get("usage"))
+        resp_headers = finalize(200, True, None, data.get("usage"))
         log(f"chat POST -> 200 ({timing['upstream_latency_ms']:.0f}ms)")
         if stream:
             return StreamingResponse(
-                _fake_stream(data, model), media_type="text/event-stream"
+                _fake_stream(data, model),
+                media_type="text/event-stream",
+                headers=resp_headers,
             )
-        return JSONResponse(content=data)
+        return JSONResponse(content=data, headers=resp_headers)
 
     async def handle_embeddings(
         self, body: dict, cfg: dict, owner: tuple, user_agent: str, log=None
@@ -447,10 +456,12 @@ class GatewayRuntime:
             base["success"] = success
             base["error_category"] = category
             base["usage"] = usage
-            self._ledger.record(base, pricing, context_map)
+            return usage_response_headers(
+                self._ledger.record(base, pricing, context_map)
+            )
 
         if timing["circuit_open"]:
-            finalize(503, False, "circuit_open")
+            resp_headers = finalize(503, False, "circuit_open")
             return JSONResponse(
                 status_code=503,
                 content={
@@ -459,9 +470,10 @@ class GatewayRuntime:
                         "type": "circuit_open",
                     }
                 },
+                headers=resp_headers,
             )
         if exc is not None:
-            finalize(502, False, classify_error(0, exc))
+            resp_headers = finalize(502, False, classify_error(0, exc))
             return JSONResponse(
                 status_code=502,
                 content={
@@ -470,9 +482,10 @@ class GatewayRuntime:
                         "type": "upstream_error",
                     }
                 },
+                headers=resp_headers,
             )
         if resp.status_code != 200:
-            finalize(
+            resp_headers = finalize(
                 resp.status_code,
                 False,
                 classify_error(resp.status_code, None, resp.text),
@@ -481,12 +494,13 @@ class GatewayRuntime:
             return JSONResponse(
                 status_code=resp.status_code,
                 content={"error": {"message": resp.text, "type": "upstream_error"}},
+                headers=resp_headers,
             )
         try:
             data = resp.json()
         except ValueError:
             self.errors += 1
-            finalize(502, False, "gateway")
+            resp_headers = finalize(502, False, "gateway")
             log("embed upstream returned 200 but body was not valid JSON")
             return JSONResponse(
                 status_code=502,
@@ -496,11 +510,12 @@ class GatewayRuntime:
                         "type": "upstream_error",
                     }
                 },
+                headers=resp_headers,
             )
         self.embed_calls += 1
-        finalize(200, True, None, data.get("usage"))
+        resp_headers = finalize(200, True, None, data.get("usage"))
         log(f"embed POST -> 200 ({timing['upstream_latency_ms']:.0f}ms)")
-        return JSONResponse(content=data)
+        return JSONResponse(content=data, headers=resp_headers)
 
     async def handle_passthrough(
         self,
@@ -559,10 +574,12 @@ class GatewayRuntime:
             base["success"] = success
             base["error_category"] = category
             base["usage"] = usage
-            self._ledger.record(base, pricing, context_map)
+            return usage_response_headers(
+                self._ledger.record(base, pricing, context_map)
+            )
 
         if timing["circuit_open"]:
-            finalize(503, False, "circuit_open")
+            resp_headers = finalize(503, False, "circuit_open")
             return JSONResponse(
                 status_code=503,
                 content={
@@ -571,9 +588,10 @@ class GatewayRuntime:
                         "type": "circuit_open",
                     }
                 },
+                headers=resp_headers,
             )
         if exc is not None:
-            finalize(502, False, classify_error(0, exc))
+            resp_headers = finalize(502, False, classify_error(0, exc))
             return JSONResponse(
                 status_code=502,
                 content={
@@ -582,6 +600,7 @@ class GatewayRuntime:
                         "type": "upstream_error",
                     }
                 },
+                headers=resp_headers,
             )
         usage = None
         if resp.status_code < 400 and "application/json" in (
@@ -591,7 +610,7 @@ class GatewayRuntime:
                 usage = resp.json().get("usage")
             except ValueError:
                 usage = None
-        finalize(
+        resp_headers = finalize(
             resp.status_code,
             resp.status_code < 400,
             None
@@ -606,6 +625,7 @@ class GatewayRuntime:
             content=resp.content,
             status_code=resp.status_code,
             media_type=resp.headers.get("content-type"),
+            headers=resp_headers,
         )
 
     def metrics(self) -> dict:

devplacepy/services/openai_gateway/usage.py

@@ -219,8 +219,38 @@ def audit_actor_for(owner_kind: str, owner_id: str) -> tuple[str, Optional[str],
     return actor_kind, actor_uid, actor_role
 
 
+def usage_response_headers(row: Optional[dict]) -> dict:
+    if not row:
+        return {}
+    headers = {
+        "X-Gateway-Model": str(row.get("model") or ""),
+        "X-Gateway-Backend": str(row.get("backend") or ""),
+        "X-Gateway-Prompt-Tokens": str(int(row.get("prompt_tokens") or 0)),
+        "X-Gateway-Completion-Tokens": str(int(row.get("completion_tokens") or 0)),
+        "X-Gateway-Total-Tokens": str(int(row.get("total_tokens") or 0)),
+        "X-Gateway-Cache-Hit-Tokens": str(int(row.get("cache_hit_tokens") or 0)),
+        "X-Gateway-Cache-Miss-Tokens": str(int(row.get("cache_miss_tokens") or 0)),
+        "X-Gateway-Reasoning-Tokens": str(int(row.get("reasoning_tokens") or 0)),
+        "X-Gateway-Cost-USD": f"{float(row.get('cost_usd') or 0.0):.8f}",
+        "X-Gateway-Input-Cost-USD": f"{float(row.get('input_cost_usd') or 0.0):.8f}",
+        "X-Gateway-Output-Cost-USD": f"{float(row.get('output_cost_usd') or 0.0):.8f}",
+        "X-Gateway-Cost-Native": "1" if row.get("native_cost") else "0",
+        "X-Gateway-Tokens-Per-Second": str(row.get("tokens_per_second") or 0),
+        "X-Gateway-Upstream-Latency-Ms": str(row.get("upstream_latency_ms") or 0),
+        "X-Gateway-Total-Latency-Ms": str(row.get("total_latency_ms") or 0),
+        "X-Gateway-Gateway-Overhead-Ms": str(row.get("gateway_overhead_ms") or 0),
+        "X-Gateway-Queue-Wait-Ms": str(row.get("queue_wait_ms") or 0),
+        "X-Gateway-Connect-Ms": str(row.get("connect_ms") or 0),
+    }
+    if row.get("context_window"):
+        headers["X-Gateway-Context-Window"] = str(int(row["context_window"]))
+    if row.get("context_utilization") is not None:
+        headers["X-Gateway-Context-Utilization"] = str(row["context_utilization"])
+    return headers
+
+
 class GatewayUsageLedger:
-    def record(self, raw: dict, pricing: Pricing, context_map: dict) -> None:
+    def record(self, raw: dict, pricing: Pricing, context_map: dict) -> Optional[dict]:
         try:
             usage = raw.get("usage") or {}
             norm = normalize_usage(usage)
@@ -278,8 +308,10 @@ class GatewayUsageLedger:
             }
             get_table(GATEWAY_LEDGER).insert(row)
             self._audit(raw, norm, cost_usd)
+            return row
         except Exception as exc:
             logger.warning("gateway usage record failed: %s", exc)
+            return None
 
     def _audit(self, raw: dict, norm: dict, cost_usd: float) -> None:
         from devplacepy.services.audit import record as audit