Fix/web docker artifact cleanup

.github/workflows/build.yml

@@ -14,6 +14,7 @@ env:
   REGISTRY: ghcr.io
   IMAGE_API: ${{ github.repository }}/api
   IMAGE_WEB: ${{ github.repository }}/web
+  IMAGE_MONITORING: ${{ github.repository }}/monitoring
 
 jobs:
   build-and-push:
@@ -82,6 +83,31 @@ jobs:
           cache-from: type=gha,scope=web
           cache-to: type=gha,mode=max,scope=web
 
+      # ── Monitoring ───────────────────────────────────────────────────────────
+
+      - name: Extract metadata for Monitoring image
+        id: meta-monitoring
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_MONITORING }}
+          tags: |
+            type=ref,event=branch
+            type=sha,prefix={{branch}}-,format=short
+
+      - name: Build and push Monitoring image
+        uses: docker/build-push-action@v7
+        with:
+          context: .
+          file: apps/monitoring/Dockerfile
+          push: true
+          tags: ${{ steps.meta-monitoring.outputs.tags }}
+          labels: ${{ steps.meta-monitoring.outputs.labels }}
+          build-args: |
+            NEXT_PUBLIC_API_BASE_URL=
+            MONITORING_BASE_PATH=/ops
+          cache-from: type=gha,scope=monitoring
+          cache-to: type=gha,mode=max,scope=monitoring
+
   # ── Deploy（按需取消注释）──────────────────────────────────────────────────
   #
   # deploy:

apps/api/app/services/source_service.py

