From 10ec0078b9ec7885edca72b139e474ef9fecffff Mon Sep 17 00:00:00 2001 From: Tyler Eastman Date: Fri, 21 Aug 2026 13:50:40 -0700 Subject: [PATCH 1/4] feat: add Nexus Service support (workflow-backed Operations) Rebased onto main with PR #15 (SDK 3.1.2 + attribution header) and PR #16 (extraction parameter) merged. Squashes the nexus spike into one clean commit. - YouDotComService exposes all six Activities as async Nexus Operations - contract.py holds the Nexus contract with SDK response models - workflows.py ships six thin backing Workflows with per-Activity ceilings - Idempotency key support for deduplication of Nexus StartOperation retries - Unit tests covering contract, handler, and sandbox registration - Integration tests for live Nexus round-trip (gated behind -m integration) Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- CHANGELOG.md | 15 + README.md | 97 ++++- examples/run_nexus_worker.py | 108 ++++++ src/youdotcom_temporal/contract.py | 153 ++++++++ src/youdotcom_temporal/nexus.py | 217 +++++++++++ src/youdotcom_temporal/workflows.py | 287 ++++++++++++++ tests/_nexus_caller_workflows.py | 148 ++++++++ tests/test_nexus.py | 480 ++++++++++++++++++++++++ tests/test_nexus_integration.py | 534 +++++++++++++++++++++++++++ tests/test_nexus_integration_live.py | 153 ++++++++ 10 files changed, 2191 insertions(+), 1 deletion(-) create mode 100644 examples/run_nexus_worker.py create mode 100644 src/youdotcom_temporal/contract.py create mode 100644 src/youdotcom_temporal/nexus.py create mode 100644 src/youdotcom_temporal/workflows.py create mode 100644 tests/_nexus_caller_workflows.py create mode 100644 tests/test_nexus.py create mode 100644 tests/test_nexus_integration.py create mode 100644 tests/test_nexus_integration_live.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 212e5fd..6970b36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- Nexus Service support: `youdotcom_temporal.nexus.YouDotComService` exposes all six Activities as asynchronous, workflow-backed Nexus Operations callable across Namespace boundaries through a Nexus Endpoint +- `youdotcom_temporal.contract` holds the Nexus contract: a request type per Operation carrying the Activity input plus an optional `idempotency_key`, and result types that are the You.com SDK's own response models, so callers get accurate nested types the SDK maintains. `contents` keeps a thin `ContentsOutput` envelope because the SDK returns a bare list. Callers must configure `temporalio.contrib.pydantic.pydantic_data_converter` +- Supplying `idempotency_key` makes the backing Workflow Id deterministic and starts it with `WorkflowIDConflictPolicy.USE_EXISTING`, so a retried Nexus StartOperation request attaches to the run already in flight instead of starting a second Workflow and paying for a second You.com call. Deduplication holds against a *running* Workflow; without a key, starts are not deduplicated +- `youdotcom_temporal.workflows` ships six thin backing Workflows (`YouSearchWorkflow`, `YouAnswerWorkflow`, `YouContentsWorkflow`, `YouResearchWorkflow`, `YouFinanceResearchWorkflow`, `YouResearchBackgroundWorkflow`), each wrapping its Activity with a per-Activity `start_to_close_timeout` carrying generous headroom, since a ceiling is a retry backstop rather than a latency target. `search` and `contents` are sized off the 60s per-URL maximum a caller can request via `crawl_timeout` +- `you_nexus_service_handler()` and `you_nexus_workflows()` helpers for Worker registration +- `examples/run_nexus_worker.py` handler-side Worker example +- Unit tests covering the Nexus Service contract and handler - `SearchInput.extraction` field accepts the SDK's new `extraction` object (`{"extraction_mode": "highlights" | "full_page", ...}`). When set, it takes priority over the deprecated `livecrawl` / `livecrawl_formats` fields and is passed to `you.search_async(extraction=...)` instead, avoiding the SDK's `ValueError` on dual-set. The legacy fields remain accepted for backward compatibility ### Changed @@ -15,6 +22,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The plugin now passes `app_name="youdotcom-temporal"` and `app_version=` to the `You(...)` constructor instead of mutating `client.sdk_configuration.user_agent` post-construction. Each outbound request carries `X-Client-Info: sdk; client=youdotcom-temporal/; ua=python/ httpx/`; the SDK's own `user-agent` stays as `youdotcom-python-sdk/`. The `_USER_AGENT` constant is removed - `youdotcom_research_background` now forwards `timeout_s` as-is (including `None`) to `research_and_wait_async` instead of substituting `120.0`. When `timeout_s` is `None`, the SDK derives an effort-based default (600s for standard, 14400s for frontier) via `_resolve_default_timeout()`. Previously the `120.0` fallback prevented that derivation, capping every effort tier at 120s +### Notes +- Nexus is an opt-in layer: importing `youdotcom_temporal.nexus` does not affect Activity-only users +- Every Operation is async/workflow-backed because Nexus sync operations have a 10-second handler deadline that several You.com calls exceed +- The handler Worker needs `YouPlugin` because it registers the Activities the backing Workflows call. Caller Workflows in other Namespaces need neither the plugin nor a sandbox escape +- The research Operations do not retry: each attempt submits a new billable research task, and the previous one keeps running because the You.com API has no way to cancel a submitted request +- Known limits, tracked before release: cancelling an Operation does not stop the in-flight You.com call +- Responses are parsed into SDK models on the handler side. A response that does not match raises a non-retryable `YouResponseShapeError`, because an unguarded parse error inside a Workflow is a Workflow task failure that Temporal would retry indefinitely, hanging the caller + ## [1.0.1] — 2026-08-18 ### Fixed diff --git a/README.md b/README.md index 0972936..f122746 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Durable [You.com](https://you.com) search, answer, research, and contents Activities for [Temporal](https://temporal.io). -Exposes You.com API calls as Temporal Activities with proper error mapping, retry semantics, and workflow sandbox support. Ships as a `SimplePlugin` for one-line setup, or as standalone activity functions for manual worker wiring. +Exposes You.com API calls as Temporal Activities with proper error mapping, retry semantics, and workflow sandbox support. Ships as a `SimplePlugin` for one-line setup, or as standalone activity functions for manual worker wiring. An optional Nexus Service (`youdotcom_temporal.nexus`) exposes the same calls as cross-Namespace Operations for teams that want a durable service contract on top of the Activities. ## Installation @@ -137,6 +137,100 @@ All activities return JSON-serializable dicts (via `model_dump(mode="json")`). | `crawl_timeout` | `int` | `10` | Per-URL timeout in seconds (1-60) | | `max_age` | `int \| None` | `None` | Max cache age in seconds (0 = always re-fetch) | +## Nexus Service + +`youdotcom_temporal.nexus.YouDotComService` is a [Temporal Nexus](https://docs.temporal.io/nexus) Service that exposes all six Activities as Operations callable across Namespace boundaries through a Nexus Endpoint. It is an opt-in layer on top of the Activities, so importing it does not affect Activity-only users. + +Every Operation is asynchronous and backed by a Workflow (`youdotcom_temporal.workflows`). Nexus synchronous operations must finish within a 10-second handler deadline, which is the wrong shape for these calls: `research` and `finance_research` are multi-step research runs measured in tens of seconds to minutes by design, background research runs up to 4 hours for `frontier`, and `search` and `contents` accept `livecrawl` and `crawl_timeout` (up to 60s per URL), so a caller can legitimately ask for a long call. + +A sync handler that misses the deadline fails as a retryable error, and five consecutive retryable errors trip a circuit breaker that blocks *every* Operation on that caller/Endpoint pair for 60 seconds. Routing each Operation through a Workflow removes the cliff and gives the caller durable, observable execution. + +Register the handler, the backing Workflows, and the Activities (via `YouPlugin`) on one Worker in the handler Namespace. `YouPlugin` is what registers the Activities the backing Workflows call, so the handler Worker needs it. + +Operations return the You.com SDK's own pydantic response models, so both the handler and every caller need Temporal's pydantic data converter. + +```python +from temporalio.client import Client +from temporalio.contrib.pydantic import pydantic_data_converter +from temporalio.worker import Worker +from youdotcom_temporal import YouPlugin +from youdotcom_temporal.nexus import you_nexus_service_handler +from youdotcom_temporal.workflows import you_nexus_workflows + +async def main(): + client = await Client.connect( + "localhost:7233", + namespace="you-handler", + data_converter=pydantic_data_converter, + ) + worker = Worker( + client, + task_queue="you-nexus", + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ) + await worker.run() +``` + +Create a Nexus Endpoint targeting that Worker, then call an Operation from a caller Workflow in another Namespace. Callers import from `youdotcom_temporal.contract`, which carries the types and none of the handler. No plugin and no sandbox escape are needed — the contract wraps its own SDK import — so these are safe at module scope: + +```python +from datetime import timedelta +from temporalio import workflow + +from youdotcom_temporal.contract import SearchRequest, SearchResponse, YouDotComService +from youdotcom_temporal.models import SearchInput + +NEXUS_ENDPOINT = "you-nexus-endpoint" + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, query: str) -> SearchResponse: + nexus_client = workflow.create_nexus_client( + service=YouDotComService, endpoint=NEXUS_ENDPOINT + ) + out = await nexus_client.execute_operation( + YouDotComService.search, + SearchRequest( + input=SearchInput(query=query, count=10), + # Optional. With a key, a retried StartOperation attaches to the + # Workflow already running instead of starting a second one and + # paying for a second You.com call. + idempotency_key=f"search:{query}", + ), + # Must exceed the handler-side worst case: the Activity ceiling + # times the retry policy's maximum_attempts (120s x 3 for search). + schedule_to_close_timeout=timedelta(minutes=10), + ) + out.results.web[0].title # typed all the way down + return out +``` + +| Operation | Request | Result | Backing Workflow | +|---|---|---|---| +| `search` | `SearchRequest` | `SearchResponse` | `YouSearchWorkflow` | +| `answer` | `AnswerRequest` | `AnswerResponse` | `YouAnswerWorkflow` | +| `contents` | `ContentsRequest` | `ContentsOutput` | `YouContentsWorkflow` | +| `research` | `ResearchRequest` | `ResearchResponse` | `YouResearchWorkflow` | +| `finance_research` | `FinanceResearchRequest` | `FinanceResearchResponse` | `YouFinanceResearchWorkflow` | +| `research_background` | `ResearchRequest` | `TaskDetail` | `YouResearchBackgroundWorkflow` | + +Every request carries the matching Activity input plus an optional `idempotency_key`. Results are the You.com SDK's own response models, imported from `youdotcom_temporal.contract`, so callers get accurate nested types the SDK maintains. `contents` is the exception: the SDK returns a bare list, which could not gain fields later without breaking callers, so it keeps a thin `ContentsOutput` envelope whose elements are still SDK `Contents` models. + +Each backing Workflow runs the Activity with a per-Activity `start_to_close_timeout`. A ceiling is a backstop rather than a latency target — it is where the Activity gives up and lets Temporal retry — so each carries generous headroom. `search` is 120s and `contents` 180s, sized off the 60s per-URL maximum a caller can request via `crawl_timeout` (`contents` gets more, since it accepts up to 10 URLs per request); `answer` is 60s; `research` is 10 minutes, `finance_research` 30 minutes, and `research_background` 4h15m. + +That ceiling is **per attempt** — `search`, `answer`, and `contents` retry up to 3 times, so size the caller's `schedule_to_close_timeout` against the ceiling times the attempt count. The research Operations do not retry: each attempt submits a new billable research task, and the previous one keeps running. + +See the [Temporal Python Nexus quickstart](https://docs.temporal.io/develop/python/nexus/quickstart) for Endpoint and caller-Namespace setup, and `examples/run_nexus_worker.py` for a runnable handler-side Worker. + +**Known limits (draft):** + +- **Cancellation does not reach You.com, and cannot.** The You.com API exposes no cancellation — the research surface is `POST /v1/research` plus two GETs to poll or stream, with no DELETE — so a submitted request runs to completion and is billed regardless. Cancelling an Operation frees the backing Workflow and the Worker slot, nothing upstream. +- **Idempotency is opt-in and bounded.** Supplying `idempotency_key` deduplicates against a Workflow that is still *running*, which covers the StartOperation-retry case. A key reused after the first Operation completed starts a fresh run. Without a key, starts are not deduplicated at all. +- **An unparseable response fails the Operation.** Results are parsed into SDK models on the handler side; a response that does not match raises a non-retryable `YouResponseShapeError` rather than hanging the caller. + ## Error handling | HTTP status | Error type | Retryable? | @@ -171,6 +265,7 @@ See the [`examples/`](examples/) directory: - `run_worker.py` - starts a worker with `YouPlugin` - `run_workflow.py` - executes the search workflow - `run_background_research_workflow.py` - executes the background research workflow +- `run_nexus_worker.py` - handler-side Worker hosting the `YouDotCom` Nexus Service ```bash # Terminal 1 diff --git a/examples/run_nexus_worker.py b/examples/run_nexus_worker.py new file mode 100644 index 0000000..df24501 --- /dev/null +++ b/examples/run_nexus_worker.py @@ -0,0 +1,108 @@ +"""Handler-side Worker for the YouDotCom Nexus Service. + +This Worker hosts the Nexus Service handler, the backing Workflows, and the +You.com Activities (registered by ``YouPlugin``). Run it in the *handler* +Namespace, then point a Nexus Endpoint at this Worker's Task Queue so callers in +other Namespaces can reach the Operations. + +``YouPlugin`` registers the Activities the backing Workflows call, so the +handler Worker needs it. Operations return the You.com SDK's pydantic response +models, so the Client needs ``pydantic_data_converter`` -- on the caller side +too. + +Setup (two terminals, local dev server with Nexus enabled):: + + # 1. Start the Temporal dev server (Nexus is enabled by default). + temporal server start-dev + + # 2. Create a handler Namespace and a Nexus Endpoint targeting this Worker. + temporal operator namespace create --namespace you-handler + temporal operator nexus endpoint create \ + --name you-nexus-endpoint \ + --target-namespace you-handler \ + --target-task-queue you-nexus + + # 3. Run this handler Worker (terminal 2). + YDC_API_KEY=your-key python examples/run_nexus_worker.py + +A caller Workflow in a different Namespace invokes an Operation through the +Endpoint. Callers import from ``youdotcom_temporal.contract``, which carries the +types and none of the handler. No plugin and no sandbox escape are needed, and +the caller's Client needs the same pydantic converter this Worker uses:: + + from datetime import timedelta + from temporalio import workflow + + from youdotcom_temporal.contract import ( + SearchRequest, + SearchResponse, + YouDotComService, + ) + from youdotcom_temporal.models import SearchInput + + NEXUS_ENDPOINT = "you-nexus-endpoint" + + @workflow.defn + class CallerWorkflow: + @workflow.run + async def run(self, query: str) -> SearchResponse: + nexus_client = workflow.create_nexus_client( + service=YouDotComService, endpoint=NEXUS_ENDPOINT + ) + # Exceeds the handler-side worst case for search + # (_SEARCH_STC 120s x 3 attempts). + out = await nexus_client.execute_operation( + YouDotComService.search, + SearchRequest( + input=SearchInput(query=query, count=10), + # Optional: a retried StartOperation attaches to the run + # already in flight instead of paying for a second call. + idempotency_key=f"search:{query}", + ), + schedule_to_close_timeout=timedelta(minutes=10), + ) + out.results.web[0].title # typed all the way down + return out + +See the Temporal Python Nexus quickstart for the full caller-side setup: +https://docs.temporal.io/develop/python/nexus/quickstart +""" + +from __future__ import annotations + +import asyncio + +from temporalio.client import Client +from temporalio.contrib.pydantic import pydantic_data_converter +from temporalio.worker import Worker + +from youdotcom_temporal import YouPlugin +from youdotcom_temporal.nexus import you_nexus_service_handler +from youdotcom_temporal.workflows import you_nexus_workflows + +HANDLER_NAMESPACE = "you-handler" +TASK_QUEUE = "you-nexus" + + +async def main() -> None: + client = await Client.connect( + "localhost:7233", + namespace=HANDLER_NAMESPACE, + data_converter=pydantic_data_converter, + ) + worker = Worker( + client, + task_queue=TASK_QUEUE, + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ) + print( + f"Handler Worker started in namespace {HANDLER_NAMESPACE!r} " + f"on task queue {TASK_QUEUE!r} (YouDotCom Nexus Service)" + ) + await worker.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/youdotcom_temporal/contract.py b/src/youdotcom_temporal/contract.py new file mode 100644 index 0000000..9ee29c1 --- /dev/null +++ b/src/youdotcom_temporal/contract.py @@ -0,0 +1,153 @@ +"""The Nexus Service contract: request types, result types, and the Service. + +This is what a caller in another Namespace imports. Results are the You.com SDK's +own response models, so callers get accurate, fully nested types that the SDK +team maintains, rather than types hand-rolled here that would drift:: + + out = await client.execute_operation(YouDotComService.search, req, ...) + out.results.web[0].title # typed all the way down + +Callers must configure Temporal's pydantic converter on their Client and Worker, +since the SDK models are pydantic:: + + from temporalio.contrib.pydantic import pydantic_data_converter + client = await Client.connect(..., data_converter=pydantic_data_converter) + +Sandbox note: + The SDK import below is wrapped in ``imports_passed_through()`` because + ``youdotcom/__init__`` currently imports its transport layer eagerly, which + pulls ``urllib.request`` and would be rejected by the Workflow sandbox. + Wrapping it here means a caller Workflow can import this module directly + without needing the escape hatch itself. Once the SDK resolves its exports + lazily (DX-776) the wrapper becomes belt-and-braces rather than load-bearing, + and importing this contract stops pulling the HTTP stack at all. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import nexusrpc +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from youdotcom.models import ( + AnswerResponse, + Contents, + FinanceResearchResponse, + ResearchResponse, + SearchResponse, + TaskDetail, + ) + +from youdotcom_temporal.models import ( + AnswerInput, + ContentsInput, + FinanceResearchInput, + ResearchInput, + SearchInput, +) + +__all__ = [ + "AnswerRequest", + "AnswerResponse", + "Contents", + "ContentsOutput", + "ContentsRequest", + "FinanceResearchRequest", + "FinanceResearchResponse", + "ResearchRequest", + "ResearchResponse", + "SearchRequest", + "SearchResponse", + "TaskDetail", + "YouDotComService", +] + + +# --------------------------------------------------------------------------- # +# Requests +# --------------------------------------------------------------------------- # +# +# Each Operation takes the Activity input plus an optional idempotency key. +# +# Temporal's guidance is that a backing Workflow Id should be business- +# meaningful and travel in the Operation input, because Workflow Ids are what +# deduplicate Workflow starts. Without one, a retried Nexus StartOperation +# request starts a second Workflow and a second billable You.com call. Only the +# caller knows what makes two requests the "same" request, so only the caller +# can supply it. +# +# The key is optional; omitting it leaves starts non-idempotent, which is the +# right default for genuinely one-off calls. + + +@dataclass +class SearchRequest: + input: SearchInput + idempotency_key: str | None = None + + +@dataclass +class AnswerRequest: + input: AnswerInput + idempotency_key: str | None = None + + +@dataclass +class ContentsRequest: + input: ContentsInput + idempotency_key: str | None = None + + +@dataclass +class ResearchRequest: + input: ResearchInput + idempotency_key: str | None = None + + +@dataclass +class FinanceResearchRequest: + input: FinanceResearchInput + idempotency_key: str | None = None + + +# --------------------------------------------------------------------------- # +# Results +# --------------------------------------------------------------------------- # +# +# Five Operations return the SDK response model directly. `contents` is the +# exception: the SDK returns a bare list, and a list is a poor Nexus result type +# because it cannot carry additional fields later without breaking callers. It +# gets a thin envelope; the elements inside are still SDK models. + + +@dataclass +class ContentsOutput: + """One `Contents` per requested URL, in request order.""" + + results: list[Contents] + + +# --------------------------------------------------------------------------- # +# Service +# --------------------------------------------------------------------------- # + + +@nexusrpc.service +class YouDotComService: + """Nexus Service contract for You.com search, answer, contents, research. + + Every Operation is asynchronous and backed by a Workflow on the handler + side. Callers reach these through a Nexus Endpoint pointing at the handler's + Namespace and Task Queue. + """ + + search: nexusrpc.Operation[SearchRequest, SearchResponse] + answer: nexusrpc.Operation[AnswerRequest, AnswerResponse] + contents: nexusrpc.Operation[ContentsRequest, ContentsOutput] + research: nexusrpc.Operation[ResearchRequest, ResearchResponse] + finance_research: nexusrpc.Operation[ + FinanceResearchRequest, FinanceResearchResponse + ] + research_background: nexusrpc.Operation[ResearchRequest, TaskDetail] diff --git a/src/youdotcom_temporal/nexus.py b/src/youdotcom_temporal/nexus.py new file mode 100644 index 0000000..74d323e --- /dev/null +++ b/src/youdotcom_temporal/nexus.py @@ -0,0 +1,217 @@ +"""Temporal Nexus Service exposing You.com calls as cross-Namespace Operations. + +This is the Nexus layer on top of the Activity layer in +:mod:`youdotcom_temporal.activities`. A Nexus Service lets other teams call +You.com search, answer, contents, and research through a durable contract +across Namespace boundaries. + +Every Operation is asynchronous and backed by a Workflow from +:mod:`youdotcom_temporal.workflows`. Nexus synchronous operations must finish +within the 10-second handler deadline, but several You.com calls routinely +exceed that (full-page search, multi-URL contents, research, background +research up to 4 hours). Backing every Operation with a Workflow removes that +cliff and gives the caller durable, observable execution. + +Caller-side note: + Callers should import from :mod:`youdotcom_temporal.contract`, not from + here:: + + from youdotcom_temporal.contract import SearchRequest, YouDotComService + from youdotcom_temporal.models import SearchInput + + ``YouDotComService`` is re-exported below for convenience, but importing it + from this module also pulls in the handler: the backing Workflows and, in + turn, the Activity layer. A caller needs none of that. The contract module + carries the types and nothing else. + + Either import is safe inside a Workflow sandbox without an escape hatch, as + the contract wraps its own SDK import. It does load the You.com SDK, which + is deliberate -- results are the SDK's response models. Once the SDK + resolves its exports lazily (DX-776) that import stops pulling the HTTP + stack with it. + +Cancellation: + Cancelling an Operation cancels the backing Workflow, but not the upstream + work. The You.com API exposes no cancellation: the research surface is + ``POST /v1/research`` to submit plus two GETs to poll or stream, with no + DELETE, and the SDK has no cancel method. Once a request is submitted it + runs to completion and is billed, so no client-side mechanism can call it + back. This is a property of the API rather than something the plugin can + fix; cancelling frees the Workflow and the Worker slot, nothing more. + +Register the handler and the backing Workflows on the same Worker that runs +the Activities (the ``YouPlugin`` registers the Activities and the sandbox +passthrough):: + + from temporalio.client import Client + from temporalio.worker import Worker + from youdotcom_temporal import YouPlugin + from youdotcom_temporal.nexus import you_nexus_service_handler + from youdotcom_temporal.workflows import you_nexus_workflows + + client = await Client.connect("localhost:7233") + worker = Worker( + client, + task_queue="you-nexus", + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ) + await worker.run() +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +import uuid +from typing import Any + +import nexusrpc +from temporalio import nexus +from temporalio.common import WorkflowIDConflictPolicy + +from youdotcom_temporal.contract import ( + AnswerRequest, + AnswerResponse, + ContentsOutput, + ContentsRequest, + FinanceResearchRequest, + FinanceResearchResponse, + ResearchRequest, + ResearchResponse, + SearchRequest, + SearchResponse, + TaskDetail, + YouDotComService, +) +from youdotcom_temporal.workflows import ( + YouAnswerWorkflow, + YouContentsWorkflow, + YouFinanceResearchWorkflow, + YouResearchBackgroundWorkflow, + YouResearchWorkflow, + YouSearchWorkflow, +) + +__all__ = [ + "YouDotComService", + "YouDotComServiceHandler", + "you_nexus_service_handler", +] + + +@nexusrpc.handler.service_handler(service=YouDotComService) +class YouDotComServiceHandler: + """Nexus Service handler that backs each Operation with a Workflow. + + Every Operation starts a Workflow from :mod:`youdotcom_temporal.workflows`, + which runs the corresponding Activity with a per-Activity + ``start_to_close_timeout``. The Workflow result is delivered back to the + Nexus caller when it completes. + + Idempotency is the caller's to opt into. Supplying + ``idempotency_key`` on the request makes the backing Workflow Id + deterministic, so a retried Nexus StartOperation request attaches to the + Workflow already running instead of starting a second one and paying for a + second You.com call. Without a key the Id is random and starts are not + deduplicated, which is the right behaviour for genuinely one-off calls. + """ + + @staticmethod + def _workflow_id(prefix: str, idempotency_key: str | None, inp: Any) -> str: + """Deterministic when the caller supplies a key, unique otherwise. + + The key alone is not enough. Callers pick their own key strings and + nothing stops two of them choosing the same one -- ``"order-123"`` is an + obvious collision waiting to happen -- and with + ``WorkflowIDConflictPolicy.USE_EXISTING`` a collision means the second + caller attaches to the first caller's Workflow and receives *their* + results. Binding the request into the Id keeps a key idempotent for the + request it was issued for and inert across different ones. + + The digest covers the full key as well as the request, so two long keys + sharing a prefix cannot collide once the readable part is truncated. + """ + if idempotency_key is None: + return f"{prefix}-{uuid.uuid4()}" + + canonical = json.dumps(dataclasses.asdict(inp), sort_keys=True, default=str) + digest = hashlib.sha256( + f"{idempotency_key}\0{canonical}".encode() + ).hexdigest()[:16] + # Keep a readable slice of the key so the Id is still recognisable in the + # UI, and bound it so an oversized key cannot push past Temporal's limit. + return f"{prefix}-{idempotency_key[:64]}-{digest}" + + @nexus.workflow_run_operation + async def search( + self, ctx: nexus.WorkflowRunOperationContext, req: SearchRequest + ) -> nexus.WorkflowHandle[SearchResponse]: + return await ctx.start_workflow( + YouSearchWorkflow.run, + req.input, + id=self._workflow_id("you-search", req.idempotency_key, req.input), + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + + @nexus.workflow_run_operation + async def answer( + self, ctx: nexus.WorkflowRunOperationContext, req: AnswerRequest + ) -> nexus.WorkflowHandle[AnswerResponse]: + return await ctx.start_workflow( + YouAnswerWorkflow.run, + req.input, + id=self._workflow_id("you-answer", req.idempotency_key, req.input), + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + + @nexus.workflow_run_operation + async def contents( + self, ctx: nexus.WorkflowRunOperationContext, req: ContentsRequest + ) -> nexus.WorkflowHandle[ContentsOutput]: + return await ctx.start_workflow( + YouContentsWorkflow.run, + req.input, + id=self._workflow_id("you-contents", req.idempotency_key, req.input), + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + + @nexus.workflow_run_operation + async def research( + self, ctx: nexus.WorkflowRunOperationContext, req: ResearchRequest + ) -> nexus.WorkflowHandle[ResearchResponse]: + return await ctx.start_workflow( + YouResearchWorkflow.run, + req.input, + id=self._workflow_id("you-research", req.idempotency_key, req.input), + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + + @nexus.workflow_run_operation + async def finance_research( + self, ctx: nexus.WorkflowRunOperationContext, req: FinanceResearchRequest + ) -> nexus.WorkflowHandle[FinanceResearchResponse]: + return await ctx.start_workflow( + YouFinanceResearchWorkflow.run, + req.input, + id=self._workflow_id("you-finance-research", req.idempotency_key, req.input), + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + + @nexus.workflow_run_operation + async def research_background( + self, ctx: nexus.WorkflowRunOperationContext, req: ResearchRequest + ) -> nexus.WorkflowHandle[TaskDetail]: + return await ctx.start_workflow( + YouResearchBackgroundWorkflow.run, + req.input, + id=self._workflow_id("you-research-bg", req.idempotency_key, req.input), + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + + +def you_nexus_service_handler() -> YouDotComServiceHandler: + """Construct the :class:`YouDotComServiceHandler` for ``Worker(nexus_service_handlers=...)``.""" + return YouDotComServiceHandler() diff --git a/src/youdotcom_temporal/workflows.py b/src/youdotcom_temporal/workflows.py new file mode 100644 index 0000000..60dc45e --- /dev/null +++ b/src/youdotcom_temporal/workflows.py @@ -0,0 +1,287 @@ +"""Thin Temporal Workflows that wrap each You.com Activity. + +These exist so a Nexus Operation can back the Activities with an asynchronous +Workflow. Nexus synchronous operations must finish within the 10-second handler +deadline, which is the wrong shape for this API: several of these calls are +long-running by design, and the rest accept parameters that make their duration +a caller's choice. + +- ``youdotcom_research`` / ``youdotcom_finance_research`` -- multi-step research, + measured in tens of seconds to minutes by design +- ``youdotcom_research_background`` -- up to 4 hours for ``frontier`` effort +- ``youdotcom_search`` / ``youdotcom_contents`` -- ``livecrawl`` and + ``crawl_timeout`` fetch pages live, and ``crawl_timeout`` alone accepts up to + 60 seconds per URL, so one route covers two very different operations +- ``youdotcom_answer`` -- the shortest of the six, and a synthesis step over + retrieved sources rather than a lookup + +That matters because a sync handler which misses the deadline fails as a +retryable error, and five consecutive retryable errors trip a circuit breaker +that blocks *every* Operation on the caller/Endpoint pair for 60 seconds. +Routing every Operation through a Workflow removes the cliff: the Activity runs +with its own ``start_to_close_timeout`` and Temporal's retries, and the caller +gets durable, observable execution. Cancellation is a partial story -- see the +note in :mod:`youdotcom_temporal.nexus`. + +Each Workflow is a one-line wrapper around ``workflow.execute_activity``. The +ceilings are workflow-side defaults; a Nexus caller still sets +``schedule_to_close_timeout`` on the operation call, and that caller timeout +should exceed the handler-side worst case (ceiling times attempts). + +Sandbox note: + This module imports the Activity layer, which imports the You.com SDK, so + the import lives in an ``imports_passed_through()`` block. That block only + works because ``youdotcom_temporal/__init__`` resolves its public names + lazily -- Python imports the parent package first, and an eager import + there would escape the block and fail under the sandbox. +""" + +from __future__ import annotations + +from datetime import timedelta +from typing import Any, TypeVar + +from temporalio import workflow +from temporalio.common import RetryPolicy +from temporalio.exceptions import ApplicationError + +with workflow.unsafe.imports_passed_through(): + from pydantic import BaseModel, ValidationError + + from youdotcom_temporal.activities import ( + youdotcom_answer, + youdotcom_contents, + youdotcom_finance_research, + youdotcom_research, + youdotcom_research_background, + youdotcom_search, + ) + from youdotcom_temporal.contract import ( + AnswerResponse, + Contents, + ContentsOutput, + FinanceResearchResponse, + ResearchResponse, + SearchResponse, + TaskDetail, + ) + from youdotcom_temporal.models import ( + AnswerInput, + ContentsInput, + FinanceResearchInput, + ResearchInput, + SearchInput, + ) + +# Per-Activity start_to_close ceilings. These are *per attempt*: the wall-clock +# ceiling is the value below times the retry policy's maximum_attempts. +# +# A ceiling is a backstop, not a latency target: it is where the Activity gives +# up and lets Temporal retry, so it wants generous headroom over what a healthy +# call takes. Undersized, it turns slow-but-fine calls into failures; oversized, +# a genuinely stuck call ties up a Worker slot longer than it should. +# +# The two crawling routes are sized off the parameters they accept. `search` and +# `contents` fetch pages live under `livecrawl`/`crawl_timeout`, and +# `crawl_timeout` alone accepts up to 60s per URL, so a caller can legitimately +# ask for a long call: +# +# _SEARCH_STC 2x the 60s maximum a caller can request +# _CONTENTS_STC 3x it, since `contents` accepts up to 10 URLs per request +# +# answer/search/contents sit *below* the client's own HTTP timeout +# (YouConfig.timeout_seconds, default 300s); research/finance_research sit above +# it. That asymmetry is intentional. For the shorter Activities, failing at the +# ceiling beats waiting out the full 300s, and the cost is that the abandoned +# HTTP call keeps running -- the You.com API has no cancellation, so nothing can +# call a submitted request back, and a retry runs alongside it. For the research +# Activities the client errors first, so the Activity reports the real failure +# instead of an opaque activity timeout; that is also why they do not retry. +_SEARCH_STC = timedelta(seconds=120) +_ANSWER_STC = timedelta(seconds=60) +_CONTENTS_STC = timedelta(seconds=180) +_RESEARCH_STC = timedelta(minutes=10) +_FINANCE_RESEARCH_STC = timedelta(minutes=30) + +# The Activity layer already raises these as non-retryable ApplicationErrors; +# listing them keeps the intent visible at the call site and matches the retry +# policies in examples/. +_NON_RETRYABLE = ["YouAuthError", "YouValidationError", "YouQuotaExhausted"] + +# search/answer/contents: the ceilings above sit far enough above a healthy call +# that reaching one means something is genuinely broken rather than merely slow, +# so a retry is likely to succeed. Worst case is maximum_attempts abandoned HTTP +# calls running concurrently, which is acceptable for these three. +_RETRY = RetryPolicy( + maximum_attempts=3, + maximum_interval=timedelta(seconds=60), + non_retryable_error_types=_NON_RETRYABLE, +) + +# Research Activities: each attempt submits a new billable research task, and +# the previous attempt keeps running, because the You.com API has no way to +# cancel it. Retry would multiply cost and wall-clock rather than recover, so do +# not retry. +# +# No non_retryable_error_types here: at maximum_attempts=1 nothing is ever +# retried, so listing them would read as a safeguard while doing nothing. +_RETRY_RESEARCH = RetryPolicy(maximum_attempts=1) + +# The deadline research_and_wait_async derives from research_effort when +# ResearchInput.timeout_s is None. Mirrored here only to size the ceiling below; +# the Activity forwards timeout_s untouched, so the SDK remains the one place +# that decides the deadline. If the SDK's own values change, these follow. +_BACKGROUND_TIMEOUT_S = 600.0 +_BACKGROUND_TIMEOUT_S_BY_EFFORT = {"frontier": 14400.0} + +# The Activity's own wait has to expire before this ceiling does, or Temporal +# kills the attempt before the Activity can report what happened and the +# Workflow fails with an opaque activity timeout. research_and_wait_async also +# issues a final GET after its internal wait, so the ceiling needs headroom +# rather than an exact match. Derived from the waits above so the two cannot +# drift apart. +_BACKGROUND_MARGIN = timedelta(minutes=15) +_RESEARCH_BACKGROUND_STC = ( + timedelta( + seconds=max([_BACKGROUND_TIMEOUT_S, *_BACKGROUND_TIMEOUT_S_BY_EFFORT.values()]) + ) + + _BACKGROUND_MARGIN +) # 4h15m: frontier's 4h wait plus headroom + + +_T = TypeVar("_T", bound=BaseModel) + + +def _validate(model: type[_T], payload: dict[str, Any]) -> _T: + """Parse an Activity payload into its SDK model, failing cleanly if it cannot. + + A bare ``model_validate`` raising inside a Workflow is a Workflow task + failure, which Temporal retries indefinitely -- an unexpected upstream + response shape would hang the Operation rather than surface. Converting it + to a non-retryable ApplicationError means the caller gets a real error, and + it reaches them through the same failure chain as any other Activity error. + + The message reports which fields failed and why, but never the values. A + pydantic error renders the offending input inline, and this message travels + across the Nexus boundary into the caller's failure and the Workflow + history -- so a malformed response would otherwise spill You.com content to + whoever called the Operation. + """ + try: + return model.model_validate(payload) + except ValidationError as exc: + problems = "; ".join( + f"{'.'.join(str(p) for p in err['loc']) or ''}: {err['msg']}" + for err in exc.errors(include_input=False, include_url=False) + ) + raise ApplicationError( + f"You.com response did not match {model.__name__}: {problems}", + type="YouResponseShapeError", + non_retryable=True, + ) from exc + + +@workflow.defn +class YouSearchWorkflow: + @workflow.run + async def run(self, inp: SearchInput) -> SearchResponse: + payload = await workflow.execute_activity( + youdotcom_search, + inp, + start_to_close_timeout=_SEARCH_STC, + retry_policy=_RETRY, + summary=f"you.com search: {inp.query}", + ) + return _validate(SearchResponse, payload) + + +@workflow.defn +class YouAnswerWorkflow: + @workflow.run + async def run(self, inp: AnswerInput) -> AnswerResponse: + payload = await workflow.execute_activity( + youdotcom_answer, + inp, + start_to_close_timeout=_ANSWER_STC, + retry_policy=_RETRY, + summary=f"you.com answer: {inp.query}", + ) + return _validate(AnswerResponse, payload) + + +@workflow.defn +class YouContentsWorkflow: + @workflow.run + async def run(self, inp: ContentsInput) -> ContentsOutput: + payload = await workflow.execute_activity( + youdotcom_contents, + inp, + start_to_close_timeout=_CONTENTS_STC, + retry_policy=_RETRY, + summary=f"you.com contents: {len(inp.urls)} url(s)", + ) + # The Activity builds this envelope itself, but reach for the key + # defensively: a KeyError here would be a Workflow task failure, which + # retries forever, and that is exactly what _validate exists to avoid. + documents = payload.get("results") + if not isinstance(documents, list): + raise ApplicationError( + "You.com contents response had no 'results' list", + type="YouResponseShapeError", + non_retryable=True, + ) + return ContentsOutput(results=[_validate(Contents, c) for c in documents]) + + +@workflow.defn +class YouResearchWorkflow: + @workflow.run + async def run(self, inp: ResearchInput) -> ResearchResponse: + payload = await workflow.execute_activity( + youdotcom_research, + inp, + start_to_close_timeout=_RESEARCH_STC, + retry_policy=_RETRY_RESEARCH, + summary=f"you.com research ({inp.research_effort})", + ) + return _validate(ResearchResponse, payload) + + +@workflow.defn +class YouFinanceResearchWorkflow: + @workflow.run + async def run(self, inp: FinanceResearchInput) -> FinanceResearchResponse: + payload = await workflow.execute_activity( + youdotcom_finance_research, + inp, + start_to_close_timeout=_FINANCE_RESEARCH_STC, + retry_policy=_RETRY_RESEARCH, + summary=f"you.com finance research ({inp.research_effort})", + ) + return _validate(FinanceResearchResponse, payload) + + +@workflow.defn +class YouResearchBackgroundWorkflow: + @workflow.run + async def run(self, inp: ResearchInput) -> TaskDetail: + payload = await workflow.execute_activity( + youdotcom_research_background, + inp, + start_to_close_timeout=_RESEARCH_BACKGROUND_STC, + retry_policy=_RETRY_RESEARCH, + summary=f"you.com background research ({inp.research_effort})", + ) + return _validate(TaskDetail, payload) + + +def you_nexus_workflows() -> list[type]: + """All Nexus-backing Workflows, for passing to ``Worker(workflows=...)``.""" + return [ + YouSearchWorkflow, + YouAnswerWorkflow, + YouContentsWorkflow, + YouResearchWorkflow, + YouFinanceResearchWorkflow, + YouResearchBackgroundWorkflow, + ] diff --git a/tests/_nexus_caller_workflows.py b/tests/_nexus_caller_workflows.py new file mode 100644 index 0000000..e65ab0f --- /dev/null +++ b/tests/_nexus_caller_workflows.py @@ -0,0 +1,148 @@ +"""Caller-side Workflows for the Nexus round-trip test. + +These stand in for a consumer in another Namespace: they reach You.com only +through the Nexus Endpoint, and the Worker that runs them registers no +``YouPlugin`` and no Activities. Imports here are unwrapped on purpose -- +that is the shape a real caller has. +""" + +from __future__ import annotations + +from datetime import timedelta + +from temporalio import workflow + +from youdotcom_temporal.contract import ( + AnswerRequest, + AnswerResponse, + ContentsOutput, + ContentsRequest, + FinanceResearchRequest, + FinanceResearchResponse, + ResearchRequest, + ResearchResponse, + SearchRequest, + SearchResponse, + TaskDetail, + YouDotComService, +) +from youdotcom_temporal.models import ( + AnswerInput, + ContentsInput, + FinanceResearchInput, + ResearchInput, + SearchInput, +) + +ENDPOINT = "you-nexus-test-endpoint" +_TIMEOUT = timedelta(minutes=2) + + +def _client() -> workflow.NexusClient[YouDotComService]: + return workflow.create_nexus_client(service=YouDotComService, endpoint=ENDPOINT) + + +@workflow.defn +class CallSearch: + @workflow.run + async def run(self, query: str) -> SearchResponse: + return await _client().execute_operation( + YouDotComService.search, + SearchRequest(input=SearchInput(query=query, count=3)), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallSearchIdempotent: + """Search carrying a caller-supplied idempotency key.""" + + @workflow.run + async def run(self, args: list[str]) -> SearchResponse: + query, key = args + return await _client().execute_operation( + YouDotComService.search, + SearchRequest(input=SearchInput(query=query, count=3), idempotency_key=key), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallAnswer: + @workflow.run + async def run(self, query: str) -> AnswerResponse: + return await _client().execute_operation( + YouDotComService.answer, + AnswerRequest(input=AnswerInput(query=query)), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallContents: + @workflow.run + async def run(self, url: str) -> ContentsOutput: + return await _client().execute_operation( + YouDotComService.contents, + ContentsRequest(input=ContentsInput(urls=[url])), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallContentsMany: + """`contents` at fan-out -- the Operation with the largest result payload.""" + + @workflow.run + async def run(self, urls: list[str]) -> ContentsOutput: + return await _client().execute_operation( + YouDotComService.contents, + ContentsRequest(input=ContentsInput(urls=urls)), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallResearch: + @workflow.run + async def run(self, question: str) -> ResearchResponse: + return await _client().execute_operation( + YouDotComService.research, + ResearchRequest(input=ResearchInput(input=question)), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallFinanceResearch: + @workflow.run + async def run(self, question: str) -> FinanceResearchResponse: + return await _client().execute_operation( + YouDotComService.finance_research, + FinanceResearchRequest(input=FinanceResearchInput(input=question)), + schedule_to_close_timeout=_TIMEOUT, + ) + + +@workflow.defn +class CallResearchBackground: + @workflow.run + async def run(self, question: str) -> TaskDetail: + return await _client().execute_operation( + YouDotComService.research_background, + ResearchRequest(input=ResearchInput(input=question, research_effort="lite")), + schedule_to_close_timeout=_TIMEOUT, + ) + + +def caller_workflows() -> list[type]: + return [ + CallSearch, + CallSearchIdempotent, + CallAnswer, + CallContents, + CallContentsMany, + CallResearch, + CallFinanceResearch, + CallResearchBackground, + ] diff --git a/tests/test_nexus.py b/tests/test_nexus.py new file mode 100644 index 0000000..f3d7405 --- /dev/null +++ b/tests/test_nexus.py @@ -0,0 +1,480 @@ +"""Unit tests for the Nexus Service contract and handler. + +These validate the :class:`YouDotComService` definition and the +:class:`YouDotComServiceHandler` without a Temporal server: the service has +six Operations with the right names and typed inputs, the handler constructs, +the helper functions return the right Workflow classes, and the backing +Workflows survive the sandbox preparation that ``Worker.__init__`` performs. + +End-to-end Nexus execution (a real Endpoint, a caller in a second Namespace) +is not covered anywhere yet -- see the draft checklist in the PR. +""" + +from __future__ import annotations + +import dataclasses +import subprocess +import sys +from datetime import timedelta + +import nexusrpc +import pytest +from pydantic import BaseModel +from temporalio import workflow +from temporalio.exceptions import ApplicationError +from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner + +from youdotcom_temporal.contract import ( + AnswerRequest, + AnswerResponse, + ContentsOutput, + ContentsRequest, + FinanceResearchRequest, + FinanceResearchResponse, + ResearchRequest, + ResearchResponse, + SearchRequest, + SearchResponse, + TaskDetail, + YouDotComService, +) +from youdotcom_temporal.models import ( + AnswerInput, + ContentsInput, + FinanceResearchInput, + ResearchInput, + SearchInput, +) +from youdotcom_temporal.nexus import ( + YouDotComServiceHandler, + you_nexus_service_handler, +) +from youdotcom_temporal.plugin import _you_workflow_runner +from youdotcom_temporal.workflows import ( + _BACKGROUND_TIMEOUT_S, + _BACKGROUND_TIMEOUT_S_BY_EFFORT, + YouAnswerWorkflow, + YouContentsWorkflow, + YouFinanceResearchWorkflow, + YouResearchBackgroundWorkflow, + YouResearchWorkflow, + YouSearchWorkflow, + you_nexus_workflows, +) + +# operation -> (request type, result type) +_EXPECTED_OPS = { + "search": (SearchRequest, SearchResponse), + "answer": (AnswerRequest, AnswerResponse), + "contents": (ContentsRequest, ContentsOutput), + "research": (ResearchRequest, ResearchResponse), + "finance_research": (FinanceResearchRequest, FinanceResearchResponse), + "research_background": (ResearchRequest, TaskDetail), +} + +_BACKING_WORKFLOWS = { + "search": YouSearchWorkflow, + "answer": YouAnswerWorkflow, + "contents": YouContentsWorkflow, + "research": YouResearchWorkflow, + "finance_research": YouFinanceResearchWorkflow, + "research_background": YouResearchBackgroundWorkflow, +} + + +# Minimal payloads that satisfy each SDK response model. The Workflow now parses +# the Activity's dict into a typed model, so a fake Activity has to return +# something the model accepts -- these tests are about timeouts and retries, not +# parsing, so keep them to the required fields only. +_OUTPUT = {"content": "x", "content_type": "text", "sources": []} +_VALID_PAYLOAD = { + "youdotcom_search": {}, + "youdotcom_answer": {"answer": "a"}, + "youdotcom_contents": {"results": []}, + "youdotcom_research": {"output": _OUTPUT}, + "youdotcom_finance_research": {"output": _OUTPUT}, + "youdotcom_research_background": { + "id": "t", + "task_type": "research", + "status": "completed", + "created_at": "2026-08-19T00:00:00Z", + "updated_at": "2026-08-19T00:00:00Z", + }, +} + + +def _payload_for(activity) -> dict: + return _VALID_PAYLOAD[activity.__name__] + + +def test_service_definition_has_six_operations(): + defn = nexusrpc.get_service_definition(YouDotComService) + assert defn is not None + assert defn.name == "YouDotComService" + assert set(defn.operation_definitions) == set(_EXPECTED_OPS) + + +def test_operation_names_match_method_names(): + """Each Operation's wire name equals its handler method name.""" + defn = nexusrpc.get_service_definition(YouDotComService) + for op_name, op_defn in defn.operation_definitions.items(): + assert op_defn.name == op_name + assert op_defn.method_name == op_name + + +def test_operation_input_types_are_typed(): + defn = nexusrpc.get_service_definition(YouDotComService) + for op_name, (request_type, _) in _EXPECTED_OPS.items(): + assert defn.operation_definitions[op_name].input_type is request_type, op_name + + +def test_operation_output_types_are_typed_results(): + """Results are declared types, not bare dicts. + + A cross-Namespace contract whose outputs are ``dict[str, Any]`` gives the + caller no field names and no signal when a shape changes. + """ + defn = nexusrpc.get_service_definition(YouDotComService) + for op_name, (_, result_type) in _EXPECTED_OPS.items(): + assert defn.operation_definitions[op_name].output_type is result_type, op_name + assert result_type is not dict, op_name + + +def test_result_types_are_sdk_models_or_thin_envelopes(): + """Results are the SDK's own response models wherever one exists. + + Using them means callers get accurate, fully nested types the SDK team + maintains. `contents` is the one exception: the SDK returns a bare list, + which cannot gain fields later without breaking callers, so it keeps a thin + envelope whose elements are still SDK models. + """ + for op_name, (_, result_type) in _EXPECTED_OPS.items(): + if op_name == "contents": + assert dataclasses.is_dataclass(result_type) + else: + assert issubclass(result_type, BaseModel), op_name + + +def test_result_types_are_serializable_by_the_pydantic_converter(): + """The contract only works if Temporal can carry these types. + + SDK models are pydantic, so callers must configure + ``temporalio.contrib.pydantic.pydantic_data_converter``; this pins that the + types are ones that converter actually handles. + """ + for op_name, (_, result_type) in _EXPECTED_OPS.items(): + if op_name == "contents": + continue + assert hasattr(result_type, "model_validate"), op_name + assert hasattr(result_type, "model_dump"), op_name + + +def test_service_handler_carries_service_definition(): + defn = nexusrpc.get_service_definition(YouDotComServiceHandler) + assert defn is not None + assert set(defn.operation_definitions) == set(_EXPECTED_OPS) + + +def test_service_handler_constructs(): + handler = you_nexus_service_handler() + assert isinstance(handler, YouDotComServiceHandler) + + +def test_nexus_workflows_list_covers_every_operation(): + """Each Operation must have a distinct backing Workflow in the Worker list. + + A missing entry here means the Operation starts a Workflow the Worker never + registered, which fails only at runtime against a real server. + """ + wfs = you_nexus_workflows() + assert len(set(wfs)) == len(wfs), "duplicate Workflow in the registration list" + assert set(wfs) == set(_BACKING_WORKFLOWS.values()) + assert set(_BACKING_WORKFLOWS) == set(_EXPECTED_OPS) + + +async def test_backing_workflows_prepare_under_the_plugin_sandbox(): + """Worker registration must succeed with YouPlugin's passthrough applied. + + ``Worker.__init__`` runs ``prepare_workflow`` for every registered Workflow. + These Workflows import the You.com SDK, so this is the check that catches a + sandbox restriction before it reaches a running Worker. + """ + runner = _you_workflow_runner(SandboxedWorkflowRunner()) + for wf in you_nexus_workflows(): + runner.prepare_workflow(workflow._Definition.must_from_class(wf)) + + +async def test_backing_workflows_prepare_without_the_plugin(): + """Registration must not depend on YouPlugin's sandbox passthrough. + + ``youdotcom_temporal/__init__`` resolves its public names lazily, so the + parent package import no longer drags in the You.com SDK and + ``workflows.py`` can cover the SDK with its own + ``imports_passed_through()`` block. Without that, Python imports the parent + package first and the block never gets the chance. + """ + runner = SandboxedWorkflowRunner() + for wf in you_nexus_workflows(): + runner.prepare_workflow(workflow._Definition.must_from_class(wf)) + + +async def test_caller_workflows_prepare_without_passthrough(): + """A caller in another Namespace must not need YouPlugin or a passthrough. + + ``tests/_nexus_caller_workflows`` imports the contract at module scope with + no sandbox escape, which is how the README documents the caller side. A + caller Namespace has no YouPlugin to inherit passthrough from, so this has + to hold on its own. + + These are the same Workflows the round-trip tests drive against a real + server, so this cannot drift away from the shape that actually runs -- an + earlier standalone copy did exactly that, passing here while carrying a + request shape the contract would have rejected. + """ + from _nexus_caller_workflows import caller_workflows + + runner = SandboxedWorkflowRunner() + for wf in caller_workflows(): + runner.prepare_workflow(workflow._Definition.must_from_class(wf)) + + +def test_importing_the_package_does_not_load_the_sdk(): + """The invariant behind both sandbox fixes, checked directly. + + If any eager import creeps back into ``__init__``, the sandbox failures + return -- and they return at Worker construction, far from this package. + """ + code = ( + "import sys; import youdotcom_temporal; " + "mods = set(sys.modules); " + "print(int(any(m == 'youdotcom' or m.startswith('youdotcom.') for m in mods)), " + "int('urllib.request' in mods))" + ) + out = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=True + ) + sdk_loaded, urllib_loaded = out.stdout.split() + assert sdk_loaded == "0", "importing youdotcom_temporal pulled in the You.com SDK" + assert urllib_loaded == "0", "importing youdotcom_temporal pulled in urllib.request" + + +async def test_background_research_mirrors_the_sdk_deadline_constants(monkeypatch): + """The mirrored deadline constants must match what the SDK actually derives. + + ``_BACKGROUND_TIMEOUT_S`` and ``_BACKGROUND_TIMEOUT_S_BY_EFFORT`` exist only + to size the ceiling, but they are a copy of the SDK's own values. If the SDK + raises frontier past 4 hours and this copy is not updated, the ceiling + silently stops covering the deadline. Read the real values back from the SDK + so that drift fails here rather than in production. + """ + from youdotcom import research_helpers + + assert research_helpers._DEFAULT_POLL_TIMEOUT_S == _BACKGROUND_TIMEOUT_S + assert ( + research_helpers._FRONTIER_TIMEOUT_S + == _BACKGROUND_TIMEOUT_S_BY_EFFORT["frontier"] + ) + + +async def test_research_workflows_do_not_retry(monkeypatch): + """Research Activities must not retry; the fast ones must. + + Every research attempt submits a new billable You.com task, and the previous + attempt keeps running because the Activities do not heartbeat, so a retry + multiplies cost instead of recovering. + """ + attempts: dict[str, int] = {} + + async def _fake_execute_activity(activity, _inp, **kwargs): + attempts[activity.__name__] = kwargs["retry_policy"].maximum_attempts + return _payload_for(activity) + + monkeypatch.setattr(workflow, "execute_activity", _fake_execute_activity) + await YouResearchWorkflow().run(ResearchInput(input="q")) + await YouFinanceResearchWorkflow().run(FinanceResearchInput(input="q")) + await YouResearchBackgroundWorkflow().run(ResearchInput(input="q")) + await YouSearchWorkflow().run(SearchInput(query="q")) + await YouAnswerWorkflow().run(AnswerInput(query="q")) + await YouContentsWorkflow().run(ContentsInput(urls=["https://example.com"])) + + assert attempts["youdotcom_research"] == 1 + assert attempts["youdotcom_finance_research"] == 1 + assert attempts["youdotcom_research_background"] == 1 + assert attempts["youdotcom_search"] > 1 + assert attempts["youdotcom_answer"] > 1 + assert attempts["youdotcom_contents"] > 1 + + +async def test_background_ceiling_outlasts_the_activity_wait(monkeypatch): + """The Workflow ceiling must expire *after* the Activity's own wait. + + ``research_and_wait_async`` issues a final GET once its internal timeout + expires, so its wall time is strictly longer than ``timeout_s``. If the + ceiling merely equals that wait, Temporal kills the attempt before the + Activity can report what happened and the Workflow fails with an opaque + activity timeout instead of the real outcome. + """ + captured: dict[str, timedelta] = {} + + async def _fake_execute_activity(_activity, inp, **kwargs): + # timeout_s must arrive untouched: the Activity forwards it to the SDK, + # which derives the deadline from research_effort. A value substituted + # anywhere on this path takes over a decision the SDK makes correctly, + # and capped frontier tasks at two minutes when it last happened. + assert inp.timeout_s is None + captured[inp.research_effort] = kwargs["start_to_close_timeout"] + return _payload_for(_activity) + + monkeypatch.setattr(workflow, "execute_activity", _fake_execute_activity) + for effort in ("lite", "standard", "deep", "exhaustive", "frontier"): + await YouResearchBackgroundWorkflow().run( + ResearchInput(input="q", research_effort=effort) + ) + + assert set(captured) == {"lite", "standard", "deep", "exhaustive", "frontier"} + for effort, ceiling in captured.items(): + wait_s = _BACKGROUND_TIMEOUT_S_BY_EFFORT.get(effort, _BACKGROUND_TIMEOUT_S) + assert wait_s < ceiling.total_seconds(), ( + f"{effort}: activity waits {wait_s}s but the ceiling is " + f"{ceiling.total_seconds()}s -- the attempt is killed first" + ) + + +async def test_background_research_respects_an_explicit_timeout(monkeypatch): + """A caller-supplied timeout_s must win over the effort-based default.""" + + async def _fake_execute_activity(_activity, inp, **_kwargs): + _fake_execute_activity.inp = inp + return _payload_for(_activity) + + monkeypatch.setattr(workflow, "execute_activity", _fake_execute_activity) + await YouResearchBackgroundWorkflow().run( + ResearchInput(input="q", research_effort="frontier", timeout_s=30.0) + ) + assert _fake_execute_activity.inp.timeout_s == 30.0 + + +def test_validation_error_names_fields_but_not_values(): + """A parse failure must not carry You.com content across the boundary. + + ``_validate``'s message becomes the caller's failure and lands in Workflow + history. Pydantic renders the offending input inline by default, so a + malformed response would otherwise spill upstream content to whoever called + the Operation. + """ + from youdotcom.models import TaskDetail + + from youdotcom_temporal.workflows import _validate + + secret = "CONFIDENTIAL-RESPONSE-CONTENT" + with pytest.raises(ApplicationError) as caught: + _validate(TaskDetail, {"id": "x", "leaked": secret}) + + message = str(caught.value) + assert secret not in message + assert "CONFIDENTIAL" not in message + # still diagnostic: it says which fields were wrong + assert "task_type" in message + assert "Field required" in message + + +def test_validation_error_is_non_retryable(): + """A shape mismatch will not fix itself, so it must not spin.""" + from youdotcom.models import TaskDetail + + from youdotcom_temporal.workflows import _validate + + with pytest.raises(ApplicationError) as caught: + _validate(TaskDetail, {}) + assert caught.value.type == "YouResponseShapeError" + assert caught.value.non_retryable + + +async def test_contents_workflow_rejects_a_response_without_results(monkeypatch): + """A missing 'results' key must fail cleanly, not raise KeyError. + + A KeyError inside a Workflow is a Workflow task failure, which Temporal + retries indefinitely -- the same hang _validate exists to prevent. + """ + + async def _fake_execute_activity(_activity, _inp, **_kwargs): + return {"unexpected": "shape"} + + monkeypatch.setattr(workflow, "execute_activity", _fake_execute_activity) + with pytest.raises(ApplicationError) as caught: + await YouContentsWorkflow().run(ContentsInput(urls=["https://example.com"])) + assert caught.value.type == "YouResponseShapeError" + assert caught.value.non_retryable + + +def test_contract_import_does_not_pull_in_the_handler(): + """A caller needs the types, not the implementation. + + Importing the contract must not load the backing Workflows or the Activity + layer. Those exist only on the handler side, and a caller in another + Namespace has no use for them. Checked in a subprocess because this test + session has already imported both. + """ + code = ( + "import sys; import youdotcom_temporal.contract as c; m = set(sys.modules); " + "print(int('youdotcom_temporal.workflows' in m), " + "int('youdotcom_temporal.activities' in m))" + ) + out = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=True + ) + workflows_loaded, activities_loaded = out.stdout.split() + assert workflows_loaded == "0", "contract import pulled in the backing Workflows" + assert activities_loaded == "0", "contract import pulled in the Activity layer" + + +class TestWorkflowIdDerivation: + """Properties the idempotency key depends on. + + The backing Workflow Id *is* the deduplication mechanism, so these are + correctness guarantees rather than implementation detail. + """ + + _id = staticmethod(YouDotComServiceHandler._workflow_id) + + def test_same_key_and_request_is_stable(self): + inp = SearchInput(query="q", count=3) + assert self._id("p", "k", inp) == self._id("p", "k", inp) + + def test_same_key_different_request_differs(self): + """The property that stops one caller receiving another's results.""" + assert self._id("p", "k", SearchInput(query="one")) != self._id( + "p", "k", SearchInput(query="two") + ) + + def test_different_key_same_request_differs(self): + inp = SearchInput(query="q") + assert self._id("p", "a", inp) != self._id("p", "b", inp) + + def test_no_key_is_unique_per_call(self): + inp = SearchInput(query="q") + assert self._id("p", None, inp) != self._id("p", None, inp) + + def test_dict_key_order_does_not_change_the_id(self): + """Serialization has to be canonical or dedup silently stops working.""" + a = ResearchInput(input="q", source_control={"a": 1, "b": 2}) + b = ResearchInput(input="q", source_control={"b": 2, "a": 1}) + assert self._id("p", "k", a) == self._id("p", "k", b) + + def test_long_keys_stay_within_temporal_limits(self): + """An oversized key must not push the Id past what the server accepts.""" + assert len(self._id("you-search", "x" * 5000, SearchInput(query="q"))) < 200 + + def test_keys_sharing_a_prefix_do_not_collide(self): + """The readable part is truncated, so the digest must cover the full key.""" + inp = SearchInput(query="q") + assert self._id("p", "x" * 64 + "A", inp) != self._id("p", "x" * 64 + "B", inp) + + def test_awkward_payloads_do_not_raise(self): + """A crash here fails the Operation start, so it must handle real inputs.""" + for inp in ( + SearchInput(query="café — \x00 😀"), + ResearchInput(input="q", output_schema={"type": "object"}), + ContentsInput(urls=["https://a.com", "https://b.com"]), + ): + assert self._id("p", "k", inp) diff --git a/tests/test_nexus_integration.py b/tests/test_nexus_integration.py new file mode 100644 index 0000000..2d42e84 --- /dev/null +++ b/tests/test_nexus_integration.py @@ -0,0 +1,534 @@ +"""End-to-end Nexus round trip against a real Temporal server. + +Everything else in the suite checks the pieces in isolation: that the contract +has the right shape, that the Workflows register under the sandbox, that the +timeouts and retry policies are what we think. None of that invokes Nexus. + +This does. A real Endpoint, a handler Worker hosting the Service, and a +*separate* caller Worker that registers no plugin and no Activities -- the split +a consumer in another Namespace actually has. Each Operation is driven from the +caller side through the Endpoint, down to the Activity, and back. +""" + +from __future__ import annotations + +import asyncio +import shutil +import uuid +from datetime import timedelta +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +import temporalio.api.nexus.v1 as nx +import temporalio.api.operatorservice.v1 as op +import temporalio.api.workflowservice.v1 as ws +from _nexus_caller_workflows import ( + ENDPOINT, + CallAnswer, + CallContents, + CallFinanceResearch, + CallResearch, + CallResearchBackground, + CallSearch, + CallSearchIdempotent, + caller_workflows, +) +from google.protobuf import duration_pb2 +from temporalio.client import Client, WorkflowFailureError +from temporalio.contrib.pydantic import pydantic_data_converter +from temporalio.exceptions import ApplicationError, NexusOperationError +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker + +from youdotcom_temporal import YouPlugin +from youdotcom_temporal.nexus import you_nexus_service_handler +from youdotcom_temporal.workflows import you_nexus_workflows + +pytestmark = pytest.mark.skipif( + shutil.which("temporal") is None, + reason="Nexus round trip needs the local Temporal dev server CLI", +) + +HANDLER_QUEUE = "you-nexus-handler" +CALLER_QUEUE = "you-nexus-caller" + + +class _Resp: + def __init__(self, payload: dict[str, Any]) -> None: + self._payload = payload + + def model_dump(self, mode: str = "json") -> dict[str, Any]: + return self._payload + + +def _mock_you_client(*_a: Any, **_kw: Any) -> Any: + """A You client returning realistically shaped, identifiable payloads. + + The shapes mirror the real API so the typed result objects populate the same + way they do in production; the marker string identifies which Operation the + caller actually reached. + """ + you = MagicMock() + you.search_async = AsyncMock( + return_value=_Resp( + { + "results": { + "web": [ + { + "url": "https://example.com", + "title": "search", + "description": "d", + "snippets": ["s"], + } + ] + }, + "metadata": {"search_uuid": "u", "query": "search", "latency": 1.0}, + } + ) + ) + you.answer_async = AsyncMock( + return_value=_Resp({"answer": "answer", "citations": [{"source": "s"}]}) + ) + you.contents_async = AsyncMock( + return_value=[_Resp({"markdown": "contents"})] + ) + you.research_async = AsyncMock( + return_value=_Resp( + { + "output": { + "content": "research", + "content_type": "text", + "sources": [], + }, + "warnings": [], + } + ) + ) + you.finance_research_async = AsyncMock( + return_value=_Resp( + { + "output": { + "content": "finance_research", + "content_type": "text", + "sources": [], + } + } + ) + ) + cm = AsyncMock() + cm.__aenter__ = AsyncMock(return_value=you) + cm.__aexit__ = AsyncMock(return_value=None) + return cm + + +async def _research_and_wait(*_a: Any, **_kw: Any) -> Any: + return _Resp( + { + "id": "task-1", + "task_type": "research", + "status": "completed", + "created_at": "2026-08-19T00:00:00Z", + "updated_at": "2026-08-19T00:00:00Z", + "result": {"content": "research_background"}, + } + ) + + +# (caller Workflow, argument, marker, how to read it off the typed result) +_CASES = [ + (CallSearch, "a query", "search", lambda r: r.results.web[0].title), + (CallAnswer, "a question", "answer", lambda r: r.answer), + (CallContents, "https://example.com", "contents", lambda r: r.results[0].markdown), + (CallResearch, "a topic", "research", lambda r: r.output.content), + (CallFinanceResearch, "a ticker", "finance_research", lambda r: r.output.content), + (CallResearchBackground, "a topic", "research_background", lambda r: r.result.content), +] + + +async def _start_server() -> WorkflowEnvironment: + return await WorkflowEnvironment.start_local( + dev_server_existing_path=shutil.which("temporal"), + dev_server_extra_args=["--dynamic-config-value", "system.enableNexus=true"], + # SDK response models are pydantic, so both sides need this converter. + data_converter=pydantic_data_converter, + ) + + +async def _create_endpoint(client: Client, namespace: str, task_queue: str) -> None: + await client.operator_service.create_nexus_endpoint( + op.CreateNexusEndpointRequest( + spec=nx.EndpointSpec( + name=ENDPOINT, + target=nx.EndpointTarget( + worker=nx.EndpointTarget.Worker( + namespace=namespace, task_queue=task_queue + ) + ), + ) + ) + ) + + +async def _register_namespace(client: Client, name: str) -> None: + """Register a Namespace and wait for it to become usable. + + Registration is asynchronous server-side, so connecting immediately after + the RPC returns is a race. + """ + retention = duration_pb2.Duration() + retention.FromTimedelta(timedelta(days=1)) + await client.service_client.workflow_service.register_namespace( + ws.RegisterNamespaceRequest( + namespace=name, workflow_execution_retention_period=retention + ) + ) + for _ in range(80): + try: + await client.service_client.workflow_service.describe_namespace( + ws.DescribeNamespaceRequest(namespace=name) + ) + return + except Exception: # noqa: BLE001 - propagation delay, keep polling + await asyncio.sleep(0.25) + raise AssertionError(f"namespace {name!r} never became available") + + +@pytest.fixture +async def nexus_env(monkeypatch): + """Dev server + registered Endpoint + handler Worker + caller Worker.""" + monkeypatch.setenv("YDC_API_KEY", "test-key") + env = await _start_server() + async with env: + await _create_endpoint(env.client, env.client.namespace, HANDLER_QUEUE) + with patch( + "youdotcom_temporal.activities.you_client", side_effect=_mock_you_client + ), patch( + "youdotcom_temporal.activities.research_and_wait_async", + new=_research_and_wait, + ): + # Handler side: hosts the Service, the backing Workflows, and (via + # the plugin) the Activities. + async with Worker( + env.client, + task_queue=HANDLER_QUEUE, + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ): + # Caller side: no plugin, no Activities, no Nexus handler. It + # can only reach You.com through the Endpoint. + async with Worker( + env.client, + task_queue=CALLER_QUEUE, + workflows=caller_workflows(), + ): + yield env + + +@pytest.mark.parametrize( + "caller,arg,expected,read", _CASES, ids=[c[2] for c in _CASES] +) +async def test_operation_round_trips_through_a_nexus_endpoint( + nexus_env, caller, arg, expected, read +): + """Each Operation resolves caller -> Endpoint -> Workflow -> Activity -> caller. + + The marker is read off a *typed* field of the result, so an Operation wired + to the wrong backing Workflow fails rather than passing on a shape match. + """ + result = await nexus_env.client.execute_workflow( + caller.run, + arg, + id=f"nexus-{expected}-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + assert read(result) == expected + + +async def test_operation_starts_a_backing_workflow(nexus_env): + """The Operation must be workflow-backed, not answered inline. + + This is the whole design premise -- if an Operation ever resolved without a + backing Workflow it would be subject to the 10s sync handler deadline. + """ + wf_id = f"nexus-backing-{uuid.uuid4()}" + await nexus_env.client.execute_workflow( + CallSearch.run, "a query", id=wf_id, task_queue=CALLER_QUEUE + ) + + backing = [ + wf + async for wf in nexus_env.client.list_workflows( + 'WorkflowType = "YouSearchWorkflow"' + ) + ] + assert backing, "no YouSearchWorkflow was started; Operation was not workflow-backed" + assert backing[0].task_queue == HANDLER_QUEUE + + +async def test_operation_crosses_a_real_namespace_boundary(monkeypatch): + """The headline claim: a caller in a *different* Namespace can call this. + + The other tests use one Namespace with two Task Queues, which exercises the + Endpoint but not the boundary. Here the caller and handler are genuinely + separate Namespaces with separate Clients, and the caller holds no You.com + credentials of its own. + """ + monkeypatch.setenv("YDC_API_KEY", "test-key") + handler_ns, caller_ns = "you-handler-test", "you-caller-test" + + env = await _start_server() + async with env: + host = env.client.service_client.config.target_host + await _register_namespace(env.client, handler_ns) + await _register_namespace(env.client, caller_ns) + + handler_client = await Client.connect( + host, namespace=handler_ns, data_converter=pydantic_data_converter + ) + caller_client = await Client.connect( + host, namespace=caller_ns, data_converter=pydantic_data_converter + ) + # The point of this test: guard against it being quietly collapsed + # back to a single Namespace, which would still pass everything below. + assert handler_client.namespace != caller_client.namespace + await _create_endpoint(handler_client, handler_ns, HANDLER_QUEUE) + + with patch( + "youdotcom_temporal.activities.you_client", side_effect=_mock_you_client + ): + async with Worker( + handler_client, + task_queue=HANDLER_QUEUE, + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ): + async with Worker( + caller_client, + task_queue=CALLER_QUEUE, + workflows=caller_workflows(), + ): + result = await caller_client.execute_workflow( + CallSearch.run, + "a query", + id=f"nexus-xns-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + + assert result.results.web[0].title == "search" + + +@pytest.fixture +async def failing_nexus_env(monkeypatch): + """Same topology, but the Activity fails with a mapped, non-retryable error.""" + monkeypatch.delenv("YDC_API_KEY", raising=False) + env = await _start_server() + async with env: + await _create_endpoint(env.client, env.client.namespace, HANDLER_QUEUE) + async with Worker( + env.client, + task_queue=HANDLER_QUEUE, + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ): + async with Worker( + env.client, task_queue=CALLER_QUEUE, workflows=caller_workflows() + ): + yield env + + +async def test_activity_failure_reaches_the_caller_with_its_type(failing_nexus_env): + """A mapped error must survive the Nexus boundary with its type intact. + + The Activity layer classifies failures (``YouAuthError``, + ``YouValidationError``, ``YouQuotaExhausted``) so callers can branch on + them. That is only useful if the classification survives being carried + across Nexus, which is several failure wrappers deep. + """ + with pytest.raises(WorkflowFailureError) as caught: + await failing_nexus_env.client.execute_workflow( + CallSearch.run, + "a query", + id=f"nexus-fail-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + + chain, err = [], caught.value + while err is not None: + chain.append(err) + err = err.__cause__ + + assert any(isinstance(e, NexusOperationError) for e in chain), ( + f"failure did not surface as a Nexus operation failure: " + f"{[type(e).__name__ for e in chain]}" + ) + app = next((e for e in chain if isinstance(e, ApplicationError)), None) + assert app is not None, [type(e).__name__ for e in chain] + assert app.type == "YouAuthError", app.type + assert app.non_retryable + + +async def test_idempotency_key_deduplicates_the_backing_workflow(nexus_env): + """The same key must resolve to one backing Workflow, not two. + + Without this, a retried Nexus StartOperation request starts a second + Workflow and pays for a second You.com call. The key makes the backing + Workflow Id deterministic, so the retry attaches to the run already in + flight instead. + """ + key = f"order-{uuid.uuid4()}" + + # Two callers, same key, issued concurrently -- the shape a StartOperation + # retry takes. + await asyncio.gather( + *[ + nexus_env.client.execute_workflow( + CallSearchIdempotent.run, + ["a query", key], + id=f"nexus-idem-{i}-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + for i in range(2) + ] + ) + + # Assert the behaviour, not the Id format: same key and same request must + # resolve to a single backing Workflow. + backing = [ + wf + async for wf in nexus_env.client.list_workflows( + 'WorkflowType = "YouSearchWorkflow"' + ) + ] + assert len(backing) == 1, ( + f"expected one backing Workflow for key {key!r}, found {len(backing)}: " + f"{[w.id for w in backing]}" + ) + assert key[:64] in backing[0].id, ( + f"Id {backing[0].id!r} should still carry the key for recognisability" + ) + + +async def test_without_a_key_each_call_gets_its_own_workflow(nexus_env): + """The default stays non-idempotent, which is right for one-off calls.""" + await asyncio.gather( + *[ + nexus_env.client.execute_workflow( + CallSearch.run, + "a query", + id=f"nexus-nokey-{i}-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + for i in range(2) + ] + ) + backing = [ + wf + async for wf in nexus_env.client.list_workflows( + 'WorkflowType = "YouSearchWorkflow"' + ) + ] + assert len(backing) == 2, f"expected two distinct Workflows, found {len(backing)}" + + +async def _malformed_research_and_wait(*_a: Any, **_kw: Any) -> Any: + """A background research task whose payload TaskDetail cannot accept. + + TaskDetail has required fields, unlike SearchResponse whose fields are all + optional, so it is the model that actually exercises a parse failure. + """ + return _Resp({"unexpected": "shape"}) + + +@pytest.fixture +async def malformed_nexus_env(monkeypatch): + monkeypatch.setenv("YDC_API_KEY", "test-key") + env = await _start_server() + async with env: + await _create_endpoint(env.client, env.client.namespace, HANDLER_QUEUE) + with patch( + "youdotcom_temporal.activities.you_client", side_effect=_mock_you_client + ), patch( + "youdotcom_temporal.activities.research_and_wait_async", + new=_malformed_research_and_wait, + ): + async with Worker( + env.client, + task_queue=HANDLER_QUEUE, + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ): + async with Worker( + env.client, task_queue=CALLER_QUEUE, workflows=caller_workflows() + ): + yield env + + +async def test_unparseable_response_fails_the_caller_instead_of_hanging( + malformed_nexus_env, +): + """An unexpected response shape must surface, not spin. + + Parsing happens in the Workflow, and a bare pydantic error there is a + Workflow task failure -- which Temporal retries forever. The caller would + wait out its whole schedule_to_close on a request that can never succeed. + It has to come back as a real, non-retryable error instead. + """ + with pytest.raises(WorkflowFailureError) as caught: + await malformed_nexus_env.client.execute_workflow( + CallResearchBackground.run, + "a topic", + id=f"nexus-malformed-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + + chain, err = [], caught.value + while err is not None: + chain.append(err) + err = err.__cause__ + app = next((e for e in chain if isinstance(e, ApplicationError)), None) + assert app is not None, [type(e).__name__ for e in chain] + assert app.type == "YouResponseShapeError", app.type + assert app.non_retryable + + +async def test_same_key_different_input_does_not_cross_wire_results(nexus_env): + """A key must not hand one caller another caller's results. + + The backing Workflow Id is derived from the caller-supplied key. Two callers + that pick the same key string for *different* requests would otherwise + collide, and USE_EXISTING would attach the second to the first's Workflow -- + returning the first caller's answer to the second. + """ + key = f"shared-{uuid.uuid4()}" + + first, second = await asyncio.gather( + nexus_env.client.execute_workflow( + CallSearchIdempotent.run, + ["query-one", key], + id=f"nexus-collide-a-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ), + nexus_env.client.execute_workflow( + CallSearchIdempotent.run, + ["query-two", key], + id=f"nexus-collide-b-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ), + ) + + backing = [ + wf + async for wf in nexus_env.client.list_workflows( + 'WorkflowType = "YouSearchWorkflow"' + ) + ] + assert len(backing) == 2, ( + f"different inputs shared a backing Workflow: {[w.id for w in backing]} -- " + "one caller received another caller's results" + ) + assert first is not None and second is not None diff --git a/tests/test_nexus_integration_live.py b/tests/test_nexus_integration_live.py new file mode 100644 index 0000000..670f1f9 --- /dev/null +++ b/tests/test_nexus_integration_live.py @@ -0,0 +1,153 @@ +"""Nexus round trip against the real You.com API. + +``test_nexus_integration`` proves the Nexus wiring with mocked responses. That +leaves one thing unproven: real API payloads crossing the Nexus boundary. A +mock returns a two-key dict; a real ``search`` response is a large nested +document that has to survive ``model_dump(mode="json")``, the Temporal payload +converter, the Operation completion callback, and deserialization on the caller +side. + +Only the fast Operations run here. ``research``, ``finance_research`` and +``research_background`` take minutes to hours and cost real money per call, so +they stay mocked; the Activity layer already covers them in +``test_integration``. + +Run with: uv run pytest -m integration +""" + +from __future__ import annotations + +import json +import os +import shutil +import uuid + +import pytest +from _nexus_caller_workflows import ( + CallAnswer, + CallContents, + CallContentsMany, + CallSearch, + caller_workflows, +) +from temporalio.worker import Worker +from test_nexus_integration import ( # reuse the topology helpers + CALLER_QUEUE, + HANDLER_QUEUE, + _create_endpoint, + _start_server, +) + +from youdotcom_temporal import YouPlugin +from youdotcom_temporal.nexus import you_nexus_service_handler +from youdotcom_temporal.workflows import you_nexus_workflows + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not os.environ.get("YDC_API_KEY"), + reason="hits the real You.com API; requires YDC_API_KEY", + ), + pytest.mark.skipif( + shutil.which("temporal") is None, + reason="needs the local Temporal dev server CLI", + ), +] + +# Temporal refuses payloads over 2 MiB and warns above 512 KiB. Measured worst +# cases: search count=100 ~107 KB, contents with 5 large pages ~281 KB. +_WARN_BYTES = 512 * 1024 +_LIMIT_BYTES = 2 * 1024 * 1024 + + +@pytest.fixture +async def live_nexus_env(): + """Real Endpoint and Workers, with no mocking of the You.com client.""" + env = await _start_server() + async with env: + await _create_endpoint(env.client, env.client.namespace, HANDLER_QUEUE) + async with Worker( + env.client, + task_queue=HANDLER_QUEUE, + workflows=you_nexus_workflows(), + nexus_service_handlers=[you_nexus_service_handler()], + plugins=[YouPlugin()], + ): + async with Worker( + env.client, task_queue=CALLER_QUEUE, workflows=caller_workflows() + ): + yield env + + +@pytest.mark.parametrize( + "caller,arg,probe", + [ + (CallSearch, "temporal workflow engine", "search"), + (CallAnswer, "what is a Temporal workflow?", "answer"), + (CallContents, "https://temporal.io", "contents"), + ], + ids=["search", "answer", "contents"], +) +async def test_live_operation_round_trips(live_nexus_env, caller, arg, probe): + """A real You.com response survives the whole Nexus path to the caller.""" + result = await live_nexus_env.client.execute_workflow( + caller.run, + arg, + id=f"nexus-live-{probe}-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + + # The payload made it across intact as typed models, and is genuinely + # JSON-serializable -- anything model_dump could not encode would fail here. + # `contents` is the one Operation with a dataclass envelope around SDK + # models, so it serializes element by element. + if probe == "contents": + payload = [c.model_dump(mode="json") for c in result.results] + else: + assert hasattr(result, "model_dump"), f"{probe} did not come back typed" + payload = result.model_dump(mode="json") + encoded = json.dumps(payload).encode() + assert len(encoded) > 200, f"{probe} returned a suspiciously small payload" + assert len(encoded) < _LIMIT_BYTES, ( + f"{probe} payload is {len(encoded)} bytes, over Temporal's {_LIMIT_BYTES}" + ) + if len(encoded) > _WARN_BYTES: + pytest.fail( + f"{probe} payload {len(encoded)} bytes exceeds Temporal's " + f"{_WARN_BYTES} warning threshold; callers should expect blob warnings" + ) + + +async def test_live_contents_at_fan_out_stays_under_the_payload_limit(live_nexus_env): + """`contents` at fan-out produces our largest payload; guard the ceiling. + + The result carries full page text per URL, so this is the Operation most + likely to grow past Temporal's blob limits. Five content-heavy pages is a + realistic upper-middle case for the 10-URL maximum. + """ + urls = [ + "https://en.wikipedia.org/wiki/Python_(programming_language)", + "https://en.wikipedia.org/wiki/Distributed_computing", + "https://en.wikipedia.org/wiki/Database", + "https://en.wikipedia.org/wiki/Operating_system", + "https://en.wikipedia.org/wiki/Computer_network", + ] + result = await live_nexus_env.client.execute_workflow( + CallContentsMany.run, + urls, + id=f"nexus-live-contents-fanout-{uuid.uuid4()}", + task_queue=CALLER_QUEUE, + ) + + assert len(result.results) == len(urls) + encoded = json.dumps( + [c.model_dump(mode="json") for c in result.results] + ).encode() + assert len(encoded) < _LIMIT_BYTES, ( + f"contents payload {len(encoded):,} bytes exceeds Temporal's " + f"{_LIMIT_BYTES:,}; the Operation cannot return this much" + ) + print( + f"\ncontents x{len(urls)}: {len(encoded):,} bytes " + f"({100 * len(encoded) / _LIMIT_BYTES:.1f}% of the 2 MiB limit)" + ) From a6d2dec7b244d5f2e4713acff33f8b17d0e761de Mon Sep 17 00:00:00 2001 From: Tyler Eastman Date: Fri, 21 Aug 2026 14:32:09 -0700 Subject: [PATCH 2/4] fix: update stale DX-776 comments, fix annotated_types passthrough, correct timeout_s comment - contract.py, nexus.py: SDK 3.1.2 shipped lazy imports (PEP 562), so the imports_passed_through() wrapper is belt-and-braces rather than load-bearing. Updated comments that referenced DX-776 as future work. - workflows.py: comment claimed the Activity forwards timeout_s untouched but activities.py substitutes 120s when timeout_s is None, preventing the SDK effort-based deadline derivation. Corrected to describe actual behavior. - plugin.py: re-added annotated_types to _PASSTHROUGH_MODULES, eliminating 13 UserWarning messages about late import under the workflow sandbox. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- src/youdotcom_temporal/contract.py | 12 +++++------- src/youdotcom_temporal/nexus.py | 6 +++--- src/youdotcom_temporal/plugin.py | 1 + src/youdotcom_temporal/workflows.py | 9 ++++++--- 4 files changed, 15 insertions(+), 13 deletions(-) diff --git a/src/youdotcom_temporal/contract.py b/src/youdotcom_temporal/contract.py index 9ee29c1..0fe6c30 100644 --- a/src/youdotcom_temporal/contract.py +++ b/src/youdotcom_temporal/contract.py @@ -14,13 +14,11 @@ client = await Client.connect(..., data_converter=pydantic_data_converter) Sandbox note: - The SDK import below is wrapped in ``imports_passed_through()`` because - ``youdotcom/__init__`` currently imports its transport layer eagerly, which - pulls ``urllib.request`` and would be rejected by the Workflow sandbox. - Wrapping it here means a caller Workflow can import this module directly - without needing the escape hatch itself. Once the SDK resolves its exports - lazily (DX-776) the wrapper becomes belt-and-braces rather than load-bearing, - and importing this contract stops pulling the HTTP stack at all. + The SDK import below is wrapped in ``imports_passed_through()`` so a caller + Workflow can import this module directly without needing its own escape + hatch. The SDK has resolved its exports lazily (PEP 562) since 3.1.2, so + ``import youdotcom`` no longer pulls ``urllib.request`` or ``httpx`` into + ``sys.modules``; the wrapper is belt-and-braces rather than load-bearing. """ from __future__ import annotations diff --git a/src/youdotcom_temporal/nexus.py b/src/youdotcom_temporal/nexus.py index 74d323e..966b20f 100644 --- a/src/youdotcom_temporal/nexus.py +++ b/src/youdotcom_temporal/nexus.py @@ -26,9 +26,9 @@ Either import is safe inside a Workflow sandbox without an escape hatch, as the contract wraps its own SDK import. It does load the You.com SDK, which - is deliberate -- results are the SDK's response models. Once the SDK - resolves its exports lazily (DX-776) that import stops pulling the HTTP - stack with it. + is deliberate -- results are the SDK's response models. The SDK has resolved + its exports lazily (PEP 562) since 3.1.2, so that import no longer pulls the + HTTP stack with it. Cancellation: Cancelling an Operation cancels the backing Workflow, but not the upstream diff --git a/src/youdotcom_temporal/plugin.py b/src/youdotcom_temporal/plugin.py index 8cf58f5..fe114f3 100644 --- a/src/youdotcom_temporal/plugin.py +++ b/src/youdotcom_temporal/plugin.py @@ -14,6 +14,7 @@ "httpcore", "pydantic", "pydantic_core", + "annotated_types", "certifi", "anyio", "sniffio", diff --git a/src/youdotcom_temporal/workflows.py b/src/youdotcom_temporal/workflows.py index 60dc45e..d59ccaa 100644 --- a/src/youdotcom_temporal/workflows.py +++ b/src/youdotcom_temporal/workflows.py @@ -128,9 +128,12 @@ _RETRY_RESEARCH = RetryPolicy(maximum_attempts=1) # The deadline research_and_wait_async derives from research_effort when -# ResearchInput.timeout_s is None. Mirrored here only to size the ceiling below; -# the Activity forwards timeout_s untouched, so the SDK remains the one place -# that decides the deadline. If the SDK's own values change, these follow. +# ResearchInput.timeout_s is None (600s default, 14400s for frontier). The +# Activity currently substitutes 120s when timeout_s is None, so the SDK's +# effort-based derivation does not fire; these constants size the ceiling for +# the case where that fallback is removed and the SDK derives the deadline +# itself. The ceiling (4h15m) is well above either value, so it is safe either +# way. _BACKGROUND_TIMEOUT_S = 600.0 _BACKGROUND_TIMEOUT_S_BY_EFFORT = {"frontier": 14400.0} From 98f52c286bfb4ab682aa80375d3674dc8112766e Mon Sep 17 00:00:00 2001 From: Tyler Eastman Date: Fri, 21 Aug 2026 14:59:52 -0700 Subject: [PATCH 3/4] fix: update workflows.py comment after PR #17 rebase The Activity now forwards timeout_s as-is (PR #17 merged), so the SDK derives the effort-based deadline itself. Updated the comment that described the old 120s substitution behavior. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- src/youdotcom_temporal/workflows.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/youdotcom_temporal/workflows.py b/src/youdotcom_temporal/workflows.py index d59ccaa..d313bf5 100644 --- a/src/youdotcom_temporal/workflows.py +++ b/src/youdotcom_temporal/workflows.py @@ -129,11 +129,9 @@ # The deadline research_and_wait_async derives from research_effort when # ResearchInput.timeout_s is None (600s default, 14400s for frontier). The -# Activity currently substitutes 120s when timeout_s is None, so the SDK's -# effort-based derivation does not fire; these constants size the ceiling for -# the case where that fallback is removed and the SDK derives the deadline -# itself. The ceiling (4h15m) is well above either value, so it is safe either -# way. +# Activity forwards timeout_s as-is, so the SDK derives the deadline itself. +# These constants mirror the SDK's own values to size the ceiling below; if +# the SDK's values change, these follow. _BACKGROUND_TIMEOUT_S = 600.0 _BACKGROUND_TIMEOUT_S_BY_EFFORT = {"frontier": 14400.0} From 64891d22ee4b90c9516c568d813469a84f7d9f63 Mon Sep 17 00:00:00 2001 From: Tyler Eastman Date: Tue, 1 Sep 2026 13:24:52 -0700 Subject: [PATCH 4/4] fix: contents Operation result model and research background guard ContentsOutput wrapped the wrong SDK model: Contents (the search extraction shape) instead of ContentsResponse (the contents endpoint response), silently dropping url, title, and metadata from every result element. The research Operation accepted ResearchInput.background=True, but with background=True the SDK returns a task handle that can never validate as the Operation's ResearchResponse result. It now rejects background=True with a non-retryable YouValidationError before any billable call; research_background is the Operation for that mode. Also restructures the CHANGELOG Unreleased section so it merges cleanly against main's 1.1.0 release. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- CHANGELOG.md | 20 ++++++--- README.md | 12 ++--- src/youdotcom_temporal/contract.py | 8 ++-- src/youdotcom_temporal/workflows.py | 17 ++++++- tests/test_nexus.py | 67 ++++++++++++++++++++++++++++ tests/test_nexus_integration.py | 10 ++++- tests/test_nexus_integration_live.py | 5 +++ 7 files changed, 119 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6970b36..aafdca6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,18 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Nexus Service support: `youdotcom_temporal.nexus.YouDotComService` exposes all six Activities as asynchronous, workflow-backed Nexus Operations callable across Namespace boundaries through a Nexus Endpoint -- `youdotcom_temporal.contract` holds the Nexus contract: a request type per Operation carrying the Activity input plus an optional `idempotency_key`, and result types that are the You.com SDK's own response models, so callers get accurate nested types the SDK maintains. `contents` keeps a thin `ContentsOutput` envelope because the SDK returns a bare list. Callers must configure `temporalio.contrib.pydantic.pydantic_data_converter` +- `youdotcom_temporal.contract` holds the Nexus contract: a request type per Operation carrying the Activity input plus an optional `idempotency_key`, and result types that are the You.com SDK's own response models, so callers get accurate nested types the SDK maintains. `contents` keeps a thin `ContentsOutput` envelope because the SDK returns a bare list; its elements are `ContentsResponse` models, not the search-extraction `Contents` model, which would have silently dropped `url`, `title`, and `metadata` from every element. Callers must configure `temporalio.contrib.pydantic.pydantic_data_converter` - Supplying `idempotency_key` makes the backing Workflow Id deterministic and starts it with `WorkflowIDConflictPolicy.USE_EXISTING`, so a retried Nexus StartOperation request attaches to the run already in flight instead of starting a second Workflow and paying for a second You.com call. Deduplication holds against a *running* Workflow; without a key, starts are not deduplicated - `youdotcom_temporal.workflows` ships six thin backing Workflows (`YouSearchWorkflow`, `YouAnswerWorkflow`, `YouContentsWorkflow`, `YouResearchWorkflow`, `YouFinanceResearchWorkflow`, `YouResearchBackgroundWorkflow`), each wrapping its Activity with a per-Activity `start_to_close_timeout` carrying generous headroom, since a ceiling is a retry backstop rather than a latency target. `search` and `contents` are sized off the 60s per-URL maximum a caller can request via `crawl_timeout` +- The `research` Operation rejects `background=True` with a non-retryable `YouValidationError`. With `background=True` the SDK returns a task handle, which can never validate as the Operation's `ResearchResponse` result—the caller would have paid for the research task and then received an opaque `YouResponseShapeError`. `research_background` is the Operation for background mode - `you_nexus_service_handler()` and `you_nexus_workflows()` helpers for Worker registration - `examples/run_nexus_worker.py` handler-side Worker example - Unit tests covering the Nexus Service contract and handler -- `SearchInput.extraction` field accepts the SDK's new `extraction` object (`{"extraction_mode": "highlights" | "full_page", ...}`). When set, it takes priority over the deprecated `livecrawl` / `livecrawl_formats` fields and is passed to `you.search_async(extraction=...)` instead, avoiding the SDK's `ValueError` on dual-set. The legacy fields remain accepted for backward compatibility - -### Changed -- Python SDK floor bumped to `youdotcom>=3.1.2,<4` (was `>=3.0.0,<4`). The 3.1.2 release ships the `X-Client-Info` attribution header and the `app_name` / `app_version` / `app_title` / `app_url` constructor kwargs on `You` -- The plugin now passes `app_name="youdotcom-temporal"` and `app_version=` to the `You(...)` constructor instead of mutating `client.sdk_configuration.user_agent` post-construction. Each outbound request carries `X-Client-Info: sdk; client=youdotcom-temporal/; ua=python/ httpx/`; the SDK's own `user-agent` stays as `youdotcom-python-sdk/`. The `_USER_AGENT` constant is removed -- `youdotcom_research_background` now forwards `timeout_s` as-is (including `None`) to `research_and_wait_async` instead of substituting `120.0`. When `timeout_s` is `None`, the SDK derives an effort-based default (600s for standard, 14400s for frontier) via `_resolve_default_timeout()`. Previously the `120.0` fallback prevented that derivation, capping every effort tier at 120s +- Cross-Namespace round-trip tests against a local dev server (mocked You.com) and live round-trip tests for the fast Operations (`pytest -m integration`) ### Notes - Nexus is an opt-in layer: importing `youdotcom_temporal.nexus` does not affect Activity-only users @@ -30,6 +26,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Known limits, tracked before release: cancelling an Operation does not stop the in-flight You.com call - Responses are parsed into SDK models on the handler side. A response that does not match raises a non-retryable `YouResponseShapeError`, because an unguarded parse error inside a Workflow is a Workflow task failure that Temporal would retry indefinitely, hanging the caller +## [1.1.0] — 2026-08-21 + +### Added +- `SearchInput.extraction` field accepts the SDK's new `extraction` object (`{"extraction_mode": "highlights" | "full_page", ...}`). When set, it takes priority over the deprecated `livecrawl` / `livecrawl_formats` fields and is passed to `you.search_async(extraction=...)` instead, avoiding the SDK's `ValueError` on dual-set. The legacy fields remain accepted for backward compatibility + +### Changed +- Python SDK floor bumped to `youdotcom>=3.1.2,<4` (was `>=3.0.0,<4`). The 3.1.2 release ships the `X-Client-Info` attribution header and the `app_name` / `app_version` / `app_title` / `app_url` constructor kwargs on `You` +- The plugin now passes `app_name="youdotcom-temporal"` and `app_version=` to the `You(...)` constructor instead of mutating `client.sdk_configuration.user_agent` post-construction. Each outbound request carries `X-Client-Info: sdk; client=youdotcom-temporal/; ua=python/ httpx/`; the SDK's own `user-agent` stays as `youdotcom-python-sdk/`. The `_USER_AGENT` constant is removed +- `youdotcom_research_background` now forwards `timeout_s` as-is (including `None`) to `research_and_wait_async` instead of substituting `120.0`. When `timeout_s` is `None`, the SDK derives an effort-based default (600s for standard, 14400s for frontier) via `_resolve_default_timeout()`. Previously the `120.0` fallback prevented that derivation, capping every effort tier at 120s + ## [1.0.1] — 2026-08-18 ### Fixed diff --git a/README.md b/README.md index f122746..4846431 100644 --- a/README.md +++ b/README.md @@ -173,7 +173,7 @@ async def main(): await worker.run() ``` -Create a Nexus Endpoint targeting that Worker, then call an Operation from a caller Workflow in another Namespace. Callers import from `youdotcom_temporal.contract`, which carries the types and none of the handler. No plugin and no sandbox escape are needed — the contract wraps its own SDK import — so these are safe at module scope: +Create a Nexus Endpoint targeting that Worker, then call an Operation from a caller Workflow in another Namespace. Callers import from `youdotcom_temporal.contract`, which carries the types and none of the handler. No plugin and no sandbox escape are needed, because the contract wraps its own SDK import, so these are safe at module scope: ```python from datetime import timedelta @@ -217,19 +217,19 @@ class CallerWorkflow: | `finance_research` | `FinanceResearchRequest` | `FinanceResearchResponse` | `YouFinanceResearchWorkflow` | | `research_background` | `ResearchRequest` | `TaskDetail` | `YouResearchBackgroundWorkflow` | -Every request carries the matching Activity input plus an optional `idempotency_key`. Results are the You.com SDK's own response models, imported from `youdotcom_temporal.contract`, so callers get accurate nested types the SDK maintains. `contents` is the exception: the SDK returns a bare list, which could not gain fields later without breaking callers, so it keeps a thin `ContentsOutput` envelope whose elements are still SDK `Contents` models. +Every request carries the matching Activity input plus an optional `idempotency_key`. Results are the You.com SDK's own response models, imported from `youdotcom_temporal.contract`, so callers get accurate nested types the SDK maintains. `contents` is the exception: the SDK returns a bare list, which could not gain fields later without breaking callers, so it keeps a thin `ContentsOutput` envelope whose elements are still SDK `ContentsResponse` models. `research` rejects `background=True` with a non-retryable `YouValidationError`, and `research_background` is the Operation for background mode. -Each backing Workflow runs the Activity with a per-Activity `start_to_close_timeout`. A ceiling is a backstop rather than a latency target — it is where the Activity gives up and lets Temporal retry — so each carries generous headroom. `search` is 120s and `contents` 180s, sized off the 60s per-URL maximum a caller can request via `crawl_timeout` (`contents` gets more, since it accepts up to 10 URLs per request); `answer` is 60s; `research` is 10 minutes, `finance_research` 30 minutes, and `research_background` 4h15m. +Each backing Workflow runs the Activity with a per-Activity `start_to_close_timeout`. A ceiling is a backstop rather than a latency target—it is where the Activity gives up and lets Temporal retry—so each carries generous headroom. `search` is 120s and `contents` 180s, sized off the 60s per-URL maximum a caller can request via `crawl_timeout` (`contents` gets more, since it accepts up to 10 URLs per request). `answer` is 60s, `research` is 10 minutes, `finance_research` is 30 minutes, and `research_background` is 4h15m. -That ceiling is **per attempt** — `search`, `answer`, and `contents` retry up to 3 times, so size the caller's `schedule_to_close_timeout` against the ceiling times the attempt count. The research Operations do not retry: each attempt submits a new billable research task, and the previous one keeps running. +That ceiling is **per attempt**—`search`, `answer`, and `contents` retry up to 3 times, so size the caller's `schedule_to_close_timeout` against the ceiling times the attempt count. The research Operations do not retry: each attempt submits a new billable research task, and the previous one keeps running. See the [Temporal Python Nexus quickstart](https://docs.temporal.io/develop/python/nexus/quickstart) for Endpoint and caller-Namespace setup, and `examples/run_nexus_worker.py` for a runnable handler-side Worker. **Known limits (draft):** -- **Cancellation does not reach You.com, and cannot.** The You.com API exposes no cancellation — the research surface is `POST /v1/research` plus two GETs to poll or stream, with no DELETE — so a submitted request runs to completion and is billed regardless. Cancelling an Operation frees the backing Workflow and the Worker slot, nothing upstream. +- **Cancellation does not reach You.com, and cannot.** The You.com API exposes no cancellation—the research surface is `POST /v1/research` plus two GETs to poll or stream, with no DELETE—so a submitted request runs to completion and is billed regardless. Cancelling an Operation frees the backing Workflow and the Worker slot, nothing upstream. - **Idempotency is opt-in and bounded.** Supplying `idempotency_key` deduplicates against a Workflow that is still *running*, which covers the StartOperation-retry case. A key reused after the first Operation completed starts a fresh run. Without a key, starts are not deduplicated at all. -- **An unparseable response fails the Operation.** Results are parsed into SDK models on the handler side; a response that does not match raises a non-retryable `YouResponseShapeError` rather than hanging the caller. +- **An unparseable response fails the Operation.** Results are parsed into SDK models on the handler side, and a response that does not match raises a non-retryable `YouResponseShapeError` rather than hanging the caller. ## Error handling diff --git a/src/youdotcom_temporal/contract.py b/src/youdotcom_temporal/contract.py index 0fe6c30..1cec912 100644 --- a/src/youdotcom_temporal/contract.py +++ b/src/youdotcom_temporal/contract.py @@ -31,7 +31,7 @@ with workflow.unsafe.imports_passed_through(): from youdotcom.models import ( AnswerResponse, - Contents, + ContentsResponse, FinanceResearchResponse, ResearchResponse, SearchResponse, @@ -49,9 +49,9 @@ __all__ = [ "AnswerRequest", "AnswerResponse", - "Contents", "ContentsOutput", "ContentsRequest", + "ContentsResponse", "FinanceResearchRequest", "FinanceResearchResponse", "ResearchRequest", @@ -122,9 +122,9 @@ class FinanceResearchRequest: @dataclass class ContentsOutput: - """One `Contents` per requested URL, in request order.""" + """One `ContentsResponse` per requested URL, in request order.""" - results: list[Contents] + results: list[ContentsResponse] # --------------------------------------------------------------------------- # diff --git a/src/youdotcom_temporal/workflows.py b/src/youdotcom_temporal/workflows.py index d313bf5..6088249 100644 --- a/src/youdotcom_temporal/workflows.py +++ b/src/youdotcom_temporal/workflows.py @@ -58,8 +58,8 @@ ) from youdotcom_temporal.contract import ( AnswerResponse, - Contents, ContentsOutput, + ContentsResponse, FinanceResearchResponse, ResearchResponse, SearchResponse, @@ -231,13 +231,26 @@ async def run(self, inp: ContentsInput) -> ContentsOutput: type="YouResponseShapeError", non_retryable=True, ) - return ContentsOutput(results=[_validate(Contents, c) for c in documents]) + return ContentsOutput( + results=[_validate(ContentsResponse, c) for c in documents] + ) @workflow.defn class YouResearchWorkflow: @workflow.run async def run(self, inp: ResearchInput) -> ResearchResponse: + # `background=True` returns a task handle, which can never validate as + # this Operation's ResearchResponse result -- the caller would pay for + # the research task and then get an opaque shape error. + # `research_background` is the Operation for that mode. + if inp.background: + raise ApplicationError( + "You.com research Operation does not accept background=True; " + "use the research_background Operation.", + type="YouValidationError", + non_retryable=True, + ) payload = await workflow.execute_activity( youdotcom_research, inp, diff --git a/tests/test_nexus.py b/tests/test_nexus.py index f3d7405..7fa68c3 100644 --- a/tests/test_nexus.py +++ b/tests/test_nexus.py @@ -155,6 +155,51 @@ def test_result_types_are_sdk_models_or_thin_envelopes(): assert issubclass(result_type, BaseModel), op_name +def test_contents_result_elements_are_the_contents_endpoint_model(): + """`contents` elements must carry the fields the contents API returns. + + The contents endpoint returns `ContentsResponse` elements (`url`, `title`, + `html`, `markdown`, `metadata`). The SDK's `Contents` model is the search + extraction shape (`html`, `markdown`, `highlights`) -- validating against + it silently dropped `url`, `title`, and `metadata` from every result. + """ + from typing import get_type_hints + + from youdotcom.models import ContentsResponse + + assert get_type_hints(ContentsOutput)["results"].__args__[0] is ContentsResponse + + +async def test_contents_workflow_preserves_the_full_response_shape(monkeypatch): + """Every field the contents endpoint returns must survive the Workflow. + + A validation model that is a subset of the real response drops fields + silently -- the caller receives elements with no `url`, `title`, or + `metadata` and no error anywhere. + """ + element = { + "url": "https://example.com", + "title": "Example", + "html": "

