feat(seap): watch-folder-router — auto-route ingest.file.added.v1 by MIME (Bud 1) - #2535
Conversation
…MIME (Bud 1) The first bud of the SEAP watch-folder constellation (SEAP_WATCHFOLDER_NATS_CONSTELLATION_2026-08-12). Consumes ingest.file.added.v1 and routes by mime_type to the right analyzer, retiring the manual media-audio /analyze calls every hand-driven ingest made: audio/* video/* -> media-audio /analyze -> ingest.transcript.ready.v1 pdf + office docs -> ingest.document.ready.v1 (pdf-ingest/langextract) else (text) -> ingest.text.ready.v1 (extract-worker) Thin NATS router (no models/GPU), :8125 health/metrics only. Follows the extract-worker pure-consumer + voice-sampler MinIO patterns. NATS default uses the standard authed URL (nats://nats:pmoves@...). All three subjects already exist in the catalog. Smoke-verified live: synthetic PDF + text ingest.file.added events routed to document.ready and text.ready respectively (SMOKE PASS); container healthy, NATS connected. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
github-actions
Bot
added
compose
Docker Hardening ValidationHardening Validation ReportValidated: Wed Aug 12 16:29:53 UTC 2026Services CheckedPMOVES.AI Docker Hardening Validation[INFO] Checking: pmoves/docker-compose.hardened.yml [INFO] Validating: hi-rag-gateway-v2 [INFO] Validating: extract-worker [INFO] Validating: langextract [INFO] Validating: presign [INFO] Validating: render-webhook [INFO] Validating: retrieval-eval [INFO] Validating: pdf-ingest [INFO] Validating: jellyfin-bridge [INFO] Validating: invidious-companion-proxy [INFO] Validating: ffmpeg-whisper [INFO] Validating: media-video [INFO] Validating: media-audio [INFO] Validating: hi-rag-gateway-v2-gpu [INFO] Validating: hi-rag-gateway-gpu [INFO] Validating: deepresearch [INFO] Validating: supaserch [INFO] Validating: publisher-discord [INFO] Validating: mesh-agent [INFO] Validating: nats-echo-req [INFO] Validating: nats-echo-res [INFO] Validating: comfy-watcher [INFO] Validating: grayjay-plugin-host [INFO] Validating: agent-zero [INFO] Validating: p7-room-orchestrator [INFO] Validating: archon [INFO] Validating: channel-monitor [INFO] Validating: pmoves-yt [INFO] Validating: notebook-sync [INFO] Validating: supabase_service_role_key [INFO] Validating: supabase_jwt_secret [INFO] Validating: p7_control_token ====================================== |
📝 WalkthroughWalkthroughAdds a NATS-driven ChangesWatch-folder routing
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant FileIngestion
participant NATS
participant WatchFolderRouter
participant MediaAudio
FileIngestion->>NATS: Publish ingest.file.added.v1
NATS->>WatchFolderRouter: Deliver file event
WatchFolderRouter->>WatchFolderRouter: Classify file
WatchFolderRouter->>MediaAudio: Request media transcription
MediaAudio-->>WatchFolderRouter: Return transcript
WatchFolderRouter->>NATS: Publish routed event
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 290e1a0769
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| bucket = event.get("bucket") | ||
| key = event.get("key") | ||
| mime = event.get("mime_type", "") |
There was a problem hiding this comment.
For every contract-compliant ingest.file.added.v1 message, the decoded object is an event envelope and the file fields are under event["payload"]; moreover, the registered payload exposes uri and kind, not top-level bucket, key, or mime_type. For example, pdf-ingest/app.py publishes exactly that envelope, so its events always hit this missing bucket/key rejection and no route runs. Unwrap the payload and parse the registered uri/kind contract, or introduce a separately versioned input contract.
AGENTS.md reference: pmoves/AGENTS.md:L34-L36
Useful? React with 👍 / 👎.
|
|
||
|
|
||
| def _envelope(topic: str, payload: Dict[str, Any], src_event: Dict[str, Any]) -> bytes: | ||
| return json.dumps( |
There was a problem hiding this comment.
When any route succeeds, _envelope() emits only topic, payload, correlation_id, source, and ts, while contracts/schemas/common/envelope.schema.json requires id and version as well. Consequently, consumers that validate the documented standard envelope will reject every event produced by this router; use services.common.events.envelope and validate the route payload before publishing.
AGENTS.md reference: pmoves/AGENTS.md:L219-L219
Useful? React with 👍 / 👎.
| elif route == "document": | ||
| # pdf-ingest / langextract own extraction; router only announces. | ||
| await _publish(SUBJECT_DOCUMENT, base, event) |
There was a problem hiding this comment.
When a PDF or Office document is classified here, the branch immediately publishes ingest.document.ready.v1 without calling pdf-ingest or langextract. Repo-wide inspection shows those services do not consume this subject—pdf-ingest publishes it only after downloading, extracting, and indexing—and the payload here also lacks the contract-required doc_id, uri, and chunk_count. Thus a watch-folder document is reported ready although no extraction or embedding occurred; invoke the document ingest path and publish its resulting completion event instead.
AGENTS.md reference: pmoves/AGENTS.md:L34-L36
Useful? React with 👍 / 👎.
| else: | ||
| await _publish(SUBJECT_TEXT, base, event) |
There was a problem hiding this comment.
For plain-text inputs, this publishes ingest.text.ready.v1, but that topic is absent from contracts/topics.json and repo-wide search finds no subscriber for it; extract-worker accepts chunks through its HTTP /ingest endpoint instead. Therefore text files are silently declared ready without ever being extracted or embedded. Route them to the existing ingest API, or add a versioned contract and an actual consumer before publishing this subject.
AGENTS.md reference: pmoves/AGENTS.md:L31-L31
Useful? React with 👍 / 👎.
| restart: unless-stopped | ||
| environment: | ||
| - NATS_URL=${NATS_URL:-nats://nats:4222} | ||
| - MEDIA_AUDIO_URL=${MEDIA_AUDIO_URL:-http://media-audio:8082} |
There was a problem hiding this comment.
Under the canonical overlay-up-workers path checked in pmoves/Makefile, Compose loads only the base, core, and workers overlays, while split_compose.py assigns media-audio exclusively to the media overlay. In that supported deployment, the configured media-audio hostname therefore has no running service, so every audio/video transcription request fails. Move the router alongside its analyzer or include/start the media dependency for the workers target.
AGENTS.md reference: AGENTS.md:L48-L50
Useful? React with 👍 / 👎.
|
Security-review note (automated scan flagged the NATS default as a hardcoded secret): acknowledged, proceeding without change — this is the documented PMOVES NATS-creds convention, verified against current code:
Ref: memory |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
pmoves/services/watch-folder-router/README.md (1)
11-16: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUpdate the required topology and service catalog files.
Add
watch-folder-router, port8125,/healthz,/metrics, and its subscribed and published subjects to.claude/context/services-catalog.mdand.claude/context/nats-subjects.md. Update the Mermaid topology usingpmoves/docs/AGENTS/PMOVES_AGENT_TOPOLOGY.md.As per coding guidelines, “Document NATS event topology in
.claude/context/nats-subjects.mdand maintain services catalog with port assignments and health endpoints in.claude/context/services-catalog.md.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pmoves/services/watch-folder-router/README.md` around lines 11 - 16, Update .claude/context/services-catalog.md with watch-folder-router, port 8125, and its /healthz and /metrics endpoints; update .claude/context/nats-subjects.md with its ingest.file.added.v1 subscription and MIME-based published subjects. Revise the Mermaid topology following PMOVES_AGENT_TOPOLOGY.md to include the router and its NATS/service connections.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pmoves/docker-compose.yml`:
- Around line 2284-2287: Replace the unauthenticated NATS fallback with the
tier-provided convention NATS_URL=${NATS_URL} in pmoves/docker-compose.yml lines
2284-2287 and regenerate the same corrected entry in
pmoves/docker-compose.workers.yml lines 97-100. Update
pmoves/services/watch-folder-router/README.md lines 37-41 to document that the
authenticated NATS URL comes from the tier environment.
In `@pmoves/services/watch-folder-router/requirements.txt`:
- Around line 2-3: Upgrade the FastAPI pin in requirements.txt to a release that
permits Starlette 1.0.1 or newer, then regenerate requirements.lock using the
dependency workflow consumed by the Docker image. Ensure the regenerated
lockfile resolves the compatible Starlette version and do not change h11 solely
for this issue.
In `@pmoves/services/watch-folder-router/worker.py`:
- Around line 84-207: Add concise docstrings to the undocumented functions in
the changed module, including _envelope, _transcribe, _handle_file_added,
_publish, lifespan, healthz, and metrics. Keep the existing behavior unchanged
and ensure at least 80% of the new functions have docstrings.
- Around line 114-120: Update the rejected-event warning in the worker’s
bucket/key validation block to stop logging the complete event payload. Log only
safe identifiers such as the presence or values of validated fields, or the
event’s field names, while preserving the rejection metric and early return.
- Around line 176-185: Ensure the router never completes startup without a
usable NATS subscription: update the lifespan connection flow in
pmoves/services/watch-folder-router/worker.py:176-185 to retry the initial
connection/subscription or re-raise the failure; update the /healthz handler at
pmoves/services/watch-folder-router/worker.py:193-202 to return an unhealthy
HTTP status when _nc is absent or closed; add depends_on with nats
service_healthy to the watch-folder-router service in
pmoves/docker-compose.yml:2283-2302 and regenerate the equivalent dependency in
pmoves/docker-compose.workers.yml:96-115.
- Around line 75-80: Update the MIME classification logic around the extension
fallback so that a non-empty mime value other than application/octet-stream
returns "text" before checking file extensions. Keep the extension-based media
and document routing only for empty MIME values or the generic octet-stream
MIME.
- Around line 106-120: Update _handle_file_added to require all
ingest.file.added.v1 fields defined by file-added.v1.schema.json, including
file_id, uri, and kind, before routing. Call
services.common.events.validate_payload() on the incoming event before
classification/publication, and validate each downstream payload immediately
before _publish.
---
Nitpick comments:
In `@pmoves/services/watch-folder-router/README.md`:
- Around line 11-16: Update .claude/context/services-catalog.md with
watch-folder-router, port 8125, and its /healthz and /metrics endpoints; update
.claude/context/nats-subjects.md with its ingest.file.added.v1 subscription and
MIME-based published subjects. Revise the Mermaid topology following
PMOVES_AGENT_TOPOLOGY.md to include the router and its NATS/service connections.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 638c9939-08c9-4864-b5d2-424866b4820c
⛔ Files ignored due to path filters (1)
pmoves/services/watch-folder-router/requirements.lockis excluded by!**/*.lock
📒 Files selected for processing (7)
pmoves/docker-compose.workers.ymlpmoves/docker-compose.ymlpmoves/scripts/split_compose.pypmoves/services/watch-folder-router/Dockerfilepmoves/services/watch-folder-router/README.mdpmoves/services/watch-folder-router/requirements.txtpmoves/services/watch-folder-router/worker.py
| environment: | ||
| - NATS_URL=${NATS_URL:-nats://nats:4222} | ||
| - MEDIA_AUDIO_URL=${MEDIA_AUDIO_URL:-http://media-audio:8082} | ||
| - ROUTER_TRANSCRIBE_TIMEOUT_SEC=${ROUTER_TRANSCRIBE_TIMEOUT_SEC:-3600} |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Use the authenticated worker NATS configuration.
The unauthenticated fallback overrides the env_file value. The router then cannot connect to an authenticated NATS server. Use the repository worker NATS_URL=${NATS_URL} convention and document that the value comes from the tier environment.
pmoves/docker-compose.yml#L2284-L2287: remove the unauthenticatednats://nats:4222fallback.pmoves/docker-compose.workers.yml#L97-L100: regenerate the overlay with the same corrected NATS entry.pmoves/services/watch-folder-router/README.md#L37-L41: document the authenticated tier-provided NATS URL instead of an unauthenticated default.
Based on learnings, worker service NATS configuration is intentionally sourced through tier env_file entries and uses NATS_URL=${NATS_URL}.
📍 Affects 3 files
pmoves/docker-compose.yml#L2284-L2287(this comment)pmoves/docker-compose.workers.yml#L97-L100pmoves/services/watch-folder-router/README.md#L37-L41
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/docker-compose.yml` around lines 2284 - 2287, Replace the
unauthenticated NATS fallback with the tier-provided convention
NATS_URL=${NATS_URL} in pmoves/docker-compose.yml lines 2284-2287 and regenerate
the same corrected entry in pmoves/docker-compose.workers.yml lines 97-100.
Update pmoves/services/watch-folder-router/README.md lines 37-41 to document
that the authenticated NATS URL comes from the tier environment.
Source: Learnings
| fastapi==0.114.2 | ||
| uvicorn[standard]==0.30.6 |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
fd -a '^requirements\.lock$' pmoves/services/watch-folder-router
rg -n '^(fastapi|uvicorn|starlette|h11)==' \
pmoves/services/watch-folder-router/requirements.lock \
pmoves/services/watch-folder-router/requirements.txtRepository: POWERFULMOVES/PMOVES.AI
Length of output: 597
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- service files ---'
git ls-files pmoves/services/watch-folder-router
printf '%s\n' '--- dependency references and lock metadata ---'
rg -n -i 'requirements\.lock|pip-compile|uv lock|starlette|h11|fastapi|uvicorn' \
pmoves/services/watch-folder-router \
pmoves/Makefile pmoves/README* README.md 2>/dev/null || true
printf '%s\n' '--- dependency entries ---'
cat -n pmoves/services/watch-folder-router/requirements.txt
cat -n pmoves/services/watch-folder-router/requirements.lockRepository: POWERFULMOVES/PMOVES.AI
Length of output: 4852
🌐 Web query:
GHSA-86qp-5c8j-p5mr Starlette affected versions patched version
💡 Result:
The vulnerability identified as GHSA-86qp-5c8j-p5mr (also tracked as CVE-2026-48710) affects the Starlette framework by allowing a malformed HTTP Host header to poison the request.url.path, which can lead to the bypass of path-based security checks [1][2][3]. Affected versions include all versions of Starlette up to and including 1.0.0 [1][2][4]. The patched version is 1.0.1 or higher [2][3][5]. Upgrading to this version ensures that the Host header is validated against the grammar of RFC 9112 §3.2 and RFC 3986 §3.2.2 when constructing request.url, falling back to scope["server"] if the header is malformed [1][4][3].
Citations:
- 1: https://osv.dev/vulnerability/GHSA-86qp-5c8j-p5mr
- 2: GHSA-86qp-5c8j-p5mr
- 3: https://nvd.nist.gov/vuln/detail/CVE-2026-48710
- 4: GHSA-86qp-5c8j-p5mr
- 5: https://www.cve.org/CVERecord?id=CVE-2026-48710
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import json
import urllib.request
for project in ("fastapi", "starlette", "uvicorn"):
with urllib.request.urlopen(f"https://pypi.org/pypi/{project}/json", timeout=10) as response:
data = json.load(response)
print(f"{project}: latest={data['info']['version']}")
if project == "fastapi":
for version in ("0.114.2", data["info"]["version"]):
release = data["releases"].get(version, [])
if not release:
print(f" {version}: unavailable")
continue
with urllib.request.urlopen(
f"https://pypi.org/pypi/{project}/{version}/json", timeout=10
) as response:
info = json.load(response)["info"]
print(f" {version} requires:")
for requirement in info.get("requires_dist") or []:
if requirement.lower().startswith("starlette"):
print(f" {requirement}")
PYRepository: POWERFULMOVES/PMOVES.AI
Length of output: 2447
🌐 Web query:
FastAPI Starlette 1.0.1 compatibility requires starlette
💡 Result:
FastAPI supports Starlette 1.0.1 and later [1][2]. While FastAPI generally manages its dependencies automatically, using Starlette 1.0.1 or higher is explicitly recommended to resolve the high-severity security vulnerability CVE-2026-48710 (Host Header Validation Bypass) present in Starlette versions 1.0.0 and earlier [3][4]. If your project is using an older, pinned version of Starlette, you should manually update to at least 1.0.1 to address this security risk [3][4]. Modern versions of FastAPI (e.g., 0.115.x and later) are configured to pull in compatible Starlette 1.0+ versions automatically, but in environments with constrained or pinned dependencies, an explicit upgrade is necessary [3]. Key points regarding compatibility: - Security Recommendation: Upgrade to Starlette 1.0.1+ immediately to patch CVE-2026-48710, which can allow attackers to bypass path-based security middleware by crafting malformed Host headers [3][4]. - Versioning: FastAPI added support for Starlette 1.0.0+ in earlier releases [1][2] and continues to support the 1.x series [2][5]. - Best Practice: Avoid pinning Starlette versions unless absolutely necessary. Letting the package manager resolve the latest version ensures that security patches are applied automatically as they become available [6][7].
Citations:
- 1: ⬆️ Add support for Starlette 1.0.0+ fastapi/fastapi#14987
- 2: https://fastapi.tiangolo.com/release-notes/
- 3: Does Starlette GHSA-86qp-5c8j-p5mr affect FastAPI installations using Starlette <= 1.0.0? fastapi/fastapi#15593
- 4: Fix to update Starlette vulnerability [HIGH] CVE-2026-48710 - Host Header Validation Bypass google/adk-python#5893
- 5: https://pypi.org/project/fastapi/
- 6: https://fastapi.tiangolo.com/deployment/versions/
- 7: starlette dependnecy seems to have been found with a CVE-2025-62727. Why does fastapi still allow the vulnerable version? fastapi/fastapi#14424
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- request URL and host-related logic ---'
rg -n -i 'host|trustedhost|request\.url|url\.path|middleware|path' \
pmoves/services/watch-folder-router/worker.py \
pmoves/services/watch-folder-router/README.md
printf '%s\n' '--- FastAPI metadata ---'
curl --fail --silent --show-error --insecure https://pypi.org/pypi/fastapi/json |
python3 -c '
import json, sys
data=json.load(sys.stdin)
print("latest:", data["info"]["version"])
for version in ("0.114.2", "0.115.0", "0.115.12", data["info"]["version"]):
if version not in data["releases"]:
continue
import urllib.request, ssl
url=f"https://pypi.org/pypi/fastapi/{version}/json"
with urllib.request.urlopen(url, context=ssl._create_unverified_context(), timeout=10) as response:
info=json.load(response)["info"]
print(version)
for requirement in info.get("requires_dist") or []:
if requirement.lower().startswith("starlette"):
print(" ", requirement)
'Repository: POWERFULMOVES/PMOVES.AI
Length of output: 480
🌐 Web query:
site:github.com/fastapi/fastapi "starlette>=1.0.0" "fastapi"
💡 Result:
FastAPI provides full support for Starlette 1.0.0 and newer versions [1][2][3]. As of August 12, 2026, FastAPI supports Starlette version 1.0.0 and beyond, with current versions of FastAPI explicitly depending on Starlette versions that meet the starlette>=0.46.0 requirement [4]. The transition to supporting Starlette 1.0.0 was introduced earlier in 2026 [5][1], and subsequent releases have maintained compatibility with the Starlette 1.x.x series, including version 1.1.0 [3]. If you are encountering issues or dependency conflicts related to Starlette 1.0.0, ensure your FastAPI installation is up to date, as older versions of FastAPI (released prior to the formal adoption of Starlette 1.0.0 support) may strictly pin Starlette to versions lower than 1.0.0 [6][5]. If you are using a modern version of FastAPI, the pyproject.toml or requirements.txt should allow Starlette 1.0.0+ automatically [4].
Citations:
- 1: ⬆️ Add support for Starlette 1.0.0+ fastapi/fastapi#14987
- 2: fastapi/fastapi@0.135.1...0.136.1
- 3: ⬆ Bump starlette from 1.0.0 to 1.1.0 fastapi/fastapi#15684
- 4: https://github.com/fastapi/fastapi/blob/master/pyproject.toml
- 5: https://github.com/fastapi/fastapi/releases/tag/0.128.3
- 6: ⬆️ Upgrade Starlette supported version range to
starlette>=0.40.0,<1.0.0fastapi/fastapi#14853
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline pmoves/services/watch-folder-router/worker.py --lang python
printf '%s\n' '--- route declarations and request URL usage ---'
rg -n -C 3 '`@app`\.|FastAPI\(|Request|request\.|url\.path|path-based|TrustedHost|middleware' \
pmoves/services/watch-folder-router/worker.pyRepository: POWERFULMOVES/PMOVES.AI
Length of output: 1730
Upgrade FastAPI and regenerate requirements.lock.
fastapi==0.114.2 constrains Starlette to <0.39.0, so lockfile regeneration alone cannot remove starlette==0.38.6. Upgrade FastAPI to a release that permits starlette>=1.0.1, then regenerate the lockfile used by the Docker image. h11==0.16.0 is unrelated to this advisory.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/services/watch-folder-router/requirements.txt` around lines 2 - 3,
Upgrade the FastAPI pin in requirements.txt to a release that permits Starlette
1.0.1 or newer, then regenerate requirements.lock using the dependency workflow
consumed by the Docker image. Ensure the regenerated lockfile resolves the
compatible Starlette version and do not change h11 solely for this issue.
Source: Linters/SAST tools
| # Fall back to extension when the MIME is generic (octet-stream). | ||
| ext = os.path.splitext(key or "")[1].lower() | ||
| if ext in (".m4a", ".mp3", ".wav", ".mp4", ".webm", ".mkv", ".mov", ".ogg", ".flac"): | ||
| return "media" | ||
| if ext in (".pdf", ".docx", ".doc", ".pptx", ".xlsx", ".rtf"): | ||
| return "document" |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Restrict extension fallback to generic MIME types.
Lines 75-80 apply the extension fallback for every non-media and non-document MIME type. A file declared as text/plain with an .mp3 suffix routes to media-audio. Return "text" before extension checks when mime is non-empty and is not application/octet-stream.
Proposed fix
if mime.startswith(DOC_MIME_PREFIXES) or mime in DOC_MIME_EXACT:
return "document"
- # Fall back to extension when the MIME is generic (octet-stream).
+ if mime and mime != "application/octet-stream":
+ return "text"
+ # Fall back to extension when the MIME is generic (octet-stream).📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # Fall back to extension when the MIME is generic (octet-stream). | |
| ext = os.path.splitext(key or "")[1].lower() | |
| if ext in (".m4a", ".mp3", ".wav", ".mp4", ".webm", ".mkv", ".mov", ".ogg", ".flac"): | |
| return "media" | |
| if ext in (".pdf", ".docx", ".doc", ".pptx", ".xlsx", ".rtf"): | |
| return "document" | |
| # Fall back to extension when the MIME is generic (octet-stream). | |
| ext = os.path.splitext(key or "")[1].lower() | |
| if ext in (".m4a", ".mp3", ".wav", ".mp4", ".webm", ".mkv", ".mov", ".ogg", ".flac"): | |
| return "media" | |
| if ext in (".pdf", ".docx", ".doc", ".pptx", ".xlsx", ".rtf"): | |
| return "document" |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/services/watch-folder-router/worker.py` around lines 75 - 80, Update
the MIME classification logic around the extension fallback so that a non-empty
mime value other than application/octet-stream returns "text" before checking
file extensions. Keep the extension-based media and document routing only for
empty MIME values or the generic octet-stream MIME.
| def _envelope(topic: str, payload: Dict[str, Any], src_event: Dict[str, Any]) -> bytes: | ||
| return json.dumps( | ||
| { | ||
| "topic": topic, | ||
| "payload": payload, | ||
| "correlation_id": src_event.get("correlation_id") or src_event.get("file_id"), | ||
| "source": "watch-folder-router", | ||
| "ts": datetime.now(timezone.utc).isoformat(), | ||
| } | ||
| ).encode() | ||
|
|
||
|
|
||
| def _transcribe(bucket: str, key: str) -> Dict[str, Any]: | ||
| resp = http.post( | ||
| f"{MEDIA_AUDIO_URL}/analyze", | ||
| json={"bucket": bucket, "key": key, "analysis_type": "transcription"}, | ||
| timeout=DIARIZE_TIMEOUT, | ||
| ) | ||
| resp.raise_for_status() | ||
| return resp.json() | ||
|
|
||
|
|
||
| async def _handle_file_added(msg) -> None: | ||
| try: | ||
| event = json.loads(msg.data.decode()) | ||
| except Exception: # noqa: BLE001 | ||
| logger.exception("unparseable ingest.file.added event") | ||
| routed_total.labels(route="unknown", status="rejected").inc() | ||
| return | ||
|
|
||
| bucket = event.get("bucket") | ||
| key = event.get("key") | ||
| mime = event.get("mime_type", "") | ||
| if not bucket or not key: | ||
| logger.warning("event missing bucket/key: %s", event) | ||
| routed_total.labels(route="unknown", status="rejected").inc() | ||
| return | ||
|
|
||
| route = _classify(mime, key) | ||
| base = { | ||
| "file_id": event.get("file_id"), | ||
| "bucket": bucket, | ||
| "key": key, | ||
| "mime_type": mime, | ||
| "namespace": event.get("namespace"), | ||
| "room_id": event.get("room_id"), | ||
| "persona": event.get("persona"), | ||
| "uploader": event.get("uploader"), | ||
| } | ||
|
|
||
| try: | ||
| if route == "media": | ||
| # Transcription is the slow step; run it off the event-loop thread. | ||
| body = await asyncio.to_thread(_transcribe, bucket, key) | ||
| if body.get("error"): | ||
| raise RuntimeError(f"media-audio error: {body['error']}") | ||
| text = body.get("text") or "" | ||
| payload = { | ||
| **base, | ||
| "text": text, | ||
| "chunks": body.get("chunks", []), | ||
| "model": body.get("model"), | ||
| "char_count": len(str(text)), | ||
| } | ||
| await _publish(SUBJECT_TRANSCRIPT, payload, event) | ||
| routed_total.labels(route="media", status="ok").inc() | ||
| logger.info("routed media %s -> transcript.ready (%d chars)", key, len(str(text))) | ||
| elif route == "document": | ||
| # pdf-ingest / langextract own extraction; router only announces. | ||
| await _publish(SUBJECT_DOCUMENT, base, event) | ||
| routed_total.labels(route="document", status="ok").inc() | ||
| logger.info("routed document %s -> document.ready", key) | ||
| else: | ||
| await _publish(SUBJECT_TEXT, base, event) | ||
| routed_total.labels(route="text", status="ok").inc() | ||
| logger.info("routed text %s -> text.ready", key) | ||
| except Exception: # noqa: BLE001 | ||
| logger.exception("routing failed for %s (route=%s)", key, route) | ||
| routed_total.labels(route=route, status="error").inc() | ||
|
|
||
|
|
||
| async def _publish(topic: str, payload: Dict[str, Any], src_event: Dict[str, Any]) -> None: | ||
| if _nc is None: | ||
| logger.warning("NATS not connected; dropping %s for %s", topic, payload.get("key")) | ||
| return | ||
| await _nc.publish(topic, _envelope(topic, payload, src_event)) | ||
|
|
||
|
|
||
| # ── Lifespan / app ────────────────────────────────────────────────────────── | ||
| @asynccontextmanager | ||
| async def lifespan(app: FastAPI): | ||
| global _nc | ||
| try: | ||
| import nats | ||
|
|
||
| _nc = await nats.connect(NATS_URL, max_reconnect_attempts=-1) | ||
| await _nc.subscribe(SUBJECT_IN, cb=_handle_file_added) | ||
| logger.info("NATS connected; subscribed %s", SUBJECT_IN) | ||
| except Exception: # noqa: BLE001 | ||
| logger.exception("NATS connect failed — router is idle until NATS is reachable") | ||
| _nc = None | ||
| yield | ||
| if _nc is not None: | ||
| await _nc.drain() | ||
|
|
||
|
|
||
| app = FastAPI(title="watch-folder-router", lifespan=lifespan) | ||
|
|
||
|
|
||
| @app.get("/healthz") | ||
| async def healthz(): | ||
| return { | ||
| "status": "healthy", | ||
| "service": "watch-folder-router", | ||
| "nats_connected": _nc is not None and not _nc.is_closed, | ||
| "media_audio_url": MEDIA_AUDIO_URL, | ||
| "subscribes": SUBJECT_IN, | ||
| "publishes": [SUBJECT_TRANSCRIPT, SUBJECT_DOCUMENT, SUBJECT_TEXT], | ||
| } | ||
|
|
||
|
|
||
| @app.get("/metrics") | ||
| async def metrics(): | ||
| return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add function docstrings to meet the coverage requirement.
Only _classify has a function docstring. The other seven functions do not meet the required 80% docstring coverage for new Python code.
As per coding guidelines, “New Python code must maintain at least 80% docstring coverage.”
🧰 Tools
🪛 ast-grep (0.45.1)
[info] 84-92: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
{
"topic": topic,
"payload": payload,
"correlation_id": src_event.get("correlation_id") or src_event.get("file_id"),
"source": "watch-folder-router",
"ts": datetime.now(timezone.utc).isoformat(),
}
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/services/watch-folder-router/worker.py` around lines 84 - 207, Add
concise docstrings to the undocumented functions in the changed module,
including _envelope, _transcribe, _handle_file_added, _publish, lifespan,
healthz, and metrics. Keep the existing behavior unchanged and ensure at least
80% of the new functions have docstrings.
Source: Coding guidelines
| async def _handle_file_added(msg) -> None: | ||
| try: | ||
| event = json.loads(msg.data.decode()) | ||
| except Exception: # noqa: BLE001 | ||
| logger.exception("unparseable ingest.file.added event") | ||
| routed_total.labels(route="unknown", status="rejected").inc() | ||
| return | ||
|
|
||
| bucket = event.get("bucket") | ||
| key = event.get("key") | ||
| mime = event.get("mime_type", "") | ||
| if not bucket or not key: | ||
| logger.warning("event missing bucket/key: %s", event) | ||
| routed_total.labels(route="unknown", status="rejected").inc() | ||
| return |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline pmoves/services/common/events.py --items all
rg -n -C 4 --glob '*.py' \
'services\.common\.events|validate.*event|ingest\.file\.added\.v1' pmoves/servicesRepository: POWERFULMOVES/PMOVES.AI
Length of output: 26718
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- worker structure ---'
ast-grep outline pmoves/services/watch-folder-router/worker.py --items all
printf '%s\n' '--- worker implementation ---'
cat -n pmoves/services/watch-folder-router/worker.py | sed -n '1,240p'
printf '%s\n' '--- shared event helper ---'
cat -n pmoves/services/common/events.py | sed -n '1,130p'
printf '%s\n' '--- relevant schemas and references ---'
find pmoves -type f \( -iname '*ingest*' -o -iname '*file*added*' -o -iname '*.json' \) -print | sort | head -200
rg -n -C 3 'file\.added|bucket|mime_type|checksum|size_bytes' pmoves --glob '*.json' --glob '*.py' | head -300Repository: POWERFULMOVES/PMOVES.AI
Length of output: 50379
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- ingest.file.added.v1 schema ---'
cat -n pmoves/contracts/schemas/ingest/file-added.v1.schema.json
printf '%s\n' '--- downstream schemas ---'
cat -n pmoves/contracts/schemas/ingest/document-ready.v1.schema.json
cat -n pmoves/contracts/schemas/ingest/transcript-ready.v1.schema.json
printf '%s\n' '--- topic registration ---'
sed -n '258,280p' pmoves/contracts/topics.json
printf '%s\n' '--- deterministic schema probe ---'
python3 - <<'PY'
import json
from pathlib import Path
try:
from jsonschema import Draft202012Validator
except Exception as exc:
print(f"jsonschema unavailable: {exc}")
raise SystemExit(0)
root = Path("pmoves")
topics = json.loads((root / "contracts/topics.json").read_text())
schema = json.loads((root / "contracts" / topics["topics"]["ingest.file.added.v1"]["schema"]).read_text())
validator = Draft202012Validator(schema)
cases = {
"minimal bucket/key": {"bucket": "assets", "key": "a.txt"},
"valid-looking event": {
"file_id": "f1", "bucket": "assets", "key": "a.txt",
"mime_type": "text/plain", "namespace": "ns",
"checksum": "abc", "size_bytes": 1, "meta": {},
},
"wrong optional types": {
"bucket": "assets", "key": "a.txt", "mime_type": 7,
"checksum": [], "size_bytes": "1", "meta": "bad",
},
"wrong required types": {"bucket": 4, "key": None},
}
for name, payload in cases.items():
errors = sorted(validator.iter_errors(payload), key=lambda e: list(e.path))
print(name, "VALID" if not errors else "INVALID")
for error in errors:
print(" ", list(error.path), error.message)
PYRepository: POWERFULMOVES/PMOVES.AI
Length of output: 4461
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- producers of ingest.file.added.v1 ---'
rg -n -C 10 'ingest\.file\.added\.v1' pmoves/services pmoves --glob '*.py' --glob '!pmoves/services/watch-folder-router/worker.py' | head -240
printf '%s\n' '--- all file-added samples and contract references ---'
rg -n -C 4 'file_id|\"uri\"|\"kind\"|\"bucket\"|\"key\"|\"mime_type\"' \
pmoves/contracts/samples pmoves/services --glob '*.json' --glob '*.py' \
| rg -B 4 -A 8 'file_id|\"uri\"|\"kind\"|\"bucket\"|\"key\"|\"mime_type\"' \
| head -320
printf '%s\n' '--- worker publication and common helper usage ---'
rg -n -C 3 'def _publish|_envelope|validate_payload|from .*common\.events' \
pmoves/services/watch-folder-router pmoves/services/commonRepository: POWERFULMOVES/PMOVES.AI
Length of output: 43794
Validate ingest.file.added.v1 before routing and publishing.
The schema requires file_id, uri, and kind, but _handle_file_added checks only bucket and key. Align the worker with pmoves/contracts/schemas/ingest/file-added.v1.schema.json, then call services.common.events.validate_payload() before classification and publication. Validate downstream payloads before _publish as well.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/services/watch-folder-router/worker.py` around lines 106 - 120, Update
_handle_file_added to require all ingest.file.added.v1 fields defined by
file-added.v1.schema.json, including file_id, uri, and kind, before routing.
Call services.common.events.validate_payload() on the incoming event before
classification/publication, and validate each downstream payload immediately
before _publish.
Source: Coding guidelines
| bucket = event.get("bucket") | ||
| key = event.get("key") | ||
| mime = event.get("mime_type", "") | ||
| if not bucket or not key: | ||
| logger.warning("event missing bucket/key: %s", event) | ||
| routed_total.labels(route="unknown", status="rejected").inc() | ||
| return |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not log the complete rejected event.
Line 118 logs all unvalidated fields. The event can contain uploader and arbitrary metadata. Log a safe identifier or field names instead of the complete payload.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/services/watch-folder-router/worker.py` around lines 114 - 120, Update
the rejected-event warning in the worker’s bucket/key validation block to stop
logging the complete event payload. Log only safe identifiers such as the
presence or values of validated fields, or the event’s field names, while
preserving the rejection metric and early return.
| try: | ||
| import nats | ||
|
|
||
| _nc = await nats.connect(NATS_URL, max_reconnect_attempts=-1) | ||
| await _nc.subscribe(SUBJECT_IN, cb=_handle_file_added) | ||
| logger.info("NATS connected; subscribed %s", SUBJECT_IN) | ||
| except Exception: # noqa: BLE001 | ||
| logger.exception("NATS connect failed — router is idle until NATS is reachable") | ||
| _nc = None | ||
| yield |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Do not accept traffic while the router has no NATS subscription.
When the initial NATS connection fails, the lifespan handler sets _nc to None and completes startup. _publish() then drops every routed event, while /healthz still returns HTTP 200 and "healthy". Retry the initial connection or fail startup. Report an unhealthy status while NATS is unavailable. Add a NATS health dependency in both Compose definitions.
pmoves/services/watch-folder-router/worker.py#L176-L185: retry the initial connection and subscription, or re-raise to fail startup.pmoves/services/watch-folder-router/worker.py#L193-L202: return an unhealthy HTTP status when_ncis absent or closed.pmoves/docker-compose.yml#L2283-L2302: adddepends_on: nats: condition: service_healthy.pmoves/docker-compose.workers.yml#L96-L115: regenerate the overlay with the same dependency.
📍 Affects 3 files
pmoves/services/watch-folder-router/worker.py#L176-L185(this comment)pmoves/services/watch-folder-router/worker.py#L193-L202pmoves/docker-compose.yml#L2283-L2302pmoves/docker-compose.workers.yml#L96-L115
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pmoves/services/watch-folder-router/worker.py` around lines 176 - 185, Ensure
the router never completes startup without a usable NATS subscription: update
the lifespan connection flow in
pmoves/services/watch-folder-router/worker.py:176-185 to retry the initial
connection/subscription or re-raise the failure; update the /healthz handler at
pmoves/services/watch-folder-router/worker.py:193-202 to return an unhealthy
HTTP status when _nc is absent or closed; add depends_on with nats
service_healthy to the watch-folder-router service in
pmoves/docker-compose.yml:2283-2302 and regenerate the equivalent dependency in
pmoves/docker-compose.workers.yml:96-115.
…ded, A0 bundle dispatched (signed) (#2536) Signed CHIT trail (HMAC-SHA256, kid chit-signing-v01). Session: #2535 watch-folder-router (SEAP Bud 1), NATS accounts/leaf-topology trio merged (#2492/93/96), #2532/33/34 admin-merged (9 GHCR builds unblocked), A0 review-refactor bundle dispatched (context 6RQ9v5PP) + harvested with caveats. Co-authored-by: Mavis <Mavis@pmoves.local> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 participant
SEAP ingestion Bud 1 — the first bud of the watch-folder constellation (
SEAP_WATCHFOLDER_NATS_CONSTELLATION_2026-08-12, operator-directed).What
A thin NATS router that consumes
ingest.file.added.v1and routes bymime_type, retiring the manualmedia-audio /analyzecalls that every hand-driven ingest made:audio/*,video/*→ media-audio/analyze(transcription) →ingest.transcript.ready.v1application/pdf+ office docs →ingest.document.ready.v1(pdf-ingest/langextract consume)ingest.text.ready.v1(extract-worker embeds)MIME is trusted first; file extension is the fallback when MIME is generic (octet-stream).
Pattern
Follows the established templates: extract-worker (pure NATS consumer, no client HTTP) + voice-sampler (thin orchestrator, MinIO/media-audio wiring). No models, no GPU.
:8125health/metrics only. NATS default aligned to the standard authed URL. All three publish subjects already exist innats-subjects.md.Verified live
Built + started healthy on 5090 (NATS connected). Routing smoke test: synthetic PDF + text-plain
ingest.file.added.v1events routed correctly todocument.readyandtext.readywith the standard envelope — SMOKE PASS.Constellation context (not in this PR)
transcript-harvester: consume the*.ready.v1family → Hi-RAG + Open Notebook + JuiceFS + CGParchon-harvest: decide what deserves an Archon work order🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation