fix: correct merge-step readiness, close Java parity gaps, and harden the client

CHANGELOG.md

@@ -5,6 +5,130 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+- Protocol diagnostics now redact credential-shaped values in request and response
+  bodies as well as headers. Workflow failure and callback logs also redact
+  credential-shaped exception text; registry search no longer logs raw intent.
+- Standard A2A HTTP and SSE error envelopes retain their status, reason,
+  retry hint and safe details. An empty response stream fails immediately with
+  `a2a.empty_stream` instead of returning an unattributed unsuccessful result.
+- A malformed subtask `input`, an activated extension missing from AgentCard,
+  or an unavailable explicitly preferred protocol now fails before dispatch.
+  Explicit TLS/mTLS/CRL options are rejected for unsupported non-HTTP transports
+  instead of being silently ignored.
+- Hosts may inject `credential_encryption_key` into `A2ATransport` rather than
+  depending on a process-wide `A2AT_CRED_KEY`; the environment remains the
+  compatibility fallback.
+- `MessageContent.with_extension(uri)` offers an immutable, Java-aligned way
+  to activate one extension for a single request.
+- Registry and PSOP HTTP helpers accept separate connect/read timeout settings
+  instead of hard-coding both to 30 seconds.
+- A **notification stream that died was never released**, and a **healthy one could be
+  reported as dead**. Liveness was judged from decoded business events only, but an SSE
+  comment heartbeat never reaches the protocol layer, so a subscription receiving only
+  heartbeats looked identical to a dead one -- `is_healthy()` returned false and a caller
+  acting on it tore down a live stream, while a genuinely dead stream leaked its
+  connection and task forever. Transport activity is now observed per body chunk and
+  recorded separately from business events.
+- **Sensitive header names were only redacted in the exact spelling the pattern
+  happened to list**: `api_key`, `x_api_key`, `access_session`, `accessSession`, `pwd`
+  and `passwd` all leaked into the opt-in protocol log, while `api-key` and `ApiKey`
+  were caught. Names are now matched after stripping `-` and `_`, and a header
+  registered through `register_sensitive_header` matches in any spelling.
+- A **merge step could start before its own predecessor had run**. Step readiness was
+  judged from *direct* predecessors, and a predecessor that had not been activated yet
+  was filtered out entirely -- so a merge whose branches became ready in different
+  batches ran early, with only part of its inputs, and still reported success. Readiness
+  is now judged from *all active ancestors*, matching the Java engine
+  (`WorkflowExecutor#getAllPredecessors`): an ancestor that is activated but has not
+  produced output is the signal that a branch is still running.
+- A Negotiation-T Propose carried on a **taskless bare message** (A2A-T pre-task
+  negotiation, where the remote proposes before creating any task) was returned to
+  the host as if it were the final business answer: `SendMessageResult.is_success`
+  reported success and `outputs()` exposed the negotiation prompt as the result.
+  Such a message now opens a negotiation, correlated by contextId, and the
+  follow-up send carries no taskId. A bare message whose Negotiation-T metadata is
+  malformed now fails loudly instead of silently passing as a result.
+- When an intermediary (API gateway, load balancer) closed an idle SSE stream
+  while the remote task was still running, the engine cancelled that task. A
+  transport fault during stream consumption now falls back to polling `get_task`
+  every `task_poll_interval_seconds` until the task leaves a non-terminal state or
+  the deadline expires, so a healthy task is awaited instead of destroyed.
+  Cancellation for a genuinely local interaction failure is unchanged.
+- A `Notification-T` subscription whose remote never acknowledged stayed pending
+  forever, pinning the caller's coroutine and the connection. The first
+  acknowledgement is now bounded by `notification_ack_timeout_seconds` (default
+  300s); SSE idleness after the acknowledgement is deliberately not bounded.
+- `sse_normalization` patched `json_format.Parse` with a duplicate keyword
+  argument, so every `StreamResponse` text parse raised `TypeError`. The patch
+  now covers only `ParseDict`, the entry point that actually performs the
+  normalization, which leaves `Parse` untouched for every other caller.
+- A `task` stream event carries a full task snapshot, yet its artifact text was
+  appended on every update, duplicating the response text. Text is now assembled
+  once from the merged artifact map, which also gives a non-empty result to
+  streams that deliver content only through `artifact_update` events.
+- `subscribe_to_task` replaced its accumulated message list on every `task`
+  event, silently dropping standalone messages seen earlier. It now merges, and
+  additionally reports status and artifact updates and terminal failure codes,
+  matching `consume_stream`.
+- `create_ssl_context` silently discarded the CA, mTLS and CRL arguments when
+  `verify_server=False`. A client identity is now still presented, while a trust
+  store or CRL without verification is rejected rather than ignored.
+- Protocol logging redacted headers by name pattern alone, so a credential in a
+  custom `auth_header` was logged in clear. Authentication interceptors now
+  register the exact header names they inject.
+- `failure_mapping` looked the A2A error mapping up by exact type, so a raised
+  subclass fell through to the generic failure code. It now walks the MRO.
+- `env_file_loader` wrote any key from a `.env` file into the process
+  environment; keys must now be valid environment-variable names.
+
+### Changed
+- Remote A2A `error_details` now uses Java-compatible `httpStatus`, `code`,
+  `status`, `reason`, `domain`, `details` and `retryAfter` fields. Callers using
+  the former `http_status` or top-level SDK metadata fields must migrate.
+- `NotificationHeartbeat.last_event_at` now reports the last **transport activity**,
+  including an SSE comment heartbeat, and `event_count` counts only decoded A2A business
+  events. The last business event is available through the new
+  `last_business_event_at`. This matches the Java engine's `NotificationSubscription`.
+- `NotificationSubscription.is_healthy()` judges liveness from transport activity, so a
+  subscription that is only receiving heartbeats is no longer reported as dead. A
+  freshly opened subscription counts as healthy, since opening the stream is activity.
+- A notification stream is now **released after `send_timeout_seconds` without any
+  transport activity** (configurable through
+  `ExtensionSender(notification_idle_timeout_seconds=...)`), mirroring the Java engine's
+  idle budget. `await subscription.completion` raises `TimeoutError` when that happens.
+  A stream that keeps receiving heartbeats is never released. **Hosts that relied on a
+  subscription staying open forever must be updated.**
+- `A2ATransport` accepts `task_poll_interval_seconds` (default 20s, floor 0.1s)
+  and `notification_ack_timeout_seconds` (default 300s), and `execute_psop`
+  forwards the former. These mirror the Java engine's `taskPollIntervalMillis`
+  and `notificationAckTimeoutSeconds`; the poll interval is used *only* after a
+  stream interruption, never while the SSE stream is alive.
+- `consume_stream` includes `task_id` in the `AGENT_STATUS_UPDATE` intermediate
+  event it forwards, so a caller can learn the remote task identity before a
+  stream ends and recover from a dropped connection.
+- `A2ATransport` caches one A2A client per agent through the new
+  `client_for(agent_name)` instead of rebuilding it for every send, task poll
+  and cancellation. `update_agent_cards` drops the cache.
+- `A2ATransport.get_extension_uris` is the public form of the former private
+  `_get_extensions`, which `ExtensionSender` reached into across modules.
+- `WorkflowEngineClient.stream_message` accepts `timeout_seconds` and bounds the
+  whole stream rather than relying on the idle read timeout alone.
+- `credential_crypto` accepts an optional `aad` that binds a ciphertext to its
+  context. It defaults to `None` to stay wire-compatible with the Java SDK.
+- Registry helpers accept a custom CA, mTLS identity and CRL, and
+  `RegistryClient` accepts an access token, so the registry path no longer has
+  less TLS capability than the transport path.
+- `workflow_engine.__version__` is the single source of the distribution version
+  (`pyproject.toml` reads it through `[tool.setuptools.dynamic]`), and the
+  `a2a-t-sdk` lower bound now matches what `verify_sdk.py` asserts.
+
+### Removed
+- Dead `A2ATransport._text_from_metadata`, and a stale
+  `extension_handlers.cpython-312.pyc` whose source was deleted in 0.0.9.
+
 ## [0.1.0] - 2026-09-18
 
 ### Changed

