梧桐平台 · AMP 可观测性说明

首页 · Agent 开发进阶

在 Agent 中接入 AMP 可观测性

这篇教程写给 Agent 开发者。读完能做三件事：在 Agent 里写出 AMP 日志、在开发机上把日志转到 Kafka、确认平台侧能看到这些日志。

  1. 用 acps-sdk 的 Emitter 往本地 NDJSON 文件追加一行
  2. 用 Fluent Bit 尾追这些文件，转发到 Kafka 的 amp.* 主题（本地开发要自己跑；完整安装包里由 Ansible 装好）
  3. 看本地文件、发现结果里的在线状态；自建了 ACPs 平台、自己是管理员时，再用 acps-cli monitor 查 Monitor

Agent 进程不连 Kafka。文件名、目录要对上 Forwarder 的采集路径，否则第 2 步会静默空转。acps-cli monitor 用的是平台 Monitor 账号，不是 Agent 自己的证书；只在别人的平台上注册了 Agent、没有管理权限，这个命令查不了。

成品参考：demo-partner/partners/generic_runner.py、demo-leader 里的装配。安装包环境的验收故事见文末 business.yml Step D。

0. 你要达到什么效果

业务跑起来之后，对某个 Agent 的 AIC，你应该能够：

SDK 只负责往本地文件追加一行。转到 Kafka、入库、存活同步是后面两段的事：

[写] Agent：Emitter → 本地 amp_*.jsonl
        ↓  Fluent Bit 尾追（开发机自己跑，或安装包已部署）
[转] Kafka 主题 amp.heartbeat / amp.metrics / amp.access / amp.message / amp.system / amp.audit
        ↓  Monitor Writer 入库；heartbeat 再做 alive-sync
[查] 发现结果里的在线状态；有管理权限时再用 acps-cli monitor

本地开发时，第 3 节把「转」跑通之后，再做第 4 节。完整安装包里 Fluent Bit 已经在业务机上，从第 4 节直接确认即可。没有管理员权限时，第 4 节以本地文件和发现结果为准，不必强求 acps-cli monitor。

1. 准备：目录、AIC、依赖

写日志之前先定三件事：日志写到哪个目录、AIC 必须和注册身份一致、SDK 要带 amp 能力。文件名跟 demo 对齐，后面 Forwarder 才好按这些文件采集。

# 开发机：安装含 amp 能力的 SDK（以你项目的依赖方式为准）
# uv add acps-sdk   或在 pyproject 中声明依赖后 uv sync

import os
from pathlib import Path

# 安装部署通常会注入 AMP_LOG_DIR（例如 /opt/acps/app/logs）
# 本地开发未设置时，可用仓库下的 logs/
AMP_LOG_DIR = Path(os.environ.get("AMP_LOG_DIR", "./logs"))
AMP_LOG_DIR.mkdir(parents=True, exist_ok=True)

# 必须与 Registry 上该 Agent 实例的 AIC 一致
AIC = os.environ["ACPS_AIC"]  # 示例：换成你的真实 AIC

约定建议（与 demo 一致，便于 Forwarder 按这些文件采集）：

类型 建议文件名示例
heartbeat amp_heartbeat_<name>.jsonl
metrics amp_metrics_<name>.jsonl
access amp_access_<name>.jsonl
message amp_message_<name>.jsonl
system amp_system_<name>.jsonl
audit amp_audit_<name>.jsonl

2. 六类 Emitter：怎么用

这一节按「什么时候用 → 怎么挂进进程 → 字段怎么填 → 示例」来讲。示例对齐 demo 的常见写法，目的是你能直接抄到自己的 Agent 里改，而不是只证明 API 能跑通。

共性习惯：

2.1 Heartbeat — 告诉平台「我还活着」

用途：持续心跳用来告诉平台「我还在」。发现查询的返回里会带上在线状态。心跳一停，别人按「只要在线」去发现时就找不到你，即使业务进程还在。平台管理员还能在 Monitor 里点查这个 AIC。

怎么挂：进程启动后开一个后台周期任务；关闭时 cancel。demo 用环境变量 AMP_HEARTBEAT_INTERVAL_SECONDS（默认约 15s）。uptimeSeconds 由 Emitter 按创建时刻自动算，你不用填。

import asyncio
import contextlib
import os

from acps_sdk.amp import HeartbeatEmitter

