UN-3636 [DEV] Hermetic LLM mock hook for execute-path critical tests

tests/compose/docker-compose.test.yaml

@@ -26,3 +26,14 @@ services:
     image: unstract/runner:${UNSTRACT_TEST_VERSION:-latest}
     environment:
       - ENVIRONMENT=test
+
+  # LLM completions run inside the worker-unified image; passing the mock env
+  # here lets execute-path e2e tests run hermetically (no real LLM/secret).
+  # Empty unless the rig/CI sets it, so production-like runs are unaffected.
+  worker-executor-v2:
+    environment:
+      - UNSTRACT_LLM_MOCK_RESPONSE=${UNSTRACT_LLM_MOCK_RESPONSE:-}
+
+  worker-file-processing-v2:
+    environment:
+      - UNSTRACT_LLM_MOCK_RESPONSE=${UNSTRACT_LLM_MOCK_RESPONSE:-}

tests/critical_paths.yaml

@@ -43,6 +43,7 @@ paths:
   - id: workflow-create-execute
     description: "Create a workflow, configure source+destination, execute, poll, fetch result."
     covered_by: [e2e-workflow]
+    proof: marker
 
   - id: api-deployment-provision
     description: "Deploying a workflow as an API mints a usable key and a resolvable endpoint."
@@ -57,6 +58,7 @@ paths:
   - id: api-deployment-run
     description: "Deploy a workflow as an API, POST a document, receive structured JSON."
     covered_by: [e2e-api-deployment]
+    proof: marker
 
   - id: prompt-studio-author
     description: "Create a Prompt Studio project and add a prompt to it."
@@ -83,7 +85,8 @@ paths:
 
   - id: usage-token-tracking
     description: "Per-execution token usage is recorded and retrievable."
-    covered_by: []   # gap
+    covered_by: [e2e-api-deployment]
+    proof: marker
 
   - id: workflow-execution-fan-out
     description: "Multi-file workflow execution fans out to file-processing workers and rejoins."

tests/e2e/api_deployment/test_api_deployment_run.py

@@ -0,0 +1,77 @@
+"""E2E: deploy a workflow as an API, POST a document, get structured JSON back.
+
+Covers two critical paths in one hermetic run (the LLM is mocked via
+UNSTRACT_LLM_MOCK_RESPONSE, so no real provider/secret is touched):
+
+  • api-deployment-run   — the public deployment endpoint executes and returns
+                           the mocked answer as structured JSON.
+  • usage-token-tracking — per-execution token usage is recorded and returned
+                           (litellm stamps a fixed 10/20/30 on the mock).
+
+Synchronous execution (timeout > 0 blocks) so the result comes back on the POST
+itself — no polling and no out-of-band result store to read.
+"""
+
+from __future__ import annotations
+
+import io
+import uuid
+
+import pytest
+
+from tests.e2e.conftest import ProvisionedWorkflow
+
+pytestmark = [pytest.mark.e2e, pytest.mark.critical]
+
+
+@pytest.mark.critical_path("api-deployment-run")
+@pytest.mark.critical_path("usage-token-tracking")
+def test_api_deployment_returns_mocked_answer(
+    provisioned_workflow: ProvisionedWorkflow, llm_mock_response: str
+) -> None:
+    pw = provisioned_workflow
+    session = pw.session
+    csrf = {"X-CSRFToken": session.cookies.get("csrftoken", "")}
+
+    api_name = f"e2edep{uuid.uuid4().hex[:8]}"
+    resp = session.post(
+        f"{pw.prefix}/api/deployment/",
+        headers=csrf,
+        json={
+            "workflow": pw.workflow_id,
+            "display_name": f"e2e {api_name}",
+            "description": "e2e api deployment",
+            "api_name": api_name,
+            "is_active": True,
+        },
+        timeout=30,
+    )
+    assert resp.status_code == 201, f"deploy: {resp.text}"
+    deployment = resp.json()
+    api_key = deployment["api_key"]
+    endpoint = deployment["api_endpoint"]
+    exec_url = endpoint if endpoint.startswith("http") else f"{pw.base}/{endpoint.lstrip('/')}"
+
+    document = io.BytesIO(b"Hello invoice 123. This is a test document about widgets.")
+    resp = session.post(
+        exec_url,
+        headers={"Authorization": f"Bearer {api_key}"},
+        data={"timeout": 300, "include_metadata": True},
+        files={"files": ("probe.txt", document, "text/plain")},
+        timeout=310,
+    )
+    assert resp.status_code == 200, f"execute: HTTP {resp.status_code}: {resp.text}"
+    message = resp.json()["message"]
+    assert message["execution_status"] == "COMPLETED", message
+
+    file_result = message["result"][0]
+    assert file_result["status"] == "Success", file_result
+    # api-deployment-run: the mocked completion surfaces as the prompt's answer.
+    assert file_result["result"]["output"]["answer"] == llm_mock_response
+
+    # usage-token-tracking: litellm stamps a deterministic 10/20/30 on the mock,
+    # recorded per-execution and returned under the file's metadata.
+    usage = file_result["result"]["metadata"]["extraction_llm"][0]
+    assert usage["input_tokens"] == 10, usage
+    assert usage["output_tokens"] == 20, usage
+    assert usage["total_tokens"] == 30, usage