@@ -540,9 +540,12 @@ async def rechunk_source(
 
         source.status = "pending"
         await self.db.flush()
+        trace_id, run = await self._create_source_ingest_run(source, origin="rechunk")
 
         self._enqueue_ingestion(
             source.id,
+            trace_id=trace_id,
+            run_id=run.id,
             job_kind="rechunk",
             job_label=f"重建索引：{source.title or source.id}",
             chunk_size=size,
@@ -628,16 +631,13 @@ async def upload_global_source(self, filename: str | None, content: bytes) -> So
         nb = await self._get_or_create_global_notebook()
 
         ext = os.path.splitext(filename or "")[1].lower()
-        type_map = {".pdf": "pdf", ".md": "md", ".txt": "md"}
+        type_map = {".pdf": "pdf", ".md": "md", ".txt": "md", ".docx": "doc"}
         source_type = type_map.get(ext, "md")
 
         file_id = str(uuid.uuid4())
         storage_key = f"global/{nb.id}/{file_id}{ext}"
         content_type = self._guess_content_type(filename or "")
 
-        from app.providers.storage import storage as get_storage
-        await get_storage().upload(storage_key, content, content_type)
-
         source = Source(
             notebook_id=nb.id,
             title=filename,
@@ -649,9 +649,30 @@ async def upload_global_source(self, filename: str | None, content: bytes) -> So
         self.db.add(source)
         await self.db.flush()
         await self.db.refresh(source)
+        trace_id, run = await self._create_source_ingest_run(source, origin="global_upload")
 
+        from app.providers.storage import storage as get_storage
+        upload_started_at = datetime.now(UTC)
+        await get_storage().upload(storage_key, content, content_type)
+        await record_completed_span(
+            self.db,
+            "source_ingest.upload",
+            run=run,
+            component="api",
+            span_kind="phase",
+            status="succeeded",
+            metadata={
+                "content_type": content_type,
+                "filename": filename,
+                "byte_length": len(content),
+            },
+            started_at=upload_started_at,
+            finished_at=datetime.now(UTC),
+        )
         self._enqueue_ingestion(
             source.id,
+            trace_id=trace_id,
+            run_id=run.id,
             job_kind="import",
             job_label=f"索引资料：{filename or '未命名文件'}",
         )
@@ -672,9 +693,23 @@ async def import_global_source_url(self, url: str, title: str | None) -> Source:
         self.db.add(source)
         await self.db.flush()
         await self.db.refresh(source)
+        trace_id, run = await self._create_source_ingest_run(
+            source, origin="global_url_import"
+        )
+        await record_instant_span(
+            self.db,
+            "source_ingest.upload",
+            run=run,
+            component="api",
+            span_kind="phase",
+            status="succeeded",
+            metadata={"mode": "url", "url": url, "skipped": True},
+        )
 
         self._enqueue_ingestion(
             source.id,
+            trace_id=trace_id,
+            run_id=run.id,
             job_kind="import",
             job_label=f"索引网页：{title or url}",
         )

apps/api/tests/unit/test_source_service_dispatch.py

@@ -104,6 +104,163 @@ async def flush() -> None:
     db.commit.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_upload_global_source_dispatches_ingestion_after_commit(monkeypatch) -> None:
+    added: list[object] = []
+
+    async def flush() -> None:
+        for obj in added:
+            if isinstance(obj, Source) and getattr(obj, "id", None) is None:
+                setattr(obj, "id", uuid.uuid4())
+            if isinstance(obj, ObservabilityRun) and getattr(obj, "id", None) is None:
+                setattr(obj, "id", uuid.uuid4())
+
+    db = SimpleNamespace(
+        add=Mock(side_effect=added.append),
+        flush=AsyncMock(side_effect=flush),
+        refresh=AsyncMock(return_value=None),
+        commit=AsyncMock(return_value=None),
+        sync_session=SimpleNamespace(info={}),
+    )
+    service = SourceService(db, uuid.uuid4())
+    service._get_or_create_global_notebook = AsyncMock(  # type: ignore[method-assign]
+        return_value=SimpleNamespace(id=uuid.uuid4())
+    )
+
+    storage = SimpleNamespace(upload=AsyncMock(return_value=None))
+    monkeypatch.setattr("app.providers.storage.storage", lambda: storage)
+
+    delayed: list[tuple[str, dict[str, object]]] = []
+    monkeypatch.setattr(
+        "app.workers.tasks.ingest_source.delay",
+        lambda source_id, **kwargs: delayed.append((source_id, kwargs)),
+    )
+
+    source = await service.upload_global_source("global.docx", b"doc-bytes")
+
+    assert source.status == "pending"
+    assert source.type == "doc"
+    assert delayed == []
+
+    callbacks = db.sync_session.info.get("_after_commit_callbacks", [])
+    assert len(callbacks) == 1
+    callbacks[0]()
+
+    assert delayed[0][0] == str(source.id)
+    assert delayed[0][1]["trace_id"]
+    assert delayed[0][1]["run_id"]
+    storage.upload.assert_awaited_once()
+    db.commit.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_import_global_source_url_dispatches_ingestion_after_commit(
+    monkeypatch,
+) -> None:
+    added: list[object] = []
+
+    async def flush() -> None:
+        for obj in added:
+            if isinstance(obj, Source) and getattr(obj, "id", None) is None:
+                setattr(obj, "id", uuid.uuid4())
+            if isinstance(obj, ObservabilityRun) and getattr(obj, "id", None) is None:
+                setattr(obj, "id", uuid.uuid4())
+
+    db = SimpleNamespace(
+        add=Mock(side_effect=added.append),
+        flush=AsyncMock(side_effect=flush),
+        refresh=AsyncMock(return_value=None),
+        commit=AsyncMock(return_value=None),
+        sync_session=SimpleNamespace(info={}),
+    )
+    service = SourceService(db, uuid.uuid4())
+    service._get_or_create_global_notebook = AsyncMock(  # type: ignore[method-assign]
+        return_value=SimpleNamespace(id=uuid.uuid4())
+    )
+
+    delayed: list[tuple[str, dict[str, object]]] = []
+    monkeypatch.setattr(
+        "app.workers.tasks.ingest_source.delay",
+        lambda source_id, **kwargs: delayed.append((source_id, kwargs)),
+    )
+
+    source = await service.import_global_source_url(
+        "https://example.com/global",
+        "Global Example",
+    )
+
+    assert source.status == "pending"
+    assert delayed == []
+
+    callbacks = db.sync_session.info.get("_after_commit_callbacks", [])
+    assert len(callbacks) == 1
+    callbacks[0]()
+
+    assert delayed[0][0] == str(source.id)
+    assert delayed[0][1]["trace_id"]
+    assert delayed[0][1]["run_id"]
+    db.commit.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_rechunk_source_dispatches_ingestion_after_commit_with_trace_context(
+    monkeypatch,
+) -> None:
+    added: list[object] = []
+
+    async def flush() -> None:
+        for obj in added:
+            if isinstance(obj, ObservabilityRun) and getattr(obj, "id", None) is None:
+                setattr(obj, "id", uuid.uuid4())
+
+    db = SimpleNamespace(
+        add=Mock(side_effect=added.append),
+        flush=AsyncMock(side_effect=flush),
+        commit=AsyncMock(return_value=None),
+        sync_session=SimpleNamespace(info={}),
+    )
+    service = SourceService(db, uuid.uuid4())
+    source = SimpleNamespace(
+        id=uuid.uuid4(),
+        notebook_id=uuid.uuid4(),
+        title="demo.pdf",
+        type="pdf",
+        url=None,
+        status="indexed",
+    )
+    monkeypatch.setattr(service, "_get_owned_source", AsyncMock(return_value=source))
+
+    apply_async_calls: list[dict[str, object]] = []
+    monkeypatch.setattr(
+        "app.workers.tasks.ingest_source.apply_async",
+        lambda *args, **kwargs: apply_async_calls.append(
+            {"args": list(args), "kwargs": kwargs}
+        ),
+    )
+
+    chunk_size, chunk_overlap = await service.rechunk_source(
+        source.id,
+        strategy="standard",
+    )
+
+    assert source.status == "pending"
+    assert chunk_size == 512
+    assert chunk_overlap == 64
+    assert apply_async_calls == []
+
+    callbacks = db.sync_session.info.get("_after_commit_callbacks", [])
+    assert len(callbacks) == 1
+    callbacks[0]()
+
+    dispatched = apply_async_calls[0]
+    assert dispatched["kwargs"]["args"] == [str(source.id)]
+    assert dispatched["kwargs"]["kwargs"]["trace_id"]
+    assert dispatched["kwargs"]["kwargs"]["run_id"]
+    assert dispatched["kwargs"]["kwargs"]["chunk_size"] == chunk_size
+    assert dispatched["kwargs"]["kwargs"]["chunk_overlap"] == chunk_overlap
+    db.commit.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_import_global_source_path_reads_file_and_delegates(
     tmp_path,

packages/cli/src/commands/prod.js

@@ -1,10 +1,43 @@
 import fs from 'fs';
 import path from 'path';
 import ora from 'ora';
-import { section, log, warn, info } from '../utils/ui.js';
-import { exec } from '../utils/proc.js';
+import { section, warn, info } from '../utils/ui.js';
+import { exec, execQ } from '../utils/proc.js';
 import { ROOT_DIR } from '../utils/paths.js';
 
+export const PROD_UPDATE_PULL_SERVICES = ['api', 'web', 'monitoring'];
+
+export function parseTrackedFilesFromGitStatus(statusOutput) {
+  return statusOutput
+    .split('\n')
+    .map((line) => line.trimEnd())
+    .filter(Boolean)
+    .filter((line) => !line.startsWith('?? '))
+    .map((line) => line.slice(3));
+}
+
+export function buildProdUpdateDirtyWorktreeGuidance(files) {
+  const visibleFiles = files.slice(0, 10);
+  const lines = [
+    '检测到本地未提交的 Git 改动，已停止更新以避免覆盖这些文件：',
+    ...visibleFiles.map((file) => `- ${file}`),
+  ];
+
+  if (files.length > visibleFiles.length) {
+    lines.push(`- 以及另外 ${files.length - visibleFiles.length} 个文件`);
+  }
+
+  lines.push('如果这些改动不需要保留，请先执行：git restore <文件>');
+  lines.push('如果这些改动需要暂存，请先执行：git stash push --include-untracked');
+  lines.push('清理完成后，再重新运行 lyra update。');
+
+  return lines;
+}
+
+export function getProdUpdatePullCommand() {
+  return `docker compose -f docker-compose.prod.yml pull ${PROD_UPDATE_PULL_SERVICES.join(' ')}`;
+}
+
 export async function startProd() {
   section('生产模式启动（ghcr.io 云端镜像）');
 
@@ -39,10 +72,19 @@ export async function updateProd() {
   section('一键更新（git pull + 拉新镜像 + 重启）');
   process.chdir(ROOT_DIR);
 
+  const dirtyTrackedFiles = parseTrackedFilesFromGitStatus(execQ('git status --porcelain'));
+  if (dirtyTrackedFiles.length > 0) {
+    warn('检测到本地改动，已取消本次更新。');
+    for (const line of buildProdUpdateDirtyWorktreeGuidance(dirtyTrackedFiles)) {
+      info(line);
+    }
+    process.exit(1);
+  }
+
   const spinner = ora('更新到最新版本...').start();
   try {
-    exec('git pull', { shell: true });
-    exec('docker compose -f docker-compose.prod.yml pull web api worker', { shell: true });
+    exec('git pull --ff-only', { shell: true });
+    exec(getProdUpdatePullCommand(), { shell: true });
     exec('docker compose -f docker-compose.prod.yml up -d', { shell: true });
     exec('docker image prune -f', { shell: true });
     spinner.succeed('更新完成');