heartbeat = HeartbeatEmitter(AMP_LOG_DIR / f"amp_heartbeat_{agent_name}.jsonl", aic=AIC)


async def run_agent() -> None:
    interval = float(os.environ.get("AMP_HEARTBEAT_INTERVAL_SECONDS", "15"))
    hb_task = asyncio.create_task(heartbeat.run_periodic(interval), name="amp-heartbeat")
    try:
        await serve_forever()  # 你的主循环
    finally:
        hb_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await hb_task

run_periodic 会立刻发首条，再按间隔续发，方便刚启动就验证链路。本地自检：文件在增长。有 Monitor 管理权限时，再用 heartbeat liveness <AIC> 点查。

2.2 Metrics — 周期性指标快照

用途：给 Monitor 提供负载与窗口统计（并发任务、CPU/内存、成功率、时延分位等），支撑 snapshots / series 查询。

怎么挂：和 heartbeat 类似，启动时 run_periodic。差别是 body 不由 Emitter 编造——你要实现 SampleProvider（只要有 sample() -> MetricsBody）。生产环境应读真实队列长度、进程资源；联调可先用 SDK 的 DemoMetricsSampler（仅 demo，勿上生产）。

resource 建议带上 OpenTelemetry 风格的服务标识，便于按服务名过滤：

import time

from acps_sdk.amp import LoadMetrics, MetricsBody, MetricsEmitter, WindowMetrics

_start = time.monotonic()


class AgentMetricsSampler:
    """示例：从本进程的任务表采真实负载（字段按你能拿到的填）。"""

    def __init__(self, tasks: dict) -> None:
        self._tasks = tasks

    def sample(self) -> MetricsBody:
        active = sum(1 for t in self._tasks.values() if t.state == "working")
        queued = sum(1 for t in self._tasks.values() if t.state == "submitted")
        return MetricsBody(
            uptime_seconds=round(time.monotonic() - _start, 3),
            load_metrics=LoadMetrics(
                active_tasks=active,
                queued_tasks=queued,
                max_active_tasks=10,
                max_queued_tasks=50,
                cpu_usage=None,  # 有 psutil / cgroup 再填
                memory_usage=None,
            ),
            window_metrics=[
                WindowMetrics(
                    window="PT5M",
                    success_rate=98.5,
                    request_total=120,
                    request_per_second=0.4,
                    p50_latency_ms=80.0,
                    p95_latency_ms=220.0,
                    p99_latency_ms=450.0,
                ),
            ],
        )


metrics = MetricsEmitter(
    AMP_LOG_DIR / f"amp_metrics_{agent_name}.jsonl",
    aic=AIC,
    sampler=AgentMetricsSampler(tasks),
    resource={
        "service.name": f"my-partner-{agent_name}",
        "service.namespace": "acps-demo",
        "deployment.environment.name": "dev",
    },
)

# 启动后：asyncio.create_task(metrics.run_periodic(30.0))

窗口字符串用 ISO 8601 Duration（如 PT1M / PT5M / PT15M）。公共指标名约定见 SDK metrics_catalog；分位务必保持 p50 ≤ p95 ≤ p99。

2.3 Access — 一次请求/响应的边

用途：记录「谁用什么方法调了谁、花了多久、成功还是失败」。典型挂点是 HTTP /rpc、出站 Partner 调用——在 finally 里写，保证异常路径也有记录。

关键字段：

字段 含义
request.method / route / url 动词与路由；route 用稳定模板（如 /rpc），便于按 endpoint 聚合
response.statusCode HTTP / RPC 状态
caller / callee 双方 AIC + serviceName
durationMs 总耗时
error 失败时的 ErrorInfo
trace_id / span_id / correlation_id 链路与会话（demo 里常用 taskId 作 correlation）

下面是 Partner 入站 RPC 的教学写法（结构对齐 demo-partner/partners/main.py）：

import time

from acps_sdk.amp import (
    AccessBody,
    AccessEmitter,
    AccessParticipant,
    AccessRequest,
    AccessResponse,
    ErrorInfo,
)

access = AccessEmitter(AMP_LOG_DIR / f"amp_access_{agent_name}.jsonl", aic=AIC)


