feat(plugins): add DeepSeek Harness agent adapter

[cosmicBboy]

Adds flyteplugins-agents-deepseek — running DeepSeek Harness (deepseek-harness-sdk) agents on Flyte, following the patterns established by the other adapters in plugins/agents/.

It passes the shared assert_adapter_conforms check, so tool + run_agent / run_agent_sync present the same call shape as every other adapter:

@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")

Why the tool layer is different

Worth a look in review, since it departs from the client-side adapters.

DeepSeek Harness has no tool-registration message. 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. (The runtime does have a first-class harness.registerTool(ctx, tool), but it's a TypeScript plugin API the Python SDK can't reach, so it's not available to an adapter working with the stock composition.)

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 dir;
3. the model runs .flyte_tools/get_weather '{"city": "Paris"}', the shim forwards the JSON args over the socket, and the adapter awaits task.aio(...) — a durable child action with its own container/resources, retries and caching.

Since there's no tool-declaration channel either, the tool manual (names, parameter types, an example invocation each) is prepended to the prompt. A failing tool comes back as a non-zero exit with the reason on stderr, so the agent can react instead of the run dying. The shims are the only thing written into the workspace and are removed when the run ends.

DeepSeekHarness.run is blocking, so it's driven via asyncio.to_thread — that's what keeps the event loop free to serve tool calls while the agent works.

Durability, memory, observability

• Durability is session resume rather than per-turn replay, for the same reason as the Claude adapter: the model loop runs in a subprocess Flyte doesn't intercept, so a turn can't be a flyte.trace leaf. The harness's JSONL session store is mirrored onto a flyte.Checkpoint, keyed by a session id derived from the task's action, so a retry continues the conversation instead of restarting. Tool durability/caching applies regardless.
• Memory: with memory_key, the same session archive lives in a keyed MemoryStore instead, giving cross-run memory — which also covers crash-resume, so it supersedes the per-run checkpoint.
• Observability: the runtime's session events (assistant turns with token usage, tool/call / tool/result, turn/end) plus the bridge's per-tool outcomes are rendered into the task report. Notifications arrive on the worker thread and are marshalled back onto the loop before touching the report, so timeline ordering stays correct.
• workspace= points the harness's own bash/editor at a real directory (e.g. a downloaded flyte.io.Dir), which is what the harness is actually built for.

Testing

61 tests, all offline — no network, no API key, no controller.

The bridge tests are the substantive ones: they run the published shim as a real subprocess, exactly as the harness's bash tool would, covering shim → socket → task.aio end to end, including concurrent calls, tool failures, and bad arguments. One run_agent test has a fake harness invoke a shim from inside its blocking run, which is the actual contract — tool calls must be served while run is in flight.

Separately, I smoke-tested that the config the adapter builds launches the real bundled runtime and completes the JSON-RPC initialize handshake (no API key needed for that part), so the subprocess/config wiring is verified against the real SDK rather than only mocks.

Not verified: no live model call was made (no DeepSeek API key available), so whether a real model reliably picks up and invokes the shims from the prompt manual is untested. That's the one part of this design that depends on model behavior rather than mechanism, and it's worth a live run before release.

Also

• Registered in plugins/agents/README.md and the publish workflow matrix. Test CI auto-discovers it (it has a tests/ dir), so no change needed there.
• deepseek-harness-sdk is currently pre-release only (0.1.0rc7), hence prerelease = "allow" in the plugin's [tool.uv]. Its pinned runtime wheel covers Linux x86-64/aarch64 and macOS 14+ arm64, so CI (ubuntu x64) resolves fine.
• Six examples: durable agent (async + sync), crash/resume, real-workspace code-fixing agent, multi-agent pipeline, cross-run memory, and bring-your-own DeepSeekHarnessConfig.

🤖 Generated with Claude Code

[samhita-alla]

can we use asyncio.create_subprocess_exec or asyncio.create_subprocess_shell instead? we could also use flyte.sandbox but i don't think it makes much of a difference in this case.

[samhita-alla]

just wondering if it’s worth supporting this given that it still seems to be in preview and the sdk isn’t as mature. looks like there are a lot of workarounds needed to get it working properly, right? i’m fine with supporting it but just wondering.

[cosmicBboy]

yeah, I think we can just sit on this one until the DeepSeek harness is out of preview