Layer 2 — Task Decomposition Format

Field	Value
Layer	2
Status	`draft`
Working Group	task-format

Scope

A standard schema for describing compute tasks that can be decomposed into sub-tasks, routed independently, and reassembled into a composed result. This layer defines the data structures; Layer 1 defines how they are transmitted.

The key words MUST, SHOULD, and MAY are interpreted as in RFC 2119.

1. Task Object

A Task is the atomic unit of work in ACMP. Every acmp/invoke request carries exactly one Task in its params.

{
  "task_id": "task_a7c923f1",
  "capability": "sentiment-analysis",
  "input": {
    "type": "text",
    "data": "The product quality has declined significantly."
  },
  "output_type": "json",
  "input_tokens_est": 500,
  "max_price_cu": 0.005,
  "preferred_tier": "B",
  "metadata": {
    "requester": "agent:hermes-7b:eu-west",
    "dag_id": "dag_f4e2a1",
    "priority": "normal"
  }
}

Task Schema

Field	Type	Required	Description
`task_id`	string	yes	Globally unique task identifier. Generated by the buyer. Format: `task_<random>`. Also acts as the Layer 1 idempotency key.
`capability`	string	yes	Capability tag matching ARD registry entries and the Layer 6 RFQ `capability` field.
`input`	object	yes	A literal input (`{type, data}`) or an input reference (`{source: {...}}`). See §1.1.
`output_type`	string	no	Expected output format. Default: `"json"`.
`input_tokens_est`	integer	no	Buyer’s estimate of input tokens. Informational, not binding.
`max_price_cu`	number	no	Maximum price in CU. Provider may reject if its minimum exceeds this.
`preferred_tier`	string	no	Preferred quality tier (`S`, `A`, `B`). Informational.
`metadata`	object	no	Arbitrary key-value metadata. Reserved keys: `requester`, `dag_id`, `priority`.

1.1 Literal Input vs. Input Reference

A task’s input is one of two shapes:

Literal input — concrete data, used when the task stands alone or is a DAG root:
```
{ "type": "text", "data": "..." }
```
Input reference — a pointer to another task’s output, used inside a DAG (see §3). The source key signals a reference, keeping it cleanly separate from the type taxonomy:
```
{ "source": { "from_task": "task_split_01", "field": "data.chunks[0]" } }
```

A literal input has a type and data; an input reference has a source and no type (the type is whatever the referenced output produces, resolved at runtime). The two forms are mutually exclusive.

Input/Output Types

The type field in a literal input and in output_type uses a simplified MIME-like taxonomy:

Type	Description	Example
`text`	Plain text	Natural language input
`json`	Structured JSON	API payloads, structured results
`image/png`	PNG image	Vision tasks
`image/jpeg`	JPEG image	Vision tasks
`audio/wav`	WAV audio	Transcription tasks
`binary`	Opaque binary	Provider-specific formats

Implementations SHOULD accept any string as type — the above are conventions, not an exhaustive list.

2. Task States

A task progresses through a defined set of states:

pending → running → completed
                  → failed
                  → cancelled

State	Description
`pending`	Task created but not yet started by the provider.
`running`	Provider has acknowledged the task and is executing.
`completed`	Task finished successfully. `acmp/result` was sent.
`failed`	Task failed. `acmp/error` was sent.
`cancelled`	Task was cancelled via `acmp/cancel`.

State transitions are one-way. A completed, failed, or cancelled task is terminal.

3. DAG Format

For composite work, ACMP defines a Directed Acyclic Graph (DAG) format. A DAG decomposes a high-level job into sub-tasks with explicit data dependencies.

The DAG is an orchestration plan, not a wire format. There is no acmp/invokeDag message. The DAG is held and executed by the buyer’s orchestrator, which walks the graph and emits one Layer 1 acmp/invoke per task — potentially to different providers. Providers never see the DAG; they only ever receive individual tasks. A DAG with an empty edges array is therefore simply a batch of independent tasks.