async def handle_rpc(command, request_headers: dict[str, str]) -> object:
    t0 = time.monotonic()
    request_bytes = len(command.model_dump_json().encode("utf-8"))
    resp = None
    status_code = 200
    error_info = None
    try:
        resp = await dispatch(command)
        status_code = status_from_rpc_response(resp)
        error_info = error_info_from_rpc_response(resp)
        return resp
    except Exception as exc:
        status_code = 500
        error_info = ErrorInfo(code=500, message=str(exc))
        raise
    finally:
        duration_ms = int((time.monotonic() - t0) * 1000)
        response_bytes = (
            len(resp.model_dump_json().encode("utf-8")) if resp is not None else 0
        )
        method = str(command.command)
        body = AccessBody(
            request=AccessRequest(
                method=method,
                url="/rpc",
                route="/rpc",
                headers=request_headers,
                bodySizeBytes=request_bytes,
            ),
            response=AccessResponse(statusCode=status_code, bodySizeBytes=response_bytes),
            caller=AccessParticipant(aic=command.senderId, serviceName="demo-leader"),
            callee=AccessParticipant(aic=AIC, serviceName=f"demo-partner-{agent_name}"),
            error=error_info,
            durationMs=duration_ms,
        )
        try:
            await access.emit(
                body,
                trace_id=trace_id,          # 从入站 headers / 上下文取
                span_id=span_id,
                parent_span_id=parent_span_id,
                correlation_id=command.taskId,
            )
        except Exception:
            logger.warning("Access emit failed", exc_info=True)

Leader 出站调用 Partner 时同样可以写 access，只是 caller / callee 对调，url 写成对方地址。有 Monitor 管理权限时，可用 acps-cli monitor access events 查询，也可按 trace_id 串整条调用链。

2.4 Message — 消息生命周期的边

用途：群组 / RabbitMQ / Kafka 场景。同一条业务消息的 send、receive、ack/nack… 各写一条独立日志，靠相同的 messageId（以及可选的 correlationId）在查询层拼成生命周期。

event_type 取值：send | receive | ack | nack | reject | timeout | dead_letter。结算类事件再用 MessageSettlement 补耗时与原因。

若走 SDK 的群组 MQ 客户端，可注入 MessageEmitter，由客户端在收发路径自动写日志；自己组 body 时也可直接调 Emitter。组包时可参考 SDK 里群组客户端使用的组装逻辑（acps_sdk/amp/_message_tap.py 中的 build_send_body / build_receive_body / build_settlement_body）：

from acps_sdk.amp import MessageBody, MessageDestination, MessageEmitter, MessageRouting, MessageSettlement

message = MessageEmitter(AMP_LOG_DIR / f"amp_message_{agent_name}.jsonl", aic=AIC)

# 生产者：发布到群组 exchange
await message.emit(
    MessageBody(
        event_type="send",
        operation_name="publish",
        system="rabbitmq",
        destination=MessageDestination(
            name="group.demo.exchange",
            kind="exchange",
            virtual_host="/",
        ),
        routing=MessageRouting(key="group.demo"),
        message_id="msg-001",
        payload_size_bytes=256,
    ),
    correlation_id="group-session-1",
    trace_id=trace_id,
)

# 消费者：收到后
await message.emit(
    MessageBody(
        event_type="receive",
        operation_name="deliver",
        system="rabbitmq",
        destination=MessageDestination(name="group.demo.exchange", kind="exchange"),
        subscription_name="partner-queue-1",
        routing=MessageRouting(key="group.demo"),
        message_id="msg-001",
        payload_size_bytes=256,
        delivery_attempt=1,
    ),
    correlation_id="group-session-1",
)

# 结算：ack（失败则 nack + reason）
await message.emit(
    MessageBody(
        event_type="ack",
        operation_name="basic.ack",
        system="rabbitmq",
        destination=MessageDestination(name="group.demo.exchange", kind="exchange"),
        subscription_name="partner-queue-1",
        message_id="msg-001",
        delivery_attempt=1,
        settlement=MessageSettlement(latency_ms=35.0),
    ),
    correlation_id="group-session-1",
)

查询：message events 按时间窗；有 messageId 时可查 lifecycle。不要把整段 payload 塞进日志，只记大小与标识即可。

2.5 System — Agent 内部事件

用途：没有强 schema 的内部可观测性——LLM 调用、Skill 执行、进程生命周期、内部错误等。body 是自由 dict；靠 severity_text / severity_number 和约定好的字段（如 category、component）在 Monitor 里过滤。

建议约定（demo 常用）：