tests/e2e/conftest.py

@@ -9,11 +9,24 @@
 from __future__ import annotations
 
 import os
+import uuid
+from dataclasses import dataclass
 
 import pytest
 import requests
 
-from tests.rig.runtime import PlatformEndpoints
+from tests.rig.runtime import LLM_MOCK_RESPONSE_ENV, PlatformEndpoints
+
+# Static adapter registry ids from unstract.sdk1 (see adapters/*/). The NoOp
+# x2text/vectordb return canned output so extraction/indexing need no real
+# service; the OpenAI LLM/embedding rows never reach a provider — the LLM is
+# mocked via UNSTRACT_LLM_MOCK_RESPONSE and, with chunk_size=0, embedding +
+# vectordb are never invoked. Adapter create does not validate connectivity,
+# so fake creds persist.
+_LLM_ADAPTER = "openai|502ecf49-e47c-445c-9907-6d4b90c5cd17"
+_EMBED_ADAPTER = "openai|717a0b0e-3bbc-41dc-9f0c-5689437a1151"
+_VECTORDB_ADAPTER = "noOpVectorDb|ca4d6056-4971-4bc8-97e3-9e36290b5bc0"
+_X2TEXT_ADAPTER = "noOpX2text|mp66d1op-7100-d340-9101-846fc7115676"
 
 
 @pytest.fixture(scope="session")
@@ -69,3 +82,192 @@ def authed_session(platform: PlatformEndpoints) -> requests.Session:
     )
     resp.raise_for_status()
     return session
