让 Agent 协作
消息送得达，
任务接得住、过程可追溯

void publishProgress(ClientServiceProvider provider, Producer producer,
                     String sessionId, String payload) throws ClientException {
    Message event = provider.newMessageBuilder()
        .setTopic("materials.progress")
        .setLiteTopic(sessionId)
        .setBody(payload.getBytes(java.nio.charset.StandardCharsets.UTF_8))
        .build();
    producer.send(event);
}

// 以下三次调用置于已初始化的业务方法中
publishProgress(provider, producer, "S-001", "缺签章页；event_seq=1");
publishProgress(provider, producer, "S-002", "核验完成；event_seq=1");
publishProgress(provider, producer, "S-001", "已收到补件；event_seq=2");

import json
import uuid

async def ask_about_missing_page(nc, response_subject):
    reply_subject = nc.new_inbox()
    answers = await nc.subscribe(reply_subject)
    await nc.flush()  # 先确认已订阅回复地址
    query = {"type": "query", "data": {
        "id": str(uuid.uuid4()),
        "reply_subject": reply_subject,
        "prompt": "缺签章页：补充材料，还是先核验其余部分？",
    }}
    await nc.publish(response_subject, json.dumps(query).encode())
    try:
        answer = await answers.next_msg(timeout=30)
        return answer.data.decode()  # 本例调用方回复纯文本选择
    finally:
        await answers.unsubscribe()

import json
import nats
from nats.js.api import RetentionPolicy

async def inspect_once():
    nc = await nats.connect("nats://localhost:4222")
    js = nc.jetstream()
    await js.add_stream(name="TASKS", subjects=["tasks.>"])
    await js.add_stream(name="EVENTS", subjects=["events.>"],
                        retention=RetentionPolicy.LIMITS)
    task = {"task_id": "T-001", "session_id": "S-001"}
    await js.publish("tasks.inspect", json.dumps(task).encode(),
                     headers={"Nats-Msg-Id": task["task_id"]})
    worker = await js.pull_subscribe("tasks.inspect", durable="inspect-workers")
    for msg in await worker.fetch(1):
        task = json.loads(msg.data)
        result = {"task_id": task["task_id"], "status": "completed"}
        await js.publish("events." + task["session_id"],
                         json.dumps(result).encode(),
                         headers={"Nats-Msg-Id": task["task_id"] + ":done"})
        await msg.ack()
    await nc.drain()

from datetime import timedelta
import slim_bindings as slim

async def consult_reviewer(local_app, conn_id, group_name):
    cfg = slim.SessionConfig(
        session_type=slim.SessionType.GROUP,
        max_retries=5, interval=timedelta(seconds=5), metadata={},
        mls_settings=slim.MlsSettings(
            header_integrity_validation_percent=100,
            max_seen_control_message_ids_size=None),
    )
    ctx = local_app.create_session(cfg, group_name)
    await ctx.completion.wait_async()
    session = ctx.session
    # 示例邀请两位已获业务授权的跨机构成员
    for name in ["bank-b/verify/agent", "bank-c/review/agent"]:
        peer = slim.Name.from_string(name)
        await local_app.set_route_async(peer, conn_id)
        invited = await session.invite_async(peer)
        await invited.wait_async()
    await session.publish_async("S-001：请复核示例摘要".encode(), None, None)
    # 实际流程应等待复核结果，再执行下面的移除
    reviewer = slim.Name.from_string("bank-c/review/agent")
    removed = await session.remove_async(reviewer)
    await removed.wait_async()

import json
from confluent_kafka import Producer, Consumer

p = Producer({"bootstrap.servers": "localhost:9092"})
p.produce("materials.events", key=b"S-001",
          value=json.dumps({"status": "missing_page", "event_seq": 1}).encode())
p.flush()

def open_reader(group_id):
    c = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": group_id,
        "enable.auto.commit": False,
        "auto.offset.reset": "earliest",
    })
    c.subscribe(["materials.events"])
    return c

orchestrator = open_reader("orchestrator")
audit = open_reader("audit")
analytics = open_reader("analytics")
# 新组无旧位点，从当前保留的最早事件开始核对；不移动其他组位点
replay = open_reader("audit-replay-B-001-new")

def recover_pending(r, inspect, idle_ms):
    # r 为 Redis(decode_responses=True)，消费组已预建
    pending = r.xpending("tasks:inspect", "inspect-workers")
    cursor = "0-0"
    while True:
        claim = r.xautoclaim(
            "tasks:inspect", "inspect-workers", "worker-b",
            min_idle_time=idle_ms, start_id=cursor, count=10)
        cursor, entries = claim[0], claim[1]
        for msg_id, task in entries:
            result = inspect(task)  # 业务按 task_id 防止重复副作用
            r.xadd("events:" + task["session_id"], result)
            r.xack("tasks:inspect", "inspect-workers", msg_id)
        if cursor == "0-0":
            break
    return pending  # 本轮接管前的待确认概况

nats --server nats://localhost:4222 request '$mq9.AI.AGENT.REGISTER' \
  '{"name":"agent.inspector","mailbox":"agent.inspector","payload":"材料完整性核验"}'
nats --server nats://localhost:4222 request '$mq9.AI.AGENT.DISCOVER' \
  '{"semantic":"材料完整性核验","limit":3}'
nats --server nats://localhost:4222 request '$mq9.AI.MAILBOX.CREATE' \
  '{"name":"agent.inspector","ttl":3600}'
nats --server nats://localhost:4222 request '$mq9.AI.MSG.SEND.agent.inspector' \
  --header 'mq9-priority:urgent' \
  '{"task_id":"T-001","session_id":"S-001","task":"inspect"}'
nats --server nats://localhost:4222 request '$mq9.AI.MSG.FETCH.agent.inspector' \
  '{"group_name":"inspect-workers","deliver":"earliest"}'