严重级别示例：成功 INFO → severity_number=9；失败 ERROR → 17（与 demo 一致即可，团队内统一比「绝对值」更重要）。

import time

from acps_sdk.amp import SystemEmitter

system = SystemEmitter(
    AMP_LOG_DIR / f"amp_system_{agent_name}.jsonl",
    aic=AIC,
    resource={
        "service.name": f"demo-partner-{agent_name}",
        "service.namespace": "acps-demo",
        "deployment.environment.name": "dev",
    },
)


async def call_llm(stage: str, model: str, task_id: str) -> str:
    t0 = time.monotonic()
    try:
        content = await client.chat.completions.create(...)
        elapsed = int((time.monotonic() - t0) * 1000)
        system.emit_sync(
            {
                "message": f"LLM call completed: stage={stage}, model={model}, elapsed={elapsed}ms",
                "category": "llm",
                "component": "llm_client",
                "module": stage,
                "model": model,
                "tags": {"model": model, "stage": stage, "task_id": task_id},
                "elapsed_ms": elapsed,
                "task_id": task_id,
                "token_total": getattr(getattr(response, "usage", None), "total_tokens", None),
            },
            severity_number=9,
            severity_text="INFO",
            correlation_id=task_id,
        )
        return content
    except Exception as e:
        elapsed = int((time.monotonic() - t0) * 1000)
        system.emit_sync(
            {
                "message": f"LLM call failed: stage={stage}, model={model}, error={type(e).__name__}",
                "category": "llm",
                "component": "llm_client",
                "module": stage,
                "model": model,
                "tags": {"model": model, "stage": stage, "error_type": type(e).__name__},
                "elapsed_ms": elapsed,
                "task_id": task_id,
                "error_type": type(e).__name__,
                "error_message": str(e)[:500],
            },
            severity_number=17,
            severity_text="ERROR",
            correlation_id=task_id,
        )
        raise

进程启停也可各写一条 lifecycle system（见 demo-partner / demo-leader 的 main）。查询可用 --severity-min 只看错误，或 --correlation-id 跟任务。

2.6 Audit — 需要留痕的关键动作

用途：谁对什么资源做了什么、结果如何——适合任务受理/终态、权限相关操作、不可抵赖的状态变更。结构固定为 actor / action / target / result。

与 system 的分工：system 偏「运行时可观测」；audit 偏「事后追责与合规」。同一业务可以两者都写（例如 LLM 细节走 system，任务终态走 audit）。

签名：生产环境应给 AuditEmitter 传 signer（demo 优先 CA 证书，失败则回退 audit_keys.json / load_signer_from_keys_json）。未签名仍能写入，但 Monitor 可能标 missing_public_key。规范要求 audit 带 integrity；联调可先通链路，上线前补齐签名与信任材料。

from pathlib import Path

from acps_sdk.amp import (
    AuditAction,
    AuditActor,
    AuditBody,
    AuditEmitter,
    AuditResult,
    AuditTarget,
    load_signer_from_keys_json,
)

signer = load_signer_from_keys_json(Path("config/audit_keys.json"), AIC)
# 或 CertificateAuditSigner(private_key_pem=..., cert_pem=...)

audit = AuditEmitter(
    AMP_LOG_DIR / f"amp_audit_{agent_name}.jsonl",
    aic=AIC,
    signer=signer,
)

# 收到 Start、完成 decision 之后（对齐 demo B1）
await audit.emit(
    AuditBody(
        actor=AuditActor(id=command.senderId or "unknown", type="agent"),
        action=AuditAction(name="receive_task_start", type="aip_protocol"),
        target=AuditTarget(type="task", id=task_id),
        result=AuditResult(
            status="success" if accepted else "failure",
            reason=None if accepted else reject_reason,
        ),
    ),
    trace_id=command.sessionId,
    correlation_id=task_id,
)

# 任务进入终态时（对齐 demo B2；同步状态机里用 emit_sync）
audit.emit_sync(
    AuditBody(
        actor=AuditActor(id=AIC, type="agent"),
        action=AuditAction(name="task_state_transition", type="aip_protocol"),
        target=AuditTarget(type="task", id=task_id),
        result=AuditResult(status="success", reason="completed"),
    ),
    trace_id=session_id,
    correlation_id=task_id,
)

result.status 只能是 success / failure / unknown。action.name / action.type 建议团队内定一套稳定词汇，便于按动作检索。

3. 把本地日志转到 Kafka