DEVELOPER_GUIDE.md

@@ -4,7 +4,7 @@
 
 - Python 3.12 or newer
 - `a2a-sdk>=1.1.2,<2`
-- `a2a-t-sdk>=1.0.9,<2` (the engine uses its core metadata contract; host callbacks use its content APIs)
+- `a2a-t-sdk>=1.1.0,<2` (the engine uses its core metadata contract; host callbacks use its content APIs)
 
 ```bash
 python -m pip install -e ".[dev]"
@@ -243,25 +243,29 @@ from a2a_t.core.standard_templates import (
 )
 
 auth_transport = A2ATransport(agent_cards, auth_provider=provider)
-sender = ExtensionSender(auth_transport)
+notification_transport = A2ATransport(agent_cards, auth_provider=provider)
+auth_sender = ExtensionSender(auth_transport)
+notification_sender = ExtensionSender(notification_transport)
 
 generated_auth = a2at_client.generate_auth_prompt_from_text(
     authorization_text,
     AUTHORIZATION_POLICY_MANAGEMENT_URI,
 )
 auth_content = A2atMessages.from_generated(generated_auth, [Part(text="authorize policy")])
-auth_result = await sender.send_authorization("domain-a", auth_content)
-if not auth_result.is_success:
-    record_independent_failure(auth_result.failure_code, auth_result.failure_message)
+try:
+    auth_result = await auth_sender.send_authorization("domain-a", auth_content)
+    if not auth_result.is_success:
+        record_independent_failure(auth_result.failure_code, auth_result.failure_message)
+finally:
+    await auth_transport.close()  # The one-shot channel is no longer needed.
 ```
 
 Notification uses a long-lived handle:
 
 ```python
 def on_notification(subscription, received):
     process(received)
-    if is_expected_final_result(received):
-        subscription.close()
+    # A single result does not end the subscription; later notifications may follow.
 
 generated_notification = a2at_client.generate_notification_prompt_from_text(
     subscription_text,
@@ -271,15 +275,18 @@ notification_content = A2atMessages.from_generated(
     generated_notification,
     [Part(text="subscribe to business event")],
 )
-subscription = sender.open_notification("domain-a", notification_content, on_notification)
-ack = await subscription.acknowledgement
-if ack.is_failure:
+subscription = notification_sender.open_notification(
+    "domain-a", notification_content, on_notification,
+)
+try:
+    await subscription.acknowledgement  # A state-bearing ACK; stream remains open.
+    await subscription.completion  # Ends on explicit close, EOF, idle timeout or error.
+finally:
     subscription.close()
-    record_independent_failure(ack.failure_code, ack.failure_message)
-await subscription.completion
+    await notification_transport.close()
 ```
 
