YoVinchen
ecdb10c37a
Some checks failed
verify / verify (push) Has been cancelled
Separate local parsing from model indexing, bind review decisions to immutable manifests, persist vectors behind active profiles, and expose retrieval, chat, evaluation, and document workflows through the React workbench. Constraint: Live Bailian authentication currently fails for all three configured capabilities Rejected: Direct upload-to-embedding flow | bypasses local review and manifest binding Confidence: high Scope-risk: broad Directive: Keep private-data deployment blocked until authentication, RBAC, and separate database roles land Tested: make verify; fresh and replay Docker document smoke; worker recovery smoke; frozen synthetic evaluation; migration 0003-0004 roundtrip Not-tested: Successful live Bailian calls, OCR, real multi-user authorization
337 lines
12 KiB
Python
337 lines
12 KiB
Python
"""Reproducible HTTP smoke for upload -> review -> vector -> retrieval."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
from app.core.demo_identity import BAILIAN_KNOWLEDGE_BASE_ID, KNOWLEDGE_BASE_ID
|
|
|
|
|
|
class DocumentPipelineSmokeError(RuntimeError):
|
|
"""A safe smoke failure without source text, paths, or response bodies."""
|
|
|
|
|
|
def _request(
|
|
base_url: str,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
body: dict[str, object] | None = None,
|
|
content: bytes | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
request_headers = {"Accept": "application/json", **(headers or {})}
|
|
payload: bytes | None = None
|
|
if body is not None:
|
|
payload = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode()
|
|
request_headers["Content-Type"] = "application/json"
|
|
elif content is not None:
|
|
payload = content
|
|
request_headers["Content-Type"] = "application/octet-stream"
|
|
request = Request( # noqa: S310 - base URL is operator-configured HTTP(S)
|
|
f"{base_url.rstrip('/')}{path}",
|
|
data=payload,
|
|
headers=request_headers,
|
|
method=method,
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=15) as response: # noqa: S310 - configured local endpoint
|
|
parsed = json.loads(response.read())
|
|
except HTTPError as exc:
|
|
code = "UNKNOWN"
|
|
try:
|
|
problem = json.loads(exc.read())
|
|
if isinstance(problem, dict) and isinstance(problem.get("code"), str):
|
|
code = problem["code"]
|
|
except (OSError, ValueError, TypeError):
|
|
pass
|
|
raise DocumentPipelineSmokeError(f"HTTP {exc.code} ({code}) for {method} {path}") from None
|
|
except (URLError, TimeoutError, OSError, ValueError, TypeError):
|
|
raise DocumentPipelineSmokeError(f"request failed for {method} {path}") from None
|
|
if not isinstance(parsed, dict):
|
|
raise DocumentPipelineSmokeError(f"invalid response for {method} {path}")
|
|
return cast(dict[str, Any], parsed)
|
|
|
|
|
|
def _wait_job(base_url: str, job_id: str, *, timeout_seconds: float) -> dict[str, Any]:
|
|
deadline = time.monotonic() + timeout_seconds
|
|
while time.monotonic() < deadline:
|
|
job = _request(base_url, "GET", f"/api/v1/document-jobs/{job_id}")
|
|
status = job.get("status")
|
|
if status == "SUCCEEDED":
|
|
return job
|
|
if status in {"FAILED", "CANCELLED"}:
|
|
code = job.get("last_error_code")
|
|
safe_code = code if isinstance(code, str) else "UNKNOWN"
|
|
raise DocumentPipelineSmokeError(f"job terminated with {status} ({safe_code})")
|
|
time.sleep(0.25)
|
|
raise DocumentPipelineSmokeError("job polling timed out")
|
|
|
|
|
|
def _wait_document_ready(
|
|
base_url: str,
|
|
document_id: str,
|
|
*,
|
|
timeout_seconds: float,
|
|
) -> dict[str, Any]:
|
|
deadline = time.monotonic() + timeout_seconds
|
|
while time.monotonic() < deadline:
|
|
detail = _request(base_url, "GET", f"/api/v1/documents/{document_id}")
|
|
document = detail.get("document")
|
|
if isinstance(document, dict) and document.get("status") == "READY":
|
|
return detail
|
|
if isinstance(document, dict) and document.get("status") in {"FAILED", "REJECTED"}:
|
|
raise DocumentPipelineSmokeError("document reached a non-ready terminal state")
|
|
time.sleep(0.25)
|
|
raise DocumentPipelineSmokeError("document activation polling timed out")
|
|
|
|
|
|
def run_smoke(
|
|
*,
|
|
base_url: str,
|
|
sample_path: Path,
|
|
timeout_seconds: float = 90.0,
|
|
run_id: uuid.UUID | None = None,
|
|
knowledge_base_id: uuid.UUID = KNOWLEDGE_BASE_ID,
|
|
) -> dict[str, object]:
|
|
endpoint = urlsplit(base_url)
|
|
if (
|
|
endpoint.scheme not in {"http", "https"}
|
|
or not endpoint.hostname
|
|
or endpoint.username is not None
|
|
or endpoint.password is not None
|
|
):
|
|
raise DocumentPipelineSmokeError("RAG base URL must be credential-free HTTP(S)")
|
|
try:
|
|
sample_content = sample_path.read_bytes()
|
|
except OSError:
|
|
raise DocumentPipelineSmokeError("synthetic upload sample is unavailable") from None
|
|
if not sample_content or len(sample_content) > 1024 * 1024:
|
|
raise DocumentPipelineSmokeError("synthetic upload sample has an invalid size")
|
|
smoke_run_id = run_id or uuid.uuid4()
|
|
content = sample_content + f"\n\nSynthetic smoke run: {smoke_run_id}\n".encode()
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
key = uuid.uuid5(uuid.NAMESPACE_URL, f"geological-rag-document-smoke:{digest}")
|
|
filename = f"upload_demo-{smoke_run_id.hex[:12]}.md"
|
|
declaration_body: dict[str, object] = {
|
|
"filename": filename,
|
|
"declared_mime_type": "text/markdown",
|
|
"expected_size": len(content),
|
|
"expected_sha256": digest,
|
|
}
|
|
declared = _request(
|
|
base_url,
|
|
"POST",
|
|
"/api/v1/document-uploads",
|
|
headers={"Idempotency-Key": str(key)},
|
|
body=declaration_body,
|
|
)
|
|
if declared.get("replayed") is not False:
|
|
raise DocumentPipelineSmokeError("fresh upload declaration was unexpectedly replayed")
|
|
upload_id = _required_uuid_text(declared, "id")
|
|
_request(
|
|
base_url,
|
|
"PUT",
|
|
f"/api/v1/document-uploads/{upload_id}/content",
|
|
content=content,
|
|
)
|
|
completed = _request(
|
|
base_url,
|
|
"POST",
|
|
f"/api/v1/document-uploads/{upload_id}/complete",
|
|
)
|
|
document = _required_mapping(completed, "document")
|
|
document_id = _required_uuid_text(document, "id")
|
|
parse_job = _required_mapping(completed, "job")
|
|
parse_job_id = _required_uuid_text(parse_job, "id")
|
|
parsed = _wait_job(base_url, parse_job_id, timeout_seconds=timeout_seconds)
|
|
if parsed.get("stage") != "LOCAL_PARSED_PENDING_CLOUD_REVIEW":
|
|
raise DocumentPipelineSmokeError("parse job did not reach the review stage")
|
|
|
|
review = _request(
|
|
base_url,
|
|
"GET",
|
|
f"/api/v1/documents/{document_id}/review-bundle?after_ordinal=-1&limit=100",
|
|
)
|
|
version = _required_mapping(review, "version")
|
|
version_id = _required_uuid_text(version, "id")
|
|
review_state = version.get("review_state")
|
|
if review_state == "LOCAL_PARSED_PENDING_CLOUD_REVIEW":
|
|
manifest = _required_hash(version, "outbound_manifest_sha256")
|
|
revision = version.get("review_revision")
|
|
if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0:
|
|
raise DocumentPipelineSmokeError("review revision is invalid")
|
|
decision = _request(
|
|
base_url,
|
|
"POST",
|
|
f"/api/v1/documents/{document_id}/review-decisions",
|
|
body={
|
|
"decision": "APPROVE",
|
|
"reason_code": "SYNTHETIC_REVIEW_APPROVED",
|
|
"expected_revision": revision,
|
|
"outbound_manifest_sha256": manifest,
|
|
},
|
|
)
|
|
embedding_job = _required_mapping(decision, "job")
|
|
embedding_job_id = _required_uuid_text(embedding_job, "id")
|
|
_wait_job(base_url, embedding_job_id, timeout_seconds=timeout_seconds)
|
|
elif review_state != "CLOUD_APPROVED":
|
|
raise DocumentPipelineSmokeError("document version is not eligible for indexing")
|
|
|
|
ready = _wait_document_ready(
|
|
base_url,
|
|
document_id,
|
|
timeout_seconds=timeout_seconds,
|
|
)
|
|
ready_document = _required_mapping(ready, "document")
|
|
if ready_document.get("active_version_id") != version_id:
|
|
raise DocumentPipelineSmokeError("ready document did not activate the reviewed version")
|
|
|
|
retrieval = _request(
|
|
base_url,
|
|
"POST",
|
|
"/api/v1/retrieval/search",
|
|
body={
|
|
"knowledge_base_id": str(knowledge_base_id),
|
|
"query": "海岳示范区萤石矿需要哪些综合找矿标志？",
|
|
"vector_top_k": 50,
|
|
"rerank_top_n": 10,
|
|
},
|
|
)
|
|
results = retrieval.get("results")
|
|
if not isinstance(results, list):
|
|
raise DocumentPipelineSmokeError("retrieval result is invalid")
|
|
match = next(
|
|
(
|
|
item
|
|
for item in results
|
|
if isinstance(item, dict) and item.get("document_id") == document_id
|
|
),
|
|
None,
|
|
)
|
|
if match is None:
|
|
raise DocumentPipelineSmokeError("uploaded document was not retrieved")
|
|
|
|
replayed = _request(
|
|
base_url,
|
|
"POST",
|
|
"/api/v1/document-uploads",
|
|
headers={"Idempotency-Key": str(key)},
|
|
body=declaration_body,
|
|
)
|
|
if replayed.get("replayed") is not True or _required_uuid_text(replayed, "id") != upload_id:
|
|
raise DocumentPipelineSmokeError("upload declaration replay contract failed")
|
|
_request(
|
|
base_url,
|
|
"PUT",
|
|
f"/api/v1/document-uploads/{upload_id}/content",
|
|
content=content,
|
|
)
|
|
replayed_completion = _request(
|
|
base_url,
|
|
"POST",
|
|
f"/api/v1/document-uploads/{upload_id}/complete",
|
|
)
|
|
replayed_document = _required_mapping(replayed_completion, "document")
|
|
replayed_job = _required_mapping(replayed_completion, "job")
|
|
if _required_uuid_text(replayed_document, "id") != document_id:
|
|
raise DocumentPipelineSmokeError("document identity changed during replay")
|
|
if _required_uuid_text(replayed_job, "id") != parse_job_id:
|
|
raise DocumentPipelineSmokeError("parse job identity changed during replay")
|
|
replayed_ready = _wait_document_ready(
|
|
base_url,
|
|
document_id,
|
|
timeout_seconds=timeout_seconds,
|
|
)
|
|
if _required_mapping(replayed_ready, "document").get("active_version_id") != version_id:
|
|
raise DocumentPipelineSmokeError("active version changed during replay")
|
|
return {
|
|
"status": "ok",
|
|
"run_id": str(smoke_run_id),
|
|
"knowledge_base_id": str(knowledge_base_id),
|
|
"document_id": document_id,
|
|
"document_version_id": version_id,
|
|
"parse_job_id": parse_job_id,
|
|
"document_status": ready_document.get("status"),
|
|
"parse_stage": parsed.get("stage"),
|
|
"retrieval_rank": match.get("rank"),
|
|
"citation_id": match.get("citation_id"),
|
|
"embedding_model": retrieval.get("embedding_model"),
|
|
"rerank_status": retrieval.get("rerank_status"),
|
|
"replay_confirmed": True,
|
|
}
|
|
|
|
|
|
def _required_mapping(value: dict[str, Any], key: str) -> dict[str, Any]:
|
|
item = value.get(key)
|
|
if not isinstance(item, dict):
|
|
raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
|
|
return cast(dict[str, Any], item)
|
|
|
|
|
|
def _required_uuid_text(value: dict[str, Any], key: str) -> str:
|
|
item = value.get(key)
|
|
if not isinstance(item, str):
|
|
raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
|
|
try:
|
|
parsed = uuid.UUID(item)
|
|
except ValueError:
|
|
raise DocumentPipelineSmokeError(f"response field is invalid: {key}") from None
|
|
if str(parsed) != item:
|
|
raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
|
|
return item
|
|
|
|
|
|
def _required_hash(value: dict[str, Any], key: str) -> str:
|
|
item = value.get(key)
|
|
if (
|
|
not isinstance(item, str)
|
|
or len(item) != 64
|
|
or any(character not in "0123456789abcdef" for character in item)
|
|
):
|
|
raise DocumentPipelineSmokeError(f"response field is invalid: {key}")
|
|
return item
|
|
|
|
|
|
def main() -> None:
|
|
base_url = os.getenv("RAG_BASE_URL", "http://127.0.0.1:8000")
|
|
sample_path = Path(os.getenv("RAG_UPLOAD_SAMPLE", "data/samples/public/upload_demo.md"))
|
|
namespace_mode = os.getenv("DOCUMENT_NAMESPACE_MODE", "fake").strip().lower()
|
|
if namespace_mode == "fake":
|
|
knowledge_base_id = KNOWLEDGE_BASE_ID
|
|
elif namespace_mode == "bailian":
|
|
knowledge_base_id = BAILIAN_KNOWLEDGE_BASE_ID
|
|
else:
|
|
sys.stdout.write(
|
|
json.dumps(
|
|
{"status": "failed", "error": "document namespace mode is invalid"},
|
|
sort_keys=True,
|
|
)
|
|
+ "\n"
|
|
)
|
|
raise SystemExit(1)
|
|
try:
|
|
result = run_smoke(
|
|
base_url=base_url,
|
|
sample_path=sample_path,
|
|
knowledge_base_id=knowledge_base_id,
|
|
)
|
|
except DocumentPipelineSmokeError as exc:
|
|
sys.stdout.write(json.dumps({"status": "failed", "error": str(exc)}, sort_keys=True) + "\n")
|
|
raise SystemExit(1) from None
|
|
sys.stdout.write(json.dumps(result, sort_keys=True) + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|