Staging

.github/workflows/manage-staging.yml

@@ -112,6 +112,68 @@ jobs:
               --services "$service_name"
           }
 
+          wait_for_healthy_workers() {
+            local worker_health_deadline="$((SECONDS + 300))"
+            local healthy_worker_count="0"
+            local -a worker_tasks=()
+
+            # The AWS CLI services-stable waiter checks deployment count and
+            # runningCount only. A cold task can therefore be RUNNING while
+            # container health remains UNKNOWN during its health-check
+            # startPeriod. Poll the task health explicitly before opening API
+            # admission, with a bounded timeout so a bad rollout still fails.
+            while (( SECONDS < worker_health_deadline )); do
+              healthy_worker_count="0"
+              worker_tasks=()
+              mapfile -t worker_tasks < <(
+                aws --profile knowhere ecs list-tasks \
+                  --cluster knowhere-fargate \
+                  --service-name knowhere-worker-staging \
+                  --desired-status RUNNING \
+                  --query 'taskArns[]' \
+                  --output text | tr '\t' '\n' | sed '/^None$/d;/^$/d'
+              )
+
+              if [ "${#worker_tasks[@]}" -eq 2 ]; then
+                healthy_worker_count="$(aws --profile knowhere ecs describe-tasks \
+                  --cluster knowhere-fargate \
+                  --tasks "${worker_tasks[@]}" \
+                  --query 'length(tasks[?lastStatus==`RUNNING` && healthStatus==`HEALTHY`])' \
+                  --output text)"
+                if [ "$healthy_worker_count" -eq 2 ]; then
+                  return 0
+                fi
+              fi
+
+              echo "Waiting for two healthy worker tasks; running=${#worker_tasks[@]}, healthy=$healthy_worker_count"
+              sleep 15
+            done
+
+            echo "Timed out waiting for two healthy worker tasks" >&2
+            return 1
+          }
+
+          wait_for_public_api_health() {
+            local api_health_deadline="$((SECONDS + 300))"
+
+            # services-stable can return as soon as the API task is RUNNING,
+            # before the load balancer has registered a healthy target. Treat
+            # transient 502/503 responses as cold-start progress, but keep a
+            # hard timeout so a broken target never reports a successful start.
+            while (( SECONDS < api_health_deadline )); do
+              if curl --fail --silent --show-error --max-time 10 \
+                https://api-staging.knowhereto.ai/health >/dev/null; then
+                return 0
+              fi
+
+              echo "Waiting for the public staging API health endpoint"
+              sleep 10
+            done
+
+            echo "Timed out waiting for the public staging API health endpoint" >&2
+            return 1
+          }
+
           read_services() {
             aws --profile knowhere ecs describe-services \
               --cluster knowhere-fargate \
@@ -129,32 +191,11 @@ jobs:
               # traffic; ECS stability alone does not prove container health.
               update_service knowhere-worker-staging 2
               wait_for_service knowhere-worker-staging
-              mapfile -t worker_tasks < <(
-                aws --profile knowhere ecs list-tasks \
-                  --cluster knowhere-fargate \
-                  --service-name knowhere-worker-staging \
-                  --desired-status RUNNING \
-                  --query 'taskArns[]' \
-                  --output text | tr '\t' '\n'
-              )
-              if [ "${#worker_tasks[@]}" -ne 2 ]; then
-                echo "Expected two running worker tasks, found ${#worker_tasks[@]}" >&2
-                exit 1
-              fi
-              healthy_workers="$(aws --profile knowhere ecs describe-tasks \
-                --cluster knowhere-fargate \
-                --tasks "${worker_tasks[@]}" \
-                --query 'length(tasks[?lastStatus==`RUNNING` && healthStatus==`HEALTHY`])' \
-                --output text)"
-              if [ "$healthy_workers" -ne 2 ]; then
-                echo "Expected two healthy workers, found $healthy_workers" >&2
-                exit 1
-              fi
+              wait_for_healthy_workers
 
               update_service knowhere-api-staging 1
               wait_for_service knowhere-api-staging
-              curl --fail --silent --show-error --max-time 30 \
-                https://api-staging.knowhereto.ai/health >/dev/null
+              wait_for_public_api_health
               startup_seconds="$(($(date +%s) - start_started_epoch))"
               ;;
             stop)

.gitignore

@@ -68,8 +68,7 @@ test_*.csv
 *.csv
 !requirements.csv
 
-# Local debugging scripts
-apps/worker/scripts/
+# Local debugging scripts (apps/worker/scripts/ is tracked)
 apps/worker/experiments/
 apps/worker/start_celery_worker.py
 apps/worker/start_celery_debug.sh

apps/api/app/services/document_ingestion/handoff_service.py

@@ -29,6 +29,7 @@
 class _UploadedFileJob(Protocol):
     job_id: str
     job_type: str
+    status: str
 
 
 class DocumentIngestionHandoffService:
@@ -66,6 +67,17 @@ async def start_uploaded_file_workflow(
                 ],
             )
 
+        # Upload completion can arrive through both the S3 notification and the
+        # confirm-upload endpoint.  Once either path has moved the job out of
+        # waiting-file, the other path must be a no-op instead of dispatching a
+        # second worker task for the same logical job.
+        if job.status != JobStatus.WAITING_FILE.value:
+            logger.info(
+                "Upload handoff already completed: "
+                f"job_id={job.job_id}, status={job.status}"
+            )
+            return
+
         outcome = await self._state_machine.transition_outcome(
             db,
             job.job_id,
@@ -75,6 +87,18 @@ async def start_uploaded_file_workflow(
             "system",
         )
         if not outcome.succeeded:
+            # A concurrent handoff may have won the CAS transition after this
+            # caller loaded the waiting-file snapshot.  The state machine
+            # reports the winner's state as ``from_state``; treat that result
+            # as an idempotent no-op and do not enqueue another task.
+            if outcome.reason == "invalid_transition" and outcome.from_state != (
+                JobStatus.WAITING_FILE.value
+            ):
+                logger.info(
+                    "Upload handoff won by another trigger: "
+                    f"job_id={job.job_id}, status={outcome.from_state}"
+                )
+                return
             logger.warning(
                 "Upload handoff transition rejected: "
                 f"job_id={job.job_id}, reason={outcome.reason}"

apps/api/tests/contract/test_s3_event_contract.py

@@ -119,6 +119,93 @@ async def start_uploaded_file_parse(
     ]
 
 
+@pytest.mark.asyncio
+async def test_should_not_dispatch_a_second_task_for_a_replayed_upload_event(
+    api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]],
+    monkeypatch: MonkeyPatch,
+) -> None:
+    workflow_calls: list[dict[str, str]] = []
+
+    class FakeDocumentIngestionWorkerDispatcher:
+        async def start_uploaded_file_parse(
+            self,
+            *,
+            job_id: str,
+            user_id: str,
+        ) -> str:
+            workflow_calls.append({"job_id": job_id, "user_id": user_id})
+            return "contract-task-id"
+
+    async with api_client_factory() as api_client:
+        user_id, job_id = await _insert_waiting_file_job()
+        handoff_service = importlib.import_module(
+            "app.services.document_ingestion.handoff_service"
+        )
+        monkeypatch.setattr(
+            handoff_service,
+            "DocumentIngestionWorkerDispatcher",
+            FakeDocumentIngestionWorkerDispatcher,
+        )
+
+        first_response = await api_client.post(
+            "/api/v1/internal/s3-events",
+            json=_build_s3_event_payload(job_id),
+        )
+        replay_response = await api_client.post(
+            "/api/v1/internal/s3-events",
+            json=_build_s3_event_payload(job_id),
+        )
+
+    assert first_response.status_code == 200
+    assert replay_response.status_code == 200
+    assert workflow_calls == [{"job_id": job_id, "user_id": user_id}]
+
+
+@pytest.mark.asyncio
+async def test_should_treat_a_concurrent_upload_handoff_cas_winner_as_a_no_op() -> None:
+    from app.services.document_ingestion.handoff_service import (
+        DocumentIngestionHandoffService,
+    )
+    from shared.core.state_machine.transition_outcome import JobTransitionOutcome
+
+    class FakeStateMachine:
+        async def transition_outcome(self, *args: object, **kwargs: object) -> object:
+            del args, kwargs
+            return JobTransitionOutcome.rejected(
+                job_id="job-race",
+                to_state="pending",
+                reason="invalid_transition",
+                attempts=1,
+                from_state="pending",
+            )
+
+    class FakeDispatcher:
+        async def start_uploaded_file_parse(
+            self,
+            *,
+            job_id: str,
+            user_id: str,
+        ) -> str:
+            del job_id, user_id
+            raise AssertionError("CAS loser must not dispatch a duplicate task")
+
+    service = DocumentIngestionHandoffService(
+        state_machine=FakeStateMachine(),
+        worker_dispatcher=FakeDispatcher(),
+    )
+
+    await service.start_uploaded_file_workflow(
+        db=cast(object, None),
+        job=SimpleNamespace(
+            job_id="job-race",
+            job_type="document_ingestion",
+            status="waiting-file",
+        ),
+        user_id="contract-user",
+        trigger="s3_upload_completed",
+    )
+
+
 @pytest.mark.asyncio
 async def test_should_accept_a_pre_rename_waiting_file_job_type_during_upload_handoff(
     api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]],