-Use a transport instance separate from workflow task traffic. Host orchestration may log an independent-operation failure, but it must not make workflow execution depend on that result unless explicit business policy says so.
+Use three transport instances for workflow tasks, Authorization-T, and Notification-T. Authorization-T is one-shot; release its transport after the result. Notification-T keeps its own transport until the subscription ends. A message or artifact without task state is not an ACK, and a failure state rejects the subscription. Host orchestration may log an independent-operation failure, but it must not make workflow execution depend on that result unless explicit business policy says so.
 
 ## 7. Existing task operations
 
@@ -332,9 +339,15 @@ The provider owns token acquisition. The engine does not read provider-specific
 
 ## 9. TLS and diagnostics
 
-Default TLS verifies the server with system trust. Optional parameters are `ca_certs_path`, `client_cert_path`, `client_key_path`, `client_key_password`, and `crl_path`. Missing or invalid configured files raise during construction.
+Default TLS verifies the server with system trust. Optional parameters are `ca_certs_path`, `client_cert_path`, `client_key_path`, `client_key_password`, and `crl_path`. Missing or invalid configured files raise during construction. Explicit TLS/mTLS/CRL options are rejected when the selected protocol cannot apply the HTTP client configuration; select an HTTP protocol or configure that transport's TLS independently. An explicitly preferred protocol absent from the AgentCard fails instead of falling back silently.
+
+For encrypted agent credentials, pass `credential_encryption_key=<64-hex-character key>` to `A2ATransport` so each host instance can receive its key from a secret manager. `A2AT_CRED_KEY` remains a fallback when no key is injected. Neither key is logged.
+
+Set `WORKFLOW_ENGINE_PROTOCOL_LOGGING=true` to log pretty-printed final A2A request/response objects. Credential-shaped values in bodies are always redacted. Sensitive headers stay redacted unless the separate sensitive-header flag is explicitly enabled. Protocol logs are diagnostic representations; transport internals may still prevent observation of exact wire bytes.
+
+Remote A2A failures use `a2a.<reason>` or `a2a.http.<status>` when no reason is present. `error_details` follows the Java SDK's `httpStatus`, `code`, `status`, `reason`, `domain`, `details`, and `retryAfter` names. A response stream with no protocol event fails as `a2a.empty_stream`.
 
-Set `WORKFLOW_ENGINE_PROTOCOL_LOGGING=true` to log pretty-printed final A2A request/response objects. Sensitive headers stay redacted unless the separate sensitive-header flag is explicitly enabled. Protocol logs are diagnostic representations; transport internals may still prevent observation of exact wire bytes.
+`RegistryClient`, `load_psop`, and `search_psop` accept `connect_timeout_seconds` and `read_timeout_seconds` (both default to 30 seconds). Both must be positive.
 
 ## 10. Workflow retrieval
 

README.md

@@ -2,7 +2,7 @@
 
 面向宿主智能体的嵌入式工作流执行引擎。引擎负责工作流 DAG、A2A 消息封装、任务与会话关联、远端任务等待、Negotiation-T 交互循环和生命周期；宿主负责业务输入解释、A2A-T 内容生成与语义校验，以及路由和协商决策。
 