完整安装包会把 AMP Forwarder（Fluent Bit）装在跑 Agent 的那台机器上，开发者不用自己维护这个进程。本地开发没有安装包时，要在开发机上自己装、自己跑，否则 jsonl 在涨，Kafka 和 Monitor 都是空的。

这一节只覆盖开发机：Homebrew 安装、仓库里现成的配置、前台运行、用 rpk 看主题水位。不要配成 systemctl 服务。生产怎么随安装包落地，见 Ansible 部署。

开发配置与安装层是同一套语义。安装层模板在 acps-infra 的 roles/amp_forwarder/templates/fluent-bit.conf.j2；开发机可把 monitor-server/config/fluent-bit/fluent-bit.conf 当完整样例（六类 INPUT/OUTPUT 都在），把里面的日志 Path 改成你第 1 节定的 AMP_LOG_DIR。下文以 macOS、Fluent Bit 5.x、本机 Redpanda（just infra up kafka，localhost:19092）为例。

3.1 安装 Fluent Bit，并拉起 Kafka

macOS：

brew install fluent-bit
which fluent-bit          # Apple Silicon 常见 /opt/homebrew/bin/fluent-bit
fluent-bit --version

Linux 开发机用 Fluent Bit 官方包或发行版包，把 fluent-bit 放到 PATH 里即可，后面同样前台跑，不必做成系统服务。

Kafka 用 monitor-server 的共享基础设施（容器名 dev-redpanda，宿主机 localhost:19092）。先开 Docker Desktop，再：

cd monitor-server
just infra up kafka
just infra wait kafka
docker exec dev-redpanda rpk cluster health
docker exec dev-redpanda rpk topic describe amp.heartbeat -c | grep message.timestamp.type
# 期望：message.timestamp.type  LogAppendTime

just infra up kafka 会创建（或修正）六类 amp.* 主题。heartbeat / metrics / access / message / system 必须是 LogAppendTime；audit 可以由 Fluent Bit 注入时间戳，允许 CreateTime。Docker 没起来时，just infra up kafka 会直接失败。

3.2 配置怎么读、怎么跑

ACPS_ROOT 指向工作区根目录（monitor-server 所在的父目录），用来定位 parsers.conf。启动时必须带上。仓库这份 conf 里的日志 Path 默认指向 demo 目录，复制后先改成你自己的 AMP_LOG_DIR，再启动：

cd <工作区根目录>
ACPS_ROOT="$(pwd)" fluent-bit -c /path/to/your/fluent-bit.conf

另开一个终端，保持这个前台窗口开着。六类日志都配了 Kafka OUTPUT 时，标准输出里应出现六个 worker #0 started：

[output:kafka:kafka.0] worker #0 started      # amp.audit
[output:kafka:kafka.1] worker #0 started      # amp.heartbeat
[output:kafka:kafka.2] worker #0 started      # amp.metrics
[output:kafka:kafka.3] worker #0 started      # amp.access
[output:kafka:kafka.4] worker #0 started      # amp.message
[output:kafka:kafka.5] worker #0 started      # amp.system

macOS 上不要加 -d。Fluent Bit v5 在 macOS 上用 daemon 模式时，进程还在，事件循环不工作，日志转不出去。每个 Kafka [OUTPUT] 必须保留 Workers 1，否则 worker 线程起不来（fluent/fluent-bit#11358）。

六类日志都是「一个 tail INPUT + 一个 kafka OUTPUT」。心跳这一段如下，其余五类只是文件名、Tag、主题不同。把 Path 换成你实际的日志目录：

[INPUT]
    Name             tail
    Path             /path/to/AMP_LOG_DIR/amp_heartbeat_*.jsonl
    Tag              amp.heartbeat
    Parser           amp_audit_json
    Refresh_Interval 2
    Read_From_Head   False
    DB               /tmp/fluentbit-heartbeat.db

[OUTPUT]
    Name                              kafka
    Match                             amp.heartbeat
    Brokers                           Brokers wt.ioa.pub:19092
    Topics                            amp.heartbeat
    Workers                           1
    rdkafka.log.connection.close      false
    rdkafka.request.required.acks     1
    rdkafka.allow.auto.create.topics  false

对照时记住这几条：