apps/worker/app/services/document_agent/agents/calibration/SKILL.md

@@ -23,8 +23,6 @@ into **page-numbering regimes** (distinct numbering systems / label shapes:
 decimal digits, roman numerals, prefixed folio labels, etc.).
 
 - Do not mix samples across regimes when computing an offset.
-- Include `entry_indices` (0-based indices into `toc_region.entries`) for each
-  regime you submit.
 - Run the same initial-calibration procedure independently for each regime that
   has usable entries.
 
@@ -53,49 +51,36 @@ offset: treat that sample / regime as **not found**, submit whatever regimes you
 already confirmed (or `status=failed`), and let production fallback handle the
 rest. Do not guess pages.
 
-## Phase 2 — Completion (deterministic after submit; production path)
-
-For each TOC region, every regime with a candidate offset is completed
-independently, then merged by **physical page**:
-
-1. Build TitleNodes via production `extract_toc_nodes` (regime-aware parse:
-   decimal / roman / prefixed labels → `printed_page` + `page_kind`).
-2. For **each** regime with an offset:
-   - Project leaves belonging to that regime
-   - Run production Phase-2: prune → tail verify → binary-search →
-     small-step recalibrate (single-leaf regimes apply offset directly)
-3. Merge all regime `match_overrides` (physical pages), then null-page parent
-   locate once on the combined tree.
-4. Emit production `SkeletonAnchor` (`offset` = primary decimal summary,
-   `match_overrides` = union of all regimes, `null_page_report`, `bulk_count`,
-   `pruned_count`, `locate_agent`).
-5. On recalibrate/budget failure inside one regime: keep that regime's complete
-   **prefix**; **drop** unresolved **suffix** leaves from the TOC tree (no TOC),
-   then run null-page parent locate on what remains. Never fall back to a fixed
-   post-TOC window.
-
-## Usability bar
-
-- Coarse structure may use the result when `SkeletonAnchor.offset_status=ok`
-  and `bulk_count > 0` (at least one complete production segment).
-- Otherwise downstream treats the document as no-TOC / Root fallback.
+## Phase 2 — Completion (deterministic after submit)
+
+Not your job and not yours to describe. After submit, production completes each
+regime independently (prune → tail verify → binary search → small-step
+recalibrate), merges the regimes by physical page, and emits the
+`SkeletonAnchor`. It recomputes segment coverage, per-regime status and the
+no-TOC entry set itself, so do not submit those.
 
 ## Tools
 
 - `inspect.pages`: primary tool for Phase 1. Open physical pages, render, answer
   your question. Prefer the progressive 1→3→5 schedule above. Per-call page
   count is capped; overall spend is limited by the calibration visual token
   budget and `max_rounds`.
-- `calibration.submit`: finish Phase 1. Pass the full result under
+- `calibration.submit`: finish Phase 1. Pass the result under
   `tool_args.result` (or result fields directly in `tool_args`).
 
 ## Output rules
 
-- Submit `status`, `regimes`, top-level `offset` / `offset_status` for the
-  primary decimal-digit regime when identifiable, `tool_calls`, `notes`.
-- Each regime must include `kind`, candidate `offset`, `offset_status`,
-  `entry_indices`, `samples` (with `title`, `printed_label`, `physical` when
-  known), and `posterior` if you already inspected a late check.
+Submit exactly the fields in the `calibration.submit` schema — `status`,
+`regimes`, `notes` — and nothing else:
+
+- Per regime: `kind` and the candidate `offset`. Add `entry_indices` only when
+  the regime is not simply the entries whose printed-label shape matches `kind`,
+  and `samples` (`title` + `physical`) only for anchors you actually confirmed.
+- `notes`: one short sentence saying why. When you found no offset, submit
+  `status=failed` and say why in that one sentence.
 - Keep `kind` values consistent within one run (`decimal`, `roman`, `prefixed`,
   or `other`).
+- Anything else — per-regime status, segment coverage, no-TOC entries, tool call
+  counts, region index — is recomputed after submit; emitting it only risks the
+  submit being cut off by the output limit, which ends the run with no result.
 - Stay within the token / round budgets announced in the payload.