A2A Task 运行时指南：运行、流式传输、恢复与排错

# A2A task trace lint for v1.0.1.
# Usage: jq -s -r -f task-trace-lint.jq trace.jsonl
# A trace is one StreamResponse (or SendMessageResponse) object per line, in arrival order.
# Checks come from the REQUIRED annotations in a2a.proto and specification sections 3.1.2, 3.1.6, 3.2.3, 3.4.3 and 11.7.
# ERROR lines break the specification; WARN lines are advisory.

def blank: . == null or . == "" or . == [];
def states: ["TASK_STATE_SUBMITTED", "TASK_STATE_WORKING", "TASK_STATE_COMPLETED", "TASK_STATE_FAILED",
  "TASK_STATE_CANCELED", "TASK_STATE_INPUT_REQUIRED", "TASK_STATE_REJECTED", "TASK_STATE_AUTH_REQUIRED"];
def terminal: ["TASK_STATE_COMPLETED", "TASK_STATE_FAILED", "TASK_STATE_CANCELED", "TASK_STATE_REJECTED"];
def interrupted: ["TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED"];
def v03_states: {"submitted": "TASK_STATE_SUBMITTED", "working": "TASK_STATE_WORKING",
  "input-required": "TASK_STATE_INPUT_REQUIRED", "completed": "TASK_STATE_COMPLETED",
  "canceled": "TASK_STATE_CANCELED", "failed": "TASK_STATE_FAILED", "rejected": "TASK_STATE_REJECTED",
  "auth-required": "TASK_STATE_AUTH_REQUIRED"};
def v03_roles: {"user": "ROLE_USER", "agent": "ROLE_AGENT"};
def members: ["task", "message", "statusUpdate", "artifactUpdate"];

def part_checks($p):
  . as $part
  | if type != "object" then "ERROR \($p): expected a Part object"
    elif has("kind") or has("file") then "ERROR \($p): v0.3 Part shape; v1.0 sets exactly one of text, raw, url, data"
    else ([("text", "raw", "url", "data") as $k | select($part | has($k))] | length) as $n
      | if $n != 1 then "ERROR \($p): exactly one of text, raw, url, data must be set, found \($n)" else empty end
    end;

def message_checks($p):
  if type != "object" then "ERROR \($p): expected a Message object"
  else
    (if .messageId | blank then "ERROR \($p)/messageId: required string is missing or empty" else empty end),
    (.role as $r
     | if $r == null then "ERROR \($p)/role: required enum is missing"
       elif v03_roles[$r | tostring] then "ERROR \($p)/role: v0.3 value \($r | tojson), use \(v03_roles[$r | tostring])"
       elif (["ROLE_USER", "ROLE_AGENT"] | index($r)) == null then "ERROR \($p)/role: not ROLE_USER or ROLE_AGENT: \($r | tojson)"
       else empty end),
    (if (.parts | type) != "array" or (.parts | length) == 0 then "ERROR \($p)/parts: required array is missing or empty"
     else (.parts | to_entries[] | .key as $k | .value | part_checks("\($p)/parts/\($k)")) end)
  end;

def state_checks($p):
  . as $s
  | if $s == null or $s == "TASK_STATE_UNSPECIFIED" then "ERROR \($p): required state is missing or TASK_STATE_UNSPECIFIED"
    elif (states | index($s)) != null then empty
    elif v03_states[$s | tostring] then "ERROR \($p): v0.3 state name \($s | tojson), use \(v03_states[$s | tostring])"
    else "ERROR \($p): not a TaskState value: \($s | tojson)" end;

def status_checks($p):
  if type != "object" then "ERROR \($p): required object is missing"
  else (.state | state_checks("\($p)/state")), (if .message != null then .message | message_checks("\($p)/message") else empty end) end;

def artifact_checks($p):
  if type != "object" then "ERROR \($p): required object is missing"
  else
    (if .artifactId | blank then "ERROR \($p)/artifactId: required string is missing or empty" else empty end),
    (if (.parts | type) != "array" or (.parts | length) == 0 then "ERROR \($p)/parts: required array is missing or empty"
     else (.parts | to_entries[] | .key as $k | .value | part_checks("\($p)/parts/\($k)")) end)
  end;