{
  "dag_id": "dag_f4e2a1",
  "tasks": [
    {
      "task_id": "task_split_01",
      "capability": "text-split",
      "input": {
        "type": "text",
        "data": "Long document text here..."
      },
      "output_type": "json"
    },
    {
      "task_id": "task_sent_02a",
      "capability": "sentiment-analysis",
      "input": {
        "source": {"from_task": "task_split_01", "field": "data.chunks[0]"}
      },
      "output_type": "json",
      "input_tokens_est": 200
    },
    {
      "task_id": "task_sent_02b",
      "capability": "sentiment-analysis",
      "input": {
        "source": {"from_task": "task_split_01", "field": "data.chunks[1]"}
      },
      "output_type": "json",
      "input_tokens_est": 200
    },
    {
      "task_id": "task_agg_03",
      "capability": "result-aggregation",
      "input": {
        "source": {"from_tasks": ["task_sent_02a", "task_sent_02b"]}
      },
      "output_type": "json"
    }
  ],
  "edges": [
    {"from": "task_split_01", "to": "task_sent_02a"},
    {"from": "task_split_01", "to": "task_sent_02b"},
    {"from": "task_sent_02a", "to": "task_agg_03"},
    {"from": "task_sent_02b", "to": "task_agg_03"}
  ]
}

DAG Schema

Field	Type	Required	Description
`dag_id`	string	yes	Unique identifier for the DAG. Format: `dag_<random>`.
`tasks`	array	yes	List of Task objects (see §1).
`edges`	array	yes	List of Edge objects defining data dependencies. MAY be empty (batch of independent tasks).

Edge Schema

Field	Type	Required	Description
`from`	string	yes	`task_id` of the upstream task.
`to`	string	yes	`task_id` of the downstream task.
`stream_eligible`	boolean	no	If `true`, the downstream task may begin processing on partial output from `from`. Default: `false`. See §5.

Conditional branching (running a downstream task only if some condition holds) is intentionally not part of the edge schema — see Design Decisions. An edge always represents an unconditional data dependency.

DAG Constraints

The graph MUST be acyclic. An implementation MUST reject DAGs with cycles.
Every task_id referenced in edges MUST exist in the tasks array.
Every from_task / from_tasks referenced in an input source MUST correspond to an upstream edge into the referencing task.
Tasks with no incoming edges are root tasks — they start immediately with literal input.
Tasks with no outgoing edges are leaf tasks — their outputs form the DAG’s result.

Input References

When a task’s input depends on another task’s output, the input uses the source form (introduced in §1.1):

{
  "source": {
    "from_task": "task_split_01",
    "field": "data.chunks[0]"
  }
}

Field	Type	Description
`from_task`	string	The task_id whose output to reference. Mutually exclusive with `from_tasks`.
`from_tasks`	array	Multiple task_ids (for aggregation inputs). The orchestrator passes their outputs as an ordered array. Mutually exclusive with `from_task`.
`field`	string	Path into the referenced output. Optional — if absent, the entire `acmp/result` output object is passed. See path grammar below.

Path grammar. The field value is evaluated against the referenced task’s full acmp/result output object (i.e. {type, data}). Therefore paths into the payload begin with data. Supported syntax (a deliberately small subset of JSONPath):

data — the whole output payload
data.foo — object member access
data.foo.bar — nested member access
data.items[0] — array index access
data.items[0].name — combined

Implementations MUST support this subset. Wildcards, slices, and filters are out of scope for v0.1.

The orchestrator resolves every source input into a concrete literal {type, data} before invoking the downstream task. The provider therefore always receives a literal input.

4. Cancellation Propagation

When a task in a DAG is cancelled:

The cancelled task transitions to cancelled state.
All downstream dependents (reachable via outgoing edges, BFS traversal) transition to cancelled.
For each in-flight downstream task, the orchestrator sends acmp/cancel (Layer 1) to the relevant provider.
Upstream tasks that are already completed are not affected — their results remain valid.

When a task fails:

The failed task transitions to failed state.
Downstream dependents that cannot execute without the failed task’s output transition to cancelled.
Sibling tasks (tasks with no dependency on the failed task) continue executing unless the orchestrator decides to cancel the entire DAG.

The orchestrator chooses the failure policy: fail-fast (cancel entire DAG on first failure) or best-effort (continue independent branches). This is a configuration choice, not specified by the protocol.

5. Streaming Within DAGs

A DAG edge with stream_eligible: true allows the downstream task to begin processing before the upstream task completes. This is built entirely on Layer 1 streaming primitives:

The upstream task MUST be invoked with stream: true (Layer 1 §7.1), and its provider must have advertised output_streaming.
The downstream task MUST be invoked with input_stream: true (Layer 1 §7.2), and its provider must have advertised input_streaming.
As the orchestrator receives each acmp/streamChunk from the upstream provider, it forwards it as an acmp/inputChunk to the downstream provider.
When the upstream sends its final chunk, the orchestrator sends a final acmp/inputChunk to the downstream.

Field extraction does not apply to streamed edges. A field path (e.g. data.chunks[0]) cannot be reliably evaluated against an incomplete JSON stream. Therefore an edge with stream_eligible: true MUST reference the whole upstream output — its source MUST omit field (or use field: "data"). If a field sub-selection is required, the edge cannot stream and MUST run in non-streaming mode regardless of stream_eligible.