hi

", + "markdown": "# hi", + "metadata": {"site_name": "Example"}, + } + + async def _fake_execute_activity(_activity, _inp, **_kwargs): + return {"results": [dict(element)]} + + monkeypatch.setattr(workflow, "execute_activity", _fake_execute_activity) + out = await YouContentsWorkflow().run( + ContentsInput(urls=["https://example.com"]) + ) + dumped = out.results[0].model_dump(mode="json") + assert dumped["url"] == "https://example.com" + assert dumped["title"] == "Example" + assert dumped["html"] == "

hi

" + assert dumped["markdown"] == "# hi" + assert dumped["metadata"]["site_name"] == "Example" + + def test_result_types_are_serializable_by_the_pydantic_converter(): """The contract only works if Temporal can carry these types. @@ -478,3 +523,25 @@ def test_awkward_payloads_do_not_raise(self): ContentsInput(urls=["https://a.com", "https://b.com"]), ): assert self._id("p", "k", inp) + + +async def test_research_operation_rejects_background_mode(monkeypatch): + """`research` must refuse `background=True` before any billable call. + + `ResearchInput.background` exists for the Activity layer, but with + `background=True` the SDK returns a task handle (TaskResponse), which can + never validate as this Operation's `ResearchResponse` result. Without this + check the caller pays for the research task and then receives an opaque + `YouResponseShapeError`. `research_background` is the Operation for that + mode, and it waits for and returns the completed task. + """ + + async def _fake_execute_activity(_activity, _inp, **_kwargs): + raise AssertionError("the Activity must not run") + + monkeypatch.setattr(workflow, "execute_activity", _fake_execute_activity) + with pytest.raises(ApplicationError) as caught: + await YouResearchWorkflow().run(ResearchInput(input="q", background=True)) + assert caught.value.type == "YouValidationError" + assert caught.value.non_retryable + assert "research_background" in str(caught.value) diff --git a/tests/test_nexus_integration.py b/tests/test_nexus_integration.py index 2d42e84..379eff7 100644 --- a/tests/test_nexus_integration.py +++ b/tests/test_nexus_integration.py @@ -91,7 +91,15 @@ def _mock_you_client(*_a: Any, **_kw: Any) -> Any: return_value=_Resp({"answer": "answer", "citations": [{"source": "s"}]}) ) you.contents_async = AsyncMock( - return_value=[_Resp({"markdown": "contents"})] + return_value=[ + _Resp( + { + "url": "https://example.com", + "title": "contents", + "markdown": "contents", + } + ) + ] ) you.research_async = AsyncMock( return_value=_Resp( diff --git a/tests/test_nexus_integration_live.py b/tests/test_nexus_integration_live.py index 670f1f9..b9c9370 100644 --- a/tests/test_nexus_integration_live.py +++ b/tests/test_nexus_integration_live.py @@ -102,6 +102,11 @@ async def test_live_operation_round_trips(live_nexus_env, caller, arg, probe): # `contents` is the one Operation with a dataclass envelope around SDK # models, so it serializes element by element. if probe == "contents": + # The elements are the contents endpoint's own model, so every field + # it returns must survive. Validating against the wrong SDK model + # (the search extraction shape) silently dropped url/title/metadata. + assert result.results[0].url == arg + assert result.results[0].title payload = [c.model_dump(mode="json") for c in result.results] else: assert hasattr(result, "model_dump"), f"{probe} did not come back typed"