+
+
+@pytest.fixture(scope="session")
+def llm_mock_response() -> str:
+    """The exact string the workers were told to return for every completion.
+
+    The rig sets this when it boots the platform (ComposeRuntime); a manual/local
+    run must export it to match what the workers got. Absent it, execute-path
+    tests can't assert a deterministic answer, so skip rather than guess.
+    """
+    value = os.environ.get(LLM_MOCK_RESPONSE_ENV)
+    if not value:
+        pytest.skip(
+            f"{LLM_MOCK_RESPONSE_ENV} not set — execute-path e2e needs the LLM "
+            "mock; the rig sets it when it boots the platform."
+        )
+    return value
+
+
+@dataclass(frozen=True)
+class ProvisionedWorkflow:
+    """Handles for a hermetic, execute-ready workflow (one Prompt Studio tool)."""
+
+    session: requests.Session
+    base: str  # backend root, e.g. http://localhost:8000
+    prefix: str  # tenant-scoped API root: {base}/api/v1/unstract/{org_id}
+    org_id: str
+    workflow_id: str
+    tool_id: str
+    prompt_registry_id: str
+
+
+def _org_id(session: requests.Session, base: str) -> str:
+    orgs = session.get(f"{base}/api/v1/organization", timeout=10)
+    orgs.raise_for_status()
+    return orgs.json()["organizations"][0]["id"]
+
+
+def _post(session: requests.Session, url: str, **kw: object) -> requests.Response:
+    headers = dict(kw.pop("headers", {}))
+    headers["X-CSRFToken"] = session.cookies.get("csrftoken", "")
+    return session.post(url, headers=headers, timeout=60, **kw)
+
+
+def _patch(session: requests.Session, url: str, **kw: object) -> requests.Response:
+    headers = dict(kw.pop("headers", {}))
+    headers["X-CSRFToken"] = session.cookies.get("csrftoken", "")
+    return session.patch(url, headers=headers, timeout=60, **kw)
+
+
+@pytest.fixture(scope="session")
+def provisioned_workflow(
+    platform: PlatformEndpoints, authed_session: requests.Session
+) -> ProvisionedWorkflow:
+    """Stand up an API workflow backed by a single Prompt Studio tool.
+
+    Chain (all HTTP, mirrors the app): create 4 adapters -> Prompt Studio project
+    (auto-creates a default profile) -> pin adapters + chunk_size=0 on that
+    profile -> add one text prompt -> export (mints a PromptStudioRegistry row) ->
+    workflow -> point both endpoints at API -> attach the exported tool. The
+    resulting workflow executes hermetically because the LLM is mocked and, with
+    chunk_size=0, embedding + vectordb are skipped.
+    """
+    s = authed_session
+    base = platform.backend_url.rstrip("/")
+    org_id = _org_id(s, base)
+    prefix = f"{base}/api/v1/unstract/{org_id}"
+    sfx = uuid.uuid4().hex[:8]  # adapter/tool names are unique per org
+
+    def create_adapter(adapter_id: str, adapter_type: str, metadata: dict) -> str:
+        name = f"e2e-{adapter_type.lower()}-{sfx}"
+        resp = _post(
+            s,
+            f"{prefix}/adapter/",
+            json={
+                "adapter_id": adapter_id,
+                "adapter_name": name,
+                "adapter_type": adapter_type,
+                "adapter_metadata": {"adapter_name": name, **metadata},
+            },
+        )
+        assert resp.status_code == 201, f"adapter {adapter_type}: {resp.text}"
+        return resp.json()["id"]
+
+    # api_base is required by the SDK's OpenAI params even though completion is
+    # mocked (validation runs before the litellm call).
+    llm_id = create_adapter(
+        _LLM_ADAPTER,
+        "LLM",
+        {"api_key": "sk-test", "model": "gpt-4o", "api_base": "https://api.openai.com/v1"},
+    )
+    embed_id = create_adapter(
+        _EMBED_ADAPTER, "EMBEDDING", {"api_key": "sk-test", "model": "text-embedding-3-small"}
+    )
+    vdb_id = create_adapter(_VECTORDB_ADAPTER, "VECTOR_DB", {"wait_time": 0})
+    x2t_id = create_adapter(_X2TEXT_ADAPTER, "X2TEXT", {"wait_time": 0})
+
+    resp = _post(
+        s,
+        f"{prefix}/prompt-studio/",
+        json={"tool_name": f"e2e-{sfx}", "description": "e2e execute", "author": "e2e"},
+    )
+    assert resp.status_code == 201, f"prompt-studio project: {resp.text}"
+    tool_id = resp.json()["tool_id"]
+
+    # Project creation auto-makes the default profile; creating a second yields
+    # two is_default rows and breaks export. Patch the auto one instead.
+    resp = s.get(f"{prefix}/prompt-studio/prompt-studio-profile/{tool_id}/", timeout=30)
+    resp.raise_for_status()
+    profile_id = next(p["profile_id"] for p in resp.json() if p.get("is_default"))
+    resp = _patch(
+        s,
+        f"{prefix}/prompt-studio/profile-manager/{profile_id}/",
+        json={
+            "llm": llm_id,
+            "embedding_model": embed_id,
+            "vector_store": vdb_id,
+            "x2text": x2t_id,
+            "chunk_size": 0,
+            "chunk_overlap": 0,
+        },
+    )
+    assert resp.status_code == 200, f"profile patch: {resp.text}"
+
+    resp = _post(
+        s,
+        f"{prefix}/prompt-studio/prompt-studio-prompt/{tool_id}/",
+        json={
+            "tool_id": tool_id,
+            "prompt_key": "answer",
+            "prompt": "What is this document about?",
+            "prompt_type": "PROMPT",
+            "enforce_type": "text",
+            "sequence_number": 1,
+            "active": True,
+            "profile_manager": profile_id,
+        },
+    )
+    assert resp.status_code == 201, f"add prompt: {resp.text}"
+
+    resp = _post(
+        s,
+        f"{prefix}/prompt-studio/export/{tool_id}",
+        json={"force_export": True, "is_shared_with_org": True},
+    )
+    assert resp.status_code == 200, f"export: {resp.text}"
+    # get_queryset returns None unless a filter arg is present -> pass custom_tool.
+    resp = s.get(
+        f"{prefix}/prompt-studio/registry/", params={"custom_tool": tool_id}, timeout=30
+    )
+    resp.raise_for_status()
+    regs = resp.json()
+    reg_list = regs if isinstance(regs, list) else regs.get("results", [])
+    assert reg_list, f"registry empty for tool {tool_id}"
+    prompt_registry_id = reg_list[0]["prompt_registry_id"]
+
+    resp = _post(s, f"{prefix}/workflow/", json={"workflow_name": f"e2e-wf-{sfx}"})
+    assert resp.status_code == 201, f"create workflow: {resp.text}"
+    workflow_id = resp.json()["id"]
+
+    resp = s.get(f"{prefix}/workflow/endpoint/", params={"workflow": workflow_id}, timeout=30)
+    resp.raise_for_status()
+    eps = resp.json()
+    eps = eps if isinstance(eps, list) else eps.get("results", [])
+    for endpoint in eps:
+        if endpoint.get("workflow") == workflow_id:
+            resp = _patch(
+                s,
+                f"{prefix}/workflow/endpoint/{endpoint['id']}/",
+                json={"connection_type": "API"},
+            )
+            assert resp.status_code == 200, f"patch endpoint: {resp.text}"
+
+    resp = _post(
+        s,
+        f"{prefix}/tool_instance/",
+        json={"workflow_id": workflow_id, "tool_id": prompt_registry_id},
+    )
+    assert resp.status_code == 201, f"attach tool: {resp.text}"
+
+    return ProvisionedWorkflow(
+        session=s,
+        base=base,
+        prefix=prefix,
+        org_id=org_id,
+        workflow_id=workflow_id,
+        tool_id=tool_id,
+        prompt_registry_id=prompt_registry_id,
+    )