If either provider lacks the required streaming feature, the orchestrator MUST fall back to non-streaming behaviour: wait for the upstream acmp/result, then invoke the downstream with literal input. stream_eligible is an optimization, never a correctness requirement.

If stream_eligible is false (default), the orchestrator always waits for the upstream task’s acmp/result before invoking the downstream task.

6. Example Payloads

6.1 Single Task: Sentiment Analysis

This matches the scenario from the RFC-0001 sequence diagram.

Invoke:

{
  "jsonrpc": "2.0",
  "method": "acmp/invoke",
  "id": "req_001",
  "params": {
    "task_id": "task_sa_001",
    "capability": "sentiment-analysis",
    "input": {
      "type": "text",
      "data": "Revenue grew 12% YoY but margins compressed due to rising input costs."
    },
    "output_type": "json",
    "input_tokens_est": 120,
    "max_price_cu": 0.005,
    "preferred_tier": "B",
    "timeout_ms": 800,
    "escrow_id": "esc_a3f9c2",
    "proof_method": "result-hash"
  }
}

Result:

{
  "jsonrpc": "2.0",
  "id": "req_001",
  "result": {
    "task_id": "task_sa_001",
    "output": {
      "type": "json",
      "data": {
        "sentiment": "mixed",
        "confidence": 0.87,
        "aspects": [
          {"topic": "revenue", "sentiment": "positive", "confidence": 0.95},
          {"topic": "margins", "sentiment": "negative", "confidence": 0.91}
        ]
      }
    },
    "tokens_used": 115,
    "cost_cu": 0.003,
    "proof": {
      "method": "result-hash",
      "hash": "sha256:a1b2c3d4e5f6..."
    }
  }
}

6.2 DAG: Text Decomposition Pipeline

A buyer splits a long document into chunks, runs sentiment analysis on each chunk in parallel, then aggregates the results.

[text-split] → [sentiment-02a] → [aggregate]
             → [sentiment-02b] ↗

See §3 for the full DAG JSON. The orchestrator (buyer-side) executes it as follows:

Invokes task_split_01 (root task) with literal input.
On completion, resolves the source inputs for task_sent_02a and task_sent_02b — extracting data.chunks[0] and data.chunks[1] from the split result.
Invokes both sentiment tasks in parallel via separate acmp/invoke calls (different providers allowed).
On completion of both, resolves the from_tasks aggregation input for task_agg_03.
Invokes the aggregation task.
Returns task_agg_03’s output as the DAG result.

Design Decisions

Question	Decision	Rationale
Explicit graph or implicit ordering?	Explicit graph (edges)	Explicit edges are unambiguous, support parallel execution, and make dependencies visible to tooling.
Cancellation model?	BFS propagation to downstream	Upstream results stay valid (already paid for). Only dependent tasks are cancelled.
How to express input references?	Separate `source` form, not `type: "ref"`	Overloading the MIME-like `type` field for control flow was ambiguous. A distinct `source` key keeps data typing and dependency wiring orthogonal.
`field` path grammar?	Small JSONPath subset, rooted at `output` (`data....`)	A pinned-down grammar makes resolvers interoperable; the full JSONPath spec is overkill for v0.1.
Conditional branching in DAGs?	No — orchestrator’s responsibility	Keeps the DAG declarative. The orchestrator can inspect results and submit follow-up DAGs. Edges are always unconditional data dependencies.
How is a DAG transmitted?	It isn’t — it’s a buyer-side orchestration plan	Providers receive only individual `acmp/invoke` tasks. This keeps Layer 1 simple and lets each task be routed independently.
Streaming within DAGs?	Built on Layer 1 `inputChunk`/`streamChunk`, with non-streaming fallback	Reuses transport primitives; `stream_eligible` is a pure optimization that never affects correctness.

Open Questions

[OPEN] What is the maximum DAG depth/breadth? Should the protocol define limits to prevent resource exhaustion?
[OPEN] How are DAG-level pricing and billing aggregated — sum of all task costs, or a single escrow for the entire DAG? (Coordinated with Layer 4.)

RFC-0001 §2 — Core Use Cases (Task Delegation, Capability Brokering)
Layer 1 — Transport & Invocation — How tasks are transmitted
Layer 3 — Proof of Execution — How task results are verified
Layer 6 — Negotiation Protocol — RFQ field naming conventions

This document is part of the A2Agora specification. Licensed under Apache 2.0.