-Python 与 Java 版本遵循同一业务契约。Python 当前要求 Python 3.12+、`a2a-sdk>=1.1.2,<2` 和 `a2a-t-sdk>=1.0.9,<2`。
+Python 与 Java 版本遵循同一业务契约。Python 当前要求 Python 3.12+、`a2a-sdk>=1.1.2,<2` 和 `a2a-t-sdk>=1.1.0,<2`。
 
 ## 安装
 
@@ -80,20 +80,31 @@ await transport.close()
 
 ## 独立协议操作
 
-Authorization-T 和 Notification-T 不属于工作流因果链，应使用独立 `A2ATransport` 和 `ExtensionSender`：
+Authorization-T 和 Notification-T 不属于工作流因果链。工作流任务、一次性授权和长订阅应分别持有 `A2ATransport`；协议处理能力可以共用，但 HTTP 通道和生命周期不能共用：
 
 ```python
-sender = ExtensionSender(independent_transport)
-authorization = await sender.send_authorization(agent_name, authorization_content)
-
-subscription = sender.open_notification(agent_name, notification_content, on_notification)
-ack = await subscription.acknowledgement
-# 收到业务结果后由集成方显式关闭：
-subscription.close()
-await subscription.completion
+auth_transport = A2ATransport(agent_cards)
+notification_transport = A2ATransport(agent_cards)
+try:
+    authorization = await ExtensionSender(auth_transport).send_authorization(
+        agent_name, authorization_content,
+    )
+finally:
+    await auth_transport.close()  # 授权操作结束即释放，不影响订阅
+
+subscription = ExtensionSender(notification_transport).open_notification(
+    agent_name, notification_content, on_notification,
+)
+try:
+    ack = await subscription.acknowledgement  # 仅表示订阅已接受，流继续保持
+    # 单条通知结果不结束订阅；保持监听，直到明确取消、停机或流异常结束。
+    await subscription.completion
+finally:
+    subscription.close()
+    await notification_transport.close()
 ```
 
-授权或订阅失败不会自动阻断后续工作流。`send_authorization` 会等待授权任务到达终态，集成方通过 `is_success` 判断结果；订阅确认通过 `is_failure` 排除失败、拒绝或取消。订阅确认与长连接结束是两个不同 Future；`heartbeat` 和 `is_healthy()` 可用于本地存活性判断。
+授权或订阅失败不会自动阻断后续工作流。`send_authorization` 会等待授权任务到达终态，集成方通过 `is_success` 判断结果；订阅拒绝会让 `acknowledgement` 抛错，普通消息和 artifact 不算 ACK。订阅确认与长连接结束是两个不同 Future；业务任务终态不会自动关闭独立订阅，持续到达的 SSE 心跳可保持长连接存活。显式关闭、远端 EOF、空闲超时或传输异常会结束流。`heartbeat` 和 `is_healthy()` 可用于本地存活性判断。
 
 ## 远端任务管理
 

README_en.md

@@ -81,7 +81,8 @@ authorization = await sender.send_authorization(agent_name, authorization_conten
 
 subscription = sender.open_notification(agent_name, notification_content, on_notification)
 ack = await subscription.acknowledgement
-subscription.close()
+# Keep receiving results; one notification does not end the subscription.
+# The host may call subscription.close() on explicit cancellation or shutdown.
 await subscription.completion
 ```
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "workflow-exec-engine"
-version = "0.1.0"
+dynamic = ["version"]
 description = "Standalone workflow execution SDK for A2A-T multi-agent orchestration"
 readme = "README.md"
 license = "Apache-2.0"
@@ -24,7 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "a2a-sdk>=1.1.2,<2",
-    "a2a-t-sdk>=1.0.9,<2",
+    "a2a-t-sdk>=1.1.0,<2",
     "httpx>=0.27.0",
     "loguru>=0.7.0",
     "protobuf>=4.25.0",
@@ -49,6 +49,11 @@ Changelog = "https://github.com/project-openan/workflow-engine-sdk-python/blob/m
 [tool.setuptools.packages.find]
 include = ["workflow_engine*"]
 
+# Single source of truth: the version lives in workflow_engine/__init__.py, so
+# the distribution metadata and __version__ can no longer drift apart.
+[tool.setuptools.dynamic]
+version = { attr = "workflow_engine.__version__" }
+
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["."]

requirements.txt

@@ -1,5 +1,5 @@
 a2a-sdk>=1.1.2,<2
-a2a-t-sdk>=1.0.9,<2
+a2a-t-sdk>=1.1.0,<2
 httpx>=0.27.0
 loguru>=0.7.0
 protobuf>=4.25.0