Experimental/brain-of-reese
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sources): real-time file progress for sync and upload — background upload with success toast

Browse Source
This commit is contained in:
ducoterra
2026-09-01 23:51:43 -04:00
parent cddc84c7db
commit 4677d86f49
103 changed files with 5914 additions and 456 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/git_sources.py
+290 -137
View File
@@ -17,13 +17,19 @@ Routes: ``GET`` (DB rows oldest-first, or the env list with
``path``, git rows — and env rows — report ``path: null``), ``POST``
(201, validated create; ``kind`` selects the validation: git → exactly
the phase-35 URL contract, local → an existing absolute directory, else
422 naming the path), ``POST /upload`` (phase 49 — admin archive upload:
``.tar``/``.tar.gz``/``.tgz``/``.zip`` streamed with a size cap, safely
unpacked, atomically swapped in over an existing folder of the same
name, row upserted, then the synchronous single-source scan — see
:func:`upload_archive`), ``DELETE /{source_id}`` (204). The whole
router sits behind :func:`app.core.auth.require_admin` — anonymous
callers get 403 on every route.
422 naming the path), ``POST /upload`` (phase 49, backgrounded in phase
64 task 03 — admin archive upload: the ``.tar``/``.tar.gz``/``.tgz``/
``.zip`` name/format gate + the 1 MiB-chunk receive with the
``upload_max_mb`` cap run **inline** and answered 202 the moment the
archive is safely on disk; unpack → swap → row upsert → model check →
single-source scan → change-gated overview then run in a **background
task** — see :func:`upload_archive` and :func:`_run_upload`),
``GET /upload/status`` (the phase-32 ``SyncStatus``-shaped in-memory
state of that run — incl. the phase-64 ``current_file`` /
``files_done`` / ``files_total`` progress fields; navigating away from
the page mid-scan no longer aborts anything), ``DELETE /{source_id}``
(204). The whole router sits behind :func:`app.core.auth.require_admin`
— anonymous callers get 403 on every route.
No credential-echo path: git URLs may embed ``user:pass@`` (phase 32's
masking discipline), so every git 409/422 detail is a fixed generic
@@ -33,21 +39,26 @@ owner sees exactly which directory failed.
Scope boundary (phase locked decisions): the CRUD routes do NOT
clone, import, or prune anything — the existing Sync button performs
that (a removal prunes on the next sync, ``prune=True``). The phase-49
upload route is the exception: it unpacks the archive and then scans
the single source synchronously in the request (``import_sources``
with ``prune=True`` + the change-gated overview refresh) and answers
with the sync-style counts.
that (a removal prunes on the next sync, ``prune=True``). The upload
route is the exception (phase 64, task 03): after the 202 receive
answer, its background task unpacks the archive, swaps it in, upserts
the row, probes the models, scans the single source
(``import_sources`` with ``prune=True`` + the change-gated overview
refresh), and lands the sync-style counts (the ``UploadOut`` fields)
in the status ``detail``.
"""
from __future__ import annotations
import asyncio
import logging
import re
import shutil
import time
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Literal, cast
from typing import Any, Literal, cast
from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile
from sqlalchemy import select
@@ -57,7 +68,7 @@ from sqlalchemy.orm import Session
from app.api.sync import _sanitize_error
from app.config import get_settings
from app.core.auth import require_admin
from app.db import get_db
from app.db import SessionLocal, get_db
from app.models import GitSource
from app.rag.archive_upload import (
ARCHIVE_SUFFIXES,
@@ -67,9 +78,15 @@ from app.rag.archive_upload import (
unpack_archive,
)
from app.rag.importer import import_sources
from app.rag.llm import LLMClient, ModelUnavailableError, check_models
from app.rag.llm import LLMClient, check_models
from app.rag.overview import regenerate_overview
from app.schemas import GitSourceIn, GitSourceList, GitSourceOut, GitSourceRow, UploadOut
from app.schemas import (
GitSourceIn,
GitSourceList,
GitSourceOut,
GitSourceRow,
UploadAccepted,
)
logger = logging.getLogger("app.api.git_sources")
@@ -79,18 +96,56 @@ router = APIRouter(
dependencies=[Depends(require_admin)], # phase 16 pattern: admin-only surface
)
#: One upload at a time (phase 49, task 02 — the phase-32 ``_task``
#: spirit): the flag is held from the name gate through the scan
#: response. A plain bool, not an ``asyncio.Lock`` — it is checked and
#: set with no await in between (a single app loop can never enter
#: One upload at a time (phase 49, backgrounded in phase 64 task 03):
#: the flag is checked and set with **no await in between, BEFORE the
#: streaming receive** — the handler now awaits (the 1 MiB-chunk stream)
#: long before the background task exists, so a task-done check alone
#: would let a concurrent POST slip through the receive window and start
#: a second run. A plain bool, not an ``asyncio.Lock``: it is checked
#: and set with no await in between (a single app loop can never enter
#: twice), and it stays correct across requests that run on separate
#: event loops (the TestClient convention).
#: event loops (the TestClient convention). Held until
#: :func:`_run_upload`'s ``finally`` (end of the background run) — or
#: cleared on the inline exception path where the task was never
#: created.
_upload_in_progress = False
#: Streaming read size while counting compressed upload bytes (1 MiB
#: chunks — the task-02 cap check granularity).
_STREAM_CHUNK = 1 << 20
@dataclass
class UploadStatus:
"""In-memory state of the (at most one) in-flight upload run.
Mirrors :class:`app.api.sync.SyncStatus` (the phase-32 pattern,
phase 64 task 03): ``state`` is the same four-state machine
(``idle`` / ``running`` / ``success`` / ``failed``); terminal states
carry the run's ``detail`` (success — the ``UploadOut`` fields) or
``error`` (failure — sanitized) so the UI can render the last result
after a page reload (the re-attach behavior, task 05).
Phase 64 (task 03) progress fields: ``current_file`` is the
``source/relative/path`` the scan is processing right now (null
outside the import phase — unpack/swap/row/model-check first — and
in terminal states); ``files_done`` / ``files_total`` carry the
hook's done/total position and survive a terminal state (the run's
last position is useful context next to the error).
"""
state: Literal["idle", "running", "success", "failed"] = "idle"
started_at: datetime | None = None
finished_at: datetime | None = None
current_file: str | None = None
files_done: int = 0
files_total: int = 0
detail: dict[str, Any] = field(default_factory=dict)
error: str | None = None
_upload_status = UploadStatus()
#: Accepted git URL shapes — the trimmed URL must *start* with one of them.
#: Covers the phase-28 real URLs (HTTPS + ``git@`` SSH); scp-style
#: ``host:repo`` is deliberately rejected (422). ASSUMPTION (task 02): the
@@ -227,46 +282,55 @@ def _create_local_row(payload: GitSourceIn, db: Session) -> GitSource:
)
@router.post("/upload", response_model=UploadOut)
@router.post("/upload", response_model=UploadAccepted, status_code=202)
async def upload_archive(
file: UploadFile = File(...), # noqa: B008
db: Session = Depends(get_db), # noqa: B008
) -> UploadOut:
"""Upload a source archive and scan it (phase 49, task 02).
) -> UploadAccepted:
"""Receive a source archive; scan it in the background (phase 49,
backgrounded in phase 64 task 03 — owner-locked A1/A2).
The scan is **synchronous in the request** (phase locked decisions,
owner-confirmed) and mirrors the admin sync pipeline:
The **inline (request) work is exactly three gates** — steps 1–3 —
everything else runs in a background task behind
``GET /upload/status`` (the phase-32 ``SyncStatus`` pattern), so
navigating away mid-scan no longer aborts anything:
1. name/format gate — only ``.tar``/``.tar.gz``/``.tgz``/``.zip``
(422 naming the accepted set) and a safe source name
(``archive_source_name`` — its message is the 422 detail);
2. one at a time — 409 ``an upload is already in progress``;
2. one at a time — 409 ``an upload is already in progress`` while
the flag is held (checked and set with no await in between,
BEFORE the receive — see ``_upload_in_progress``);
3. stream the upload in 1 MiB chunks into a dotfile temp with the
``upload_max_mb`` cap — 413 naming the cap, temp deleted;
``upload_max_mb`` cap — 413 naming the cap, temp deleted.
Then the archive is **safely on disk** — 202 + ``UploadAccepted``
(the "successfully uploaded" moment the UI toasts on, A2) and
:func:`_run_upload` runs the rest on the app's event loop:
4. unpack to a temp sibling (traversal/symlink/device/corrupt/
over-cap all 422 with the task-01 user-safe message, temps
deleted); a zero-entry archive is 422 ``the archive contains no
files`` — an archive with only non-A9 files is a VALID
replacement (the scan indexes nothing, prune removes the
over-cap → ``failed`` with the task-01 user-safe message, temps
deleted); a zero-entry archive is ``failed`` ``the archive
contains no files`` — an archive with only non-A9 files is a
VALID replacement (the scan indexes nothing, prune removes the
source's docs);
5. atomic swap-in — a same-name re-upload replaces the previous
folder in place; a failure leaves the previous folder/row/KB
untouched (422);
untouched;
6. upsert the row by ``path`` (``kind='local'``; an existing row is
left as-is — ``added_at`` preserved — and the unique index is
the 409 backstop);
7. fail-fast ``check_models`` — 503 with the sanitized
model-unavailable message; the folder/row are already committed,
so the next sync/re-upload retries idempotently;
8. ``import_sources([folder], llm, prune=True)`` + the change-gated
``regenerate_overview``;
9. one INFO log line (PLAN §9 / AGENTS.md rule 10);
10. 200 with the sync-detail count keys (``UploadOut``).
the backstop: a concurrent insert lands ``failed`` with
``a local source with this path already exists: <path>``);
7. fail-fast ``check_models`` — ``ModelUnavailableError`` →
``failed`` with the sanitized message (the phase-49 503 becomes
a status state, A5); the folder/row are already committed, so
the next sync/re-upload retries idempotently;
8. ``import_sources([folder], llm, prune=True, progress=<hook>)``
+ the change-gated ``regenerate_overview`` — the hook feeds the
status ``current_file`` / ``files_done`` / ``files_total``;
9. one INFO log line (PLAN §9 / AGENTS.md rule 10 — ``total_ms`` is
the background run's duration);
10. ``success`` — ``detail`` = the ``UploadOut`` fields.
"""
started = time.monotonic()
settings = get_settings()
total = 0
# 1. Name/format gate — the accepted formats first (the 422 names
# them), then the task-01 safe-name derivation. A BARE suffix
# ("tar.gz") is an accepted format with no usable stem — it
@@ -288,12 +352,15 @@ async def upload_archive(
raise HTTPException(status_code=422, detail=str(e)) from None
# 2. One at a time — the flag is checked and set with no await
# between, so the single app loop can never enter twice.
# between, BEFORE the streaming receive: the background task
# does not exist yet, so the flag (not a task-done check) is the
# gate (see ``_upload_in_progress``).
global _upload_in_progress
if _upload_in_progress:
raise HTTPException(status_code=409, detail="an upload is already in progress")
_upload_in_progress = True
settings = get_settings()
upload_root = Path(settings.upload_dir).expanduser()
upload_root.mkdir(parents=True, exist_ok=True)
max_bytes = settings.upload_max_mb * 1024 * 1024
@@ -302,77 +369,192 @@ async def upload_archive(
try:
# 3. Stream with the compressed-size cap — dotfile temps are
# hidden from the upload dir's listing.
try:
with open(temp_upload, "wb") as out:
while chunk := await file.read(_STREAM_CHUNK):
total += len(chunk)
if total > max_bytes:
raise HTTPException(
status_code=413,
detail=f"the upload exceeds the {settings.upload_max_mb} MiB limit",
)
out.write(chunk)
except HTTPException:
temp_upload.unlink(missing_ok=True)
raise
# 4. Unpack to a temp sibling; the compressed bytes are no
# longer needed once unpacked (phase locked decision: only
# the unpacked content is kept).
try:
unpack_archive(temp_upload, temp_unpack, max_bytes)
except ArchiveUploadError as e:
temp_upload.unlink(missing_ok=True)
shutil.rmtree(temp_unpack, ignore_errors=True)
raise HTTPException(status_code=422, detail=str(e)) from None
total = 0
with open(temp_upload, "wb") as out:
while chunk := await file.read(_STREAM_CHUNK):
total += len(chunk)
if total > max_bytes:
raise HTTPException(
status_code=413,
detail=f"the upload exceeds the {settings.upload_max_mb} MiB limit",
)
out.write(chunk)
# The archive is safely on disk — 202 is the "successfully
# uploaded" moment (A2). Steps 4–10 run in the background:
asyncio.create_task(
_run_upload(name, filename, total, upload_root, temp_upload, temp_unpack)
)
except BaseException:
# The background task was never created (cap 413, a broken
# pipe, cancellation, or create_task itself): release the flag
# so the next upload is not refused, and make sure no temp
# survives the failed receive.
temp_upload.unlink(missing_ok=True)
_upload_in_progress = False
raise
return UploadAccepted(name=name)
@router.get("/upload/status")
def upload_status() -> dict[str, Any]:
"""Current upload state (the UI polls this — the phase-32
``GET /api/sync/status`` contract, identical key set).
``started_at`` / ``finished_at`` are ISO-8601 strings or null.
``current_file`` (phase 64) is the ``source/relative/path`` the
scan is processing right now — null during the unpack/swap/row/
model phases and in terminal states; ``files_done`` / ``files_total``
carry the hook's position (0/0 idle). The router dependency makes
it admin-only like every other route here.
"""
return {
"state": _upload_status.state,
"started_at": (
_upload_status.started_at.isoformat() if _upload_status.started_at else None
),
"finished_at": (
_upload_status.finished_at.isoformat() if _upload_status.finished_at else None
),
"detail": _upload_status.detail,
"error": _upload_status.error,
"current_file": _upload_status.current_file,
"files_done": _upload_status.files_done,
"files_total": _upload_status.files_total,
}
async def _run_upload(
name: str,
filename: str,
total_bytes: int,
upload_root: Path,
temp_upload: Path,
temp_unpack: Path,
) -> None:
"""The post-202 upload pipeline, one in-process background task
(the phase-32 ``_run_sync`` shape — A1).
Every failure mode (unpack, zero entries, swap, row, models,
import, anything else) lands in the ``failed`` state with a
sanitized ``error`` string — a background task must die in state,
never as an unobserved exception (A5: post-202 failures are status
states, never HTTP errors). ``CancelledError`` is deliberately *not*
caught: app shutdown cancels the task, and swallowing that would
mask a real stop. The ``finally`` cleans both temps (defensive —
each step already cleans its own) and clears ``_upload_in_progress``.
"""
global _upload_in_progress
started = time.monotonic()
_upload_status.state = "running"
_upload_status.started_at = datetime.now(UTC)
_upload_status.finished_at = None
_upload_status.current_file = None
_upload_status.files_done = 0
_upload_status.files_total = 0
_upload_status.detail = {}
_upload_status.error = None
try:
settings = get_settings()
max_bytes = settings.upload_max_mb * 1024 * 1024
# Step 4 — unpack to a temp sibling; the compressed bytes are
# no longer needed once unpacked (phase 49 locked decision:
# only the unpacked content is kept).
unpack_archive(temp_upload, temp_unpack, max_bytes)
temp_upload.unlink(missing_ok=True)
if not any(temp_unpack.iterdir()):
# Zero entries = a user error. (Only non-A9 files is NOT an
# error — it still has entries and is a valid replacement.)
shutil.rmtree(temp_unpack, ignore_errors=True)
raise HTTPException(status_code=422, detail="the archive contains no files")
# 5. Swap in — a same-name re-upload replaces the previous
# folder atomically; a failure leaves it, the row, and the
# KB untouched.
raise ArchiveUploadError("the archive contains no files")
# Step 5 — swap in — a same-name re-upload replaces the
# previous folder atomically; a failure leaves it, the row,
# and the KB untouched (the ``failed`` state carries the
# user-safe message).
final_dir = upload_root / name
try:
swap_in(temp_unpack, final_dir)
except ArchiveUploadError as e:
shutil.rmtree(temp_unpack, ignore_errors=True)
raise HTTPException(status_code=422, detail=str(e)) from None
# 6. Upsert the row by path — no duplicates: an existing row is
# left exactly as it is (``added_at`` preserved); the unique
# index is the 409 backstop for a concurrent insert the
# pre-check missed.
swap_in(temp_unpack, final_dir)
# Step 6 — upsert the row by path in a SHORT-LIVED session
# (open/close around it — the ``effective_sources`` /
# ``bump_sources_version`` pattern in ``app.api.sync``): the
# background task has no request session to leak locks from
# (the old inline ``db.close()`` discipline, now structural).
# No duplicates: an existing row is left exactly as it is
# (``added_at`` preserved); the unique index is the backstop
# for a concurrent insert the pre-check missed.
path = str(final_dir)
if db.scalar(select(GitSource).where(GitSource.path == path)) is None:
_commit_new(
GitSource(url=path, kind="local", path=path),
f"a local source with this path already exists: {path}",
db,
)
# Release the request session NOW — the handler never touches
# ``db`` again (the scan below uses its own sessions). If the
# session stayed open, its uncommitted transaction (the
# ``_commit_new`` refresh SELECT) would hold ``git_sources``
# locks for the whole scan, and any concurrent TRUNCATE of the
# KB tables (the E2E isolation fixtures) would deadlock against
# the scan's own document locks — a cycle Postgres cannot see.
# ``get_db``'s teardown close() is idempotent.
db.close()
# 7. Fail-fast models (phase 41) — 503 with the sanitized
# message; nothing else is rolled back (the folder/row are
# committed and the next sync/re-upload retries idempotently).
llm = LLMClient()
db = SessionLocal()
try:
await check_models(llm)
except ModelUnavailableError as e:
raise HTTPException(status_code=503, detail=_sanitize_error(str(e))) from None
# 8. Scan — single source, prune (dropped files leave the KB),
# then the change-gated overview refresh (phases 31/32).
summary = await import_sources([final_dir], llm, prune=True)
if db.scalar(select(GitSource).where(GitSource.path == path)) is None:
db.add(GitSource(url=path, kind="local", path=path))
try:
db.commit()
except IntegrityError:
db.rollback()
raise ValueError(
f"a local source with this path already exists: {path}"
) from None
finally:
db.close()
# Step 7 — fail-fast models (phase 41): ``ModelUnavailableError``
# lands in the ``failed`` state sanitized (the phase-49 503
# becomes a status state, A5). Nothing is rolled back — the
# folder/row are committed and the next sync/re-upload retries
# idempotently.
llm = LLMClient()
await check_models(llm)
# Step 8 — scan — single source, prune (dropped files leave
# the KB), with the phase-64 progress hook feeding the status,
# then the change-gated overview refresh (phases 31/32). The
# closure captures the module ``_upload_status`` exactly like
# the state assignments above.
def _hook(source: str, rel: str, done: int, total: int) -> None:
_upload_status.current_file = f"{source}/{rel}"
_upload_status.files_done = done
_upload_status.files_total = total
summary = await import_sources([final_dir], llm, prune=True, progress=_hook)
overview = False
if summary.added + summary.updated > 0:
overview = await regenerate_overview(llm)
# Step 9 — per-upload log line (PLAN §9 / AGENTS.md rule 10)
# — moved with the scan: ``total_ms`` is the background run's
# duration.
logger.info(
"upload: name=%s file=%s bytes_in=%d files=%d added=%d updated=%d "
"unchanged=%d pruned=%d errors=%d overview=%s total_ms=%d",
name,
filename,
total_bytes,
summary.files,
summary.added,
summary.updated,
summary.unchanged,
summary.pruned,
summary.errors,
overview,
round((time.monotonic() - started) * 1000),
)
# Step 10 — success: the ``UploadOut`` fields ride in the
# status ``detail`` (the UI renders the same result line from
# the status that the sync button renders from its own).
_upload_status.state = "success"
_upload_status.finished_at = datetime.now(UTC)
_upload_status.current_file = None # phase 64: keep the final counts
_upload_status.detail = {
"source": name,
"files": summary.files,
"added": summary.added,
"updated": summary.updated,
"unchanged": summary.unchanged,
"pruned": summary.pruned,
"errors": summary.errors,
"chunks": summary.chunks,
"overview": overview,
}
except Exception as e: # noqa: BLE001 — a background task dies in state, see above
logger.exception("upload: failed")
_upload_status.state = "failed"
_upload_status.finished_at = datetime.now(UTC)
_upload_status.error = _sanitize_error(str(e))
_upload_status.current_file = None # phase 64: keep the final counts
finally:
_upload_in_progress = False
# No temp may survive any failure path (defensive — each step
@@ -380,35 +562,6 @@ async def upload_archive(
temp_upload.unlink(missing_ok=True)
shutil.rmtree(temp_unpack, ignore_errors=True)
# 9. Per-upload log line (PLAN §9 / AGENTS.md rule 10).
logger.info(
"upload: name=%s file=%s bytes_in=%d files=%d added=%d updated=%d "
"unchanged=%d pruned=%d errors=%d overview=%s total_ms=%d",
name,
filename,
total,
summary.files,
summary.added,
summary.updated,
summary.unchanged,
summary.pruned,
summary.errors,
overview,
round((time.monotonic() - started) * 1000),
)
# 10. Respond 200 with the sync-style counts.
return UploadOut(
source=name,
files=summary.files,
added=summary.added,
updated=summary.updated,
unchanged=summary.unchanged,
pruned=summary.pruned,
errors=summary.errors,
chunks=summary.chunks,
overview=overview,
)
@router.delete("/{source_id}", status_code=204)
def delete_git_source(