Task Adapter

The typed-task half of the capability-unit fracture (pass-2 Thread 3) —

TaskAdapter

Layer-1 of the fractured capability unit is a PAIR: a tool capability (core.capability.ToolCapability — manage the tool) + a task adapter (this base — give one task a typed contract against that tool). The two are separate registered units composed at runtime; co-packaging in one library is authoring convenience, never fusing.

  • The per-task adapter interface library (cjm-<task>-adapter-interface) subclasses this with the typed task method (e.g. transcribe(audio, ...) -> TranscriptionResult), the task’s result DTOs (wire-registered via core.wire), and the task’s persistence helpers (save_with_logging / get_cached storage classes).
  • Adapter implementations run IN-WORKER, co-located with their tool capability (the tool-touching work is in-worker anyway: models on GPU, API secrets pinned per CR-12 WORKER_ENV).
  • required_tool_protocol names the structural contract the adapter needs from a tool (a typing.Protocol defined in the dep-light interface lib). The substrate matches it against each capability’s recorded structural surface (manifest code.structural_surface) — host-side, against UNLOADED capabilities (surface-based, adapter-driven compatibility).
  • Composition logic is explicitly NOT an adapter: code that spans multiple capability outputs or assembles/projects graph data is layer-2 composition (CR-16 ports + CR-18 graph-aware layer) and lives in the workflow cores.

Adapter registry + composition routing land with CR-17 pt 2 (execution stage 4); this base fixes the SHAPE so the first adapter-interface libraries can be born against it.

TaskAdapter


def TaskAdapter(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Base for task adapters — the typed-task half of the capability-unit fracture (pass-2 Thread 3).

Subclasses (one ABC per task, in cjm-<task>-adapter-interface libraries) declare:

  • the TYPED task method (the contract execute(*args, **kwargs) never gave the task), abstract on the per-task ABC;
  • task_name: the task this adapter serves (e.g. “transcription”);
  • required_tool_protocol: the structural contract required of a tool capability (a typing.Protocol; provisional None until the protocol is evidence-locked — Q5 posture: declare the slot, let stage-4/8 tool-splitting evidence finalize the protocol bodies);
  • the task’s persistence helpers (storage classes), beside the task method rather than on it.

Implementations run in-worker beside their tool capability. The base is deliberately mechanism-light: registry/routing is CR-17 pt 2 (stage 4).

# Shape test: a per-task ABC subclasses TaskAdapter with a typed method and
# declares the two ClassVars; a concrete impl fills them in.
from abc import abstractmethod
from typing import Protocol, runtime_checkable


@runtime_checkable
class _EchoToolProtocol(Protocol):
    def echo_native(self, text: str) -> str: ...


class _EchoAdapter(TaskAdapter):
    task_name = "echo"
    required_tool_protocol = _EchoToolProtocol

    @abstractmethod
    def echo(self, text: str) -> str: ...


class _EchoImpl(_EchoAdapter):
    def echo(self, text: str) -> str:
        return text


# The per-task ABC keeps its abstract set (ABCMeta freezes at class creation
# — the NB-1 hazard the fracture had to get right up front).
try:
    _EchoAdapter()  # type: ignore[abstract]
    raise AssertionError("abstract _EchoAdapter must not instantiate")
except TypeError:
    pass

impl = _EchoImpl()
assert impl.echo("hi") == "hi"
assert _EchoImpl.task_name == "echo"
assert _EchoImpl.required_tool_protocol is _EchoToolProtocol
assert TaskAdapter.required_tool_protocol is None  # provisional default
print("TaskAdapter shape test OK")
TaskAdapter shape test OK