def unwrap: if type == "object" and has("jsonrpc") and (.result | type) == "object" then .result else . end;
def kind: if type == "object" then [members[] as $m | select(has($m)) | $m] | .[0] else null end;
def state_of: if type == "object" then ((.task | objects | .status | objects | .state) // (.statusUpdate | objects | .status | objects | .state) // null) else null end;
def shaped: type == "object" and (has("kind") | not) and ([members[] as $m | select(has($m) and (.[$m] | type) == "object")] | length) == 1;

def event_checks($i):
  "event[\($i)]" as $p
  | if type != "object" then "ERROR \($p): expected a JSON object"
    elif has("kind") or (.statusUpdate | type) == "object" and (.statusUpdate | has("final")) then
      "ERROR \($p): v0.3 event shape (kind or final); v1.0 wraps the payload in exactly one of task, message, statusUpdate, artifactUpdate"
    elif ([members[] as $m | select(has($m))] | length) != 1 then
      "ERROR \($p): expected exactly one of task, message, statusUpdate, artifactUpdate, found \([members[] as $m | select(has($m))] | length)"
    elif ([members[] as $m | select(has($m) and (.[$m] | type) != "object")] | length) > 0 then
      "ERROR \($p): the payload must be a JSON object"
    elif has("task") then
      (if .task.id | blank then "ERROR \($p)/task/id: required string is missing or empty" else empty end),
      (.task.status | status_checks("\($p)/task/status"))
    elif has("message") then (.message | message_checks("\($p)/message"))
    elif has("statusUpdate") then
      (.statusUpdate | (if .taskId | blank then "ERROR \($p)/statusUpdate/taskId: required string is missing or empty" else empty end),
        (if .contextId | blank then "ERROR \($p)/statusUpdate/contextId: required string is missing or empty" else empty end),
        (.status | status_checks("\($p)/statusUpdate/status")))
    else
      (.artifactUpdate | (if .taskId | blank then "ERROR \($p)/artifactUpdate/taskId: required string is missing or empty" else empty end),
        (if .contextId | blank then "ERROR \($p)/artifactUpdate/contextId: required string is missing or empty" else empty end),
        (.artifact | artifact_checks("\($p)/artifactUpdate/artifact")))
    end;

# Trace-level rules: order, identity, termination and artifact chunking. They run only when every event is well formed.
def trace_checks:
  . as $e
  | ($e | length) as $n
  | ($e[0].task // {}) as $first
  | ($e | map(state_of)) as $st
  | ($e | map(select(kind == "task" or kind == "statusUpdate") | state_of) | .[-1]) as $last
  | ([range(0; $n) | select($st[.] as $s | $s != null and (terminal | index($s)) != null)] | .[0]) as $end
  | (if $n > 1 and ($e[0] | kind) != "task" and ($e[0] | kind) != "message" then
       "ERROR event[0]: a stream starts with a task or a message, found \($e[0] | kind // "nothing")" else empty end),
    (if $n > 1 and ($e[0] | kind) == "message" then "ERROR event[1]: a message-only stream holds exactly one message" else empty end),
    (range(1; $n) as $i | $e[$i] as $x
      | (if ($x | kind) == "message" and ($e[0] | kind) == "task" then "ERROR event[\($i)]: message inside a task stream; only status and artifact updates follow the task" else empty end),
        (if ($x | kind) == "task" and ($e[0] | kind) == "task" and ($end == null or $i != $end + 1 or $x.task.status.state != $st[$end])
           then "ERROR event[\($i)]: task snapshot inside a stream; only one final snapshot after the terminal state is allowed" else empty end),
        (($x.statusUpdate // $x.artifactUpdate) as $u
          | if $u != null and ($first.id // null) != null and $u.taskId != null and $u.taskId != $first.id
            then "ERROR event[\($i)]/\($x | kind)/taskId: \($u.taskId | tojson) differs from the stream task \($first.id | tojson)" else empty end,
            if $u != null and ($first.contextId // null) != null and $u.contextId != null and $u.contextId != $first.contextId
            then "ERROR event[\($i)]/\($x | kind)/contextId: \($u.contextId | tojson) differs from the stream context \($first.contextId | tojson)" else empty end),
        (if $end != null and $i > $end and ($x | kind) != "task" then "ERROR event[\($i)]: event after terminal state \($st[$end]) at event[\($end)]" else empty end)),
    (reduce range(0; $n) as $i ({seen: {}, closed: {}, out: []};
        ($e[$i]) as $x
        | if ($x | kind) == "task" then .seen += ([$x.task.artifacts[]? | {(.artifactId | tostring): true}] | add // {})
          elif ($x | kind) == "artifactUpdate" then
            ($x.artifactUpdate.artifact.artifactId | tostring) as $id
            | (if $x.artifactUpdate.append == true and (.seen[$id] | not) then .out += ["WARN event[\($i)]/artifactUpdate: append for artifact \($id | tojson) before any earlier chunk"]
               elif $x.artifactUpdate.append == true and .closed[$id] then .out += ["WARN event[\($i)]/artifactUpdate: append for artifact \($id | tojson) after its lastChunk"] else . end)
            | .seen[$id] = true
            | (if $x.artifactUpdate.lastChunk == true then .closed[$id] = true else . end)
          else . end) | .out[]),
    (if $n > 1 and ($e[0] | kind) == "task" and (["TASK_STATE_SUBMITTED", "TASK_STATE_WORKING"] | index($last)) != null
     then "WARN event[\($n - 1)]: the trace ends in \($last) before a terminal or interrupted state; the stream was cut, resubscribe" else empty end);

map(unwrap) as $e
| if ($e | length) == 0 then "ERROR trace: no events"
  else
    ([range(0; $e | length) as $i | $e[$i] | event_checks($i)] + (if $e | all(shaped) then [$e | trace_checks] else [] end))
    | if length == 0 then "OK: \($e | length) event(s), last state \($e | map(state_of) | map(select(. != null)) | .[-1] // "none")" else .[] end
  end