feat(plugins): add DeepSeek Harness agent adapter

.github/workflows/publish.yml

@@ -266,6 +266,7 @@ jobs:
           - "plugins/agents/langgraph"
           - "plugins/agents/pydantic_ai"
           - "plugins/agents/hermes"
+          - "plugins/agents/deepseek"
           - "plugins/lance"
         include:
           - workdir: "plugins/sglang"

plugins/agents/README.md

@@ -19,9 +19,10 @@ Each SDK is a separate, co-located package on a shared core:
 | [`flyteplugins.agents.crewai`](crewai/) | `flyteplugins-agents-crewai` | [CrewAI](https://docs.crewai.com/) |
 | [`flyteplugins.agents.pydantic_ai`](pydantic_ai/) | `flyteplugins-agents-pydantic-ai` | [Pydantic AI](https://ai.pydantic.dev/) (2.x) |
 | [`flyteplugins.agents.hermes`](hermes/) | `flyteplugins-agents-hermes` | [Hermes Agent](https://pypi.org/project/hermes-agent/) (Nous Research; Python ≥3.11) |
+| [`flyteplugins.agents.deepseek`](deepseek/) | `flyteplugins-agents-deepseek` | [DeepSeek Harness](https://deepseek-harness.github.io/deepseek-harness/en/guide/python-sdk) (`deepseek-harness-sdk`) |
 
 ```bash
-pip install flyteplugins-agents-openai   # or -claude / -mistral / -google / -deepagents / -langchain / -langgraph / -crewai / -pydantic-ai
+pip install flyteplugins-agents-openai   # or -claude / -mistral / -google / -deepagents / -langchain / -langgraph / -crewai / -pydantic-ai / -deepseek
 ```
 
 Each adapter has its own README (linked above) with the SDK-specific details.
@@ -38,7 +39,7 @@ across SDKs:
 - Each model turn is recorded for replay by tracing the seam below the SDK's
   loop (`durable=True`), so a crashed/retried run replays completed turns instead of
   re-calling (and re-billing) the model. (Where the SDK runs its loop in a subprocess
-  — Claude — durability is the SDK's own session-resume instead.)
+  — Claude, DeepSeek Harness — durability is the SDK's own session-resume instead.)
 - Cross-run memory via `memory_key` — the conversation continues across separate
   runs and workers, backed by a durable keyed store.
 

plugins/agents/deepseek/README.md

@@ -0,0 +1,173 @@
+# flyteplugins-agents-deepseek
+
+Run [DeepSeek Harness](https://deepseek-harness.github.io/deepseek-harness/en/guide/python-sdk)
+agents on Flyte. You keep writing DeepSeek Harness code; Flyte is the runtime
+underneath.
+
+```bash
+pip install flyteplugins-agents-deepseek
+```
+
+```python
+import flyte
+from flyteplugins.agents.deepseek import tool, run_agent
+
+env = flyte.TaskEnvironment(
+    "deepseek-agent",
+    secrets=[flyte.Secret(key="deepseek_api_key", as_env_var="DEEPSEEK_API_KEY")],
+)
+
+
+@tool
+@env.task(cache="auto", retries=3)
+async def get_weather(city: str) -> str:
+    """Get the current weather for a city."""
+    return f"The weather in {city} is sunny, 22°C."
+
+
+@env.task(report=True, retries=3)
+async def city_agent(question: str) -> str:
+    return await run_agent(question, tools=[get_weather], model="deepseek-v4-flash")
+```
+
+## How it maps to Flyte
+
+- The loop runs inside the harness runtime subprocess (JSON-RPC over stdio).
+  `run_agent` drives it from inside your `@env.task` — the blocking
+  `DeepSeekHarness.run` is bridged off the event loop with `asyncio.to_thread`, so
+  the task stays responsive while the agent works.
+- Tools are Flyte tasks. When the agent calls one it runs as a durable Flyte child
+  action (its own container/resources, retries, caching). The input schema is
+  derived via the Flyte type engine.
+- `workspace=` points the harness's own tools (bash, string editor) at a real
+  directory — for example a `flyte.io.Dir` you downloaded — so the agent can read
+  and edit actual files and you can hand the result downstream as an artifact.
+
+### How tools work here
+
+This adapter differs from the client-side ones, and it's worth knowing why.
+
+DeepSeek Harness has **no tool-registration channel**: its wire protocol is
+`initialize` / `session/prompt`, and its tool surface is whatever its Cordis
+plugin composition provides *inside the runtime subprocess*. There is nowhere to
+hand a Python function.
+
+What every composition does provide is local bash, scoped to a working directory
+the adapter chooses. So the bridge meets it there:
+
+1. each tool is published into `<workspace>/.flyte_tools/<name>` as a small
+   executable shim (stdlib-only Python, run under this process's own interpreter,
+   so the harness runtime needs nothing installed);
+2. `run_agent` listens on a Unix domain socket in a private temp directory;
+3. the model runs `.flyte_tools/get_weather '{"city": "Paris"}'`, the shim forwards
+   the JSON arguments over the socket, and the adapter awaits `task.aio(...)` — a
+   durable Flyte child action — before writing the result back for the shim to print.
+
+Because the harness has no tool-declaration message either, the tool manual
+(names, parameter types, an example invocation per tool) is prepended to the
+prompt. A failing tool comes back as a non-zero exit with the reason on stderr, so
+the agent sees the error and can react rather than the run dying.
+
+The shims are the only thing the adapter writes into your workspace, and they are
+removed when the run ends.
+
+If you own the Cordis composition, the runtime does have a first-class in-process
+tool API (`harness.registerTool(ctx, tool)`) — but it is a TypeScript plugin API
+inside the runtime, not something the Python SDK can reach, so it is not an option
+for an adapter that has to work with the stock composition.
+
+## Durability
+
+Two layers, both real:
+
+- Tool calls are durable Flyte child actions (own container/resources, retries,
+  caching) — always, regardless of `durable`.
+- The conversation survives a crash. With `durable=True`, `run_agent` mirrors the
+  harness's JSONL session store (`session_root` / `DSH_SESSION_ROOT`) onto a
+  `flyte.Checkpoint`. The session id is derived from the task's action, so it is
+  stable across retries; on a retry the prior attempt's transcript is restored into
+  the session root and the same session is prompted, so the harness continues the
+  conversation instead of starting over.
+
+We delegate to the harness's own session persistence because the model loop runs
+in its runtime subprocess (which Flyte doesn't intercept), so a model turn can't be
+a `flyte.trace` leaf the way it is for other client-side SDKs. Session resume is the
+coarser-grained equivalent — whole-session, not per-turn — and it no-ops cleanly
+when there's no checkpoint context (e.g. local runs).
+
+## Observability
+
+`run_agent` renders a timeline into the task report (`report=True`): assistant
+turns and turn endings from the harness's streamed session events, the harness's
+own tool activity, and each Flyte-task tool's arguments and result (or error),
+recorded by the bridge as it dispatches them.
+
+Session notifications arrive on the worker thread running the blocking harness
+call; each is marshalled back onto the event loop before touching the report, so
+the timeline stays correctly ordered against the tool rows.
+
+## Memory
+
+Pass `memory_key` (a user/thread id) for cross-run memory — the agent resumes the
+same conversation across separate runs:
+
+```python
+await run_agent(message, model="deepseek-v4-flash", memory_key="user-alice")
+```
+
+The session archive is persisted to a durable, keyed `MemoryStore` and restored on
+the next run with the same key, which then prompts the same harness session id.
+That also covers crash-resume, so it supersedes the per-run `durable` checkpoint.
+
+## Bring your own configuration
+
+Pass a fully-built `DeepSeekHarnessConfig` as `config=` to keep SDK-native setup —
+a custom `cordis` composition, `base_url` / `api_key`, timeouts. The adapter layers
+only `cwd` (the workspace, where shims are published) and `session_root` (what
+resume and memory mirror) on top; `model` / `provider` / `max_tokens` and any extra
+keyword arguments override fields on it.
+
+## Runtime
+
+The `deepseek-harness-sdk` wheel pins the matching `deepseek-harness-runtime-bin`
+platform wheel, which bundles the single-file `dsh-jsonrpc-agent` executable — so
+`pip install flyteplugins-agents-deepseek` is all the runtime image needs (no
+separate Node.js install). Wheels are published for Linux x86-64/aarch64 and
+macOS 14+ arm64.
+
+Set `DEEPSEEK_API_KEY` in the environment (a Flyte secret, as above), and
+`DEEPSEEK_BASE_URL` if you route through a proxy.
+
+The SDK is currently published as a pre-release, so installing it may require
+`--prerelease=allow` (uv) / `--pre` (pip) depending on your resolver settings.
+
+## Examples
+
+See [`examples/`](examples/):
+
+- [`deepseek_durable_agent.py`](examples/deepseek_durable_agent.py) — a single
+  durable agent: tools as Flyte tasks, assistant turns and tool outcomes in the report,
+  in both the async and sync call forms.
+- [`deepseek_crash_resume.py`](examples/deepseek_crash_resume.py) — crash & resume:
+  the task crashes on its first attempt; on retry the conversation resumes from the
+  `flyte.Checkpoint`-backed session and completed tool calls are cache hits. Run on a
+  backend to see resume.
+- [`deepseek_workspace_agent.py`](examples/deepseek_workspace_agent.py) — a real
+  workspace: the agent reads, edits and tests an actual `flyte.io.Dir` with the
+  harness's bash and editor, verifies its own work through a Flyte-task tool, and the
+  patched directory becomes the task's output.
+- [`deepseek_multi_agent.py`](examples/deepseek_multi_agent.py) — multi-agent
+  orchestration: a planner agent decomposes a topic, researcher agents fan out in
+  parallel, an editor agent synthesizes — each agent its own durable action.
+- [`deepseek_memory.py`](examples/deepseek_memory.py) — cross-run memory: two
+  separate runs share a `memory_key`; the agent learns a fact in run 1 and recalls it
+  in run 2.
+- [`deepseek_custom_agent.py`](examples/deepseek_custom_agent.py) — bring your own
+  `DeepSeekHarnessConfig` (custom Cordis composition, proxied endpoint, timeouts).
+
+## Conformance
+
+This adapter passes the shared `flyteplugins.agents.core.testing.assert_adapter_conforms`
+check — the same one every adapter runs — so it follows the common format
+(`tool` + `run_agent`, tool tasks wired to the resolver) despite a very different
+underlying SDK shape.

plugins/agents/deepseek/examples/deepseek_crash_resume.py

@@ -0,0 +1,85 @@
+"""Crash and resume: durable DeepSeek Harness agent recovery on Flyte.
+
+Shows that a crash mid-run does not restart the agent from scratch. On the first
+attempt the agent does real work (model turns inside the harness runtime + durable
+tool calls), then the worker is killed (simulated). Flyte retries the task; on the
+second attempt:
+
+- the conversation resumes — with ``durable=True`` (the default) ``run_agent``
+  mirrors the harness's JSONL session store to a ``flyte.Checkpoint``, and on the
+  retry it restores that transcript and prompts the same session id, so the
+  harness continues the prior conversation instead of starting over (the session
+  id is derived from the task's action, so every attempt points at one session);
+- completed tool calls are cache hits — each tool is a durable Flyte child action
+  with ``cache="auto"``, so it isn't re-executed on the retry.
+
+Backend only: per-attempt numbers and the previous attempt's checkpoint are provided
+by the platform per attempt — that is where resume is exercised. In ``local`` mode the
+crash is skipped and the example just runs once.
+
+Run:  flyte run deepseek_crash_resume.py resilient_agent --question "What's the weather and population of Paris?"
+      (add `--local` right after `run` to execute locally instead of on the backend)
+"""
+
+import os
+
+import flyte
+
+from flyteplugins.agents.deepseek import run_agent, tool
+
+env = flyte.TaskEnvironment(
+    "deepseek-crash-resume",
+    resources=flyte.Resources(cpu=1),
+    secrets=[flyte.Secret(key="deepseek_api_key", as_env_var="DEEPSEEK_API_KEY")],
+    image=flyte.Image.from_debian_base(name="deepseek-crash-resume").with_local_v2_plugins(
+        ["flyteplugins-agents-core", "flyteplugins-agents-deepseek"]
+    ),
+)
+
+
+@tool
+@env.task(cache="auto", retries=3)
+async def get_weather(city: str) -> str:
+    """Get the current weather for a city."""
+    print(f"  🛠  get_weather EXECUTED for {city} (cache MISS)", flush=True)
+    return f"sunny, 22°C in {city}"
+
+
+@tool
+@env.task(cache="auto", retries=3)
+async def get_population(city: str) -> int:
+    """Get the population of a city."""
+    print(f"  🛠  get_population EXECUTED for {city} (cache MISS)", flush=True)
+    return {"San Francisco": 808988, "Paris": 2102650, "Tokyo": 13929286}.get(city, 1_000_000)
+
+
+@env.task(report=True, retries=2)
+async def resilient_agent(question: str) -> str:
+    attempt = flyte.ctx().attempt_number if flyte.ctx() else 0
+    print(f"▶ resilient_agent attempt {attempt}", flush=True)
+
+    answer = await run_agent(
+        question,
+        tools=[get_weather, get_population],
+        instructions="You are a concise city-facts assistant. Use the provided tools to answer.",
+        model="deepseek-v4-flash",
+        durable=True,  # default — mirror the session to a checkpoint and resume on retry
+    )
+
+    # Simulate a worker crash after the agent did real work, but only on the first
+    # attempt on a backend. ``FLYTE_ATTEMPT_NUMBER`` is only set per attempt on a
+    # backend, so local runs skip the crash and just complete.
+    on_backend = os.environ.get("FLYTE_ATTEMPT_NUMBER") is not None
+    if on_backend and attempt == 0:
+        raise RuntimeError("💥 simulated worker crash (first attempt only)")
+
+    print("✅ completed on retry — tools were cache hits, conversation resumed", flush=True)
+    return answer
+
+
+if __name__ == "__main__":
+    flyte.init_from_config()
+    run = flyte.run(resilient_agent, question="What's the weather and population of Paris?")
+    print(f"View at: {run.url}")
+    run.wait()
+    print(f"Result: {run.outputs()}")

plugins/agents/deepseek/examples/deepseek_custom_agent.py

@@ -0,0 +1,71 @@
+"""Bring your own harness configuration — SDK-native config, Flyte underneath.
+
+The adapter is a thin layer, not a wrapper that hides the SDK. When you need the
+harness configured your way — a custom Cordis plugin composition, a proxied
+endpoint, tighter timeouts — build a ``DeepSeekHarnessConfig`` exactly as you
+would outside Flyte and pass it as ``config=``.
+
+The adapter layers only what it must on top:
+
+- ``cwd`` — the workspace, because that is where the tool shims are published;
+- ``session_root`` — because that is what durable resume and cross-run memory
+  mirror.
+
+Everything else on the config is yours, and ``model`` / ``provider`` /
+``max_tokens`` / ``**kwargs`` on ``run_agent`` override fields on top of it.
+
+Run:  flyte run deepseek_custom_agent.py configured_agent --question "What's the population of Tokyo?"
+      (add `--local` right after `run` to execute locally instead of on the backend)
+"""
+
+import flyte
+from deepseek_harness import DeepSeekHarnessConfig
+
+from flyteplugins.agents.deepseek import run_agent, tool
+
+env = flyte.TaskEnvironment(
+    "deepseek-custom-agent",
+    resources=flyte.Resources(cpu=1),
+    secrets=[flyte.Secret(key="deepseek_api_key", as_env_var="DEEPSEEK_API_KEY")],
+    image=flyte.Image.from_debian_base(name="deepseek-custom-agent").with_local_v2_plugins(
+        ["flyteplugins-agents-core", "flyteplugins-agents-deepseek"]
+    ),
+)
+
+
+@tool
+@env.task(cache="auto", retries=3)
+async def get_population(city: str) -> int:
+    """Get the population of a city."""
+    return {"San Francisco": 808988, "Paris": 2102650, "Tokyo": 13929286}.get(city, 1_000_000)
+
+
+@env.task(report=True, retries=3)
+async def configured_agent(question: str) -> str:
+    """Drive the agent from a hand-built SDK config."""
+    # Native SDK configuration — the same object you would use standalone. Point
+    # ``cordis`` at your own plugin composition to change the harness's own tools,
+    # persistence or provider routes (keep the JSON-RPC server entry in it).
+    config = DeepSeekHarnessConfig(
+        provider="deepseek-official",
+        model="deepseek-v4-flash",
+        max_tokens=49_152,
+        request_timeout_seconds=300.0,
+        # cordis="/opt/harness/my.cordis.yml",
+        # base_url="https://my-proxy.internal/v1",
+    )
+
+    return await run_agent(
+        question,
+        tools=[get_population],
+        instructions="You are a concise city-facts assistant. Use the provided tools to answer.",
+        config=config,
+    )
+
+
+if __name__ == "__main__":
+    flyte.init_from_config()
+    run = flyte.run(configured_agent, question="What's the population of Tokyo?")
+    print(f"View at: {run.url}")
+    run.wait()
+    print(f"Result: {run.outputs()}")