项 取值 含义
多实例路径 AMP_LOG_DIR/amp_<类型>_*.jsonl 每个 Agent 一个文件，glob 采集
单文件路径 AMP_LOG_DIR/amp_<类型>.jsonl 进程只有一份日志时用固定名
Read_From_Head False 只转发 Fluent Bit 启动之后追加的行
Refresh_Interval 2 新文件、新行大约两秒内会被看到
Brokers localhost:19092 本机 Redpanda 对外监听
Workers 1 macOS 必填
Parser amp_audit_json（在 parsers.conf） 按 JSON 解析，并用记录里的 timestamp

第 1 节那种 amp_heartbeat_<name>.jsonl 对得上 glob amp_heartbeat_*.jsonl。INPUT 写了 glob，却只写 amp_heartbeat.jsonl（没有下划线和名字），就匹配不到。

Parser 要求 timestamp 带小数秒和时区，和 SDK 默认写法一致，例如 2026-09-17T13:12:21.043924+00:00。手写测试行时不要写成没有小数秒的时间。

3.3 怎么知道转发成功

顺序必须是：Kafka 已健康 → Fluent Bit 六个 worker 已 started → 再让 Agent 写新行。配了 Read_From_Head False，启动前已经在文件里的旧行不会进 Kafka。

  1. 记一笔追加前的水位：
docker exec dev-redpanda rpk topic describe amp.heartbeat -p
# 看 PARTITION 行里的 HIGH-WATERMARK，下面记为旧 HW

  1. 让 Agent 写出新的心跳。进程在跑时，等一个心跳周期即可（第 2.1 节默认大约 15 秒）。新行必须落在 conf 里 Path 能匹配到的文件上。

  2. 再看水位。HIGH-WATERMARK 增加了，用旧 HW 当 --offset 把新消息读出来（不要从 0 扫，主题里可能已有历史消息）：

docker exec dev-redpanda rpk topic describe amp.heartbeat -p
docker exec dev-redpanda rpk topic consume amp.heartbeat --num 1 --offset <旧HW> --format json

payload 里应能看到你的 aic、log_type、log_id。Fluent Bit 常会多注入一个 @timestamp（秒.小数）；Monitor 入库前会剥掉 @ / _fb 前缀，开发者用 rpk 眼看时不要被这个字段吓到。

同时看一眼死信主题，这次追加之后水位不该增长：

docker exec dev-redpanda rpk topic describe amp.heartbeat.dlq -p

access / message / system 同理，主题名换成对应的 amp.*。要停 Forwarder 时，在前台窗口 Ctrl-C 即可。

3.4 自己的 Agent 要改什么

仓库样例改到你的进程，通常只动这些：

  1. 每个 [INPUT] 的 Path，改成你的 AMP_LOG_DIR，文件名对齐第 1 节的表。多实例用 amp_heartbeat_*.jsonl 这种 glob。
  2. Brokers：本机 Redpanda 保持 localhost:19092；连别的集群再改。
  3. 每个 INPUT 的 DB 用不同文件。两个 tail 共用一个 DB 会把位点搅乱。
  4. 启动命令里的 ACPS_ROOT 仍指向工作区根目录（parsers.conf 靠它定位）。conf 拷到别处时，按拷贝后的路径改 -c 和 Parsers_File。

Tag、Match、Topics 保持 amp.heartbeat 这一套六类名字，和 Monitor 消费约定一致。不要在 Agent 里直连 Kafka。

3.5 文件在写、Kafka 没有：先查这些

这些情况常常不报业务错误，或只在 Fluent Bit 启动日志里闪一下：

现象 原因
进程立刻退出：could not open parser configuration file 没设 ACPS_ROOT。Parsers_File 变成无效路径，Fluent Bit 5.1.2 会直接退出，不会空转
六个 worker 都 started，水位不动 Path 对不上实际文件：目录错了，或 glob 写成 amp_heartbeat_*.jsonl 却只写了 amp_heartbeat.jsonl
Agent 早就在写，刚启动 Fluent Bit，水位仍不动 Read_From_Head False：只转启动之后的新行。等下一条心跳，或再 emit 一次
macOS 上进程在、Kafka 没有新消息 用了 -d，或 [OUTPUT] 里没有 Workers 1
主题能写入，Monitor 仍为空、DLQ 在涨 主题不是 LogAppendTime（再执行一次 just infra up kafka 做幂等修正）；这已经越过 Forwarder，下一步才是 Monitor

安装包环境里 Forwarder 由运维托管。文件在写但平台查不到，先确认业务机上 Fluent Bit 在跑、采集目录就是部署注入的 AMP_LOG_DIR，再查 日常运维。

4. 怎么确认能查到

本地文件在涨，只说明 Emitter 在写。Kafka 水位在涨，说明第 3 节的转发通了。没有平台管理权限时，再看第 4.3 节发现结果里的在线状态。acps-cli monitor 只给自建平台的管理员用，见第 4.2 节。

4.1 先看本地文件（最快自检）

ls -la "${AMP_LOG_DIR:-./logs}"/amp_*.jsonl
tail -n 1 "${AMP_LOG_DIR:-./logs}"/amp_heartbeat_demo.jsonl

文件在增长，说明 Agent 侧已经在写。

4.2 用 acps-cli 查 Monitor（自建平台的管理员）

acps-cli monitor 访问的是 Monitor 的查询接口，要用平台侧的 Monitor 账号登录，不是 Agent 的证书，也不是你在 Registry 注册 Agent 的那个普通用户。

如果你自建了完整 ACPs 平台、你自己就是管理员时，才能跑下面这些命令。只是在别人的平台上登记了一个 Agent、没有管理员权限，请跳过这一节：看 4.1 的本地文件和第 4.3 节的发现结果。Monitor 里有没有你的日志，要请平台管理员帮你看。

登录和地址配置见 CLI 参考 的 acps-cli monitor；OIDC 登录步骤见 Device 登录 的 monitor 用户登录。

时间窗用**北京时间（东八区）**即可，写成带 +08:00 的 ISO 8601；CLI 会原样交给 Monitor，服务端按带时区的时间解析。把区间换成你实际发日志前后的时段（access / message / system / audit 的 events / records 未提供完整 request 时，一般需要 --start / --end）：

# 例：2026-07-25 当天（北京时间）
START=2026-07-25T00:00:00+08:00
END=2026-07-25T23:59:59+08:00

# 健康
acps-cli monitor status

# 心跳（点查单个 AIC）
acps-cli monitor heartbeat liveness "$AIC"

# 指标快照
acps-cli monitor metrics snapshots --aic "$AIC"

# 访问 / 消息 / 系统 / 审计（按 AIC + 时间窗）
acps-cli monitor access events --aic "$AIC" --start "$START" --end "$END"
acps-cli monitor message events --start "$START" --end "$END"
acps-cli monitor system events --aic "$AIC" --start "$START" --end "$END"
acps-cli monitor audit records --aic "$AIC" --start "$START" --end "$END"

刚写入时可能有短暂延迟，稍等再查。更多子命令（trace、series、verify 等）见 CLI 参考。

4.3 发现结果里的在线状态

当你使用发现服务做发现查询，返回里会带一份 aliveMap，用 AIC 当键，标明这个 Agent 现在在不在线。

要先有心跳进平台（第 3 节转到 Kafka，Monitor 再同步给发现），aliveMap 里才会出现你。停了心跳，过一会儿会变成离线；筛选条件写成「只要在线」时就看不见它，即使进程其实还在跑。

aliveMap 怎么读、怎么按在线状态筛选，见 发现服务查询。

5. 端到端故事：business.yml 的 Step D

这一节用安装包里的 business.yml Step D，说明 demo 写日志、Forwarder 转发、Monitor 查询怎样串成一次验收。

  1. 运维用安装包跑通 site.yml，打开 demo-leader / demo-partner（见 部署教程）。
  2. 写：demo 进程启动时已创建六类 Emitter（Partner 见 generic_runner.py）。跑业务验收 A/B/C 时：
    • RPC 路径写 access
    • 群组路径写 message
    • LLM / 生命周期写 system
    • 周期任务写 heartbeat / metrics
    • 任务终态等写 audit
  3. 中间：安装包里 Fluent Bit 由 Ansible 的 amp_forwarder 拉起，采集 demo 的 amp_*.jsonl 送到 Redpanda。本地开发没有安装包时，自己按第 3 节跑 Forwarder。
  4. 查：business.yml 的 Step D 轮询 Monitor：demo 相关 AIC 是否出现 metrics / system / access / message / audit 等记录，并检查 Discovery 存活。

因此：你自己写 Agent 时，对齐的是 同类 Emitter + 正确 AIC，而不是去改 business.yml。Step D 只是平台侧的「验收剧本」。

6. 开发者自检清单

7. 接下来做什么